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
5.7 KiB
Python
153 lines
5.7 KiB
Python
"""
|
||
极简策略测试 — 一条均线 + 一个止损
|
||
|
||
核心理念：多加过滤条件往往不如简洁的信号。
|
||
|
||
策略：价格上穿 EMA(N) → 买入，价格下穿 EMA(N) → 卖出，ATR 动态止损。
|
||
无成交量、无多周期、无ADX、无双均线交叉。
|
||
|
||
对比 N=10/20/30，4个币种，4h周期，2024-2026。
|
||
|
||
用法：
|
||
source .venv/bin/activate && python example/strategy_simple.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 ema, atr
|
||
|
||
|
||
class SingleEMAConfig(StrategyConfig):
|
||
period: int = 20
|
||
atr_stop: float = 2.5
|
||
|
||
|
||
class SingleEMAStrategy(BaseStrategy):
|
||
"""一条EMA均线 + ATR止损。没有更多了。"""
|
||
|
||
strategy_type = "single_ema"
|
||
|
||
def __init__(self, c: SingleEMAConfig):
|
||
super().__init__(c)
|
||
self.cfg = c
|
||
self._closes: list[float] = []
|
||
self._highs: list[float] = []
|
||
self._lows: list[float] = []
|
||
self._highest: float = 0.0
|
||
self._in_position = False
|
||
|
||
async def on_kline(self, k: Kline) -> Optional[Signal]:
|
||
self._closes.append(k.close)
|
||
self._highs.append(k.high)
|
||
self._lows.append(k.low)
|
||
n = len(self._closes)
|
||
if n < self.cfg.period + 5:
|
||
return None
|
||
|
||
ema_vals = ema(self._closes, self.cfg.period)
|
||
atr_vals = atr(self._highs, self._lows, self._closes, 14)
|
||
cur_ema, cur_atr = ema_vals[-1], atr_vals[-1]
|
||
prev_ema = ema_vals[-2]
|
||
if cur_ema == 0 or cur_atr == 0:
|
||
return None
|
||
|
||
# ── 出场 ──
|
||
if self._in_position:
|
||
self._highest = max(self._highest, k.high)
|
||
stop = self._highest - self.cfg.atr_stop * cur_atr
|
||
cross_down = self._closes[-2] >= prev_ema and k.close < cur_ema
|
||
|
||
if k.close < stop:
|
||
self._in_position = False
|
||
return Signal(symbol=self.cfg.symbol, side="SELL", reason="ATR止损", timestamp=k.open_time)
|
||
if cross_down:
|
||
self._in_position = False
|
||
return Signal(symbol=self.cfg.symbol, side="SELL",
|
||
reason=f"下穿EMA{self.cfg.period}", timestamp=k.open_time)
|
||
|
||
# ── 入场 ──
|
||
if not self._in_position:
|
||
cross_up = self._closes[-2] <= prev_ema and k.close > cur_ema
|
||
if cross_up:
|
||
self._in_position = True
|
||
self._highest = k.close
|
||
return Signal(symbol=self.cfg.symbol, side="BUY",
|
||
reason=f"上穿EMA{self.cfg.period}", timestamp=k.open_time)
|
||
|
||
return None
|
||
|
||
|
||
# ════════════════════════════════════════════════════════
|
||
|
||
SYMBOLS = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"]
|
||
PERIODS = [10, 20, 30]
|
||
DATE_START = datetime(2024, 1, 1)
|
||
DATE_END = datetime(2026, 1, 1)
|
||
|
||
|
||
async def main():
|
||
print()
|
||
print("═" * 105)
|
||
print(" 极简策略 — 一条 EMA + ATR 止损 | 4h | 2024-2026")
|
||
print("═" * 105)
|
||
print(f" {'EMA':<8} {'币种':<10} {'收益%':>7} {'夏普':>6} {'回撤%':>7} {'交易':>5} {'胜率%':>6} {'盈亏比':>6}")
|
||
print("─" * 105)
|
||
|
||
all_rows = []
|
||
for period in PERIODS:
|
||
for symbol in SYMBOLS:
|
||
sc = SingleEMAConfig(symbol=symbol, period=period)
|
||
bt = BacktestConfig(symbol=symbol, interval="4h",
|
||
start_time=DATE_START, end_time=DATE_END, initial_capital=10_000.0)
|
||
engine = BacktestEngine(bt, db_config=config.db)
|
||
r = await engine.run(SingleEMAStrategy, sc)
|
||
m = r.metrics
|
||
all_rows.append((period, symbol, m))
|
||
print(f" EMA({period:<2}) {symbol:<10} {m.total_return_pct:>6.1f}% {m.sharpe_ratio:>6.2f} {m.max_drawdown_pct:>6.1f}% {m.total_trades:>5} {m.win_rate*100:>5.1f}% {m.profit_factor:>6.2f}")
|
||
|
||
# ── 对比之前最优结果 ──
|
||
print("─" * 105)
|
||
print("\n ■ 对比：极简 vs 之前最优 (EMA v3 双均线)")
|
||
V3 = {
|
||
"BTCUSDT": (39.9, 1.03, -11.5),
|
||
"ETHUSDT": (53.6, 1.04, -15.3),
|
||
"BNBUSDT": (26.0, 0.64, -23.4),
|
||
"SOLUSDT": (73.6, 1.18, -25.7),
|
||
}
|
||
print(f" {'币种':<10} {'策略':<20} {'收益%':>7} {'夏普':>6} {'回撤%':>7}")
|
||
for symbol in SYMBOLS:
|
||
v3 = V3[symbol]
|
||
print(f" {symbol:<10} {'EMA v3(最优参数)':<20} {v3[0]:>6.1f}% {v3[1]:>6.2f} {v3[2]:>6.1f}%")
|
||
# 找极简最佳
|
||
best = max([(p, m) for p, s, m in all_rows if s == symbol], key=lambda x: x[1].sharpe_ratio)
|
||
print(f" {'':<10} {'单EMA('+str(best[0])+') 极简':<20} {best[1].total_return_pct:>6.1f}% {best[1].sharpe_ratio:>6.2f} {best[1].max_drawdown_pct:>6.1f}%")
|
||
print()
|
||
|
||
# ── 汇总 ──
|
||
print("─" * 105)
|
||
ranked = sorted(all_rows, key=lambda x: x[2].sharpe_ratio, reverse=True)
|
||
print(" ■ 按夏普 TOP 5:")
|
||
for i, (p, s, m) in enumerate(ranked[:5]):
|
||
print(f" {i+1}. {s} EMA({p:<2}) 夏普={m.sharpe_ratio:.2f} 收益={m.total_return_pct:+.1f}% 回撤={m.max_drawdown_pct:.1f}% 胜率={m.win_rate*100:.0f}% 交易={m.total_trades}")
|
||
|
||
avg_sh = sum(x[2].sharpe_ratio for x in all_rows) / len(all_rows)
|
||
print(f"\n 12组平均夏普: {avg_sh:.2f}")
|
||
|
||
print("\n═" * 105)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|