Chan/research/live/verify_lean_parity.py

"""在影子信号路径上实测 lean 与 full 是否等价，并量提速。

## 为什么不能直接引用 step46 的对拍结论

step46 固化的是 klc/笔/中枢/`bsp_list`/被消费列的哈希，走的是 `bsp_list` 那条
链。影子路径走的是另一条：`find_fast_bsp3` + `build_htf_zones` +
`htf_fx_timeline` + `attach_htf_context`。两条链读的东西不完全一样，所以
「lean ≡ full」在 step46 用例上成立，不等于在这条路径上成立。

静态检查显示这条链只读 `chan.dataframe` 与 `chan.klc_list`（lean 都不跳），
但静态检查漏不掉间接依赖——`build_htf_zones` 收的是 chan 对象本体。所以逐根
实测：同一个窗口分别用 full 和 lean 跑 `compute()`，比对返回的每一个字段。

判据是**逐字段完全相同**，不是「信号数量相同」。数量相同而方向或标志不同，
会让影子测的是另一批信号，且不报错。

    python research/live/verify_lean_parity.py --syms BTC,ETH,SOL --n 150
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
import warnings
from pathlib import Path

import numpy as np
import pandas as pd

warnings.filterwarnings("ignore")
for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
    os.environ.setdefault(_v, "1")

HERE = Path(__file__).resolve()
sys.path.insert(0, str(HERE.parents[1]))
sys.path.insert(0, str(HERE.parents[2]))
sys.path.insert(0, str(HERE.parent))

LTF_BARS, HTF_BARS = 2001, 801       # 与 shadow_hb 同窗口
SKIP = ("inner_ms", "queue_ms")      # 计时字段本就不同，不参与比对


def load(sym: str, tf: str, cache: Path) -> pd.DataFrame:
    c = sorted(cache.glob(f"bitget_{sym}_{tf}_*.feather"),
               key=lambda p: p.stat().st_size, reverse=True)
    if not c:
        raise FileNotFoundError(f"没有 {sym} {tf} 缓存")
    return pd.read_feather(c[0])


# 随窗口长度必然改变的记账字段。比「窗口长度会不会改信号」时要排除它们，
# 否则一定 0/12 不一致，而那是记账字段在变，不是信号在变
WINDOW_KEYS = ("last_idx", "n_bars")


def canon(d: dict, skip: tuple = SKIP) -> str:
    """把返回值规范化成可比较的字符串。

    浮点直接比会被末位差异误判，但 lean 走的是同一段算术、不该有任何差异，
    所以这里**不设容差**：round 到 12 位只是为了消掉 repr 差异，真有数值
    分歧一定会被抓到。
    """
    def norm(v):
        if isinstance(v, float):
            return None if not np.isfinite(v) else round(v, 12)
        if isinstance(v, dict):
            return {k: norm(x) for k, x in sorted(v.items())}
        if isinstance(v, (list, tuple)):
            return [norm(x) for x in v]
        if isinstance(v, (np.integer, np.floating, np.bool_)):
            return norm(v.item())
        return v
    return json.dumps({k: norm(v) for k, v in sorted(d.items())
                       if k not in skip}, sort_keys=True, ensure_ascii=False)


def signal_bars(sym: str, cache: Path) -> np.ndarray:
    """全量历史里过三滤网的信号根下标。

    随机取窗口几乎测不到信号分支——信号密度约 1/2000 根，12 个窗口命中 0 个。
    只测早退路径的「一致」是很弱的证据：lean 若真影响了中枢或笔，分歧恰恰
    出现在有信号的那些根上。所以把窗口对齐到这些根。

    注意这些下标来自**全量历史**建的中枢，而 compute 只看 2000 根窗口，所以
    对齐后未必真的命中（这正是尚未收口的窗口左边界效应）。但命中率会从
    1/2000 提到可用水平。
    """
    from step43_fill_aware_budget import signals_for
    _, sig = signals_for(sym, cache)
    return sig["entry_idx"].astype(int).to_numpy()


def run_one(sym: str, cache: Path, n: int, step: int,
            at_signals: bool = True) -> dict:
    from shadow_signal import NUM_COLS, compute

    l_all, h_all = load(sym, "1m", cache), load(sym, "5m", cache)
    for df in (l_all, h_all):
        for c in NUM_COLS:
            if c not in df.columns:
                raise RuntimeError(f"{sym} 缺列 {c}")

    l_ts = l_all["timestamp"].to_numpy("int64")
    h_ts = h_all["timestamp"].to_numpy("int64")
    # 一半窗口对齐到已知信号根（测信号分支），一半均匀铺开（测早退路径）
    ends: list[int] = []
    if at_signals:
        try:
            sb = signal_bars(sym, cache)
            sb = sb[(sb > LTF_BARS) & (sb < len(l_all) - 1)]
            ends += list(sb[-(n // 2 or 1):])
            print(f"  对齐到信号根 {len(ends)} 个")
        except Exception as e:
            print(f"  取信号根失败，只用均匀窗口：{e!r}")
    ends += list(range(len(l_all) - 1, LTF_BARS, -step))[:max(n - len(ends), 1)]
    ends = sorted(set(ends))
    if not ends:
        raise RuntimeError("数据不足一个窗口")

    n_same = n_diff = 0
    t_full = t_lean = 0.0
    first_diff = None
    n_hits_full = n_hits_lean = 0

    for e in ends:
        df_l = l_all.iloc[e - LTF_BARS + 1:e + 1]
        # 5m 只取已收盘且不晚于 1m 窗口末尾的根，与实盘一致
        hi = int(np.searchsorted(h_ts, l_ts[e], side="right"))
        df_h = h_all.iloc[max(0, hi - HTF_BARS):hi]
        # 次根开盘价：窗口末尾的下一根，与实盘 entry_px 同义
        entry_px = float(l_all["open"].to_numpy(float)[e + 1]) \
            if e + 1 < len(l_all) else None

        t0 = time.perf_counter()
        rf = compute(df_l.copy(), df_h.copy(), entry_px, lean=False)
        t_full += time.perf_counter() - t0

        t0 = time.perf_counter()
        rl = compute(df_l.copy(), df_h.copy(), entry_px, lean=True)
        t_lean += time.perf_counter() - t0

        n_hits_full += len(rf.get("hits") or [])
        n_hits_lean += len(rl.get("hits") or [])
        if canon(rf) == canon(rl):
            n_same += 1
        else:
            n_diff += 1
            if first_diff is None:
                first_diff = (e, canon(rf), canon(rl))

    k = len(ends)
    print(f"  窗口 {k} 个 · 完全一致 {n_same} · 不一致 {n_diff}")
    print(f"  命中数 full {n_hits_full} / lean {n_hits_lean}")
    print(f"  单窗耗时 full {t_full / k * 1000:.0f}ms · "
          f"lean {t_lean / k * 1000:.0f}ms · "
          f"提速 {t_full / max(t_lean, 1e-9):.2f}x")
    if first_diff:
        e, a, b = first_diff
        print(f"  ⚠ 首个分歧在窗口末尾 idx={e}")
        print(f"    full: {a[:400]}")
        print(f"    lean: {b[:400]}")
    return {"sym": sym, "n": k, "same": n_same, "diff": n_diff,
            "full_ms": t_full / k * 1000, "lean_ms": t_lean / k * 1000,
            "hits_full": n_hits_full, "hits_lean": n_hits_lean}


def recall(syms: list[str], cache: Path, k: int = 30) -> None:
    """全量历史找出的信号，在 2000 根窗口里还能不能复现。

    这是窗口左边界效应的**一半**答案。窗口只有 2000 根，中枢是在窗口内重建
    的，理论上可能与全量历史建的中枢不同，从而漏掉信号。实测最近 k 笔全部
    复现，说明窗口不丢信号。

    ⚠ 另一半没答：窗口会不会**多造出**全量历史没有的信号。那个方向对实盘更
    危险（会多开仓），但要反向扫描——遍历窗口找命中、再回全量历史核对，成本
    高得多。lean 之后单窗 130ms，抽样 2 万个窗口约 43 分钟，已经可做。
    """
    from shadow_signal import compute

    print("全量历史找出的信号，在 2000 根窗口里的复现率\n")
    for sym in syms:
        try:
            l_all, h_all = load(sym, "1m", cache), load(sym, "5m", cache)
            l_ts = l_all["timestamp"].to_numpy("int64")
            h_ts = h_all["timestamp"].to_numpy("int64")
            sb = signal_bars(sym, cache)
            sb = sb[(sb > LTF_BARS) & (sb < len(l_all) - 1)][-k:]
        except Exception as e:
            print(f"  {sym} 跳过：{e!r}")
            continue
        hit = 0
        for e in sb:
            df_l = l_all.iloc[e - LTF_BARS + 1:e + 1]
            hi = int(np.searchsorted(h_ts, l_ts[e], side="right"))
            df_h = h_all.iloc[max(0, hi - HTF_BARS):hi]
            r = compute(df_l.copy(), df_h.copy(),
                        float(l_all["open"].to_numpy(float)[e + 1]))
            if any(h.get("pass_all") for h in (r.get("hits") or [])):
                hit += 1
        print(f"  {sym}  {hit}/{len(sb)} 复现（过全部滤网）")
    print("\n  只说明窗口不丢信号；会不会多造信号需反向扫描，见 docstring。")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--recall", action="store_true",
                    help="只查窗口对全量历史信号的复现率")
    ap.add_argument("--syms", default="BTC,ETH,SOL")
    ap.add_argument("--cache", default="research/live/cache")
    ap.add_argument("--n", type=int, default=150, help="每币比对多少个窗口")
    ap.add_argument("--step", type=int, default=37,
                    help="窗口间隔根数。取质数避免与任何周期共振")
    ap.add_argument("--no-signals", action="store_true",
                    help="不对齐信号根（快，但测不到信号分支）")
    a = ap.parse_args()

    if a.recall:
        recall(a.syms.split(","), Path(a.cache))
        return

    rows = []
    for sym in a.syms.split(","):
        print(f"\n{'=' * 70}\n{sym}")
        try:
            rows.append(run_one(sym, Path(a.cache), a.n, a.step,
                                at_signals=not a.no_signals))
        except Exception as e:
            print(f"  跳过：{e!r}")

    if not rows:
        return
    d = pd.DataFrame(rows)
    print(f"\n\n{'=' * 70}\n汇总\n")
    print(f"  比对窗口 {int(d['n'].sum()):,} 个 · "
          f"不一致 {int(d['diff'].sum())} 个")
    print(f"  单窗耗时 full {d['full_ms'].mean():.0f}ms → "
          f"lean {d['lean_ms'].mean():.0f}ms "
          f"（{d['full_ms'].sum() / max(d['lean_ms'].sum(), 1e-9):.2f}x）")
    if int(d["diff"].sum()) == 0:
        print("\n  逐字段完全一致，可以开 lean。")
    else:
        print("\n  ⛔ 存在分歧，不要开 lean。lean 跳掉的东西这条路径确实在用。")


if __name__ == "__main__":
    main()