""" 夏普比率专项优化 — 波动率自适应 + ADX过滤 + 分批止盈 优化手段（核心目标：提高收益/波动比）： 1. 波动率自适应仓位 — ATR大→confidence小（少买），ATR小→confidence大（多买） 2. ADX 趋势过滤 — ADX>20 才交易，避开震荡市的反复假突破 3. 分批止盈 — RSI过热先出一半锁利，剩下一半ATR跟踪 对比：v1基线 / v3优化 / v4夏普优化 用法： source .venv/bin/activate && python example/strategy_optimize3.py """ import asyncio import sys from datetime import datetime, timezone from pathlib import Path from typing import Optional _project_root = Path(__file__).resolve().parent.parent.parent if str(_project_root) not in sys.path: sys.path.insert(0, str(_project_root)) from engine.common.base import BaseStrategy, Signal, StrategyConfig from engine.common.models import Kline from engine.common.config import config from engine.backtest import BacktestEngine, BacktestConfig, BacktestResult from engine.indicators import ema, atr, rsi as calc_rsi, adx # ════════════════════════════════════════════════════════ # EMA v4: 波动率自适应 + ADX过滤 + 分批止盈 # ════════════════════════════════════════════════════════ class EmaV4Config(StrategyConfig): fast: int = 20 slow: int = 50 atr_period: int = 14 atr_stop_mult: float = 3.0 adx_period: int = 14 adx_threshold: float = 20.0 # ADX 趋势阈值 vol_base: float = 20.0 # ATR% 基准（周期数用于标准化） rsi_period: int = 14 rsi_take_profit: float = 72.0 # 止盈 RSI 线 partial_exit_pct: float = 0.5 # 分批止盈比例（0=不分批） class EmaV4Strategy(BaseStrategy): """EMA v4: 以夏普比率为目标的全方位优化""" strategy_type = "ema_v4" def __init__(self, c: EmaV4Config): super().__init__(c) self.cfg = c self._closes: list[float] = [] self._highs: list[float] = [] self._lows: list[float] = [] self._highest_since_entry: float = 0.0 self._in_position = False self._partial_done = False # 是否已部分止盈 async def on_kline(self, k: Kline) -> Optional[Signal]: self._closes.append(k.close) self._highs.append(k.high) self._lows.append(k.low) n = len(self._closes) need = max(self.cfg.slow, self.cfg.adx_period * 2) if n < need + 5: return None fast = ema(self._closes, self.cfg.fast) slow = ema(self._closes, self.cfg.slow) atr_vals = atr(self._highs, self._lows, self._closes, self.cfg.atr_period) adx_vals = adx(self._highs, self._lows, self._closes, self.cfg.adx_period) rsi_vals = calc_rsi(self._closes, self.cfg.rsi_period) cur_f, cur_s = fast[-1], slow[-1] cur_atr, cur_adx, cur_rsi = atr_vals[-1], adx_vals[-1], rsi_vals[-1] if cur_f == 0 or cur_s == 0 or cur_atr == 0 or cur_adx == 0 or cur_rsi == 0: return None # ── 波动率自适应仓位系数 ── # ATR/价格 = 当前波动率，波动率越高→仓位越小 atr_pct = cur_atr / k.close if k.close > 0 else 0.02 # 基准波动率约 2%，波动率翻倍时仓位减半 vol_conf = min(1.0, max(0.2, 0.02 / max(atr_pct, 0.005))) # ADX 趋势强度加成：强趋势更有信心 trend_bonus = min(1.3, max(0.7, cur_adx / 25)) position_conf = min(1.0, vol_conf * trend_bonus) # ADX 趋势过滤 in_trend = cur_adx > self.cfg.adx_threshold # ── 出场 ── if self._in_position: self._highest_since_entry = max(self._highest_since_entry, k.high) stop_price = self._highest_since_entry - self.cfg.atr_stop_mult * cur_atr death_cross = fast[-2] >= slow[-2] and cur_f < cur_s # ATR 止损 if k.close < stop_price: self._in_position = False return Signal(symbol=self.cfg.symbol, side="SELL", confidence=1.0, reason=f"ATR止损 P={k.close:.0f}", timestamp=k.open_time) # EMA 死叉 if death_cross: self._in_position = False return Signal(symbol=self.cfg.symbol, side="SELL", confidence=1.0, reason="EMA死叉", timestamp=k.open_time) # 分批止盈：RSI过热先出一半 if not self._partial_done and cur_rsi > self.cfg.rsi_take_profit and self.cfg.partial_exit_pct > 0: self._partial_done = True return Signal(symbol=self.cfg.symbol, side="SELL", quantity=None, # None=全部，但我们用confidence控制比例 confidence=self.cfg.partial_exit_pct, reason=f"半仓止盈 RSI={cur_rsi:.0f}", timestamp=k.open_time) # ── 入场 ── if not self._in_position: golden = fast[-2] <= slow[-2] and cur_f > cur_s if golden and in_trend: self._in_position = True self._highest_since_entry = k.close self._partial_done = False return Signal( symbol=self.cfg.symbol, side="BUY", confidence=position_conf, reason=f"金叉 ADX={cur_adx:.0f} vol_conf={vol_conf:.2f}", timestamp=k.open_time, ) return None # ════════════════════════════════════════════════════════ # 对比运行 # ════════════════════════════════════════════════════════ SYMBOLS = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"] # 各币种最优参数（来自 v3 扫描） BEST_PARAMS = { "BTCUSDT": (10, 50), "ETHUSDT": (10, 75), "BNBUSDT": (10, 40), "SOLUSDT": (30, 50), } # v1 基线 V1 = { "BTCUSDT": (45.5, 0.74, -31.6, 42, 26.2), "ETHUSDT": (24.4, 0.47, -54.8, 41, 24.4), "BNBUSDT": (52.0, 0.71, -39.8, 41, 39.0), "SOLUSDT": (27.8, 0.49, -39.5, 45, 40.0), } # v3 最优参数结果 V3 = { "BTCUSDT": (39.9, 1.03, -11.5, 20, 55.0), "ETHUSDT": (53.6, 1.04, -15.3, 18, 38.9), "BNBUSDT": (26.0, 0.64, -23.4, 23, 34.8), "SOLUSDT": (73.6, 1.18, -25.7, 13, 46.2), } async def run_v4(symbol: str, fast: int, slow: int) -> BacktestResult: bt = BacktestConfig( symbol=symbol, interval="4h", start_time=datetime(2024, 1, 1), end_time=datetime(2026, 1, 1), initial_capital=10_000.0, ) sc = EmaV4Config(symbol=symbol, fast=fast, slow=slow) engine = BacktestEngine(bt, db_config=config.db) return await engine.run(EmaV4Strategy, sc) async def main(): print() print("═" * 120) print(" 夏普比率专项优化 — 波动率自适应 + ADX过滤 + 分批止盈") print("═" * 120) # ── 扫描 partial_exit_pct 参数 ── print("\n ▸ 分批止盈参数扫描 (SOLUSDT 为例)") print(f" {'partial%':>10} {'收益%':>8} {'夏普':>6} {'回撤%':>8} {'交易':>5} {'胜率%':>6}") print(" " + "─" * 50) for pct in [0.0, 0.3, 0.5, 0.7]: bt = BacktestConfig(symbol="SOLUSDT", interval="4h", start_time=datetime(2024, 1, 1), end_time=datetime(2026, 1, 1), initial_capital=10_000.0) sc = EmaV4Config(symbol="SOLUSDT", fast=30, slow=50, partial_exit_pct=pct) engine = BacktestEngine(bt, db_config=config.db) r = await engine.run(EmaV4Strategy, sc) m = r.metrics print(f" {pct:>10.0%} {m.total_return_pct:>7.1f}% {m.sharpe_ratio:>6.2f} {m.max_drawdown_pct:>7.1f}% {m.total_trades:>5} {m.win_rate*100:>5.1f}%") # ── 全部币种 v4 运行 ── print("\n" + "═" * 120) print(" ■ v1 → v3 → v4 夏普进化 | 使用各币种最优参数") print(f" {'币种':<10} {'版本':<18} {'收益%':>7} {'夏普':>6} {'Δ夏普':>7} {'回撤%':>7} {'交易':>5} {'胜率%':>6}") print("─" * 120) v4_results = {} for symbol in SYMBOLS: fast, slow = BEST_PARAMS[symbol] r = await run_v4(symbol, fast, slow) v4_results[symbol] = r v1 = V1[symbol] v3 = V3[symbol] m = r.metrics # v1 print(f" {symbol:<10} {'v1 原始':<18} {v1[0]:>6.1f}% {v1[1]:>6.2f} {'—':>7} {v1[2]:>6.1f}% {v1[3]:>5} {v1[4]:>5.1f}%") # v3 print(f" {symbol:<10} {'v3 最优参数':<18} {v3[0]:>6.1f}% {v3[1]:>6.2f} {v3[1]-v1[1]:>+6.2f} {v3[2]:>6.1f}%") # v4 delta_sh = m.sharpe_ratio - v1[1] print(f" {symbol:<10} {'v4 夏普优化':<18} {m.total_return_pct:>6.1f}% {m.sharpe_ratio:>6.2f} {delta_sh:>+6.2f} {m.max_drawdown_pct:>6.1f}% {m.total_trades:>5} {m.win_rate*100:>5.1f}%") print() # ── 夏普改善汇总 ── print("─" * 120) print(" ■ 夏普比率改善汇总:") for symbol in SYMBOLS: v1_sh = V1[symbol][1] v3_sh = V3[symbol][1] v4_sh = v4_results[symbol].metrics.sharpe_ratio print(f" {symbol}: v1={v1_sh:.2f} → v3={v3_sh:.2f} → v4={v4_sh:.2f} (总提升 {v4_sh-v1_sh:+.2f})") # 平均夏普 avg_v1 = sum(V1[s][1] for s in SYMBOLS) / 4 avg_v3 = sum(V3[s][1] for s in SYMBOLS) / 4 avg_v4 = sum(v4_results[s].metrics.sharpe_ratio for s in SYMBOLS) / 4 print(f" 平均: v1={avg_v1:.2f} → v3={avg_v3:.2f} → v4={avg_v4:.2f} (总提升 {avg_v4-avg_v1:+.2f})") print("\n═" * 120) if __name__ == "__main__": asyncio.run(main())