""" 多周期牛熊共识策略 — 四周期协同判定 核心思路： 1w 定宏观方向（只做多 / 只做空）→ 1d 确认中周期 → 4h EMA 金叉/死叉入场 配合 ATR 动态止损 + 多周期投票确认出场 对比基准：单周期 RegimeEmaStrategy（regime_all.py）、无过滤 LongShortEmaStrategy 用法： source .venv/bin/activate && python example/regime_mtf_strategy.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.incremental import EmaInc, AtrInc from engine.indicators.regime import MultiTimeframeRegime from engine.example.long_short import LongShortEngine from engine.example.regime_all import RegimeEmaConfig as OldRegimeCfg from engine.example.regime_all import RegimeEmaStrategy as OldRegimeS from engine.example.long_short import LongShortEmaConfig as LSCfg from engine.example.long_short import LongShortEmaStrategy as LSS # ════════════════════════════════════════════════════════ # 多周期牛熊共识策略 # ════════════════════════════════════════════════════════ class MTFConfig(StrategyConfig): fast: int = 10 slow: int = 50 atr_stop: float = 2.5 class MTFRegimeStrategy(BaseStrategy): """四周期协同牛熊策略 判定层（MultiTimeframeRegime，即时法）： 每个 4h bar 的 close 同时更新 1h / 4h / 1d / 1w 四个检测器， 1w 定方向宏调，1d 确认中周期，1h 预警微观背离。 入场： 做多 — 1w bull + 1d not bear + 4h EMA 金叉 做空 — 1w bear + 1d not bull + 4h EMA 死叉 出场： EMA 交叉反转、ATR 跟踪止损、高周期方向逆转 """ strategy_type = "mtf_regime" def __init__(self, c: MTFConfig): super().__init__(c) self.cfg = c self._mtf = MultiTimeframeRegime(["1h", "4h", "1d", "1w"]) self._ema_fast = EmaInc(c.fast) self._ema_slow = EmaInc(c.slow) self._atr = AtrInc(14) self._side: str = "" self._hp: float = 0.0 self._lp: float = float("inf") async def on_kline(self, k: Kline) -> Optional[Signal]: self._mtf.update(k.close) self._ema_fast.update(k.close) self._ema_slow.update(k.close) self._atr.update(k.high, k.low, k.close) n = len(self._mtf._prices) if n < 220: return None regimes = self._mtf.detect_all() r1h = regimes.get("1h", "unknown") r4h = regimes.get("4h", "unknown") r1d = regimes.get("1d", "unknown") r1w = regimes.get("1w", "unknown") cf, cs = self._ema_fast[-1], self._ema_slow[-1] ca = self._atr[-1] pf, ps = self._ema_fast[-2], self._ema_slow[-2] if cf == 0 or cs == 0 or ca == 0: return None golden = pf <= ps and cf > cs death = pf >= ps and cf < cs # ── 多头持仓 ── if self._side == "long": self._hp = max(self._hp, k.high) stop = self._hp - self.cfg.atr_stop * ca exit_reason = None if death: exit_reason = "死叉" elif k.close < stop: exit_reason = "ATR止损" elif r1w == "bear": exit_reason = "1w转熊→平多" elif r1d == "bear" and r4h == "bear": exit_reason = "1d+4h双双转熊→平多" if exit_reason: self._side = "" return Signal(symbol=self.cfg.symbol, side="SELL", reason=exit_reason, timestamp=k.open_time) # ── 空头持仓 ── elif self._side == "short": self._lp = min(self._lp, k.low) stop = self._lp + self.cfg.atr_stop * ca exit_reason = None if golden: exit_reason = "金叉" elif k.close > stop: exit_reason = "ATR止损" elif r1w == "bull": exit_reason = "1w转牛→平空" elif r1d == "bull" and r4h == "bull": exit_reason = "1d+4h双双转牛→平空" if exit_reason: self._side = "" return Signal(symbol=self.cfg.symbol, side="BUY", reason=exit_reason, timestamp=k.open_time) # ── 空仓等信号 ── else: # 做多条件：1w牛 + 1d不熊 + 4h金叉 long_ok = r1w == "bull" and r1d != "bear" and golden # 做空条件：1w熊 + 1d不牛 + 4h死叉 short_ok = r1w == "bear" and r1d != "bull" and death if long_ok: self._side = "long" self._hp = k.close return Signal(symbol=self.cfg.symbol, side="BUY", reason=f"多周期做多({r1w}/{r1d}/{r4h})", timestamp=k.open_time) if short_ok: self._side = "short" self._lp = k.close return Signal(symbol=self.cfg.symbol, side="SELL", reason=f"多周期做空({r1w}/{r1d}/{r4h})", timestamp=k.open_time) return None # ════════════════════════════════════════════════════════ # 对照策略 2：1w 绝对过滤（更强约束） # ════════════════════════════════════════════════════════ class MTFStrictConfig(StrategyConfig): fast: int = 10 slow: int = 50 atr_stop: float = 2.5 class MTFStrictStrategy(BaseStrategy): """严格多周期策略 — 必须 1w + 1d 方向一致才入场 入场： 做多 — 1w bull + 1d bull + 4h 金叉 做空 — 1w bear + 1d bear + 4h 死叉 出场： 4h 死叉/金叉 或 ATR 止损 或 1w 方向逆转 """ strategy_type = "mtf_strict" def __init__(self, c: MTFStrictConfig): super().__init__(c) self.cfg = c self._mtf = MultiTimeframeRegime(["1h", "4h", "1d", "1w"]) self._ema_fast = EmaInc(c.fast) self._ema_slow = EmaInc(c.slow) self._atr = AtrInc(14) self._side: str = "" self._hp: float = 0.0 self._lp: float = float("inf") async def on_kline(self, k: Kline) -> Optional[Signal]: self._mtf.update(k.close) self._ema_fast.update(k.close) self._ema_slow.update(k.close) self._atr.update(k.high, k.low, k.close) n = len(self._mtf._prices) if n < 220: return None regimes = self._mtf.detect_all() r4h = regimes.get("4h", "unknown") r1d = regimes.get("1d", "unknown") r1w = regimes.get("1w", "unknown") cf, cs = self._ema_fast[-1], self._ema_slow[-1] ca = self._atr[-1] pf, ps = self._ema_fast[-2], self._ema_slow[-2] if cf == 0 or cs == 0 or ca == 0: return None golden = pf <= ps and cf > cs death = pf >= ps and cf < cs # 多头持仓 if self._side == "long": self._hp = max(self._hp, k.high) stop = self._hp - self.cfg.atr_stop * ca if death or k.close < stop or r1w == "bear": self._side = "" reason = "死叉" if death else ("ATR止损" if k.close < stop else "1w转熊") return Signal(symbol=self.cfg.symbol, side="SELL", reason=reason, timestamp=k.open_time) # 空头持仓 elif self._side == "short": self._lp = min(self._lp, k.low) stop = self._lp + self.cfg.atr_stop * ca if golden or k.close > stop or r1w == "bull": self._side = "" reason = "金叉" if golden else ("ATR止损" if k.close > stop else "1w转牛") return Signal(symbol=self.cfg.symbol, side="BUY", reason=reason, timestamp=k.open_time) # 空仓 else: if r1w == "bull" and r1d == "bull" and golden: self._side = "long" self._hp = k.close return Signal(symbol=self.cfg.symbol, side="BUY", reason=f"严格做多({r1w}+{r1d}+{r4h})", timestamp=k.open_time) if r1w == "bear" and r1d == "bear" and death: self._side = "short" self._lp = k.close return Signal(symbol=self.cfg.symbol, side="SELL", reason=f"严格做空({r1w}+{r1d}+{r4h})", timestamp=k.open_time) return None # ════════════════════════════════════════════════════════ # 回测入口 # ════════════════════════════════════════════════════════ SYMBOLS = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"] PARAMS = { "BTCUSDT": (10, 50), "ETHUSDT": (10, 75), "BNBUSDT": (20, 50), "SOLUSDT": (30, 50), } DATE_START = datetime(2024, 1, 1) DATE_END = datetime(2026, 1, 1) async def run_one(engine, strategy_cls, config_cls, symbol, desc, params=None): fast, slow = PARAMS[symbol] if params is None else params sc = config_cls(symbol=symbol, fast=fast, slow=slow) bt = BacktestConfig( symbol=symbol, interval="4h", start_time=DATE_START, end_time=DATE_END, initial_capital=10_000.0, warmup_bars=250, ) eng = engine(bt, db_config=config.db) r = await eng.run(strategy_cls, sc) return r async def main(): print() print("═" * 120) print(" 多周期牛熊共识策略 — MTF Regime × 4h EMA | 2024-2026") print("═" * 120) print() print(" 对比策略:") print(" • MTF Consensus — 1w定方向 + 1d确认 + 4h金叉/死叉入场（新）") print(" • MTF Strict — 1w+1d必须同向才入场（新）") print(" • Old Regime — 单周期(4h)三法投票（regime_all.py 基线）") print(" • Long/Short — 无牛熊过滤，始终多空（long_short.py 基线）") print() for symbol in SYMBOLS: print(f" ── {symbol} ──") print(f" {'策略':<18} {'总收益%':>7} {'年化%':>7} {'夏普':>6} {'回撤%':>7} {'交易':>5} {'胜率%':>6} {'盈亏比':>6} {'多头P&L':>9} {'空头P&L':>9}") print(" " + "─" * 110) results = {} # 1. MTF Consensus (new) try: r1 = await run_one(LongShortEngine, MTFRegimeStrategy, MTFConfig, symbol, "MTF Consensus") m = r1.metrics long1 = [t for t in r1.trades if t.pnl is not None and t.side == "SELL"] short1 = [t for t in r1.trades if t.pnl is not None and t.side == "BUY"] lp1 = sum(t.pnl for t in long1) if long1 else 0 sp1 = sum(t.pnl for t in short1) if short1 else 0 results["mtf"] = m print(f" {'MTF Consensus':<18} {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} {m.win_rate*100:>5.1f}% {m.profit_factor:>6.2f} {lp1:>+9.0f} {sp1:>+9.0f}") except Exception as e: print(f" {'MTF Consensus':<18} 错误: {e}") # 2. MTF Strict (new) try: r2 = await run_one(LongShortEngine, MTFStrictStrategy, MTFStrictConfig, symbol, "MTF Strict") m = r2.metrics long2 = [t for t in r2.trades if t.pnl is not None and t.side == "SELL"] short2 = [t for t in r2.trades if t.pnl is not None and t.side == "BUY"] lp2 = sum(t.pnl for t in long2) if long2 else 0 sp2 = sum(t.pnl for t in short2) if short2 else 0 results["strict"] = m print(f" {'MTF Strict':<18} {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} {m.win_rate*100:>5.1f}% {m.profit_factor:>6.2f} {lp2:>+9.0f} {sp2:>+9.0f}") except Exception as e: print(f" {'MTF Strict':<18} 错误: {e}") # 3. Old Regime (baseline) try: r3 = await run_one(LongShortEngine, OldRegimeS, OldRegimeCfg, symbol, "Old Regime") m = r3.metrics long3 = [t for t in r3.trades if t.pnl is not None and t.side == "SELL"] short3 = [t for t in r3.trades if t.pnl is not None and t.side == "BUY"] lp3 = sum(t.pnl for t in long3) if long3 else 0 sp3 = sum(t.pnl for t in short3) if short3 else 0 results["old"] = m print(f" {'Old Regime':<18} {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} {m.win_rate*100:>5.1f}% {m.profit_factor:>6.2f} {lp3:>+9.0f} {sp3:>+9.0f}") except Exception as e: print(f" {'Old Regime':<18} 错误: {e}") # 4. Long/Short (baseline, no filter) try: r4 = await run_one(LongShortEngine, LSS, LSCfg, symbol, "Long/Short") m = r4.metrics long4 = [t for t in r4.trades if t.pnl is not None and t.side == "SELL"] short4 = [t for t in r4.trades if t.pnl is not None and t.side == "BUY"] lp4 = sum(t.pnl for t in long4) if long4 else 0 sp4 = sum(t.pnl for t in short4) if short4 else 0 results["ls"] = m print(f" {'Long/Short':<18} {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} {m.win_rate*100:>5.1f}% {m.profit_factor:>6.2f} {lp4:>+9.0f} {sp4:>+9.0f}") except Exception as e: print(f" {'Long/Short':<18} 错误: {e}") # 简要对比 if len(results) >= 2: best_sharpe = max(results.items(), key=lambda x: x[1].sharpe_ratio) worst_dd = min(results.items(), key=lambda x: x[1].max_drawdown_pct) print(f" └ 最佳夏普: {best_sharpe[0]} ({best_sharpe[1].sharpe_ratio:.2f}) | " f"最小回撤: {worst_dd[0]} ({worst_dd[1].max_drawdown_pct:.1f}%)") print() print("═" * 120) if __name__ == "__main__": asyncio.run(main())