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: 数据读取示例
153 lines
6.9 KiB
Python
153 lines
6.9 KiB
Python
"""
|
|
全周期回测 — 2017-2026，覆盖完整牛熊
|
|
|
|
多空双向 EMA 趋势跟踪，展示牛市/熊市/全周期分段表现。
|
|
|
|
用法：
|
|
source .venv/bin/activate && python example/full_cycle.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 import ema, atr
|
|
from engine.example.long_short import LongShortEngine, LongShortEmaConfig, LongShortEmaStrategy
|
|
|
|
|
|
# ════════════════════════════════════════════════════════
|
|
|
|
SYMBOLS = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"]
|
|
|
|
PARAMS = {
|
|
"BTCUSDT": (10, 50),
|
|
"ETHUSDT": (10, 75),
|
|
"BNBUSDT": (20, 50),
|
|
"SOLUSDT": (30, 50),
|
|
}
|
|
|
|
# 牛熊分段（以 BTC 为参考）
|
|
PERIODS = [
|
|
("2017-2018 牛市", datetime(2017, 1, 1), datetime(2018, 1, 1)),
|
|
("2018 熊市", datetime(2018, 1, 1), datetime(2019, 1, 1)),
|
|
("2019 反弹", datetime(2019, 1, 1), datetime(2020, 1, 1)),
|
|
("2020 牛初+312", datetime(2020, 1, 1), datetime(2021, 1, 1)),
|
|
("2021 牛市", datetime(2021, 1, 1), datetime(2022, 1, 1)),
|
|
("2022 熊市", datetime(2022, 1, 1), datetime(2023, 1, 1)),
|
|
("2023 复苏", datetime(2023, 1, 1), datetime(2024, 1, 1)),
|
|
("2024-2025 牛市", datetime(2024, 1, 1), datetime(2026, 1, 1)),
|
|
]
|
|
|
|
DATE_START = datetime(2017, 1, 1)
|
|
DATE_END = datetime(2026, 1, 1)
|
|
|
|
|
|
async def run_backtest(symbol, fast, slow, start, end):
|
|
sc = LongShortEmaConfig(symbol=symbol, fast=fast, slow=slow)
|
|
bt = BacktestConfig(symbol=symbol, interval="4h", start_time=start, end_time=end,
|
|
initial_capital=10_000.0)
|
|
engine = LongShortEngine(bt, db_config=config.db)
|
|
return await engine.run(LongShortEmaStrategy, sc)
|
|
|
|
|
|
async def main():
|
|
print()
|
|
print("═" * 125)
|
|
print(" 全周期多空回测 — 2017-2026 完整牛熊 | 4h EMA趋势")
|
|
print("═" * 125)
|
|
|
|
# ── 全周期汇总 ──
|
|
print(f"\n ■ 全周期 2017-2026 汇总")
|
|
print(f" {'币种':<10} {'总收益%':>7} {'年化%':>7} {'夏普':>6} {'回撤%':>7} {'交易':>5} {'多头P&L':>10} {'空头P&L':>10}")
|
|
print(" " + "─" * 105)
|
|
|
|
for symbol in SYMBOLS:
|
|
fast, slow = PARAMS[symbol]
|
|
r = await run_backtest(symbol, fast, slow, DATE_START, DATE_END)
|
|
m = r.metrics
|
|
long_t = [t for t in r.trades if t.pnl is not None and t.side == "SELL"]
|
|
short_t = [t for t in r.trades if t.pnl is not None and t.side == "BUY"]
|
|
long_pnl = sum(t.pnl for t in long_t) if long_t else 0
|
|
short_pnl = sum(t.pnl for t in short_t) if short_t else 0
|
|
print(f" {symbol:<10} {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} {long_pnl:>+9.0f} {short_pnl:>+9.0f}")
|
|
|
|
# ── BTC 分段 ──
|
|
print(f"\n ■ BTC 各阶段表现 (参数 EMA{10},{50})")
|
|
print(f" {'阶段':<22} {'总收益%':>7} {'年化%':>7} {'夏普':>6} {'回撤%':>7} {'多头P&L':>10} {'空头P&L':>10}")
|
|
print(" " + "─" * 105)
|
|
|
|
for period_name, p_start, p_end in PERIODS:
|
|
try:
|
|
r = await run_backtest("BTCUSDT", 10, 50, p_start, p_end)
|
|
m = r.metrics
|
|
long_t = [t for t in r.trades if t.pnl is not None and t.side == "SELL"]
|
|
short_t = [t for t in r.trades if t.pnl is not None and t.side == "BUY"]
|
|
long_pnl = sum(t.pnl for t in long_t) if long_t else 0
|
|
short_pnl = sum(t.pnl for t in short_t) if short_t else 0
|
|
print(f" {period_name:<22} {m.total_return_pct:>6.1f}% {m.annual_return_pct:>6.1f}% {m.sharpe_ratio:>6.2f} {m.max_drawdown_pct:>6.1f}% {long_pnl:>+9.0f} {short_pnl:>+9.0f}")
|
|
except Exception as e:
|
|
print(f" {period_name:<22} 数据不足或错误: {e}")
|
|
|
|
# ── 只做多 vs 多空全周期对比 ──
|
|
print(f"\n ■ BTC 只做多 vs 多空 (全周期)")
|
|
# 只做多需要单跑一次（LongShortEngine 本身就支持只做多：不开空就行）
|
|
# 简单做法：用原版 BacktestEngine 跑一次只做多
|
|
from engine.backtest import BacktestEngine
|
|
from engine.common.base import BaseStrategy as BS, Signal as Sig, StrategyConfig as SC
|
|
|
|
class LongOnlyEMAConfig(SC):
|
|
fast: int = 10; slow: int = 50; atr_stop: float = 2.5
|
|
|
|
class LongOnlyEMAStrategy(BS):
|
|
strategy_type = "long_only"
|
|
def __init__(self, c): super().__init__(c); self.cfg = c
|
|
async def on_start(self): self._c = []; self._h = []; self._l = []; self._hp = 0.0; self._in = False; await super().on_start()
|
|
async def on_kline(self, k):
|
|
self._c.append(k.close); self._h.append(k.high); self._l.append(k.low)
|
|
n = len(self._c)
|
|
if n < self.cfg.slow + 5: return None
|
|
f = ema(self._c, self.cfg.fast); s = ema(self._c, self.cfg.slow)
|
|
a = atr(self._h, self._l, self._c, 14)
|
|
cf, cs, ca = f[-1], s[-1], a[-1]; pf, ps = f[-2], s[-2]
|
|
if cf == 0 or cs == 0 or ca == 0: return None
|
|
if self._in:
|
|
self._hp = max(self._hp, k.high); stop = self._hp - self.cfg.atr_stop * ca
|
|
if (pf >= ps and cf < cs) or k.close < stop:
|
|
self._in = False
|
|
return Sig(symbol=self.cfg.symbol, side="SELL", reason="死叉" if pf >= ps else "ATR止损", timestamp=k.open_time)
|
|
else:
|
|
if pf <= ps and cf > cs:
|
|
self._in = True; self._hp = k.close
|
|
return Sig(symbol=self.cfg.symbol, side="BUY", reason="金叉", timestamp=k.open_time)
|
|
return None
|
|
|
|
lo_sc = LongOnlyEMAConfig(symbol="BTCUSDT")
|
|
lo_bt = BacktestConfig(symbol="BTCUSDT", interval="4h", start_time=DATE_START, end_time=DATE_END, initial_capital=10_000.0)
|
|
lo_eng = BacktestEngine(lo_bt, db_config=config.db)
|
|
lo_r = await lo_eng.run(LongOnlyEMAStrategy, lo_sc)
|
|
lo_m = lo_r.metrics
|
|
|
|
# 多空
|
|
ls_r = await run_backtest("BTCUSDT", 10, 50, DATE_START, DATE_END)
|
|
ls_m = ls_r.metrics
|
|
|
|
print(f" {'':<10} {'总收益%':>7} {'年化%':>7} {'夏普':>6} {'回撤%':>7} {'交易':>5}")
|
|
print(f" {'只做多':<10} {lo_m.total_return_pct:>6.1f}% {lo_m.annual_return_pct:>6.1f}% {lo_m.sharpe_ratio:>6.2f} {lo_m.max_drawdown_pct:>6.1f}% {lo_m.total_trades:>5}")
|
|
print(f" {'多空':<10} {ls_m.total_return_pct:>6.1f}% {ls_m.annual_return_pct:>6.1f}% {ls_m.sharpe_ratio:>6.2f} {ls_m.max_drawdown_pct:>6.1f}% {ls_m.total_trades:>5}")
|
|
|
|
print("\n═" * 125)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|