diff --git a/research/HANDOFF.md b/research/HANDOFF.md index 928f51c..f70643c 100644 --- a/research/HANDOFF.md +++ b/research/HANDOFF.md @@ -1086,6 +1086,59 @@ B4 跑出 PF 1.83~2.06 / t +8.5~+10.7。所以 fast B1 的负数不是管线接 头早就写了「B4 不只是 B3 提前几根，两者统计性质符号相反」，一类的实验说明 反过来也成立：**把反转信号提前，它还是反转信号。** +### 3.395 ⭐ 一类反手：识别没错，但发出的时点让它变成了反向信号（step55/57） + +用户问「是不是一类二类的识别错了」。查下来结论是：**代码没写错，但信号该反着用。** + +**先排除掉一个我自己的错误猜测。** `check_bi_div` 只比 MACD 面积、不检查创新极值， +这是缠论背驰定义（创新极值 **且** 力度减弱）丢了一半。看着像致命缺陷，但实测 +**93.2% 的 B1 / 93.4% 的 S1 离开笔本来就创了新极值**，缺口只影响约 7%， +解释不了 PF 0.2。**先量再断，别拿理论缺陷当结论。** + +**真正的线索是胜率的方向**：B1 胜率 18.6%，而 SL2/TP3 下随机入场约 40%。 +**远低于随机说明样本带信息，只是指反了**——真正没用的信号会落在 40% 附近。 +于是拿相反方向重跑出场（路径依赖，不能把 g 取负号了事）： + +| 反手 | 5m 胜率 | 5m PF | 5m t值 | 15m PF | 15m t值 | +|---|---|---|---|---|---| +| **B1（反手做空）** | 64.4% | **2.35** | +10.4 | **3.03** | +12.6 | +| **S1（反手做多）** | 67.1% | **2.75** | +12.8 | 2.63 | +9.6 | +| B3/S3（反手） | 45~49% | 1.14~1.24 | +4 | 1.15~1.20 | +4 | + +**稳健性（step57）——这是本项目做过的最稳的结果** + +| 切法 | 结论 | +|---|---| +| 逐币 | 5 个币全正。15m PF 2.25~4.12，5m 2.31~2.70 | +| 前后半段 | 15m 2.81 / 2.81，5m 2.55 / 2.51，几乎不变 | +| 逐年 | 2021~2026 **连续六年全正**，15m PF 2.57~2.99 | +| 多空 | 两边都正，排除单边行情 | +| 余量 | 15m 33~76bp、5m 18~28bp。对比 B4 只有约 3bp | + +**为什么不是 bug 而要反着用** + +代码是照缠论写的：向下离开中枢 + 背驰 → B1 买。**教科书没错，错的是它发信号的 +时刻。** 信号在 `leave_bi.sure_time` 发出，中位滞后 8~9 根，此时价格已从低点 +**反弹了 1.68 ATR**（§3.393 表）。引擎想说「抄底」，但它给出的时间戳对应的是 +「反弹已走完」。在下跌趋势里于此处做空 —— 那是标准的**顺势回调入场**。 + +所以这一族信号的几何识别是对的，方向标注按理论也是对的，**是滞后把它变成了 +相反的交易**。它是一个穿着反转外衣的顺势回调信号。 + +**它与 B4 不是一回事**：B4 是突破延续（§3.391 实测 mom10 中位 +3.47 ATR， +根本不是回调入场），这个是回调延续。两种入场原型不同，**可能互补**。 + +**仍未验证，落地前必须做完（否则不许上实盘）** + +1. **因果性**。`zs.is_sure` / `bi.sure_time` 在全量数据上算出，§5.41 已知中枢 + 右边缘会重画。必须用增量/流式回放确认信号在当时真的发得出来。 + *反证据（不能替代验证）*：同一套机器算出的 B3 是 PF 0.64 的烂数， + 若存在通用的未来函数，B3 不该这么难看。 +2. **消融：是不是「反弹就做空」都赚**。照 §3.4 对 B4 做过的那样，把背驰条件 + 拿掉、把中枢换成伪阻力位，看 PF 掉不掉。掉不下来就说明 alpha 在几何和 + 时点上，与背驰无关，那这个信号就该重新命名和重新设计。 +3. **与 B4 的重叠度与相关性**。若只是换名字的 B4，就没有增量价值。 + ### 3.4 alpha 的来源（step32 消融） 逐条拆掉 `fast_bsp3` 的条件后发现：**alpha 完全来自缠论中枢的上下文定位， diff --git a/research/step55_bsp12_probe.py b/research/step55_bsp12_probe.py index feef910..09f728e 100644 --- a/research/step55_bsp12_probe.py +++ b/research/step55_bsp12_probe.py @@ -151,6 +151,15 @@ def collect(sym: str, rows: int, tf: str = "1m") -> pd.DataFrame | None: return None for k in ("g", "r", "c", "b"): r[k] = res[f"{cfg}_{k}"].to_numpy() + # 反向对照：胜率远低于随机说明信号带信息但可能指反了。出场是路径依赖的， + # 不能把 g 取负号了事，必须拿相反方向重跑一遍 + flip = pd.DataFrame({"entry_idx": r.i_sure.values, + "direction": -r.dir.values}) + rf = walk_exits(cdf, flip, [SL], [RUNNER], [MAXB], scale_at=SCALE_AT, + runners=(RUNNER,), runner_stops=(RSTOP,)) + if len(rf) == len(r): + 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] return r except Exception as e: @@ -258,6 +267,31 @@ def profit(d: pd.DataFrame) -> None: print("\n⚠️ B3 是口径校验锚：HANDOFF §4 记的是 PF 0.66 / 胜率 27.4%。" "这里若明显更好，先怀疑接错了再信 B1/B2。") + if "f_g" not in x.columns: + return + print("\n" + "=" * 100) + print("########## 六、反向对照：信号是「指反了」还是「没有边」 ##########") + print("胜率 18.6% 远低于 SL2/TP3 的随机基准约 40%，说明样本带信息。若只是" + "方向标反了，\n反手做应显著为正；若反手也不赚，那就是入场时点本身" + "已经过期，两边都吃滑点。") + y = x.copy() + y["fnet"] = y.f_g.values - fee_of(y.f_r.values, y.f_c.values) + y["fgR"] = y.f_g.values / (SL * y.atr_pct.values) + rows = [] + for name, g in y.groupby("type"): + if len(g) < 40: + continue + w, o = g.fnet[g.fnet > 0].sum(), -g.fnet[g.fnet <= 0].sum() + rows.append({ + "类型(反手)": name, "笔数": len(g), + "胜率": f"{(g.fnet > 0).mean()*100:.1f}%", + "毛R": round(g.fgR.mean(), 3), + "PF": round(w / o, 2) if o > 0 else np.inf, + "t值": round(g.fgR.mean() / (g.fgR.std(ddof=1) / np.sqrt(len(g))), + 2), + }) + print(pd.DataFrame(rows).to_string(index=False)) + def main() -> None: ap = argparse.ArgumentParser() diff --git a/research/step57_b1_flip.py b/research/step57_b1_flip.py new file mode 100644 index 0000000..3312cfc --- /dev/null +++ b/research/step57_b1_flip.py @@ -0,0 +1,84 @@ +"""一类反手（fade B1/S1）是不是真的——逐币、分时段、与 B4 的重叠。 + +step55 的反向对照给出 5m/15m 上 PF 2.35~3.03、t +10~+13，且多空两边都正。 +数字太好，先按 §3.392 的教训做证伪：那次「周末效应」在 1m 上显著，换到 +7 年 5m/15m 样本就消失且符号反转。任何单一样本上的漂亮结果都要先过三关。 + + 逐币 5 个币是不是都成立，还是被某一个币带起来的 + 分时段 前后半段样本各自是否成立（时间外样本） + 与 B4 重叠 如果它只是换个名字的 B4，那就没有增量价值 +""" +from __future__ import annotations + +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.exit_model import fee_of, taker_notional # noqa: E402 + +SRC = HERE / "out" / "step55_bsp12.feather" +SL = 2.0 + + +def stats(g: pd.DataFrame) -> dict: + """反手口径的统计。f_* 是 step55 里用相反方向重跑出场得到的。""" + net = g.f_g.values - fee_of(g.f_r.values, g.f_c.values) + gR = g.f_g.values / (SL * g.atr_pct.values) + R = net / (SL * g.atr_pct.values) + tn = taker_notional(g.f_r.values, g.f_c.values) + w, o = net[net > 0].sum(), -net[net <= 0].sum() + return { + "笔数": len(g), "胜率": f"{(net > 0).mean()*100:.1f}%", + "毛R": round(gR.mean(), 3), "净均R": round(R.mean(), 3), + "PF": round(w / o, 2) if o > 0 else np.inf, + "余量bp": round(net.mean() / tn.mean() * 1e4, 2), + "t值": round(gR.mean() / (gR.std(ddof=1) / np.sqrt(len(g))), 2), + } + + +def main() -> None: + d = pd.read_feather(SRC) + if "f_g" not in d.columns: + print("step55 数据里没有反手列，先重跑 step55") + return + d["date_ext"] = pd.to_datetime(d["date_ext"]) + one = d[d.type.isin(["B1", "S1"])].copy() + + for tf, x in one.groupby("tf"): + print("\n" + "#" * 92) + print(f"########## {tf} · 一类反手 ##########") + + print("\n【逐币】任何一个币独自撑起来的结果都不能要") + print(pd.DataFrame([{"币": s, **stats(g)} + for s, g in x.groupby("sym")]).to_string(index=False)) + + print("\n【分时段】按信号时间中位数切两半，后半段是时间外样本") + cut = x.date_ext.median() + for nm, g in (("前半", x[x.date_ext <= cut]), ("后半", x[x.date_ext > cut])): + r = stats(g) + print(f" {nm} {g.date_ext.min():%Y-%m-%d}~{g.date_ext.max():%Y-%m-%d}" + f" {r}") + + print("\n【分方向】B1反手=做空，S1反手=做多。只有一边成立就是单边行情") + print(pd.DataFrame([{"原类型": t, **stats(g)} + for t, g in x.groupby("type")]).to_string(index=False)) + + print("\n【逐年】") + rows = [] + for y, g in x.groupby(x.date_ext.dt.year): + if len(g) >= 40: + rows.append({"年": y, **stats(g)}) + if rows: + print(pd.DataFrame(rows).to_string(index=False)) + + +if __name__ == "__main__": + main()