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: 数据读取示例
236 lines
7.8 KiB
Python
236 lines
7.8 KiB
Python
"""
|
||
多周期策略 v2 — 双周期同指标（EMA）
|
||
|
||
策略逻辑：
|
||
4h 和 30m 使用同一个技术指标 EMA，不同参数：
|
||
- 4h EMA50 → 判断主趋势方向
|
||
- 30m EMA20 → 寻找入场/出场时机
|
||
|
||
入场：4h 多头（Price > EMA50）+ 30m 价格上穿 EMA20
|
||
出场：30m 价格下穿 EMA20 或 4h 趋势转空
|
||
|
||
用法：
|
||
source .venv/bin/activate && python example/multi_tf_demo2.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.data import DataService
|
||
from engine.indicators import ema
|
||
|
||
|
||
class DualEMATFConfig(StrategyConfig):
|
||
"""双周期 EMA 策略配置"""
|
||
|
||
# 4h 主趋势
|
||
trend_ema_period: int = 50
|
||
|
||
# 30m 交易信号
|
||
entry_ema_period: int = 20
|
||
|
||
# 数据范围
|
||
data_start: Optional[datetime] = None
|
||
data_end: Optional[datetime] = None
|
||
|
||
|
||
class DualEMATFStrategy(BaseStrategy):
|
||
"""双周期 EMA 均线策略
|
||
|
||
┌──────────────────────────────────────────────┐
|
||
│ 4h EMA50 → 价格在上=多头趋势 │
|
||
│ 30m EMA20 → 价格在上+EMA上行+收阳=买入 │
|
||
│ EMA下行+收阴=卖出 │
|
||
└──────────────────────────────────────────────┘
|
||
"""
|
||
|
||
strategy_type = "dual_ema_tf"
|
||
|
||
def __init__(self, config: DualEMATFConfig):
|
||
super().__init__(config)
|
||
self.cfg: DualEMATFConfig = config
|
||
|
||
# 4h 预加载数据
|
||
self._klines_4h: list[Kline] = []
|
||
self._ema_4h: list[float] = []
|
||
|
||
# 30m 数据积累
|
||
self._closes_30m: list[float] = []
|
||
self._ema_30m: list[float] = []
|
||
|
||
# 持仓
|
||
self._has_position: bool = False
|
||
|
||
async def on_start(self) -> None:
|
||
"""预加载 4h 数据并计算 EMA50"""
|
||
from engine.common.config import config as app_config
|
||
|
||
ds = DataService(app_config.db)
|
||
await ds.connect()
|
||
try:
|
||
self._klines_4h = await ds.fetch_klines(
|
||
symbol=self.cfg.symbol,
|
||
interval="4h",
|
||
start_time=self.cfg.data_start,
|
||
end_time=self.cfg.data_end,
|
||
limit=1_000_000,
|
||
)
|
||
closes_4h = [k.close for k in self._klines_4h]
|
||
self._ema_4h = ema(closes_4h, self.cfg.trend_ema_period)
|
||
finally:
|
||
await ds.close()
|
||
|
||
await super().on_start()
|
||
|
||
def _get_4h_trend(self, ts: float) -> tuple[bool, float, float]:
|
||
"""4h 趋势判断（仅用已完成 K 线，close_time <= ts）"""
|
||
if not self._klines_4h:
|
||
return False, 0.0, 0.0
|
||
|
||
for i in range(len(self._klines_4h) - 1, -1, -1):
|
||
if self._klines_4h[i].close_time <= ts:
|
||
price = self._klines_4h[i].close
|
||
ema_val = self._ema_4h[i]
|
||
if ema_val == 0.0:
|
||
return False, price, ema_val
|
||
return price > ema_val, price, ema_val
|
||
|
||
return False, 0.0, 0.0
|
||
|
||
async def on_kline(self, kline: Kline) -> Optional[Signal]:
|
||
self._closes_30m.append(kline.close)
|
||
self._ema_30m = ema(self._closes_30m, self.cfg.entry_ema_period)
|
||
|
||
n = len(self._closes_30m)
|
||
if n < 2:
|
||
return None
|
||
|
||
cur_ema = self._ema_30m[-1]
|
||
prev_ema = self._ema_30m[-2]
|
||
|
||
if cur_ema == 0.0 or prev_ema == 0.0:
|
||
return None
|
||
|
||
cur_price = kline.close
|
||
|
||
# 30m K线收阳
|
||
is_bullish_bar = kline.close > kline.open
|
||
|
||
# 30m EMA 斜率（最近3根是否递增）
|
||
ema_sloping_up = (
|
||
n >= 4
|
||
and self._ema_30m[-1] > self._ema_30m[-2]
|
||
and self._ema_30m[-2] > self._ema_30m[-3]
|
||
)
|
||
|
||
# 4h 趋势
|
||
is_uptrend, price_4h, ema_4h = self._get_4h_trend(kline.open_time)
|
||
|
||
# ── 出场 ──
|
||
if self._has_position:
|
||
# 仅 4h 趋势转空时出场（让 30m 波动自然消化）
|
||
if not is_uptrend and price_4h > 0:
|
||
self._has_position = False
|
||
return Signal(
|
||
symbol=self.cfg.symbol,
|
||
side="SELL",
|
||
signal_type="MARKET",
|
||
confidence=0.9,
|
||
reason=f"4h转空 P={price_4h:.0f}<EMA{self.cfg.trend_ema_period}={ema_4h:.0f}",
|
||
timestamp=kline.open_time,
|
||
)
|
||
|
||
# ── 入场 ──
|
||
if not self._has_position:
|
||
if not is_uptrend:
|
||
return None
|
||
|
||
# 30m 价格在 EMA 上方 + EMA 上行 + 收阳 → 买入
|
||
price_above_ema = cur_price > cur_ema
|
||
if price_above_ema and ema_sloping_up and is_bullish_bar:
|
||
self._has_position = True
|
||
return Signal(
|
||
symbol=self.cfg.symbol,
|
||
side="BUY",
|
||
signal_type="MARKET",
|
||
confidence=0.7,
|
||
reason=(
|
||
f"30m多头确认 | "
|
||
f"4hP={price_4h:.0f}>E={ema_4h:.0f}"
|
||
),
|
||
timestamp=kline.open_time,
|
||
)
|
||
|
||
return None
|
||
|
||
|
||
# ============================================================
|
||
# 主函数
|
||
# ============================================================
|
||
|
||
|
||
async def main():
|
||
bt_config = BacktestConfig(
|
||
symbol="ETHUSDT",
|
||
interval="30m",
|
||
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 = DualEMATFConfig(
|
||
name="dual_ema_eth",
|
||
symbol="ETHUSDT",
|
||
trend_ema_period=50,
|
||
entry_ema_period=20,
|
||
data_start=bt_config.start_time,
|
||
data_end=bt_config.end_time,
|
||
)
|
||
|
||
print()
|
||
print("╔" + "═" * 60 + "╗")
|
||
print("║" + " 多周期策略 v2 — 双周期同指标 (EMA)".center(54) + "║")
|
||
print("╠" + "═" * 60 + "╣")
|
||
print(f"║ {'交易对:':<8} {bt_config.symbol:<14} {'周期:':<6} {bt_config.interval:<12} ║")
|
||
print(f"║ {'时间:':<8} {bt_config.start_time.date()} ~ {bt_config.end_time.date()} ║")
|
||
print("╠" + "═" * 60 + "╣")
|
||
print("║ 同指标 · 双周期： ║")
|
||
print(f"║ 4h EMA{strategy_config.trend_ema_period} → 趋势方向（价格在上=多头） ║")
|
||
print(f"║ 30m EMA{strategy_config.entry_ema_period} → 价格在上+EMA上行+收阳=买入 ║")
|
||
print("╚" + "═" * 60 + "╝")
|
||
print()
|
||
|
||
engine = BacktestEngine(bt_config, db_config=config.db)
|
||
result = await engine.run(DualEMATFStrategy, strategy_config)
|
||
|
||
print(result.summary())
|
||
|
||
sells = [t for t in result.trades if t.pnl is not None]
|
||
if sells:
|
||
print(f"\n最近 10 笔平仓:")
|
||
print(f"{'时间':<22} {'方向':<6} {'价格':>10} {'盈亏':>10} 原因")
|
||
print("-" * 80)
|
||
for t in sells[-10:]:
|
||
dt = datetime.fromtimestamp(t.timestamp / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
|
||
print(f"{dt:<22} {t.side:<6} {t.price:>10.2f} {t.pnl:>+10.2f} {t.reason}")
|
||
|
||
print("\n回测完成。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|