Chan/research/live/aggregate_robustness.py

"""最关键的一步：逐笔清单不可复现，总体期望还在不在。

signal_sensitivity 证明 0.25bp 的数据扰动就能换掉一半信号。这本身不判死刑——
趋势跟随策略允许「成交的具体是哪几笔」随机，只要总体期望稳定就仍可交易。
但如果扰动后 PF 与毛均收益也跟着塌，那回测测的就是噪声。

两种结果对应完全不同的下一步：
  总体稳定  -> 改成「Bitget 原生信号测滑点 + Bitget 原生回测基线」，主线继续
  总体也塌  -> 滑点根本不是瓶颈，1m 腿的问题在信号定义本身，影子交易器白写

口径与 step23 一致：SL/TP/MAX_BARS = 1.5/3.0/48，ATR 取信号根，
入场为信号次根开盘价（entry_delay=1），成本 4bp 手续费 + 1bp 滑点。
3.91bp 的滑点预算就是从「毛均收益 +0.0991%」推出来的，所以毛均收益是主看指标。

输出 out/aggregate_robustness.csv。
"""
from __future__ import annotations

import argparse
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().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
sys.path.insert(0, str(HERE))
pd.set_option("display.width", 260)

from signal_sensitivity import TICK, perturb  # noqa: E402

SYMS = ("BTC", "ETH", "SOL")
LEVELS = (0.0, 0.5, 1.0, 2.0)
SL, TP, MAX_BARS = 1.5, 3.0, 48
FEE, SLIP = 0.0004, 0.0001


def run_once(df_l: pd.DataFrame, df_h: pd.DataFrame) -> pd.DataFrame:
    """跑完整管线并逐笔模拟，返回交易表。"""
    from chanlun import TF_DF
    from lib.breakout import run_trades
    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

    chan_l = TF_DF(df_l, 1, "1m")
    cdf = chan_l.dataframe
    zones = build_htf_zones(cdf, "1m", chan=chan_l)
    if zones.empty:
        return pd.DataFrame()
    sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
    if sig.empty:
        return pd.DataFrame()

    chan_h = TF_DF(df_h, 1, "5m")
    hdf = chan_h.dataframe
    tl = htf_fx_timeline(signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
    full = attach_htf_context(sig, cdf, tl, "h1")
    fin = full[full["h1_agree"] == 1]
    if fin.empty:
        return pd.DataFrame()

    entries = list(zip(fin["entry_idx"].astype(int), fin["direction"].astype(int)))
    return run_trades(cdf, entries, SL, TP, MAX_BARS,
                      fee=FEE + SLIP, entry_delay=1)


def stats(tr: pd.DataFrame) -> dict:
    if tr.empty:
        return {"笔数": 0}
    g = tr["gross"].to_numpy(dtype=float)
    n = tr["ret"].to_numpy(dtype=float)
    win, loss = n[n > 0].sum(), -n[n < 0].sum()
    return {
        "笔数": len(tr),
        "胜率": f"{(n > 0).mean() * 100:.1f}%",
        "毛均收益": f"{g.mean() * 100:+.4f}%",
        "净均收益": f"{n.mean() * 100:+.4f}%",
        "PF": round(win / loss, 2) if loss > 0 else np.inf,
        "t值": round(n.mean() / n.std(ddof=1) * np.sqrt(len(n)), 2) if len(n) > 1 else np.nan,
        "滑点余量bp": round(g.mean() * 1e4 - 6.0, 2),
    }


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbols", default="BTC,ETH,SOL")
    ap.add_argument("--bars", type=int, default=200_000, help="每币用多少根 1m")
    ap.add_argument("--seeds", type=int, default=2)
    ap.add_argument("--levels", default=None,
                    help="逗号分隔的噪声档（bp）；只给 0 就是纯基线复现")
    ap.add_argument("--tag", default="", help="产物文件名后缀")
    args = ap.parse_args()

    levels = (tuple(float(x) for x in args.levels.split(","))
              if args.levels else LEVELS)

    from lib.data import load_local

    syms = [s.strip() for s in args.symbols.split(",")]
    print(f"[总体稳健性] {syms} · 每币 {args.bars} 根 1m "
          f"({args.bars / 1440:.0f} 天) · 噪声档 {levels} bp\n", flush=True)

    rows = []
    for sym in syms:
        df_l = load_local(f"{sym}/USDT:USDT", "1m")
        df_h = load_local(f"{sym}/USDT:USDT", "5m")
        if df_l is None or df_h is None:
            print(f"{sym}: 本地无数据，跳过")
            continue
        df_l = df_l.tail(args.bars).reset_index(drop=True)
        lo = int(df_l["timestamp"].iloc[0])
        df_h = df_h[df_h.timestamp >= lo].reset_index(drop=True)
        print(f"── {sym}  {len(df_l)} 根 1m / {len(df_h)} 根 5m  "
              f"{df_l['date'].iloc[0]:%Y-%m-%d} ~ {df_l['date'].iloc[-1]:%Y-%m-%d}",
              flush=True)

        for bp in levels:
            for k in range(1 if bp == 0 else args.seeds):
                t0 = time.perf_counter()
                tr = run_once(perturb(df_l, bp, TICK[sym], 2000 + k), df_h)
                s = stats(tr)
                rows.append({"品种": sym, "噪声bp": bp, "种子": k, **s})
                print(f"   噪声 {bp:>4.2f}bp 种子{k}: " +
                      " · ".join(f"{k2} {v}" for k2, v in s.items()) +
                      f"  [{time.perf_counter() - t0:.0f}s]", flush=True)

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

    tb = pd.DataFrame(rows)
    print("\n" + "=" * 130)
    print("########## 1. 逐币 × 噪声档 ##########")
    print(tb.to_string(index=False))

    print("\n########## 2. 三币合并（同噪声档取均值）##########")
    num = tb.copy()
    num["毛均bp"] = num["毛均收益"].str.rstrip("%").astype(float) * 100
    num["净均bp"] = num["净均收益"].str.rstrip("%").astype(float) * 100
    num["胜率_"] = num["胜率"].str.rstrip("%").astype(float)
    agg = num.groupby("噪声bp").agg(
        笔数=("笔数", "mean"), 胜率=("胜率_", "mean"),
        毛均bp=("毛均bp", "mean"), 净均bp=("净均bp", "mean"),
        PF=("PF", "mean"), t值=("t值", "mean")).round(2)
    agg["滑点余量bp"] = (agg["毛均bp"] - 6.0).round(2)
    print(agg.to_string())

    print("\n########## 结论 ##########")
    print("  step23 的 1m 基线（3 币 3578 笔 / 2.28 年）：毛均 9.91bp · PF 2.31 · "
          "t 19.92 · 余量 3.91bp")
    if 0.0 not in agg.index:
        print("  本次未跑无噪声档，无法给出相对基线的比例")
        return
    base = agg.loc[0.0]
    print(f"  无噪声基线：毛均 {base['毛均bp']:.2f}bp · PF {base['PF']:.2f} · "
          f"t {base['t值']:.2f} · 滑点余量 {base['滑点余量bp']:.2f}bp")
    for bp in levels[1:]:
        if bp not in agg.index:
            continue
        r = agg.loc[bp]
        print(f"  噪声 {bp}bp：毛均 {r['毛均bp']:.2f}bp "
              f"({r['毛均bp'] / base['毛均bp'] * 100:.0f}% of 基线) · "
              f"PF {r['PF']:.2f} · t {r['t值']:.2f} · 余量 {r['滑点余量bp']:.2f}bp")
    print("\n  毛均与 PF 若基本持平 → 逐笔身份随机但总体期望稳定，主线继续，")
    print("  但必须换成 Bitget 原生回测基线，Binance 的逐笔清单不可用于对照。")
    print("  若毛均随噪声单调下滑 → 回测吃的是数据噪声，滑点不是瓶颈。")

    out = RESEARCH / "out" / f"aggregate_robustness{args.tag}.csv"
    tb.to_csv(out, index=False)
    print(f"\n产物写入 {out}")


if __name__ == "__main__":
    main()