mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 20:24:19 +08:00
release: v1.30.1 — 移除 ZIG 未来函数指标与 zig_breakout 策略：回测数字可信优先
This commit is contained in:
Justin Gu
@@ -571,117 +571,4 @@ def FSL(CLOSE, VOL, CAPITAL): # 分水岭指标：多空趋势强弱分界（SW
|
||||
return RD(SWL), RD(SWS)
|
||||
|
||||
|
||||
def ZIG(S, X=35): # 之字转向指标（未来函数）：S为价格序列，X为转向阈值百分比（如10表示10%）
|
||||
"""之字转向指标 (ZigZag) — 经典未来函数。
|
||||
|
||||
当价格从前一个极值点反向变动超过 X% 时确立波峰/波谷拐点并转向，
|
||||
拐点之间线性插值，返回与 S 等长的拟合序列。
|
||||
|
||||
注意：拐点只有在**其后**的走势确认了转向才会回溯标出，序列中波峰/
|
||||
波谷位置含有未来信息。把 ZIG 拐点直接当买卖信号回测会严重高估收益
|
||||
（前视偏差）；如需使用，必须配合右侧确认或止损保护（参见内置策略
|
||||
``zig_breakout`` 的做法）。
|
||||
|
||||
Args:
|
||||
S: 价格序列（通常为 CLOSE）
|
||||
X: 转向阈值百分比。10 表示 10%；也可传小数形式 0.1（以 1.0 为界
|
||||
自动区分，故阈值本身小于 1% 时请用小数形式）
|
||||
|
||||
Returns:
|
||||
np.ndarray: 与 S 等长的 ZIG 之字转向插值序列
|
||||
"""
|
||||
S = np.asarray(S, dtype=float)
|
||||
n = len(S)
|
||||
if n == 0:
|
||||
return np.array([], dtype=float)
|
||||
if n == 1:
|
||||
return S.copy()
|
||||
|
||||
x = float(X) / 100.0 if float(X) > 1.0 else float(X)
|
||||
if x <= 0:
|
||||
return S.copy()
|
||||
|
||||
ZIG_STATE_START = 0
|
||||
ZIG_STATE_RISE = 1
|
||||
ZIG_STATE_FALL = 2
|
||||
|
||||
peer_i = 0
|
||||
candidate_i = None
|
||||
peers = [0]
|
||||
state = ZIG_STATE_START
|
||||
|
||||
for scan_i in range(1, n):
|
||||
if scan_i == n - 1:
|
||||
# 扫描到序列尾部：未确立的候选极值按当前方向收尾
|
||||
if candidate_i is None:
|
||||
peers.append(scan_i)
|
||||
else:
|
||||
if state == ZIG_STATE_RISE:
|
||||
if S[scan_i] >= S[candidate_i]:
|
||||
peers.append(scan_i)
|
||||
else:
|
||||
peers.append(candidate_i)
|
||||
if candidate_i != scan_i:
|
||||
peers.append(scan_i)
|
||||
elif state == ZIG_STATE_FALL:
|
||||
if S[scan_i] <= S[candidate_i]:
|
||||
peers.append(scan_i)
|
||||
else:
|
||||
peers.append(candidate_i)
|
||||
if candidate_i != scan_i:
|
||||
peers.append(scan_i)
|
||||
else:
|
||||
peers.append(scan_i)
|
||||
break
|
||||
|
||||
if state == ZIG_STATE_START:
|
||||
if S[peer_i] != 0:
|
||||
if S[scan_i] >= S[peer_i] * (1.0 + x):
|
||||
candidate_i = scan_i
|
||||
state = ZIG_STATE_RISE
|
||||
elif S[scan_i] <= S[peer_i] * (1.0 - x):
|
||||
candidate_i = scan_i
|
||||
state = ZIG_STATE_FALL
|
||||
elif state == ZIG_STATE_RISE:
|
||||
if S[scan_i] >= S[candidate_i]:
|
||||
candidate_i = scan_i
|
||||
elif S[candidate_i] != 0 and S[scan_i] <= S[candidate_i] * (1.0 - x):
|
||||
peer_i = candidate_i
|
||||
peers.append(peer_i)
|
||||
state = ZIG_STATE_FALL
|
||||
candidate_i = scan_i
|
||||
elif state == ZIG_STATE_FALL:
|
||||
if S[scan_i] <= S[candidate_i]:
|
||||
candidate_i = scan_i
|
||||
elif S[candidate_i] != 0 and S[scan_i] >= S[candidate_i] * (1.0 + x):
|
||||
peer_i = candidate_i
|
||||
peers.append(peer_i)
|
||||
state = ZIG_STATE_RISE
|
||||
candidate_i = scan_i
|
||||
|
||||
# 去除重复拐点并确保末端对齐
|
||||
clean_peers = []
|
||||
for p in peers:
|
||||
if not clean_peers or p != clean_peers[-1]:
|
||||
clean_peers.append(p)
|
||||
if clean_peers[-1] != n - 1:
|
||||
clean_peers.append(n - 1)
|
||||
|
||||
# 拐点间线性插值
|
||||
z = np.zeros(n, dtype=float)
|
||||
for i in range(len(clean_peers) - 1):
|
||||
p_start = clean_peers[i]
|
||||
p_end = clean_peers[i + 1]
|
||||
v_start = S[p_start]
|
||||
v_end = S[p_end]
|
||||
if p_end == p_start:
|
||||
z[p_start] = v_start
|
||||
else:
|
||||
slope = (v_end - v_start) / (p_end - p_start)
|
||||
for j in range(p_end - p_start + 1):
|
||||
z[p_start + j] = v_start + slope * j
|
||||
|
||||
return RD(z)
|
||||
|
||||
|
||||
# 望大家能提交更多指标和函数 https://github.com/mpquant/MyTT
|
||||
|
||||
@@ -30,7 +30,6 @@ from easy_tdx.MyTT import (
|
||||
EMA,
|
||||
EMV,
|
||||
FSL,
|
||||
HHV,
|
||||
KDJ,
|
||||
KTN,
|
||||
MA,
|
||||
@@ -39,7 +38,6 @@ from easy_tdx.MyTT import (
|
||||
TAQ,
|
||||
TRIX,
|
||||
WR,
|
||||
ZIG,
|
||||
)
|
||||
|
||||
__all__: list[str] = [] # 注册副作用即可，无需导出符号
|
||||
@@ -693,113 +691,3 @@ class FslStrategy(ParametrizedStrategy):
|
||||
def entry_exit_masks(self) -> tuple[Any, Any]:
|
||||
"""与 next() 同源（gold/dead 即 next() 判定用的同一组掩码数组）。"""
|
||||
return self.gold, self.dead
|
||||
|
||||
|
||||
# ── ZIG 右侧突破回补 ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@register_strategy(
|
||||
name="zig_breakout",
|
||||
label="ZIG 右侧突破回补",
|
||||
description=(
|
||||
"ZIG 向上启动（波谷确认）全仓买入；ZIG 见顶回落清仓并记录 N 日最高点，"
|
||||
"其后收盘突破前高×(1+确认比例) 时右侧回补。两路径买入均带硬止损，"
|
||||
"对冲 ZIG 波谷确认的前视偏差（未来函数，实盘信号会滞后）。"
|
||||
),
|
||||
)
|
||||
class ZigBreakoutStrategy(ParametrizedStrategy):
|
||||
"""ZIG 右侧突破回补（Re-entry on Breakout + 硬止损保护）。
|
||||
|
||||
ZIG 是未来函数：波峰/波谷只有在其后走势确认转向才回溯标出，回测里
|
||||
"波谷启动"信号天然偷看未来。本策略用两层保护缓解而非消除该偏差：
|
||||
|
||||
1. 买入即挂 ``stop_loss_pct`` 硬止损（引擎逐 bar 监控，跌破自动平仓），
|
||||
假波谷不至于深套；
|
||||
2. 卖出后不追 ZIG 新波谷，而是等价格**右侧突破**前高确认_pct 再回补，
|
||||
把"猜底"换成"确认后进场"。
|
||||
|
||||
交易逻辑::
|
||||
|
||||
空仓 + ZIG 上行 → 全仓买入（带止损）
|
||||
持仓 + ZIG 下行（见顶） → 全仓卖出，记录 HHV(high, N) 为前高
|
||||
空仓 + 收盘 ≥ 前高×(1+确认) → 右侧回补（带止损）
|
||||
|
||||
注意：``_breakout_level`` 随持仓路径变化，信号不可向量化，故不实现
|
||||
``entry_exit_masks``——引擎自动走逐 bar 回放路径（与 next() 完全一致）。
|
||||
"""
|
||||
|
||||
params = [
|
||||
Param(
|
||||
"zig_delta",
|
||||
float,
|
||||
default=10.0,
|
||||
min_value=0.5,
|
||||
max_value=50.0,
|
||||
label="ZIG转向阈值%",
|
||||
),
|
||||
Param(
|
||||
"confirm_pct",
|
||||
float,
|
||||
default=2.0,
|
||||
min_value=0.1,
|
||||
max_value=20.0,
|
||||
label="突破确认比例%",
|
||||
),
|
||||
Param("hhv_period", int, default=20, min_value=5, max_value=120, label="前高周期"),
|
||||
Param(
|
||||
"stop_loss_pct",
|
||||
float,
|
||||
default=3.0,
|
||||
min_value=0.0,
|
||||
max_value=30.0,
|
||||
label="硬止损%",
|
||||
),
|
||||
]
|
||||
|
||||
def init(self) -> None:
|
||||
self.zig = self.I(ZIG, self.data.close, self.p["zig_delta"])
|
||||
self.hhv = self.I(HHV, self.data.high, self.p["hhv_period"])
|
||||
# 见顶清仓时记录的前高（0 = 未记录，等待首次建仓-见顶周期）
|
||||
self._breakout_level: float = 0.0
|
||||
|
||||
def next(self) -> None:
|
||||
i = self._bar_index
|
||||
if i == 0:
|
||||
return
|
||||
|
||||
cur_close = float(self.data.close[0])
|
||||
cur_zig = float(self.zig[i])
|
||||
prev_zig = float(self.zig[i - 1])
|
||||
cur_pos = self.position["size"]
|
||||
|
||||
# 持仓：ZIG 见顶回落 → 清仓，并记录突破位（HHV 含未来 bar 已确认的高点）
|
||||
if cur_pos > 0 and cur_zig < prev_zig:
|
||||
self._breakout_level = float(self.hhv[i])
|
||||
self.sell(size=0)
|
||||
return
|
||||
|
||||
if cur_pos == 0:
|
||||
# 路径 1：ZIG 向上启动（波谷确认）→ 初始建仓
|
||||
if cur_zig > prev_zig:
|
||||
self._breakout_level = 0.0
|
||||
self._buy_with_stop()
|
||||
return
|
||||
|
||||
# 路径 2：右侧突破前高 → 回补（洗盘结束、主升确立）
|
||||
if self._breakout_level > 0:
|
||||
threshold = self._breakout_level * (1.0 + self.p["confirm_pct"] / 100.0)
|
||||
if cur_close >= threshold:
|
||||
self._breakout_level = 0.0
|
||||
self._buy_with_stop()
|
||||
|
||||
def _buy_with_stop(self) -> None:
|
||||
"""市价全仓买入并按 ``stop_loss_pct`` 挂硬止损（0 = 不挂）。
|
||||
|
||||
市价单（price=None）由引擎在下一根开盘成交，与本地其他内置策略
|
||||
口径一致，避免信号 bar 收盘价成交的前视味道。
|
||||
"""
|
||||
pct = self.p["stop_loss_pct"] / 100.0
|
||||
if pct > 0:
|
||||
self.buy(size=0, stop_loss_pct=pct)
|
||||
else:
|
||||
self.buy(size=0)
|
||||
|
||||
@@ -98,12 +98,6 @@ STRATEGY_PRESETS: dict[str, dict[str, list[Any]]] = {
|
||||
# capital 仅作粗档扫描（1千万/1亿/10亿股），覆盖小盘→大盘
|
||||
"capital": [1e7, 1e8, 1e9, 1e10],
|
||||
}, # 4
|
||||
# ── 之字转向类 ───────────────────────────────────────────────────────────
|
||||
"zig_breakout": {
|
||||
# 转向阈值×确认比例 = 4×3；hhv/止损用默认（HHV20 / 3%）
|
||||
"zig_delta": [5.0, 8.0, 10.0, 15.0],
|
||||
"confirm_pct": [1.0, 2.0, 3.0],
|
||||
}, # 12
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user