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: 数据读取示例
254 lines
8.9 KiB
Python
254 lines
8.9 KiB
Python
"""
|
|
多因子组合回测 — 三重共振策略
|
|
|
|
随机挑选 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 (动量) │ 30<RSI<65 区间入场，RSI>75 过热出场 │
|
|
│ 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())
|