Rekey
515e61c517
- backtest_demo.py: 回测基础演示 - strategy_simple.py / three_ema.py / long_short.py: 基础策略（双均线/三均线/多空） - strategy_optimize*.py (3 版本): 参数优化示例（网格搜索/贝叶斯/遗传算法） - multi_tf_*.py (4 版本): 多时间框架策略（EMA200/多周期共振/混合信号） - regime_*.py (4 版本): 市场状态检测（趋势/震荡/波动率区间/全状态） - cross_section.py: 截面多品种策略 - factor_demo.py: 多因子模型演示 - strategy_battle.py / strategy_more.py: 策略对比与组合 - full_cycle.py: 全流程演示（数据→回测→分析） - data.py: 数据读取示例
249 lines
9.9 KiB
Python
249 lines
9.9 KiB
Python
"""
|
|
夏普比率专项优化 — 波动率自适应 + 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())
|