Chan/research/step45_repaint.py

"""Step 43：重画率——实时看到过的信号，有多少后来消失了。

step39 也报过「假阳性」，但那个测法是无效的：它从全序列随机抽非信号点，
看时点重建会不会凭空冒信号。1m 上信号密度约 474 根 1 个，180 个随机点里
本来就只期望撞上 0.38 个，测出 0% 几乎不含信息量。

重画不发生在随机点上，只发生在「差一点就成型」的结构附近。所以要
**逐根**做时点重建，把「当根确实出现了信号」的位置全收集起来，
再看它们在全量视角里还在不在。

  实时信号   窗口只喂到第 T 根，重建后信号恰好落在第 T 根（实盘会下单的那些）
  重画       该信号在全量重建里不存在（图上后来消失，但实盘已经开了仓）
  漏看       全量有、实时当根没有（step39 已验证 ~0，这里顺带复核）

只有**深色信号**（h1_agree ∧ ladder_ok）才会真下单，所以分层报告：
浅色重画无所谓，深色重画才影响实盘。
"""
from __future__ import annotations

import argparse
import os
import sys
import time
import warnings
from concurrent.futures import ProcessPoolExecutor, as_completed
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().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))
pd.set_option("display.width", 320)

LTF, HTF = "1m", "5m"
WINDOW = 2000          # step39 证明 1m 在 2000 根就饱和，实盘也用这个
HTF_WINDOW = 800       # 5m 侧窗口，同 step39
MAX_ROWS = 1_200_000


def _load(sym: str):
    from chanlun import TF_DF
    from lib.data import fetch_ohlcv
    from lib.fx_signal import extract_fx_signals, signals_to_frame
    from lib.nested_bsp import htf_fx_timeline

    pair = f"{sym}/USDT:USDT"
    df_l = fetch_ohlcv(pair, LTF, MAX_ROWS)
    df_h = fetch_ohlcv(pair, HTF, 10 ** 9)
    chan_l = TF_DF(df_l, 1, LTF)
    cdf = chan_l.dataframe
    chan_h = TF_DF(df_h, 1, HTF)
    hdf = chan_h.dataframe
    tl = htf_fx_timeline(signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
    return chan_l, cdf, hdf, tl


def _full_signals(chan_l, cdf, tl):
    """全量视角的信号集（带两个过滤器）。"""
    from chanlun.analysis.fast_bsp import attach_zone_ladder
    from lib.fast_bsp3 import find_fast_bsp3
    from lib.nested_bsp import attach_htf_context
    from lib.nested_level import build_htf_zones

    zones = build_htf_zones(cdf, LTF, chan=chan_l).reset_index(drop=True)
    if zones.empty:
        return pd.DataFrame()
    sig = find_fast_bsp3(cdf, zones)
    if sig.empty:
        return pd.DataFrame()
    sig = attach_htf_context(sig, cdf, tl, "h1")
    sig = attach_zone_ladder(sig, zones)
    sig["ts"] = cdf["timestamp"].to_numpy()[sig["entry_idx"].astype(int)]
    return sig


def scan_chunk(task: tuple) -> dict:
    """逐根时点重建，只记录信号恰好落在当根的位置。"""
    import warnings as _w
    _w.filterwarnings("ignore")
    sys.path.insert(0, str(HERE))
    sys.path.insert(0, str(HERE.parent))

    from chanlun import TF_DF
    from chanlun.analysis.fast_bsp import attach_zone_ladder
    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

    sym, lo, hi = task
    try:
        chan_l, cdf, hdf, _tl_full = _load(sym)
        ltf_ts = cdf["timestamp"].to_numpy()
        htf_ts = hdf["timestamp"].to_numpy()
        hi = min(hi, len(cdf))
        lo = max(lo, WINDOW)

        hits, t0 = [], time.perf_counter()
        for T in range(lo, hi):
            sl = cdf.iloc[T - WINDOW + 1: T + 1].reset_index(drop=True)
            z = build_htf_zones(sl, LTF)
            if z.empty:
                continue
            z = z.reset_index(drop=True)
            sig = find_fast_bsp3(sl, z)
            if sig.empty:
                continue
            last = len(sl) - 1
            row = sig[sig["entry_idx"].astype(int) == last]
            if row.empty:
                continue

            # 只在真的出信号时才算过滤器——信号稀疏，这部分开销可忽略
            h_end = int(np.searchsorted(htf_ts, ltf_ts[T], side="right"))
            agree = np.nan
            if h_end >= HTF_WINDOW:
                hsl = hdf.iloc[h_end - HTF_WINDOW: h_end].reset_index(drop=True)
                ch = TF_DF(hsl, 1, HTF)
                tl_p = htf_fx_timeline(
                    signals_to_frame(extract_fx_signals(ch, ch.dataframe)), ch.dataframe)
                got = attach_htf_context(row.copy(), sl, tl_p, "h1")
                agree = float(got["h1_agree"].iloc[0] == 1)
            lad = attach_zone_ladder(row.copy(), z)
            hits.append({
                "sym": sym, "T": int(T), "ts": int(ltf_ts[T]),
                "direction": int(row["direction"].iloc[0]),
                "h1_agree": agree,
                "ladder_ok": bool(lad["ladder_ok"].iloc[0]),
            })
        return {"sym": sym, "lo": lo, "hi": hi, "n_bars": hi - lo,
                "hits": pd.DataFrame(hits), "secs": time.perf_counter() - t0}
    except Exception as e:
        return {"sym": sym, "error": repr(e)[:300]}


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbols", default="BTC,ETH,SOL,DOGE,XRP,LINK")
    ap.add_argument("--bars", type=int, default=30000, help="每币逐根扫描的根数")
    ap.add_argument("--workers", type=int, default=10)
    ap.add_argument("--chunk", type=int, default=2500)
    args = ap.parse_args()

    syms = [s.strip() for s in args.symbols.split(",")]
    out_dir = HERE / "out"
    out_dir.mkdir(exist_ok=True)

    # 先拿各币全量信号做基准，同时确定扫描区间
    print(f"[重画审计] {len(syms)} 币 × 每币 {args.bars} 根逐根重建，"
          f"窗口 {WINDOW}\n", flush=True)
    full_map, tasks = {}, []
    for s in syms:
        try:
            chan_l, cdf, hdf, tl = _load(s)
            full = _full_signals(chan_l, cdf, tl)
            full_map[s] = full
            hi = len(cdf) - 1
            lo = max(WINDOW, hi - args.bars)
            for a in range(lo, hi, args.chunk):
                tasks.append((s, a, min(a + args.chunk, hi)))
            print(f"  {s}: 全量 {len(cdf)} 根，全量信号 {len(full)}，"
                  f"扫描 [{lo}, {hi})", flush=True)
        except Exception as e:
            print(f"  {s}: 载入失败 {e!r}", flush=True)

    est = sum(t[2] - t[1] for t in tasks) * 0.202 / args.workers
    print(f"\n共 {len(tasks)} 个分块，预计 {est / 60:.0f} 分钟\n", flush=True)

    res, done = [], 0
    with ProcessPoolExecutor(max_workers=args.workers) as ex:
        futs = {ex.submit(scan_chunk, t): t for t in tasks}
        for f in as_completed(futs):
            r = f.result()
            done += 1
            if "error" in r:
                print(f"  [{done}/{len(tasks)}] {r['sym']} 出错 {r['error']}", flush=True)
                continue
            res.append(r)
            print(f"  [{done}/{len(tasks)}] {r['sym']} [{r['lo']},{r['hi']}) "
                  f"实时信号 {len(r['hits'])} 个，{r['secs']:.0f}s", flush=True)

    if not res:
        print("无结果")
        return

    live = pd.concat([r["hits"] for r in res if len(r["hits"])], ignore_index=True)
    live.to_feather(out_dir / "step43_live_signals.feather")
    n_bars = sum(r["n_bars"] for r in res)

    # 对齐：实时信号的时间戳是否出现在全量信号集里
    rows = []
    for s, g in live.groupby("sym"):
        full = full_map.get(s)
        fts = set(full["ts"].astype(int).tolist()) if full is not None and len(full) else set()
        g = g.copy()
        g["survived"] = g["ts"].isin(fts)
        rows.append(g)
    live = pd.concat(rows, ignore_index=True)
    live["dark"] = (live["h1_agree"] == 1.0) & live["ladder_ok"]

    print("\n" + "=" * 100)
    print(f"########## 1. 总体（扫描 {n_bars} 根）##########")
    n, sv = len(live), int(live["survived"].sum())
    print(f"  实时出现过的信号 {n} 个，全量视角仍在 {sv} 个，"
          f"重画 {n - sv} 个 = {(n - sv) / max(n, 1) * 100:.2f}%")

    print("\n########## 2. 按过滤器分层（只有深色会真下单）##########")
    rows = []
    for lab, m in (("深色（双过滤通过）", live["dark"]),
                   ("浅色（未通过）", ~live["dark"])):
        gg = live[m]
        if not len(gg):
            continue
        k = int((~gg["survived"]).sum())
        # Wilson 95% 上界，样本小的时候点估计没意义
        from math import sqrt
        nn, p = len(gg), k / len(gg)
        z = 1.96
        hi_b = (p + z * z / (2 * nn) + z * sqrt(p * (1 - p) / nn + z * z / (4 * nn * nn))) / (1 + z * z / nn)
        rows.append({"分层": lab, "实时信号": nn, "重画": k,
                     "重画率": f"{p * 100:.2f}%", "95%上界": f"{hi_b * 100:.2f}%"})
    print(pd.DataFrame(rows).to_string(index=False))

    print("\n########## 3. 分币种 ##########")
    rows = []
    for s, g in live.groupby("sym"):
        d = g[g["dark"]]
        rows.append({"币": s, "实时信号": len(g), "其中深色": len(d),
                     "深色重画": int((~d["survived"]).sum()) if len(d) else 0,
                     "全部重画": int((~g["survived"]).sum())})
    print(pd.DataFrame(rows).to_string(index=False))

    print("\n########## 4. 漏看（全量有、实时当根没有）复核 ##########")
    for s, g in live.groupby("sym"):
        full = full_map.get(s)
        if full is None or not len(full):
            continue
        lo = min(r["lo"] for r in res if r["sym"] == s)
        hi = max(r["hi"] for r in res if r["sym"] == s)
        inrange = full[(full["entry_idx"] >= lo) & (full["entry_idx"] < hi)]
        seen = set(g["ts"].astype(int).tolist())
        miss = int((~inrange["ts"].astype(int).isin(seen)).sum())
        print(f"  {s}: 扫描区间内全量信号 {len(inrange)}，实时当根未出现 {miss} 个"
              f"（{miss / max(len(inrange), 1) * 100:.1f}%）")

    print("\n########## 结论 ##########")
    d = live[live["dark"]]
    if len(d):
        k = int((~d["survived"]).sum())
        print(f"  深色信号 {len(d)} 个，重画 {k} 个。")
        print("  重画的仓位是真实成交的，但出场（止损/止盈/超时）不依赖信号是否还在图上，")
        print("  所以不会卡仓；影响仅限于「实盘比回测多开的这部分，质量不在回测统计里」。")


if __name__ == "__main__":
    main()