diff --git a/research/HANDOFF.md b/research/HANDOFF.md index f70643c..c6e04de 100644 --- a/research/HANDOFF.md +++ b/research/HANDOFF.md @@ -1128,6 +1128,38 @@ B4 跑出 PF 1.83~2.06 / t +8.5~+10.7。所以 fast B1 的负数不是管线接 **它与 B4 不是一回事**:B4 是突破延续(§3.391 实测 mom10 中位 +3.47 ATR, 根本不是回调入场),这个是回调延续。两种入场原型不同,**可能互补**。 +**止损位置:真问题,但不是主因(step58)** + +用户提出:B1 做多的止损可能挂在了 B2 低点上方,于是在那次「回踩不破前低」的 +正常回抽处被打掉,而不是等趋势真正走完。查下来**观察成立且比预估更严重**: + +| 量 | 5m 一类 | +|---|---| +| 入场价到结构极值(ATR,用入场根 ATR 归一) | 中位 **2.90** | +| 2ATR 止损在结构位外侧留的余地 | 中位 **−0.90**(在结构**以内**) | +| **止损落在结构位以内的占比** | **88.6%** | + +即近九成的一类做多,价格连回踩到前低都不需要就已出局。 + +修掉之后确实有改善,但救不活。⚠️ 做这个对照时目标必须**随止损等比放大** +(恒定 1.5R 减半 / 4R 收尾);放宽止损却不放大目标会把盈亏比压到 1 以下, +测出来的「放宽无效」是自证的: + +| 方向 | 止损 | 中位宽度 | 胜率 | PF | t值 | +|---|---|---|---|---|---| +| 原方向做多 | 固定 2ATR | 2.00 | 16.7% | 0.22 | −31.0 | +| 原方向做多 | 结构+1.5 | 4.40 | 33.1% | **0.36** | −16.6 | +| 反手做空 | 固定 2ATR | 2.00 | 65.7% | **2.53** | +14.4 | +| 反手做空 | 结构+1.5 | 4.40 | 58.2% | 2.15 | +10.7 | + +**结论:趋势确实越过第二类买卖点继续走原方向。** 最有说服力的读法是 +「结构+1.5」那一档——止损宽达 4.4 ATR,仍有 **67% 被打掉**,即三分之二的一类 +信号会在 48 根内跌破结构低点再多走 1.5 ATR。缠论里「B2 回踩不破前低」这条 +在本市场多数时候不成立。 + +佐证:反手做空在**最窄的 2ATR 止损上最好**(2.53),放宽反而降到 2.15。 +这与顺势回调入场自洽——价格若重新涨过反弹高点,论点当场失效,不该留余地。 + **仍未验证,落地前必须做完(否则不许上实盘)** 1. **因果性**。`zs.is_sure` / `bi.sure_time` 在全量数据上算出,§5.41 已知中枢 diff --git a/research/step55_bsp12_probe.py b/research/step55_bsp12_probe.py index 09f728e..e6a2589 100644 --- a/research/step55_bsp12_probe.py +++ b/research/step55_bsp12_probe.py @@ -126,6 +126,10 @@ def collect(sym: str, rows: int, tf: str = "1m") -> pd.DataFrame | None: "i_ext": i_ext, "i_sure": i_sure, "lag_bars": i_sure - i_ext, "atr_bp": a / px_ideal * 1e4, + # 结构位:笔末 KLC 的实际极值,不是收盘。止损要挂在它外面 + # 才谈得上「守住结构」,用收盘价会把插针那一段漏掉 + "ext_px": float(b.klc.low if d == 1 else b.klc.high), + "entry_px": px_real, # 等待期间价格顺着信号方向跑掉了多少(正 = 你追高/追空,吃亏) "drift_atr": (px_real - px_ideal) * d / a, "drift_bp": (px_real - px_ideal) * d / px_ideal * 1e4, @@ -161,6 +165,8 @@ def collect(sym: str, rows: int, tf: str = "1m") -> pd.DataFrame | None: for k in ("g", "r", "c", "b"): r["f_" + k] = rf[f"{cfg}_{k}"].to_numpy() r["atr_pct"] = atr[r.i_sure.values] / cl[r.i_sure.values] + # 原始 ATR,供结构止损换算用。atr_pct 除过价格,换不回来 + r["atr_at_entry"] = atr[r.i_sure.values] return r except Exception as e: print(f" {sym} 失败: {e!r}", flush=True) diff --git a/research/step58_struct_stop.py b/research/step58_struct_stop.py new file mode 100644 index 0000000..fe4f01b --- /dev/null +++ b/research/step58_struct_stop.py @@ -0,0 +1,190 @@ +"""结构止损:一类做多是被「止损太紧」打死的,还是趋势真的继续? + +用户的观察:B1 做多的入场价 P 在结构低点 L 上方约 1.68 ATR,而固定止损是 +P − 2 ATR —— **只在 L 下方 0.32 ATR**。缠论里 B2 正是「回踩不破 L」的那次 +机会,属于预期之内的正常回抽。止损只留 0.32 ATR,一根插针就打掉,而那恰恰 +是该加仓的位置。 + +这与「趋势继续向下」是两个不同的失败模式,且**预测相反**: + + 止损太紧 把止损放到 L 下方足够远 -> 多头应被救活 + 趋势继续 放宽止损只是亏得更多,且反手做空应持续为正 + +本脚本用逐笔的结构止损(挂在 L 外侧 margin 个 ATR)重跑,直接区分这两者。 +`walk_exits` 只支持全局固定止损,故这里自带模拟器;出场结构与实盘一致: +减半于 SCALE_AT、剩余半仓目标 RUNNER、剩余半仓止损回到初始止损、MAXB 超时。 +""" +from __future__ import annotations + +import argparse +import sys +import warnings +from pathlib import Path + +import numpy as np +import pandas as pd + +warnings.filterwarnings("ignore") +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) +sys.path.insert(0, str(HERE.parent)) + +from lib.data import fetch_ohlcv # noqa: E402 +from lib.exit_model import fee_of, taker_notional # noqa: E402 + +SRC = HERE / "out" / "step55_bsp12.feather" +SCALE_AT, RUNNER, MAXB = 3.0, 8.0, 48 + + +def simulate(cdf: pd.DataFrame, trades: pd.DataFrame, + stop_atr: np.ndarray, r_scale: bool = True) -> pd.DataFrame: + """逐笔止损宽度的出场模拟。stop_atr 是每笔各自的初始止损(ATR 倍数)。 + + 与 `lib.exit_model.walk_exits` 的结构对齐:先减半仓于 SCALE_AT,剩余半仓 + 看 RUNNER 或回到初始止损;未减仓则整仓在止损/超时了结。同根内止损优先于 + 获利目标 —— 分辨不了根内先后时,按不利的一侧算,不给回测送分。 + + `r_scale=True` 时目标按 **R 倍数**而非 ATR 固定值放置:实盘的 2/3/8 等于 + 在 1.5R 减半、4R 收尾,放宽止损却不放大目标会把盈亏比压到 1 以下, + 那样测出来的「放宽无效」是自证的。默认按 R 等比放大才是公平对照。 + """ + k_scale = SCALE_AT / 2.0 if r_scale else None + k_run = RUNNER / 2.0 if r_scale else None + high = cdf["high"].to_numpy(float) + low = cdf["low"].to_numpy(float) + open_ = cdf["open"].to_numpy(float) + close = cdf["close"].to_numpy(float) + n = len(cdf) + out = [] + ent = trades.entry_idx.astype(int).to_numpy() + dirs = trades.direction.astype(int).to_numpy() + atrs = trades.atr_at_entry.to_numpy(float) + for k in range(len(trades)): + s, d, a, sl = ent[k], dirs[k], atrs[k], stop_atr[k] + e = s + 1 + if e >= n - 1 or not np.isfinite(sl) or sl <= 0 or not np.isfinite(a): + out.append((np.nan, "skip", 0, 0)) + continue + tgt = sl * k_scale if r_scale else SCALE_AT + run = sl * k_run if r_scale else RUNNER + entry = open_[e] + end = min(e + MAXB, n - 1) + half_done = False + g, why, bars = None, "timeout", end - e + for j in range(e, end + 1): + adv = (high[j] - entry) / a if d == 1 else (entry - low[j]) / a + ret = (entry - low[j]) / a if d == 1 else (high[j] - entry) / a + if ret >= sl: # 同根内止损优先 + g = ((-sl * a) / entry) * (0.5 if half_done else 1.0) + if half_done: + g += (tgt * a / entry) * 0.5 + why, bars = "stop", j - e + break + if not half_done and adv >= tgt: + half_done = True + if half_done and adv >= run: + g = (tgt * 0.5 + run * 0.5) * a / entry + why, bars = "target", j - e + break + if g is None: + px = close[end] + r = (px - entry) * d / entry + g = (tgt * a / entry) * 0.5 + r * 0.5 if half_done else r + why, bars = ("timeout_half" if half_done else "timeout"), end - e + out.append((g, why, int(half_done), bars)) + return pd.DataFrame(out, columns=["g", "r", "c", "b"]) + + +def stats(g: pd.Series, r: pd.Series, c: pd.Series, atr_pct: np.ndarray, + sl_used: np.ndarray) -> dict: + net = g.values - fee_of(r.values, c.values) + # R 用每笔自己的止损宽度归一,否则宽止损会被系统性低估风险 + denom = sl_used * atr_pct + R = net / denom + tn = taker_notional(r.values, c.values) + w, o = net[net > 0].sum(), -net[net <= 0].sum() + return { + "笔数": len(g), "胜率": f"{(net > 0).mean()*100:.1f}%", + "净均R": round(np.nanmean(R), 3), + "PF": round(w / o, 2) if o > 0 else np.inf, + "余量bp": round(net.mean() / tn.mean() * 1e4, 2), + "t值": round(np.nanmean(R) / (np.nanstd(R, ddof=1) + / np.sqrt(len(R))), 2), + } + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--tf", default="5m") + ap.add_argument("--rows", type=int, default=200_000) + args = ap.parse_args() + + d = pd.read_feather(SRC) + if "ext_px" not in d.columns: + print("step55 数据缺 ext_px/entry_px,先重跑 step55") + return + d = d[(d.tf == args.tf) & d.type.isin(["B1", "S1"])].copy() + + print(f"[结构止损] {args.tf} · 一类 {len(d)} 笔\n") + print("=" * 92) + print("########## 一、几何:固定 2ATR 止损离结构位有多远 ##########") + # 入场价到结构极值的距离,用入场根的 ATR 归一(与 walk_exits 同口径) + d["to_ext"] = (d.entry_px - d.ext_px) * d.dir / d.atr_at_entry + d["margin_2atr"] = 2.0 - d.to_ext + print(f"入场价到结构极值 (ATR) 中位 {d.to_ext.median():.2f} " + f"P25 {d.to_ext.quantile(.25):.2f} P75 {d.to_ext.quantile(.75):.2f}") + print(f"2ATR 止损在结构位外侧留的余地 (ATR) 中位 " + f"{d.margin_2atr.median():.2f}") + inside = (d.margin_2atr <= 0).mean() + print(f"**止损落在结构位以内(还没到极值就被打掉)的占比:{inside*100:.1f}%**") + print(" —— 这部分交易,价格连回踩到前低都不用,就已经出局了。") + + print("\n" + "=" * 92) + print("########## 二、把止损挂到结构位外侧,多头能不能救活 ##########") + print("margin = 止损挂在结构极值外侧几个 ATR。对照组是现行的固定 2 ATR。") + from chanlun import TF_DF + rows = [] + cache = {} + for sym in sorted(d.sym.unique()): + df = fetch_ohlcv(f"{sym}/USDT:USDT", args.tf, args.rows) + cache[(sym, args.tf)] = TF_DF(df, 1, args.tf).dataframe + + print("目标随止损等比放大,恒定 1.5R 减半 / 4R 收尾,与实盘的 2/3/8 同构。") + for flip in (False, True): + for label, margin in [("固定2ATR", None), ("结构+0.25", 0.25), + ("结构+0.5", 0.5), ("结构+1.0", 1.0), + ("结构+1.5", 1.5)]: + gs, rs, cs, ap_, sl_ = [], [], [], [], [] + for sym, x in d.groupby("sym"): + cdf = cache[(sym, args.tf)] + x = x.reset_index(drop=True) + t = pd.DataFrame({ + "entry_idx": x.i_sure.values, + "direction": (-x.dir.values if flip else x.dir.values), + "atr_at_entry": x.atr_at_entry.values}) + # 反手时结构位在**盈利**方向,不能拿它当止损;止损改挂在 + # 入场价另一侧同样宽度处,否则两组比的不是同一个东西 + sl = (np.full(len(x), 2.0) if margin is None + else (x.to_ext.values + margin)) + res = simulate(cdf, t, sl) + ok = res.g.notna().values + gs.append(res.g[ok]) + rs.append(res.r[ok]) + cs.append(res.c[ok]) + ap_.append((x.atr_at_entry.values / x.entry_px.values)[ok]) + sl_.append(sl[ok]) + rows.append({"方向": "反手" if flip else "原方向", "止损": label, + "中位宽度ATR": round(float(np.median( + np.concatenate(sl_))), 2), + **stats(pd.concat(gs, ignore_index=True), + pd.concat(rs, ignore_index=True), + pd.concat(cs, ignore_index=True), + np.concatenate(ap_), np.concatenate(sl_))}) + print(pd.DataFrame(rows).to_string(index=False)) + print("\n判读:若「止损太紧」是主因,放宽到结构位外侧应让原方向 PF 越过 1。" + "\n若原方向放宽后仍在 1 以下、而反手始终显著为正," + "说明趋势确实在继续,\n那么该做的是反手,加宽止损只是少亏一点。") + + +if __name__ == "__main__": + main()