rekey/trade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(engine): 添加策略示例集（18 个 Demo）

Browse Source 
- 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: 数据读取示例
This commit is contained in:
Rekey
2026-06-12 10:27:04 +08:00
parent 4da520c14b
commit 515e61c517
21 changed files with 5194 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engine/example/backtest_demo.py
+218
View File
@@ -0,0 +1,218 @@
"""
回测引擎使用示例 — 双策略演示
用法：
source .venv/bin/activate && python example/backtest_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 sma, rsi
# ============================================================
# 策略 1：双均线交叉
# ============================================================
class MACrossConfig(StrategyConfig):
fast_period: int = 7
slow_period: int = 25
class MACrossStrategy(BaseStrategy):
"""双均线交叉策略 — 使用 engine.indicators.sma 计算均线"""
strategy_type = "ma_cross"
def __init__(self, config: MACrossConfig):
super().__init__(config)
self.config: MACrossConfig = config
self._closes: list[float] = []
self._last_signal: Optional[str] = None
async def on_kline(self, kline: Kline) -> Optional[Signal]:
self._closes.append(kline.close)
# 使用指标库计算 SMA
fast_ma = sma(self._closes, self.config.fast_period)
slow_ma = sma(self._closes, self.config.slow_period)
fast = fast_ma[-1]
slow = slow_ma[-1]
if fast == 0.0 or slow == 0.0:
return None
if fast > slow and self._last_signal != "BUY":
self._last_signal = "BUY"
return Signal(
symbol=self.config.symbol,
side="BUY",
signal_type="MARKET",
confidence=0.8,
reason=f"金叉 MA{self.config.fast_period}>{self.config.slow_period}",
timestamp=kline.open_time,
)
if fast < slow and self._last_signal != "SELL":
self._last_signal = "SELL"
return Signal(
symbol=self.config.symbol,
side="SELL",
signal_type="MARKET",
confidence=0.8,
reason=f"死叉 MA{self.config.fast_period}<{self.config.slow_period}",
timestamp=kline.open_time,
)
return None
# ============================================================
# 策略 2：RSI 超买超卖
# ============================================================
class RSIStrategyConfig(StrategyConfig):
period: int = 14
oversold: float = 30.0 # 超卖阈值
overbought: float = 70.0 # 超买阈值
class RSIStrategy(BaseStrategy):
"""RSI 超买超卖策略 — 使用 engine.indicators.rsi 计算 RSI
RSI 低于超卖线 → 买入；RSI 高于超买线 → 卖出。
"""
strategy_type = "rsi"
def __init__(self, config: RSIStrategyConfig):
super().__init__(config)
self.config: RSIStrategyConfig = config
self._closes: list[float] = []
self._has_position = False
async def on_kline(self, kline: Kline) -> Optional[Signal]:
self._closes.append(kline.close)
# 使用指标库计算 RSI
rsi_vals = rsi(self._closes, self.config.period)
current_rsi = rsi_vals[-1]
if current_rsi == 0.0:
return None
# 超卖 → 买入
if current_rsi < self.config.oversold and not self._has_position:
self._has_position = True
return Signal(
symbol=self.config.symbol,
side="BUY",
signal_type="MARKET",
confidence=0.7,
reason=f"RSI超卖 ({current_rsi:.1f} < {self.config.oversold})",
timestamp=kline.open_time,
)
# 超买 → 卖出
if current_rsi > self.config.overbought and self._has_position:
self._has_position = False
return Signal(
symbol=self.config.symbol,
side="SELL",
signal_type="MARKET",
confidence=0.7,
reason=f"RSI超买 ({current_rsi:.1f} > {self.config.overbought})",
timestamp=kline.open_time,
)
return None
# ============================================================
# 主函数
# ============================================================
async def run_backtest(
engine: BacktestEngine,
strategy_cls,
strategy_config: StrategyConfig,
label: str,
):
"""运行一次回测并打印结果"""
print(f"\n{'━' * 60}")
print(f" {label}")
print(f"{'━' * 60}")
result = await engine.run(strategy_cls, strategy_config)
print(result.summary())
# 最近 5 笔交易
if result.trades:
print(f"\n 最近 5 笔交易:")
print(f" {'时间':<22} {'方向':<6} {'价格':>10} {'数量':>10} {'盈亏':>10} 原因")
for t in result.trades[-5:]:
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.4f} {t.quantity:>10.6f} {pnl_str:>10} {t.reason}")
return result
async def main():
# ── 回测配置 ──
bt_config = BacktestConfig(
symbol="ETHUSDT",
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,
)
print(f"\n回测: {bt_config.symbol} {bt_config.interval}")
print(f"时间: {bt_config.start_time.date()} ~ {bt_config.end_time.date()}")
print(f"初始资金: {bt_config.initial_capital:.2f} USDT")
engine = BacktestEngine(bt_config, db_config=config.db)
# ── 策略 1：双均线交叉 ──
ma_config = MACrossConfig(
name="ma_cross_eth",
symbol="ETHUSDT",
fast_period=7,
slow_period=25,
)
await run_backtest(engine, MACrossStrategy, ma_config, "策略 1：双均线交叉 (MA7/MA25)")
# ── 策略 2：RSI 超买超卖 ──
rsi_config = RSIStrategyConfig(
name="rsi_eth",
symbol="ETHUSDT",
period=14,
oversold=30.0,
overbought=70.0,
)
await run_backtest(engine, RSIStrategy, rsi_config, "策略 2：RSI 超买超卖 (30/70)")
print("\n全部回测完成。")
if __name__ == "__main__":
asyncio.run(main())