Chan/research/live/verify_signal_path.py

"""验证 live 信号路径与 step42 的批量过滤等价。

live 侧每根只看最后一根、且只喂 2000 根窗口；研究侧一次性跑全量。两者
用同一套滤网（同向 + 中枢阶梯 + ATR 门控），但**不保证逐笔一致**——
缠论结构依赖历史，2000 根窗口是 step39 定的命中率饱和点，不是无损截断。

每个信号根上比三种口径：

  批量      全量历史 + 完整 5m。这是预算的来源，是基准
  剔partial 窗口 2000 根 1m + 800 根**已收盘** 5m
  含partial 窗口，5m 末尾保留那根**尚未收盘**的。这是 live 现行做法

## 结论：partial 根要保留，不能剔

live 的 `df_h[df_h["timestamp"] < kline_ts]` 里 kline_ts 是 1m 的收盘时刻，
而正在走的那根 5m 开盘更早，于是被保留下来——五根里有四根如此。乍看像是
「把未收盘的根当完整根用」的口径错误，实测**反过来**：

  BTC 25/25、ETH 24/25、SOL 23/25 与批量一致（合计 96%）
  剔掉则只有 21/25、21/25、18/25（合计 80%）

原因是批量口径里那根 5m 是存在的。缠论的包含处理与分型检测吃整条序列，
凭空少一根会把结构整体挪位；保留一根「开盘价正确、高低点尚不完整」的
近似根，比直接删掉更接近批量。

这也暴露了研究侧的一处残留：批量的 HTF 结构用到了那根 5m 的**最终**高低点，
而实盘在该时刻不可能知道。`htf_fx_timeline` 的 `confirm_ts += period` 只挡住了
分型**选取**上的未来函数，挡不住结构构建。live 用 partial 根逼近，落在 96%，
差的那 4% 是这条残留的下界，不是可以修掉的 bug。

    .venv/bin/python research/live/verify_signal_path.py --symbol BTC
"""
from __future__ import annotations

import argparse
import sys
import warnings
from pathlib import Path

warnings.filterwarnings("ignore")
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))
sys.path.insert(0, str(HERE.parents[1]))

import numpy as np  # noqa: E402
import pandas as pd  # noqa: E402

WINDOW_L, WINDOW_H = 2000, 800


HTF_MS = 300_000


def partial_htf_bar(df_l: pd.DataFrame, kline_ts: int) -> dict | None:
    """用 1m 合成「此刻正在走的那根 5m」，复现 live 曾经喂进去的 partial 根。"""
    bucket = kline_ts // HTF_MS * HTF_MS
    if bucket >= kline_ts:            # 正好落在 5m 边界，没有未收盘的根
        return None
    part = df_l[(df_l["timestamp"] >= bucket) & (df_l["timestamp"] < kline_ts)]
    if part.empty:
        return None
    date = pd.to_datetime(bucket, unit="ms", utc=True) \
             .tz_convert("Asia/Shanghai")
    return {"timestamp": bucket, "date": date,
            "open": float(part["open"].iloc[0]),
            "high": float(part["high"].max()), "low": float(part["low"].min()),
            "close": float(part["close"].iloc[-1]),
            "volume": float(part["volume"].sum())}


def load(sym: str, tf: str, days: int) -> pd.DataFrame:
    f = HERE / "cache" / f"bitget_{sym}_{tf}_{days}d.feather"
    if not f.exists():
        raise SystemExit(f"缺数据 {f}")
    return pd.read_feather(f)


def batch_flags(df_l: pd.DataFrame, df_h: pd.DataFrame) -> pd.DataFrame:
    """step42_exit_tp_1m.run_one 的滤网，逐行照搬。"""
    from chanlun import TF_DF
    from lib.fast_bsp3 import find_fast_bsp3
    from lib.fx_signal import extract_fx_signals, signals_to_frame
    from lib.nested_bsp import attach_htf_context, htf_fx_timeline
    from lib.nested_level import build_htf_zones
    from lib.shadow_budget import ATR_GATE_BP

    chan_l = TF_DF(df_l, 1, "1m")
    cdf = chan_l.dataframe
    zones = build_htf_zones(cdf, "1m", chan=chan_l).reset_index(drop=True)
    z = zones.copy()
    pg, pdn = z["zg"].shift(), z["zd"].shift()
    z["z_above"], z["z_below"] = z["zd"] > pg, z["zg"] < pdn
    z["zone_i"] = np.arange(len(z))

    chan_h = TF_DF(df_h, 1, "5m")
    hdf = chan_h.dataframe
    tl = htf_fx_timeline(
        signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)

    sig = find_fast_bsp3(cdf, zones)
    sig = sig.merge(z[["zone_i", "z_above", "z_below"]], on="zone_i", how="left")
    sig = attach_htf_context(sig, cdf, tl, "h1")

    d = sig["direction"].astype(int)
    push = np.where(d == 1, sig["z_above"], sig["z_below"])
    idx = sig["entry_idx"].astype(int).to_numpy()
    entry = cdf["open"].to_numpy(float)[np.minimum(idx + 1, len(cdf) - 1)]
    atr_pct = cdf["atr"].to_numpy(float)[idx] / entry

    out = pd.DataFrame({
        "entry_idx": idx,
        "ts": cdf["timestamp"].to_numpy()[idx],
        "direction": d.to_numpy(),
        "h1_agree": sig["h1_agree"].fillna(0).astype(int).to_numpy(),
        "ladder_ok": pd.Series(push).fillna(False).astype(int).to_numpy(),
        "atr_bp": atr_pct * 1e4,
    })
    out["gate_ok"] = (out["atr_bp"] >= ATR_GATE_BP).astype(int)
    out["pass_all"] = ((out["h1_agree"] == 1) & (out["ladder_ok"] == 1)
                       & (out["gate_ok"] == 1)).astype(int)
    return out, cdf


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbol", default="BTC")
    ap.add_argument("--days", type=int, default=30)
    ap.add_argument("--checks", type=int, default=12)
    a = ap.parse_args()

    df_l, df_h = load(a.symbol, "1m", a.days), load(a.symbol, "5m", a.days)
    print(f"[{a.symbol}] 1m {len(df_l)} 根 / 5m {len(df_h)} 根，跑批量滤网…",
          flush=True)
    batch, cdf = batch_flags(df_l, df_h)
    n = len(batch)
    print(f"  原始 B4/S4 {n} 个 · 同向 {int((batch.h1_agree == 1).sum())}"
          f" · 同向+阶梯 "
          f"{int(((batch.h1_agree == 1) & (batch.ladder_ok == 1)).sum())}"
          f" · 三项全过 {int(batch.pass_all.sum())}")
    print(f"  ATR 中位 {batch.atr_bp.median():.2f}bp · "
          f"门控刷掉 {(1 - batch.gate_ok.mean()) * 100:.1f}%\n")

    # 挑最近的若干个信号根做窗口复现
    from shadow_signal import compute
    cand = batch[batch["entry_idx"] >= WINDOW_L].tail(a.checks)
    if cand.empty:
        raise SystemExit("窗口内没有可核对的信号")

    def run_window(r, with_partial: bool):
        i = int(r["entry_idx"])
        kline_ts = int(r["ts"]) + 60_000        # 信号根的收盘时刻
        wl = df_l[df_l["timestamp"] < kline_ts].tail(WINDOW_L)
        wh = df_h[df_h["timestamp"] + HTF_MS <= kline_ts].tail(WINDOW_H)
        if with_partial:
            p = partial_htf_bar(df_l, kline_ts)
            if p is not None:
                wh = pd.concat([wh, pd.DataFrame([p])], ignore_index=True)
        entry_px = float(cdf["open"].to_numpy(float)[min(i + 1, len(cdf) - 1)])
        res = compute(wl.reset_index(drop=True), wh.reset_index(drop=True),
                      entry_px)
        if res.get("error"):
            raise SystemExit(f"compute 报错，测试本身有问题：{res['error']}\n"
                             f"{res.get('traceback', '')}")
        return next((h for h in res["hits"]
                     if h["direction"] == int(r["direction"])), None)

    def fmt(h):
        if h is None:
            return f"{'未复现':>18}"
        return (f"{h['h1_agree']:>6}{h['ladder_ok']:>5}{h['gate_ok']:>5}")

    print(f"{'K线时刻':<15}{'方向':>4}{'  批量':>18}{'  窗口':>18}"
          f"{'  含未收盘':>18}{'  截断':>7}{'partial':>9}")
    print(f"{'':<15}{'':>4}{'同向 阶梯 门控':>20}{'同向 阶梯 门控':>20}"
          f"{'同向 阶梯 门控':>20}")
    n_trunc = n_part = n_ok = n_bad = 0
    for _, r in cand.iterrows():
        h_ok = run_window(r, False)
        h_bad = run_window(r, True)
        t = pd.to_datetime(r["ts"], unit="ms", utc=True) \
              .tz_convert("Asia/Shanghai").strftime("%m-%d %H:%M")
        ref = (int(r.h1_agree), int(r.ladder_ok), int(r.gate_ok))
        got = None if h_ok is None else (h_ok["h1_agree"], h_ok["ladder_ok"],
                                         h_ok["gate_ok"])
        bad = None if h_bad is None else (h_bad["h1_agree"], h_bad["ladder_ok"],
                                          h_bad["gate_ok"])
        n_trunc += got != ref
        n_part += bad != got
        n_ok += got == ref
        n_bad += bad == ref
        print(f"{t:<15}{int(r.direction):>+4}"
              f"{ref[0]:>6}{ref[1]:>5}{ref[2]:>5}"
              f"{fmt(h_ok):>18}{fmt(h_bad):>18}"
              f"{'' if got == ref else '差':>7}"
              f"{'' if bad == got else '差':>9}")
    n = len(cand)
    print(f"\n  与批量（预算口径）一致：")
    print(f"    剔掉未收盘 5m 根  {n_ok}/{n}")
    print(f"    保留未收盘 5m 根  {n_bad}/{n}   ← live 现行做法")
    print(f"  两种窗口口径互不相同 {n_part}/{n}")


if __name__ == "__main__":
    main()