"""那 72~77% 的「额外信号」到底怎么来的：中枢重画，还是扫描窗口错位？

我在 §3.3994 里把它归因为「中枢重画」，**这个归因没验证过，而且与已有结论矛盾**
（step39 的假阳性 0%、`verify_window_sens` 的窗口 +200/+500/+1000 逐字段一致）。
用户指出中枢不重画，代码注释也支持他：

    # -1 = 最后一笔（历史默认）。中枢每吸收一根K线，最后一笔就可能后移，
    #      于是 available_ts 跟着漂——这是右边缘重画的根因
                                        —— chanlun/analysis/fast_bsp.py:47

漂的不是中枢**边界**（zg/zd），是它的**可用时刻**。而 `find_fast_bsp3` 只从
`available_ts` 往后扫 `scan=200` 根。两种口径的窗口因此错位：

    实时  中枢没吸收完，available_ts 偏早 -> 窗口开得早
    全量  中枢吸收完了，available_ts 偏晚（§5.41 实测中位晚 62 分钟）-> 窗口开得晚

落在「实时窗口内、全量窗口外」的信号，全量根本没扫到那个时段，于是显示为「额外」。

两种机制的修法完全不同，所以必须分清：

  边界重画   结构本身不稳，只能用滞后换稳定性，代价大
  窗口错位   中枢是同一个真中枢，信号也是真信号，只是**开得太早、确认不足**
             —— 这正好解释它们为什么亏（PF 0.26~0.36），且修法是调 available_ts

判据：逐个额外信号，去全量中枢表里按 (zg, zd) 找它的中枢。
  找得到且边界一致 -> 窗口错位（用户是对的，我的归因错了）
  找不到或边界不同 -> 边界重画（我的归因成立）
"""
from __future__ import annotations

import argparse
import sys
import warnings
from concurrent.futures import ProcessPoolExecutor, as_completed
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))

OUT = HERE / "out" / "step69_mech.feather"
WIN, MAX_GROW, SCAN = 2001, 500, 200
TOL = 1e-6


def replay(sym: str, ltf: str, rows: int, steps: int) -> pd.DataFrame | None:
    from chanlun import TF_DF
    from chanlun.analysis.fast_bsp import (
        ensure_timestamp,
        find_fast_bsp3,
        zones_from_zs_list,
    )
    from lib.data import fetch_ohlcv

    try:
        df = fetch_ohlcv(f"{sym}/USDT:USDT", ltf, rows)
        if df is None or len(df) < WIN + steps + 100:
            return None
        df = df.iloc[-(WIN + steps):].reset_index(drop=True)

        full = TF_DF(df, 1, ltf)
        cdf = ensure_timestamp(full.dataframe)
        zf = zones_from_zs_list(full.cal_bi_zs_list_pure(full.bi_list), cdf)
        if zf is None or zf.empty:
            return None
        sig_full = find_fast_bsp3(cdf, zf)
        full_keys = {(int(r.entry_idx), int(r.direction))
                     for r in sig_full.itertuples()} if not sig_full.empty \
            else set()
        fzg = zf["zg"].to_numpy(float)
        fzd = zf["zd"].to_numpy(float)
        fav = zf["available_ts"].to_numpy()
        ts_all = cdf["timestamp"].to_numpy()

        rec: dict[tuple, dict] = {}
        chan, anchor = None, 0
        for i in range(WIN, len(df)):
            if chan is None or (i - anchor) >= MAX_GROW:
                w = df.iloc[i - WIN + 1:i + 1].copy()
                chan = TF_DF(w, 1, ltf)
                chan.init_stream(w, 1, ltf)
                anchor = i
            else:
                chan.append_bar(df.iloc[i])
            try:
                zl = chan.cal_bi_zs_list_pure(chan.bi_list)
                if not zl:
                    continue
                sub = ensure_timestamp(chan.dataframe)
                z = zones_from_zs_list(zl, sub)
                if z is None or z.empty:
                    continue
                s = find_fast_bsp3(sub, z)
                if s is None or s.empty:
                    continue
                last = len(sub) - 1
                s = s[s["entry_idx"].astype(int) == last]
                if s.empty or "zone_i" not in s.columns:
                    continue
                # find_fast_bsp3 的输出自带 zg/zd，直接 merge 会加后缀，
                # 所以中枢侧的列全部改名再接
                zc = z[["zg", "zd", "available_ts"]].rename(columns={
                    "zg": "z_zg", "zd": "z_zd", "available_ts": "z_av"})
                zc["zone_i"] = np.arange(len(z))
                s = s.drop(columns=[c for c in ("z_zg", "z_zd", "z_av")
                                    if c in s.columns])
                s = s.merge(zc, on="zone_i", how="left")
            except Exception:                                 # noqa: BLE001
                continue
            for r in s.itertuples():
                k = (i, int(r.direction))
                if k in rec:
                    continue
                # 该中枢在全量表里是否存在（按边界匹配，边界是不该漂的量）
                zg_, zd_ = float(r.z_zg), float(r.z_zd)
                m = (np.abs(fzg - zg_) <= TOL * max(1.0, abs(zg_))) \
                    & (np.abs(fzd - zd_) <= TOL * max(1.0, abs(zd_)))
                j = int(np.argmax(m)) if m.any() else -1
                rec[k] = {
                    "sym": sym, "entry_idx": i, "direction": int(r.direction),
                    "in_full": k in full_keys,
                    "zone_found": bool(m.any()),
                    "z_zg": zg_, "z_zd": zd_,
                    "avail_rt": int(r.z_av),
                    "avail_full": int(fav[j]) if j >= 0 else -1,
                    "ts_entry": int(ts_all[i]) if i < len(ts_all) else -1,
                }
        return pd.DataFrame(list(rec.values())) if rec else None
    except Exception as e:                                    # noqa: BLE001
        print(f"  {sym} 失败: {type(e).__name__}: {e}", flush=True)
        return None


def report(d: pd.DataFrame, ltf: str) -> None:
    ex = d[~d.in_full]
    print("\n" + "=" * 92)
    print("【一】判据：额外信号所在的中枢，在全量表里找得到吗")
    print("=" * 92)
    print(f"回放信号 {len(d)} · 额外 {len(ex)}")
    print(f"**额外信号中，其中枢按 (zg,zd) 在全量表里找得到的："
          f"{ex.zone_found.mean()*100:.1f}%**")
    print(f"（对照：非额外信号 {d[d.in_full].zone_found.mean()*100:.1f}%）")
    print("""
  ≈100% -> 中枢边界没变，是**扫描窗口错位**，用户对、我的「重画」归因错
  明显偏低 -> 中枢确实消失或改边界，「重画」成立""")

    print("\n" + "=" * 92)
    print("【一b】数量级对账：中枢层面只有 6.5% 被改，信号层面却 74% 是额外的")
    print("=" * 92)
    print("§5.41 的 A/B 实测：`bis[-1]` 口径下确认时刻被改 6.5%、中枢消失 9.0%。")
    print("若信号层面的 74% 成立，必然有放大机制。怀疑是 `max_per_zone=1`：")
    print("  每个中枢只返回**第一个**入场点，而扫描起点随 available_ts 棘轮后移，")
    print("  越过旧入场点后，同一中枢会重新产出一个「第一个」——全量只用最终值，")
    print("  所以每中枢至多一个信号，实时却能反复触发。")
    for nm, x in [("额外信号", ex), ("非额外信号", d[d.in_full])]:
        if x.empty:
            continue
        nz = x.groupby(["sym", "z_zg", "z_zd"]).size() \
            if "z_zg" in x.columns else None
        if nz is None:
            print("  （缺 z_zg/z_zd 列，跳过）")
            break
        print(f"\n{nm}：{len(x)} 个信号，落在 {len(nz)} 个不同中枢上 "
              f"-> 每中枢 {len(x)/len(nz):.2f} 次")
        print(f"  同一中枢触发次数分布  中位 {nz.median():.0f}  "
              f"P90 {nz.quantile(.9):.0f}  最大 {nz.max()}")
    print("""
  若「额外信号」的每中枢次数显著 > 1 而「非额外」≈ 1，放大机制坐实：
  6.5% 的中枢改动通过棘轮重扫，放大成信号层面的几百个。""")

    g = ex[ex.zone_found & (ex.avail_full > 0)].copy()
    if g.empty:
        return
    bar_ms = {"1m": 60_000, "5m": 300_000, "15m": 900_000}.get(ltf, 300_000)
    g["drift_bars"] = (g.avail_full - g.avail_rt) / bar_ms
    g["from_rt"] = (g.ts_entry - g.avail_rt) / bar_ms
    g["from_full"] = (g.ts_entry - g.avail_full) / bar_ms

    print("\n" + "=" * 92)
    print("【二】available_ts 漂了多少，以及入场落在谁的扫描窗口里")
    print("=" * 92)
    print(f"avail 漂移（全量 − 实时，根）  中位 {g.drift_bars.median():.0f}  "
          f"P25 {g.drift_bars.quantile(.25):.0f}  "
          f"P75 {g.drift_bars.quantile(.75):.0f}")
    print(f"  §5.41 记的是中位晚 62 分钟，本表 {ltf} 下即 "
          f"{62*60_000/bar_ms:.0f} 根，可交叉验证")
    print(f"\n入场距实时 avail（根）  中位 {g.from_rt.median():.0f} "
          f"（应落在 0~{SCAN} 内，否则实时也扫不到）")
    print(f"入场距全量 avail（根）  中位 {g.from_full.median():.0f}")
    out = ((g.from_full < 0) | (g.from_full > SCAN)).mean()
    print(f"\n**入场落在全量扫描窗口 [0,{SCAN}] 之外的比例：{out*100:.1f}%**")
    print("""
  这是机制的直接证据：比例高 -> 全量根本没扫到那个时段，所以「没有」这个信号，
  与中枢是否重画无关。其中 from_full < 0 表示入场早于全量的可用时刻——
  即**实时抢跑了**，中枢还没吸收完就下单。""")
    early = (g.from_full < 0).mean()
    print(f"  其中抢跑（早于全量 avail）：{early*100:.1f}%")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE")
    ap.add_argument("--ltf", default="5m")
    ap.add_argument("--rows", type=int, default=45_000)
    ap.add_argument("--steps", type=int, default=20_000)
    ap.add_argument("--workers", type=int, default=5)
    ap.add_argument("--reuse", action="store_true")
    args = ap.parse_args()

    if args.reuse and OUT.exists():
        report(pd.read_feather(OUT), args.ltf)
        return
    syms = [s.strip() for s in args.symbols.split(",")]
    print(f"[机制判定] {len(syms)} 币 × {args.steps} 根 · {args.ltf}\n",
          flush=True)
    parts = []
    with ProcessPoolExecutor(max_workers=args.workers) as ex_:
        fut = {ex_.submit(replay, s, args.ltf, args.rows, args.steps): s
               for s in syms}
        for i, f in enumerate(as_completed(fut), 1):
            r = f.result()
            print(f"  [{i}/{len(syms)}] {fut[f]} "
                  f"{0 if r is None else len(r)}", flush=True)
            if r is not None:
                parts.append(r)
    if not parts:
        print("无结果")
        return
    d = pd.concat(parts, ignore_index=True)
    OUT.parent.mkdir(exist_ok=True)
    d.to_feather(OUT)
    report(d, args.ltf)


if __name__ == "__main__":
    main()