From 2903d8c80039a2984896037cd7b6cf39861a0875 Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Fri, 3 Jul 2026 04:36:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(strategies):=20=E5=86=85=E7=BD=AE=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E4=BB=8E=205=20=E4=B8=AA=E6=89=A9=E5=85=85=E5=88=B0?= =?UTF-8?q?=2018=20=E4=B8=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 13 个经典策略，覆盖趋势/通道/震荡/均线四大类： 趋势类： - ema_cross EMA 双线交叉（比 MA 反应更灵敏） - triple_ma 三均线系统（多头/空头排列） - dmi DMI 趋向指标（+DI/-DI 交叉） - trix TRIX 三重平滑（过滤短期波动） 通道/突破类： - donchian 唐安奇通道突破（海龟交易法） - keltner 肯特纳通道（ATR-based） - atr_breakout ATR 通道突破（均线±K×ATR） 震荡/反转类： - cci CCI 超卖反弹 - wr_reversal WR 威廉超卖 - bias_reversal BIAS 乖离反弹 - emv EMV 简易波动（量价结合） - dpo DPO 区间震荡 均线类： - bbi BBI 多空指标（4 均线综合） 全部基于 MyTT 现成指标实现，每个策略声明参数 schema 供 Web 表单动态渲染。 18/18 策略冒烟测试通过（合成数据回测无运行时错误）。 --- src/easy_tdx/backtest/strategies/builtin.py | 379 +++++++++++++++++++- tests/unit/test_web_backtest.py | 4 +- 2 files changed, 380 insertions(+), 3 deletions(-) diff --git a/src/easy_tdx/backtest/strategies/builtin.py b/src/easy_tdx/backtest/strategies/builtin.py index 98cc8d8..6ab0939 100644 --- a/src/easy_tdx/backtest/strategies/builtin.py +++ b/src/easy_tdx/backtest/strategies/builtin.py @@ -14,7 +14,26 @@ from easy_tdx.backtest.strategies.registry import ( ParametrizedStrategy, register_strategy, ) -from easy_tdx.MyTT import BOLL, CROSS, KDJ, MA, MACD, RSI +from easy_tdx.MyTT import ( + ATR, + BBI, + BIAS, + BOLL, + CCI, + CROSS, + DMI, + DPO, + EMA, + EMV, + KDJ, + KTN, + MA, + MACD, + RSI, + TAQ, + TRIX, + WR, +) __all__: list[str] = [] # 注册副作用即可，无需导出符号 @@ -175,3 +194,361 @@ class KdjCrossStrategy(ParametrizedStrategy): self.buy() elif self.dead[i] and self.position["size"] > 0: self.sell() + + +# ── EMA 双线交叉 ────────────────────────────────────────────────────────────── + + +@register_strategy( + name="ema_cross", + label="EMA 双线交叉", + description="指数均线金叉买入、死叉卖出。比简单均线反应更灵敏。", +) +class EmaCrossStrategy(ParametrizedStrategy): + params = [ + Param("fast", int, default=12, min_value=2, max_value=60, label="快线周期"), + Param("slow", int, default=26, min_value=5, max_value=120, label="慢线周期"), + ] + + def init(self) -> None: + self.ema_fast = self.I(EMA, self.data.close, self.p["fast"]) + self.ema_slow = self.I(EMA, self.data.close, self.p["slow"]) + self.gold = self.I(CROSS, self.ema_fast, self.ema_slow) + self.dead = self.I(CROSS, self.ema_slow, self.ema_fast) + + def next(self) -> None: + i = self._bar_index + if self.gold[i]: + self.buy() + elif self.dead[i] and self.position["size"] > 0: + self.sell() + + +# ── 三均线系统 ──────────────────────────────────────────────────────────────── + + +@register_strategy( + name="triple_ma", + label="三均线系统", + description="短中长期均线多头排列买入、空头排列卖出。", +) +class TripleMaStrategy(ParametrizedStrategy): + params = [ + Param("short", int, default=5, min_value=1, max_value=30, label="短期"), + Param("mid", int, default=20, min_value=5, max_value=60, label="中期"), + Param("long", int, default=60, min_value=20, max_value=250, label="长期"), + ] + + def init(self) -> None: + self.ma_s = self.I(MA, self.data.close, self.p["short"]) + self.ma_m = self.I(MA, self.data.close, self.p["mid"]) + self.ma_l = self.I(MA, self.data.close, self.p["long"]) + + def next(self) -> None: + i = self._bar_index + if self.ma_s[i] > self.ma_m[i] > self.ma_l[i] and self.position["size"] == 0: + self.buy() + elif self.ma_s[i] < self.ma_m[i] < self.ma_l[i] and self.position["size"] > 0: + self.sell() + + +# ── 唐安奇通道（海龟）──────────────────────────────────────────────────────── + + +@register_strategy( + name="donchian", + label="唐安奇通道突破", + description="突破N日最高价买入，跌破N日最低价卖出。海龟交易法核心。", +) +class DonchianStrategy(ParametrizedStrategy): + params = [ + Param("n", int, default=20, min_value=5, max_value=100, label="通道周期"), + ] + + def init(self) -> None: + self.upper, self._mid, self.lower = self.I(TAQ, self.data.high, self.data.low, self.p["n"]) + + def next(self) -> None: + i = self._bar_index + close = self.data.close[0] + if close >= self.upper[i] and self.position["size"] == 0: + self.buy() + elif close <= self.lower[i] and self.position["size"] > 0: + self.sell() + + +# ── 肯特纳通道 ──────────────────────────────────────────────────────────────── + + +@register_strategy( + name="keltner", + label="肯特纳通道", + description="收盘价突破上轨买入，跌破下轨卖出。ATR-based 通道。", +) +class KeltnerStrategy(ParametrizedStrategy): + params = [ + Param("n", int, default=20, min_value=5, max_value=100, label="均线周期"), + Param("m", int, default=10, min_value=2, max_value=50, label="ATR周期"), + ] + + def init(self) -> None: + self.upper, self._mid, self.lower = self.I( + KTN, self.data.close, self.data.high, self.data.low, self.p["n"], self.p["m"] + ) + + def next(self) -> None: + i = self._bar_index + close = self.data.close[0] + if close >= self.upper[i] and self.position["size"] == 0: + self.buy() + elif close <= self.lower[i] and self.position["size"] > 0: + self.sell() + + +# ── BBI 多空指标 ────────────────────────────────────────────────────────────── + + +@register_strategy( + name="bbi", + label="BBI 多空指标", + description="收盘价上穿BBI买入，下穿BBI卖出。多空综合指标。", +) +class BbiStrategy(ParametrizedStrategy): + params = [ + Param("m1", int, default=3, min_value=1, max_value=20, label="均线1"), + Param("m2", int, default=6, min_value=2, max_value=30, label="均线2"), + Param("m3", int, default=12, min_value=5, max_value=60, label="均线3"), + Param("m4", int, default=20, min_value=10, max_value=120, label="均线4"), + ] + + def init(self) -> None: + self.bbi = self.I( + BBI, self.data.close, self.p["m1"], self.p["m2"], self.p["m3"], self.p["m4"] + ) + + def next(self) -> None: + i = self._bar_index + close = self.data.close[0] + if close > self.bbi[i] and self.position["size"] == 0: + self.buy() + elif close < self.bbi[i] and self.position["size"] > 0: + self.sell() + + +# ── CCI 顺势指标 ────────────────────────────────────────────────────────────── + + +@register_strategy( + name="cci", + label="CCI 超卖反弹", + description="CCI 跌破-100后回升买入，涨破+100卖出。", +) +class CciStrategy(ParametrizedStrategy): + params = [ + Param("n", int, default=14, min_value=2, max_value=50, label="CCI周期"), + Param("oversold", int, default=-100, min_value=-200, max_value=0, label="超卖线"), + Param("overbought", int, default=100, min_value=0, max_value=200, label="超买线"), + ] + + def init(self) -> None: + self.cci = self.I(CCI, self.data.close, self.data.high, self.data.low, self.p["n"]) + + def next(self) -> None: + i = self._bar_index + cci = self.cci[i] + if cci <= self.p["oversold"] and self.position["size"] == 0: + self.buy() + elif cci >= self.p["overbought"] and self.position["size"] > 0: + self.sell() + + +# ── WR 威廉指标 ─────────────────────────────────────────────────────────────── + + +@register_strategy( + name="wr_reversal", + label="WR 威廉超卖", + description="WR 进入超卖区（<-80）买入，进入超买区（>-20）卖出。", +) +class WrReversalStrategy(ParametrizedStrategy): + params = [ + Param("n", int, default=14, min_value=2, max_value=50, label="WR周期"), + Param("oversold", int, default=-80, min_value=-100, max_value=-40, label="超卖线"), + Param("overbought", int, default=-20, min_value=-60, max_value=0, label="超买线"), + ] + + def init(self) -> None: + self.wr, self._wr1 = self.I(WR, self.data.close, self.data.high, self.data.low, self.p["n"]) + + def next(self) -> None: + i = self._bar_index + wr = self.wr[i] + if wr <= self.p["oversold"] and self.position["size"] == 0: + self.buy() + elif wr >= self.p["overbought"] and self.position["size"] > 0: + self.sell() + + +# ── BIAS 乖离率 ─────────────────────────────────────────────────────────────── + + +@register_strategy( + name="bias_reversal", + label="BIAS 乖离反弹", + description="乖离率低于负阈值（超跌）买入，高于正阈值（超涨）卖出。", +) +class BiasReversalStrategy(ParametrizedStrategy): + params = [ + Param("n", int, default=6, min_value=2, max_value=30, label="均线周期"), + Param("threshold", float, default=5.0, min_value=1.0, max_value=20.0, label="乖离阈值%"), + ] + + def init(self) -> None: + self.bias, self._b2, self._b3 = self.I(BIAS, self.data.close, self.p["n"], 12, 24) + + def next(self) -> None: + i = self._bar_index + bias_pct = self.bias[i] * 100 + threshold = self.p["threshold"] + if bias_pct <= -threshold and self.position["size"] == 0: + self.buy() + elif bias_pct >= threshold and self.position["size"] > 0: + self.sell() + + +# ── DMI 趋向指标 ────────────────────────────────────────────────────────────── + + +@register_strategy( + name="dmi", + label="DMI 趋向指标", + description="+DI 上穿-DI 买入（多头趋强），+DI 下穿-DI 卖出。", +) +class DmiStrategy(ParametrizedStrategy): + params = [ + Param("m1", int, default=14, min_value=2, max_value=30, label="DI周期"), + Param("m2", int, default=6, min_value=2, max_value=20, label="ADX周期"), + ] + + def init(self) -> None: + self.pdi, self.mdi, self._adx, self._adxr = self.I( + DMI, self.data.close, self.data.high, self.data.low, self.p["m1"], self.p["m2"] + ) + self.gold = self.I(CROSS, self.pdi, self.mdi) + self.dead = self.I(CROSS, self.mdi, self.pdi) + + def next(self) -> None: + i = self._bar_index + if self.gold[i]: + self.buy() + elif self.dead[i] and self.position["size"] > 0: + self.sell() + + +# ── TRIX 三重平滑 ───────────────────────────────────────────────────────────── + + +@register_strategy( + name="trix", + label="TRIX 三重平滑", + description="TRIX 上穿信号线买入，下穿卖出。过滤短期波动的趋势指标。", +) +class TrixStrategy(ParametrizedStrategy): + params = [ + Param("m1", int, default=12, min_value=2, max_value=30, label="TRIX周期"), + Param("m2", int, default=20, min_value=5, max_value=60, label="信号周期"), + ] + + def init(self) -> None: + self.trix, self.trma = self.I(TRIX, self.data.close, self.p["m1"], self.p["m2"]) + self.gold = self.I(CROSS, self.trix, self.trma) + self.dead = self.I(CROSS, self.trma, self.trix) + + def next(self) -> None: + i = self._bar_index + if self.gold[i]: + self.buy() + elif self.dead[i] and self.position["size"] > 0: + self.sell() + + +# ── EMV 简易波动 ────────────────────────────────────────────────────────────── + + +@register_strategy( + name="emv", + label="EMV 简易波动", + description="EMV 上穿0轴买入，下穿0轴卖出。量价结合指标。", +) +class EmvStrategy(ParametrizedStrategy): + params = [ + Param("n", int, default=14, min_value=2, max_value=30, label="EMV周期"), + ] + + def init(self) -> None: + self.emv, self._maemv = self.I( + EMV, self.data.high, self.data.low, self.data.vol, self.p["n"] + ) + + def next(self) -> None: + i = self._bar_index + if self.emv[i] > 0 and self.position["size"] == 0: + self.buy() + elif self.emv[i] < 0 and self.position["size"] > 0: + self.sell() + + +# ── DPO 区间震荡 ────────────────────────────────────────────────────────────── + + +@register_strategy( + name="dpo", + label="DPO 区间震荡", + description="DPO 上穿信号线买入，下穿卖出。去除趋势的震荡指标。", +) +class DpoStrategy(ParametrizedStrategy): + params = [ + Param("m1", int, default=20, min_value=5, max_value=60, label="DPO周期"), + ] + + def init(self) -> None: + self.dpo, self.madpo = self.I(DPO, self.data.close, self.p["m1"]) + self.gold = self.I(CROSS, self.dpo, self.madpo) + self.dead = self.I(CROSS, self.madpo, self.dpo) + + def next(self) -> None: + i = self._bar_index + if self.gold[i]: + self.buy() + elif self.dead[i] and self.position["size"] > 0: + self.sell() + + +# ── ATR 通道突破 ────────────────────────────────────────────────────────────── + + +@register_strategy( + name="atr_breakout", + label="ATR 通道突破", + description="收盘价突破 均线+K×ATR 买入，跌破 均线-K×ATR 卖出。", +) +class AtrBreakoutStrategy(ParametrizedStrategy): + params = [ + Param("n_ma", int, default=20, min_value=5, max_value=100, label="均线周期"), + Param("n_atr", int, default=20, min_value=5, max_value=50, label="ATR周期"), + Param("k", float, default=2.0, min_value=0.5, max_value=5.0, label="ATR倍数"), + ] + + def init(self) -> None: + self.ma = self.I(MA, self.data.close, self.p["n_ma"]) + self.atr = self.I(ATR, self.data.close, self.data.high, self.data.low, self.p["n_atr"]) + + def next(self) -> None: + i = self._bar_index + close = self.data.close[0] + upper = self.ma[i] + self.p["k"] * self.atr[i] + lower = self.ma[i] - self.p["k"] * self.atr[i] + if close >= upper and self.position["size"] == 0: + self.buy() + elif close <= lower and self.position["size"] > 0: + self.sell() diff --git a/tests/unit/test_web_backtest.py b/tests/unit/test_web_backtest.py index f96c4a1..8d69c0b 100644 --- a/tests/unit/test_web_backtest.py +++ b/tests/unit/test_web_backtest.py @@ -59,7 +59,7 @@ def test_registry_has_builtin_strategies(): assert "boll_breakout" in names assert "rsi_reversal" in names assert "kdj_cross" in names - assert len(names) >= 5 + assert len(names) >= 18 def test_strategy_schema_serialization(): @@ -522,7 +522,7 @@ def test_list_strategies_endpoint(client): resp = client.get("/api/v1/backtest/strategies") assert resp.status_code == 200 body = resp.json() - assert body["count"] >= 5 + assert body["count"] >= 18 names = [s["name"] for s in body["strategies"]] assert "ma_cross" in names # 每个策略的 schema 结构完整