Chan/research/live/verify_window_sens.py

"""compute() 的输出对窗口长度是否不变——增量路径的前提。

## 为什么这是增量路径的前提

`init_stream/append_bar` 没有 trim：`dataframe` 靠 `pd.concat` 无界增长。所以
增量方案必然意味着**窗口会长大**（每根 +1），只能靠周期性 `init_stream` 重建
拉回。于是在两次重建之间，实际窗口是 [W, W+slack] 而不是恒定 W。

这就把一个问题摆在前面：如果 `compute()` 的输出随窗口长度变化，增量路径等于
静默把信号换了一批——不报错、不崩，只是测的不再是回测那批信号。

step39 的结论是「命中率在 2000 根饱和」。**饱和不等于不变**：再加根数不再提高
命中率，与「结果逐字段相同」是两回事。所以要单独验。

判据是逐字段相同，不是命中数相同。数量相同而方向或滤网标志不同，会让影子测
的是另一批信号。

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

import argparse
import os
import sys
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))

BASE_L, BASE_H = 2001, 801
# 增量在两次重建之间会长这么多。取 500 是因为它对应约 8 小时，
# 重建摊薄后单根成本仍接近纯追加
GROW = (0, 200, 500, 1000)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--syms", default="BTC,ETH,SOL")
    ap.add_argument("--cache", default="research/live/cache")
    ap.add_argument("--n", type=int, default=40)
    a = ap.parse_args()

    from shadow_signal import compute
    from verify_lean_parity import (SKIP, WINDOW_KEYS, canon, load,
                                    signal_bars)

    # last_idx/n_bars 必然随窗口长度变。不排除的话结果一定是「全不一致」，
    # 而那说明的是记账字段在变，不是信号在变
    skip = SKIP + WINDOW_KEYS

    cache = Path(a.cache)
    print("同一根上，只改窗口长度，比对 compute() 的全部返回字段")
    print(f"基准窗口 1m×{BASE_L} + 5m×{BASE_H}；增量会让它长大，故试 "
          f"+{GROW[1:]}\n")

    tot = {g: [0, 0] for g in GROW[1:]}     # [相同, 不同]
    for sym in a.syms.split(","):
        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")
        try:
            sb = signal_bars(sym, cache)
        except Exception as e:
            print(f"{sym} 取信号根失败：{e!r}")
            continue
        need = BASE_L + max(GROW)
        sb = sb[(sb > need) & (sb < len(l_all) - 1)][-a.n:]
        if len(sb) == 0:
            print(f"{sym} 可用信号根不足")
            continue

        res: dict[int, list[str]] = {g: [] for g in GROW}
        for e in sb:
            hi = int(np.searchsorted(h_ts, l_ts[e], side="right"))
            entry = float(l_all["open"].to_numpy(float)[e + 1])
            for g in GROW:
                df_l = l_all.iloc[e - (BASE_L + g) + 1:e + 1]
                df_h = h_all.iloc[max(0, hi - (BASE_H + g // 5)):hi]
                res[g].append(canon(compute(df_l.copy(), df_h.copy(), entry),
                                    skip=skip))

        print(f"{sym}  {len(sb)} 根信号窗口")
        for g in GROW[1:]:
            same = sum(1 for x, y in zip(res[0], res[g]) if x == y)
            tot[g][0] += same
            tot[g][1] += len(sb) - same
            print(f"  +{g:>4} 根 → 一致 {same}/{len(sb)}"
                  + ("" if same == len(sb) else "   ⚠ 有分歧"))
        del l_all, h_all

    print(f"\n{'=' * 66}\n汇总\n")
    for g in GROW[1:]:
        s, d = tot[g]
        print(f"  窗口 +{g:>4} 根：一致 {s} · 不一致 {d}")
    worst = max(GROW[1:], key=lambda g: tot[g][1])
    if tot[worst][1] == 0:
        print("\n  窗口长度不影响输出，增量路径的前提成立。")
        print("  可以按「长到 +N 根再 init_stream 重建」摊薄成本。")
    else:
        print("\n  ⛔ 窗口长度会改变输出，增量路径会静默换掉一批信号。")
        print("  此时要么每根都重建（等于没有增量），要么先把窗口效应本身收口。")


if __name__ == "__main__":
    main()