""" 全周期回测 — 2017-2026，覆盖完整牛熊 多空双向 EMA 趋势跟踪，展示牛市/熊市/全周期分段表现。 用法： source .venv/bin/activate && python example/full_cycle.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 BacktestConfig from engine.indicators import ema, atr from engine.example.long_short import LongShortEngine, LongShortEmaConfig, LongShortEmaStrategy # ════════════════════════════════════════════════════════ SYMBOLS = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"] PARAMS = { "BTCUSDT": (10, 50), "ETHUSDT": (10, 75), "BNBUSDT": (20, 50), "SOLUSDT": (30, 50), } # 牛熊分段（以 BTC 为参考） PERIODS = [ ("2017-2018 牛市", datetime(2017, 1, 1), datetime(2018, 1, 1)), ("2018 熊市", datetime(2018, 1, 1), datetime(2019, 1, 1)), ("2019 反弹", datetime(2019, 1, 1), datetime(2020, 1, 1)), ("2020 牛初+312", datetime(2020, 1, 1), datetime(2021, 1, 1)), ("2021 牛市", datetime(2021, 1, 1), datetime(2022, 1, 1)), ("2022 熊市", datetime(2022, 1, 1), datetime(2023, 1, 1)), ("2023 复苏", datetime(2023, 1, 1), datetime(2024, 1, 1)), ("2024-2025 牛市", datetime(2024, 1, 1), datetime(2026, 1, 1)), ] DATE_START = datetime(2017, 1, 1) DATE_END = datetime(2026, 1, 1) async def run_backtest(symbol, fast, slow, start, end): sc = LongShortEmaConfig(symbol=symbol, fast=fast, slow=slow) bt = BacktestConfig(symbol=symbol, interval="4h", start_time=start, end_time=end, initial_capital=10_000.0) engine = LongShortEngine(bt, db_config=config.db) return await engine.run(LongShortEmaStrategy, sc) async def main(): print() print("═" * 125) print(" 全周期多空回测 — 2017-2026 完整牛熊 | 4h EMA趋势") print("═" * 125) # ── 全周期汇总 ── print(f"\n ■ 全周期 2017-2026 汇总") print(f" {'币种':<10} {'总收益%':>7} {'年化%':>7} {'夏普':>6} {'回撤%':>7} {'交易':>5} {'多头P&L':>10} {'空头P&L':>10}") print(" " + "─" * 105) for symbol in SYMBOLS: fast, slow = PARAMS[symbol] r = await run_backtest(symbol, fast, slow, DATE_START, DATE_END) m = r.metrics long_t = [t for t in r.trades if t.pnl is not None and t.side == "SELL"] short_t = [t for t in r.trades if t.pnl is not None and t.side == "BUY"] long_pnl = sum(t.pnl for t in long_t) if long_t else 0 short_pnl = sum(t.pnl for t in short_t) if short_t else 0 print(f" {symbol:<10} {m.total_return_pct:>6.1f}% {m.annual_return_pct:>6.1f}% {m.sharpe_ratio:>6.2f} {m.max_drawdown_pct:>6.1f}% {m.total_trades:>5} {long_pnl:>+9.0f} {short_pnl:>+9.0f}") # ── BTC 分段 ── print(f"\n ■ BTC 各阶段表现 (参数 EMA{10},{50})") print(f" {'阶段':<22} {'总收益%':>7} {'年化%':>7} {'夏普':>6} {'回撤%':>7} {'多头P&L':>10} {'空头P&L':>10}") print(" " + "─" * 105) for period_name, p_start, p_end in PERIODS: try: r = await run_backtest("BTCUSDT", 10, 50, p_start, p_end) m = r.metrics long_t = [t for t in r.trades if t.pnl is not None and t.side == "SELL"] short_t = [t for t in r.trades if t.pnl is not None and t.side == "BUY"] long_pnl = sum(t.pnl for t in long_t) if long_t else 0 short_pnl = sum(t.pnl for t in short_t) if short_t else 0 print(f" {period_name:<22} {m.total_return_pct:>6.1f}% {m.annual_return_pct:>6.1f}% {m.sharpe_ratio:>6.2f} {m.max_drawdown_pct:>6.1f}% {long_pnl:>+9.0f} {short_pnl:>+9.0f}") except Exception as e: print(f" {period_name:<22} 数据不足或错误: {e}") # ── 只做多 vs 多空全周期对比 ── print(f"\n ■ BTC 只做多 vs 多空 (全周期)") # 只做多需要单跑一次（LongShortEngine 本身就支持只做多：不开空就行） # 简单做法：用原版 BacktestEngine 跑一次只做多 from engine.backtest import BacktestEngine from engine.common.base import BaseStrategy as BS, Signal as Sig, StrategyConfig as SC class LongOnlyEMAConfig(SC): fast: int = 10; slow: int = 50; atr_stop: float = 2.5 class LongOnlyEMAStrategy(BS): strategy_type = "long_only" def __init__(self, c): super().__init__(c); self.cfg = c async def on_start(self): self._c = []; self._h = []; self._l = []; self._hp = 0.0; self._in = False; await super().on_start() async def on_kline(self, k): self._c.append(k.close); self._h.append(k.high); self._l.append(k.low) n = len(self._c) if n < self.cfg.slow + 5: return None f = ema(self._c, self.cfg.fast); s = ema(self._c, self.cfg.slow) a = atr(self._h, self._l, self._c, 14) cf, cs, ca = f[-1], s[-1], a[-1]; pf, ps = f[-2], s[-2] if cf == 0 or cs == 0 or ca == 0: return None if self._in: self._hp = max(self._hp, k.high); stop = self._hp - self.cfg.atr_stop * ca if (pf >= ps and cf < cs) or k.close < stop: self._in = False return Sig(symbol=self.cfg.symbol, side="SELL", reason="死叉" if pf >= ps else "ATR止损", timestamp=k.open_time) else: if pf <= ps and cf > cs: self._in = True; self._hp = k.close return Sig(symbol=self.cfg.symbol, side="BUY", reason="金叉", timestamp=k.open_time) return None lo_sc = LongOnlyEMAConfig(symbol="BTCUSDT") lo_bt = BacktestConfig(symbol="BTCUSDT", interval="4h", start_time=DATE_START, end_time=DATE_END, initial_capital=10_000.0) lo_eng = BacktestEngine(lo_bt, db_config=config.db) lo_r = await lo_eng.run(LongOnlyEMAStrategy, lo_sc) lo_m = lo_r.metrics # 多空 ls_r = await run_backtest("BTCUSDT", 10, 50, DATE_START, DATE_END) ls_m = ls_r.metrics print(f" {'':<10} {'总收益%':>7} {'年化%':>7} {'夏普':>6} {'回撤%':>7} {'交易':>5}") print(f" {'只做多':<10} {lo_m.total_return_pct:>6.1f}% {lo_m.annual_return_pct:>6.1f}% {lo_m.sharpe_ratio:>6.2f} {lo_m.max_drawdown_pct:>6.1f}% {lo_m.total_trades:>5}") print(f" {'多空':<10} {ls_m.total_return_pct:>6.1f}% {ls_m.annual_return_pct:>6.1f}% {ls_m.sharpe_ratio:>6.2f} {ls_m.max_drawdown_pct:>6.1f}% {ls_m.total_trades:>5}") print("\n═" * 125) if __name__ == "__main__": asyncio.run(main())