Chan/research/step49_cluster_oos.py

"""Step 49：簇内「有无前序信号」的样本外验证。

step48 在最近 208 天上发现：开仓时若过去 5 分钟内已有别的币发过信号，
该笔的滑点余量 21.88bp，明显高于无前序的 14.13bp（+55%）。毛R 只差 8.8%，
所以值得追的是余量这一条——1m 的生死线就在执行成本上。

但那是发现期内的数字，且「过去 5 分钟」这个窗口是我挑的。本步用更长的历史，
把发现期之外的时段单独拿出来看同一个差距还在不在。

只验这一个假设。158 个簇撑不起更多——每多验一个，假阳性概率就涨一截。
判据：OOS 各时段里 `有前序` 的余量应稳定高于 `无前序`，且方向一致。
"""
from __future__ import annotations

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

SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48
GATE_BP = 8.0
WIN_MIN = 5                      # 前序窗口，与 step48 一致
OUT = HERE / "out" / "step49_cluster_oos.feather"
# 发现期：step48 用的就是最近这段
IS_START = pd.Timestamp("2026-01-30", tz="Asia/Shanghai")


def collect(sym: str, rows: int):
    import warnings as _w
    _w.filterwarnings("ignore")
    sys.path.insert(0, str(HERE))
    sys.path.insert(0, str(HERE.parent))
    from step48_signal_timing import collect as _c
    return _c(sym, rows)


def split_stats(d: pd.DataFrame, label: str) -> pd.DataFrame:
    from lib.exit_model import fee_of, taker_notional

    d = d.sort_values("date").reset_index(drop=True)
    t = d.date.values.astype("datetime64[m]").astype(np.int64)
    # 只看过去：严格早于本笔、且在 WIN_MIN 分钟内。这是开仓时真正可知的信息。
    prev = np.searchsorted(t, t, "left") - np.searchsorted(t, t - WIN_MIN, "left")
    net = d.g.values - fee_of(d.r.values, d.c.values)
    gR = d.g.values / (SL * d.atr_pct.values)
    R = net / (SL * d.atr_pct.values)
    tn = taker_notional(d.r.values, d.c.values)

    rows = []
    for m, lab in ((prev == 0, "无前序"), (prev >= 1, "有前序")):
        if m.sum() < 15:
            rows.append({"时段": label, "分组": lab, "笔数": int(m.sum()), "备注": "样本不足"})
            continue
        nn, rr, gg, tt = net[m], R[m], gR[m], tn[m]
        w, o = nn[nn > 0].sum(), -nn[nn <= 0].sum()
        rows.append({
            "时段": label, "分组": lab, "笔数": int(m.sum()),
            "占比": f"{m.mean()*100:.0f}%",
            "胜率": f"{(nn > 0).mean()*100:.1f}%",
            "毛R": round(gg.mean(), 3), "净均R": round(rr.mean(), 3),
            "PF": round(w / o, 2) if o > 0 else np.inf,
            "余量bp": round(nn.mean() / tt.mean() * 1e4, 2),
        })
    return pd.DataFrame(rows)


def report(d: pd.DataFrame) -> None:
    print("\n" + "=" * 100)
    print("########## 逐时段：有前序 vs 无前序 ##########")
    tabs, deltas = [], []
    d["半年"] = d.date.dt.to_period("2Q").astype(str)
    periods = sorted(d["半年"].unique())
    for p in periods:
        g = d[d["半年"] == p]
        if len(g) < 60:
            continue
        t = split_stats(g, p)
        tabs.append(t)
        if len(t) == 2 and "余量bp" in t.columns and t["余量bp"].notna().all():
            a = t[t.分组 == "无前序"]["余量bp"].iloc[0]
            b = t[t.分组 == "有前序"]["余量bp"].iloc[0]
            ga = t[t.分组 == "无前序"]["毛R"].iloc[0]
            gb = t[t.分组 == "有前序"]["毛R"].iloc[0]
            deltas.append({"时段": p, "笔数": len(g),
                           "无前序余量": a, "有前序余量": b, "余量差": round(b - a, 2),
                           "无前序毛R": ga, "有前序毛R": gb, "毛R差": round(gb - ga, 3),
                           "方向": "✅ 同向" if b > a else "❌ 反向"})
    if tabs:
        print(pd.concat(tabs, ignore_index=True).to_string(index=False))
    if deltas:
        print("\n########## 判据：余量差是否稳定为正 ##########")
        dd = pd.DataFrame(deltas)
        print(dd.to_string(index=False))
        ok = (dd["余量差"] > 0).sum()
        print(f"\n{len(dd)} 个时段中 {ok} 个方向一致（有前序余量更高）")
        print(f"余量差 中位 {dd['余量差'].median():.2f}bp、均值 {dd['余量差'].mean():.2f}bp")
        print(f"毛R差 中位 {dd['毛R差'].median():.3f}、均值 {dd['毛R差'].mean():.3f}")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbols", default="BTC,BNB,ETH,SOL,LINK,LTC,AVAX,XRP,DOGE,ADA")
    ap.add_argument("--rows", type=int, default=800_000)
    ap.add_argument("--workers", type=int, default=3)
    ap.add_argument("--reuse", action="store_true")
    args = ap.parse_args()

    if args.reuse and OUT.exists():
        d = pd.read_feather(OUT)
    else:
        syms = [s.strip() for s in args.symbols.split(",")]
        print(f"[簇内前序 样本外验证] {len(syms)} 币 × {args.rows} 根 1m\n", flush=True)
        parts = []
        with ProcessPoolExecutor(max_workers=args.workers) as ex:
            fut = {ex.submit(collect, s, args.rows): s for s in syms}
            for i, f in enumerate(as_completed(fut), 1):
                r = f.result()
                print(f"  [{i}/{len(syms)}] {fut[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)
        d.to_feather(OUT)

    d["date"] = pd.to_datetime(d["date"])
    d = d[(d.htf == 1.0) & d.lad & (d.atr_bp >= GATE_BP)].copy()
    print(f"实盘口径 {len(d)} 笔，{d.date.min():%Y-%m-%d} ~ {d.date.max():%Y-%m-%d}")

    oos = d[d.date < IS_START]
    ins = d[d.date >= IS_START]
    print(f"  发现期（step48 用过）: {len(ins)} 笔")
    print(f"  样本外            : {len(oos)} 笔")

    print("\n" + "=" * 100)
    print("########## 样本外整体（发现期之前的全部数据）##########")
    if len(oos) > 60:
        print(split_stats(oos, "样本外").to_string(index=False))
    print("\n########## 发现期（复现 step48）##########")
    if len(ins) > 60:
        print(split_stats(ins, "发现期").to_string(index=False))
    report(d)


if __name__ == "__main__":
    main()