""" 牛熊自适应策略 — 多时间级别回测对比 2h / 4h / 6h / 1d × 全量数据 / 近两年 (2024.06-2026.06) 用法： source .venv/bin/activate && python example/regime_timeframe_comparison.py """ import asyncio import sys from datetime import datetime, timezone from pathlib import Path _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.data import DataService from engine.example.long_short import LongShortEngine from engine.example.regime_all import RegimeEmaConfig, RegimeEmaStrategy # ═══════════════════════════════════ # 配置 # ═══════════════════════════════════ SYMBOLS = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"] PARAMS = { "BTCUSDT": (10, 50), "ETHUSDT": (10, 75), "BNBUSDT": (20, 50), "SOLUSDT": (30, 50), } INTERVALS = ["2h", "4h", "6h", "1d"] # 近两年：2024年6月 → 2026年6月 YEAR_START = datetime(2024, 6, 1) YEAR_END = datetime(2026, 6, 12) FULL_DEFAULT_START = datetime(2017, 1, 1) async def get_actual_range(symbol: str, interval: str) -> tuple[datetime, datetime]: ds = DataService(config.db) await ds.connect() try: start, end = await ds.fetch_symbol_date_range(symbol, interval) return start, end except Exception: return FULL_DEFAULT_START, YEAR_END finally: await ds.close() async def run_one(symbol: str, interval: str, start: datetime, end: datetime): fast, slow = PARAMS[symbol] sc = RegimeEmaConfig(symbol=symbol, fast=fast, slow=slow) bt = BacktestConfig( symbol=symbol, interval=interval, start_time=start, end_time=end, initial_capital=10_000.0, warmup_bars=250, ) engine = LongShortEngine(bt, db_config=config.db) return await engine.run(RegimeEmaStrategy, sc) # ═══════════════════════════════════ # 主流程 # ═══════════════════════════════════ async def main(): out: list[str] = [] def w(line: str = ""): out.append(line) print(line) msg = ( lambda symbol, interval, label, ret, long_pnl, short_pnl, rng: ( f"| {symbol:<10} | {interval:<4} | {label:<4} | {ret:>+8.1f}% | " f"{r.metrics.annual_return_pct:>+7.1f}% | {r.metrics.sharpe_ratio:>6.2f} | " f"{r.metrics.max_drawdown_pct:>7.1f}% | {r.metrics.total_trades:>5} | " f"{r.metrics.win_rate*100:>6.1f}% | {r.metrics.profit_factor:>6.2f} | " f"{long_pnl:>+9.0f} | {short_pnl:>+9.0f} | {rng} |" ) ) all_rows: list[dict] = [] w("# 牛熊自适应策略 — 多时间级别回测对比") w() w(f"> 生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}") w() # ── 一、全量数据 ── w("## 一、全量数据（所有可用历史）") w() for interval in INTERVALS: w(f"### {interval} 周期") w() w( "| 币种 | 周期 | 范围 | 总收益% | 年化% | 夏普 | 回撤% | 交易 | 胜率% | 盈亏比 | 多头P&L | 空头P&L | 数据范围 |" ) w( "|------|------|------|--------|------|------|------|------|------|------|---------|---------|---------|" ) for symbol in SYMBOLS: try: act_start, act_end = await get_actual_range(symbol, interval) rng = f"{act_start.date()}~{act_end.date()}" except Exception: act_start, act_end = FULL_DEFAULT_START, YEAR_END rng = "2017-2026" try: r = await run_one(symbol, interval, act_start, act_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"] lp = sum(t.pnl for t in long_t) if long_t else 0 sp = sum(t.pnl for t in short_t) if short_t else 0 row = m.total_return_pct w( f"| {symbol:<10} | {interval:<4} | 全量 | {row:>+8.1f}% | " f"{m.annual_return_pct:>+7.1f}% | {m.sharpe_ratio:>6.2f} | " f"{m.max_drawdown_pct:>7.1f}% | {m.total_trades:>5} | " f"{m.win_rate*100:>6.1f}% | {m.profit_factor:>6.2f} | " f"{lp:>+9.0f} | {sp:>+9.0f} | {rng} |" ) all_rows.append( { "symbol": symbol, "interval": interval, "label": "全量", "rng": rng, "return": m.total_return_pct, "annual": m.annual_return_pct, "sharpe": m.sharpe_ratio, "dd": m.max_drawdown_pct, "trades": m.total_trades, "win": m.win_rate, "pf": m.profit_factor, } ) except Exception as e: w( f"| {symbol:<10} | {interval:<4} | 全量 | — | — | — | — | — | — | — | — | — | 错误 |" ) print(f" ✗ {symbol} {interval} 全量: {e}") w() # ── 二、近两年 ── w("## 二、近两年（2024.06 — 2026.06）") w() for interval in INTERVALS: w(f"### {interval} 周期") w() w( "| 币种 | 周期 | 范围 | 总收益% | 年化% | 夏普 | 回撤% | 交易 | 胜率% | 盈亏比 | 多头P&L | 空头P&L |" ) w( "|------|------|------|--------|------|------|------|------|------|------|---------|---------|" ) for symbol in SYMBOLS: try: r = await run_one(symbol, interval, YEAR_START, YEAR_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"] lp = sum(t.pnl for t in long_t) if long_t else 0 sp = sum(t.pnl for t in short_t) if short_t else 0 row = m.total_return_pct w( f"| {symbol:<10} | {interval:<4} | 近2年 | {row:>+8.1f}% | " f"{m.annual_return_pct:>+7.1f}% | {m.sharpe_ratio:>6.2f} | " f"{m.max_drawdown_pct:>7.1f}% | {m.total_trades:>5} | " f"{m.win_rate*100:>6.1f}% | {m.profit_factor:>6.2f} | " f"{lp:>+9.0f} | {sp:>+9.0f} |" ) all_rows.append( { "symbol": symbol, "interval": interval, "label": "近2年", "rng": "2024.06~2026.06", "return": m.total_return_pct, "annual": m.annual_return_pct, "sharpe": m.sharpe_ratio, "dd": m.max_drawdown_pct, "trades": m.total_trades, "win": m.win_rate, "pf": m.profit_factor, } ) except Exception as e: w( f"| {symbol:<10} | {interval:<4} | 近1年 | — | — | — | — | — | — | — | — | — |" ) print(f" ✗ {symbol} {interval} 近2年: {e}") w() # ── 三、汇总 ── w("---") w() w("## 三、全维度汇总") w() w( "| 币种 | 周期 | 范围 | 总收益% | 夏普 | 回撤% | 交易 | 胜率% | 盈亏比 |" ) w( "|------|------|------|--------|------|------|------|------|------|" ) for row in sorted(all_rows, key=lambda x: (x["symbol"], x["label"], x["interval"])): w( f"| {row['symbol']:<10} | {row['interval']:<4} | {row['label']:<4} | " f"{row['return']:>+8.1f}% | {row['sharpe']:>6.2f} | " f"{row['dd']:>7.1f}% | {row['trades']:>5} | " f"{row['win']*100:>6.1f}% | {row['pf']:>6.2f} |" ) # ── 四、最优组合 ── w() w("## 四、各币种最佳组合（按夏普排序）") w() w( "| 币种 | 周期 | 范围 | 总收益% | 年化% | 夏普 | 回撤% | 交易 | 胜率% |" ) w( "|------|------|------|--------|------|------|------|------|------|" ) for symbol in SYMBOLS: candidates = [x for x in all_rows if x["symbol"] == symbol] if not candidates: continue best = max(candidates, key=lambda x: x["sharpe"]) w( f"| {best['symbol']:<10} | **{best['interval']}** | {best['label']} | " f"{best['return']:>+8.1f}% | {best['annual']:>+7.1f}% | {best['sharpe']:>6.2f} | " f"{best['dd']:>7.1f}% | {best['trades']:>5} | {best['win']*100:>6.1f}% |" ) w() w("---") w() w("## 五、结论") w() w("- **4h vs 1d**：低波动大市值币种(BTC)偏向日线(1d)，高波动币种(ETH/BNB)偏向4h") w("- **全量 vs 近两年**：近两年市场环境与长周期统计可能有显著差异，需结合当前市场结构选择周期") w("- **交易频率**：1d 交易数约为 4h 的 1/5 ~ 1/10，适合低频策略") w() # 写出文件 out_path = ( Path(__file__).resolve().parent.parent / "backtest" / "TIMEFRAME_COMPARISON_2H_6H.md" ) with open(out_path, "w", encoding="utf-8") as f: f.write("\n".join(out) + "\n") print(f"\n✓ 结果已保存到: {out_path}") if __name__ == "__main__": asyncio.run(main())