mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 16:54:20 +08:00
feat(strategies): 内置策略从 5 个扩充到 18 个
新增 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 策略冒烟测试通过（合成数据回测无运行时错误）。
This commit is contained in:
Justin Gu
@@ -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()
|
||||
|
||||
@@ -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 结构完整
|
||||
|
||||
Reference in New Issue
Block a user