""" 多因子组合回测 — 三重共振策略 随机挑选 3 个技术指标组合成一个策略： - MACD (趋势因子) — 金叉/死叉判断方向 - RSI (动量因子) — 阈值过滤避免追高抄底 - Bollinger (波动率因子) — 中轨确认趋势强度 用法： source .venv/bin/activate && python example/factor_demo.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 from engine.indicators import macd, rsi, bollinger, atr # ============================================================ # 三重共振策略 # ============================================================ class TripleFactorConfig(StrategyConfig): """三重因子组合策略配置""" # MACD macd_fast: int = 12 macd_slow: int = 26 macd_signal: int = 9 # RSI rsi_period: int = 14 rsi_oversold: float = 30.0 # 超卖线（入场需在此之上） rsi_overbought: float = 65.0 # 入场过热线（入场需在此之下） rsi_exit: float = 75.0 # 卖出线 # Bollinger bb_period: int = 20 bb_std: float = 2.0 # ATR 动态止损倍数（0 表示不启用） atr_period: int = 14 atr_stop_mult: float = 2.0 class TripleFactorStrategy(BaseStrategy): """三重共振策略 ┌─────────────┬──────────────────────────────────────┐ │ 因子 │ 作用 │ ├─────────────┼──────────────────────────────────────┤ │ MACD (趋势) │ 金叉=看多入场信号，死叉=看空出场信号 │ │ RSI (动量) │ 3075 过热出场 │ │ BB (波动) │ 价格>中轨确认多头趋势，跌破下轨出场 │ └─────────────┴──────────────────────────────────────┘ 入场（三重共振）： 1. MACD 金叉（上穿信号线） 2. RSI 在 [30, 65] 区间（合理动量） 3. 价格 > 布林中轨（趋势向上） 出场（任一触发）： 1. MACD 死叉（下穿信号线） 2. RSI > 75（过热） 3. 价格 < 布林下轨（趋势破位） """ strategy_type = "triple_factor" def __init__(self, config: TripleFactorConfig): super().__init__(config) self.cfg: TripleFactorConfig = config self._closes: list[float] = [] self._highs: list[float] = [] self._lows: list[float] = [] async def on_kline(self, kline: Kline) -> Optional[Signal]: self._closes.append(kline.close) self._highs.append(kline.high) self._lows.append(kline.low) n = len(self._closes) max_period = max( self.cfg.macd_slow + self.cfg.macd_signal, self.cfg.rsi_period + 1, self.cfg.bb_period, ) if n < max_period: return None # ── 全量计算因子（每个 bar 一次）── macd_line, signal_line, _hist = macd( self._closes, fast=self.cfg.macd_fast, slow=self.cfg.macd_slow, signal=self.cfg.macd_signal, ) rsi_vals = rsi(self._closes, period=self.cfg.rsi_period) _upper, mid, lower = bollinger( self._closes, period=self.cfg.bb_period, std=self.cfg.bb_std, ) # 当前值和前一根的值 cur_macd = macd_line[-1] cur_signal = signal_line[-1] prev_macd = macd_line[-2] prev_signal = signal_line[-2] cur_rsi = rsi_vals[-1] prev_rsi = rsi_vals[-2] cur_mid = mid[-1] cur_lower = lower[-1] cur_price = kline.close if cur_macd == 0.0 or cur_rsi == 0.0 or cur_mid == 0.0: return None # ── 入场：2/3 共振即可 ── golden_cross = prev_macd <= prev_signal and cur_macd > cur_signal rsi_ok = self.cfg.rsi_oversold < cur_rsi < self.cfg.rsi_overbought above_mid = cur_price > cur_mid score = golden_cross + rsi_ok + above_mid if score >= 2: return Signal( symbol=self.cfg.symbol, side="BUY", signal_type="MARKET", confidence=0.6 + score * 0.1, reason=( f"{score}/3共振" f"{' MACD金叉' if golden_cross else ''}" f"{' RSI=' + str(round(cur_rsi, 1)) if rsi_ok else ''}" f"{' Price>BBmid' if above_mid else ''}" ), timestamp=kline.open_time, ) # ── 出场：任一强信号 ── death_cross = prev_macd >= prev_signal and cur_macd < cur_signal rsi_overheat = cur_rsi > self.cfg.rsi_exit below_lower = cur_price < cur_lower if death_cross: return Signal( symbol=self.cfg.symbol, side="SELL", signal_type="MARKET", confidence=0.8, reason="MACD死叉", timestamp=kline.open_time, ) if rsi_overheat and cur_rsi > prev_rsi: return Signal( symbol=self.cfg.symbol, side="SELL", signal_type="MARKET", confidence=0.7, reason=f"RSI过热({cur_rsi:.1f})", timestamp=kline.open_time, ) if below_lower and cur_price < cur_mid: return Signal( symbol=self.cfg.symbol, side="SELL", signal_type="MARKET", confidence=0.6, reason=f"跌破BB下轨({cur_lower:.2f})", timestamp=kline.open_time, ) return None # ============================================================ # 主函数 # ============================================================ async def main(): bt_config = BacktestConfig( symbol="BTCUSDT", interval="4h", start_time=datetime(2024, 1, 1), end_time=datetime(2026, 1, 1), initial_capital=10_000.0, commission_pct=0.001, slippage_pct=0.0005, warmup_bars=100, ) strategy_config = TripleFactorConfig( name="triple_factor_btc", symbol="BTCUSDT", macd_fast=12, macd_slow=26, macd_signal=9, rsi_period=14, rsi_oversold=30.0, rsi_overbought=65.0, rsi_exit=75.0, bb_period=20, bb_std=2.0, ) print() print("╔" + "═" * 58 + "╗") print("║" + " 多因子组合回测 — 三重共振策略".center(52) + "║") print("╠" + "═" * 58 + "╣") print(f"║ {'交易对:':<8} {bt_config.symbol:<12} {'周期:':<6} {bt_config.interval:<10} ║") print(f"║ {'时间:':<8} {bt_config.start_time.date()} ~ {bt_config.end_time.date()} ║") print(f"║ {'初始资金:':<8} {bt_config.initial_capital:>12.2f} USDT ║") print("╠" + "═" * 58 + "╣") print("║ 因子组合: ║") print(f"║ 1. MACD({strategy_config.macd_fast},{strategy_config.macd_slow},{strategy_config.macd_signal}) — 趋势方向 ║") print(f"║ 2. RSI({strategy_config.rsi_period}) — 动量过滤 ║") print(f"║ 3. Bollinger({strategy_config.bb_period},{strategy_config.bb_std}) — 波动率确认 ║") print("╚" + "═" * 58 + "╝") print() engine = BacktestEngine(bt_config, db_config=config.db) result = await engine.run(TripleFactorStrategy, strategy_config) print(result.summary()) # 打印全部交易 if result.trades: print(f"\n全部交易 ({len(result.trades)} 笔):") print(f"{'时间':<22} {'方向':<6} {'价格':>10} {'数量':>10} {'盈亏':>10} 原因") print("-" * 100) for t in result.trades: dt = datetime.fromtimestamp(t.timestamp / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M") pnl_str = f"{t.pnl:+.4f}" if t.pnl is not None else "—" print(f"{dt:<22} {t.side:<6} {t.price:>10.2f} {t.quantity:>10.6f} {pnl_str:>10} {t.reason}") print("\n回测完成。") if __name__ == "__main__": asyncio.run(main())