Chan/research/step35_audit.py

"""Step 35：禁用回抽跳过后跑出 Sharpe 9 / 年化 906%，先当作可疑数字来审。

这个量级通常意味着某个假设塌了，本步专查三件事：

  1 持仓重叠：滞后压到 2.2 根、年 334 笔后，同时在场的仓位是否变多。
    若重叠严重，t=19 是虚高的（收益不独立），且实盘保证金撑不住。
  2 滑点脆弱性：入场越早越接近突破根，成交价优势可能全靠那 2 根。
    逐档加 5/10/20/30bp，看优势是否被吃光。
  3 复利假设：年化 906% 是「每笔按 1% 风险、20 倍杠杆上限、连续复利」的产物。
    改成固定名义仓位、不复利，看真实量级。

另外做一次未来函数的独立复核：把入场索引整体后移 1/2/3 根，
若收益随后移平滑衰减而非断崖归零，说明信号真实且优势来自时间价值。
"""
from __future__ import annotations

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, TP, MAXB = 1.5, 3.0, 48
FEE = 0.0004
BEST = {"5m": "30m", "15m": "1h", "30m": "2h"}
SLIPS = [0.0, 0.0005, 0.0010, 0.0020, 0.0030]
DELAYS = [1, 2, 3, 4]


def run_one(task: tuple) -> dict | None:
    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 lib.breakout import run_trades
    from lib.data import fetch_ohlcv
    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, ltf = task
    pair = f"{sym}/USDT:USDT"
    try:
        df_l = fetch_ohlcv(pair, ltf, 10**9)
        if df_l is None or len(df_l) < 3000:
            return None
        chan_l = TF_DF(df_l, 1, ltf)
        cdf = chan_l.dataframe
        zones = build_htf_zones(cdf, ltf, chan=chan_l).reset_index(drop=True)
        if zones.empty:
            return None
        z = zones.copy()
        pg, pdn = z["zg"].shift(), z["zd"].shift()
        z["z_above"], z["z_below"] = z["zd"] > pg, z["zg"] < pdn
        z["zone_i"] = np.arange(len(z))

        df_h = fetch_ohlcv(pair, BEST[ltf], 10**9)
        if df_h is None or len(df_h) < 300:
            return None
        chan_h = TF_DF(df_h, 1, BEST[ltf])
        s = signals_to_frame(extract_fx_signals(chan_h, chan_h.dataframe))
        tl = htf_fx_timeline(s, chan_h.dataframe)

        sig = find_fast_bsp3(cdf, zones, tol=-1.0)   # 禁用回抽跳过
        if sig.empty or len(sig) < 20:
            return None
        sig = sig.merge(z[["zone_i", "z_above", "z_below"]], on="zone_i", how="left")
        sig = attach_htf_context(sig, cdf, tl, "h1")
        sig["push"] = np.where(sig["direction"] == 1, sig["z_above"], sig["z_below"])
        sig = sig[(sig["h1_agree"] == 1) & sig["push"].fillna(False).astype(bool)]
        if len(sig) < 20:
            return None

        entries = list(zip(sig["entry_idx"].astype(int), sig["direction"].astype(int)))
        bars = int(np.median(np.diff(cdf["timestamp"].to_numpy())))
        out = []
        for delay in DELAYS:
            for slip in SLIPS:
                if delay != 1 and slip != 0.0:
                    continue           # 后移只在零滑点下扫，避免组合爆炸
                tr = run_trades(cdf, entries, SL, TP, MAXB, fee=0.0,
                                entry_delay=delay, slippage=slip)
                if tr.empty:
                    continue
                tr["delay"], tr["slip"] = delay, slip
                tr["symbol"], tr["ltf"] = sym, ltf
                tr["date"] = cdf["date"].to_numpy()[tr["entry_idx"].to_numpy()]
                tr["bar_ms"] = bars
                out.append(tr)
        if not out:
            return None
        return {"task": f"{sym} {ltf}", "trades": pd.concat(out, ignore_index=True)}
    except Exception as e:
        return {"task": f"{sym} {ltf}", "error": repr(e)[:250]}


def stat(g: pd.DataFrame, label: str) -> dict:
    r = g["ret"].to_numpy()
    if len(r) < 25:
        return {}
    w, o = r[r > 0], r[r <= 0]
    sd = r.std(ddof=1)
    t10 = r[r <= np.quantile(r, 0.90)]
    return {"口径": label, "笔数": len(r),
            "胜率": f"{(r > 0).mean() * 100:.1f}%",
            "中位": f"{np.median(r) * 100:+.3f}%",
            "PF": f"{w.sum() / abs(o.sum()):.2f}" if len(o) else "inf",
            "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}",
            "剔10%PF": f"{t10[t10 > 0].sum() / abs(t10[t10 <= 0].sum()):.2f}"}


def overlap(g: pd.DataFrame) -> dict:
    """按真实时间轴统计同时在场仓位数。"""
    g = g.sort_values("date")
    bar = g["bar_ms"].iloc[0] / 1000 / 60          # 每根多少分钟
    st = pd.to_datetime(g["date"]).astype("int64") // 10**9 / 60
    en = st + g["bars_held"].to_numpy() * bar
    ev = np.concatenate([st.to_numpy(), en])
    kind = np.concatenate([np.ones(len(g)), -np.ones(len(g))])
    o = np.argsort(ev, kind="stable")
    live = np.cumsum(kind[o])
    # 按持续时长加权的平均并发
    dur = np.diff(np.concatenate([ev[o], ev[o][-1:]]))
    wt = dur[:len(live)]
    return {"笔数": len(g), "峰值并发": int(live.max()),
            "时长加权平均并发": round(float((live * wt).sum() / max(wt.sum(), 1e-9)), 2),
            "有仓位时间占比": f"{(wt[live > 0].sum() / max(wt.sum(), 1e-9)) * 100:.1f}%"}


def main() -> None:
    tasks = [(s, l) for l in BEST for s in ("BTC", "ETH", "SOL")]
    print(f"[审计] {len(tasks)} 个任务\n", flush=True)
    res = []
    with ProcessPoolExecutor(max_workers=5) as ex:
        futs = {ex.submit(run_one, t): t for t in tasks}
        for i, f in enumerate(as_completed(futs), 1):
            r = f.result()
            if r is None or "error" in (r or {}):
                print(f"  [{i}] 跳过 {(r or {}).get('error', '')}", flush=True)
                continue
            res.append(r)
            print(f"  [{i}/{len(tasks)}] {r['task']}", flush=True)
    if not res:
        return
    allt = pd.concat([r["trades"] for r in res], ignore_index=True)
    allt["date"] = pd.to_datetime(allt["date"])
    allt.to_csv(HERE / "out" / "step35_audit.csv", index=False)
    base = allt[(allt["delay"] == 1) & (allt["slip"] == 0.0)]

    print("=" * 110)
    print("########## 1. 滑点脆弱性：优势是否全靠提前那两根 ##########")
    rows = []
    for slip in SLIPS:
        g = allt[(allt["delay"] == 1) & (allt["slip"] == slip)]
        rows.append(stat(g, f"次根开盘 +{slip * 1e4:.0f}bp"))
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))
    print("  含 4bp 手续费。若加到 20~30bp 仍显著为正，说明不是靠抢那两根。")

    print("\n########## 2. 入场整体后移：真信号应平滑衰减，未来函数会断崖 ##########")
    rows = []
    for d in DELAYS:
        g = allt[(allt["delay"] == d) & (allt["slip"] == 0.0)]
        rows.append(stat(g, f"信号后 {d} 根开盘入场"))
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 3. 持仓重叠：t 值是否被虚高 ##########")
    rows = []
    for (sym, ltf), g in base.groupby(["symbol", "ltf"]):
        rows.append({"品种": sym, "级别": ltf, **overlap(g)})
    print(pd.DataFrame(rows).to_string(index=False))
    print("  单品种单级别若平均并发接近 1，说明各笔基本独立，t 值可信。")

    print("\n########## 4. 组合层面并发（三品种三级别一起做）##########")
    print(pd.DataFrame([{"全组合": "9 条腿", **overlap(base)}]).to_string(index=False))

    print("\n########## 5. 剥掉复利与杠杆假设，看真实量级 ##########")
    g = base.sort_values("date")
    r = g["ret"].to_numpy()
    yrs = (g["date"].max() - g["date"].min()).days / 365.25
    lev = np.clip(0.01 / np.clip(g["risk_pct"].to_numpy(), 0.002, None), 0, 20)
    for name, pnl, comp in (
        ("A 1%风险+20倍上限+复利", r * lev, True),
        ("B 1%风险+20倍上限+不复利", r * lev, False),
        ("C 固定名义1倍+不复利", r, False),
        ("D 固定名义3倍+不复利", r * 3, False),
    ):
        if comp:
            eq = np.cumprod(1 + pnl)
            ann = (eq[-1] ** (1 / yrs) - 1) * 100
            dd = (1 - eq / np.maximum.accumulate(eq)).max() * 100
        else:
            eq = 1 + np.cumsum(pnl)
            ann = (eq[-1] - 1) / yrs * 100
            dd = (np.maximum.accumulate(eq) - eq).max() * 100
        sh = pnl.mean() / pnl.std(ddof=1) * np.sqrt(len(pnl) / yrs)
        print(f"  {name:<26}: 年化 {ann:+8.1f}%  回撤 {dd:5.1f}%  Sharpe {sh:5.2f}")
    print(f"  平均单笔杠杆 {lev.mean():.1f}x（上限20），中位 {np.median(lev):.1f}x")
    print("  未含资金费率、未含同时持仓的保证金约束，C/D 才是可直接对照的量级。")


if __name__ == "__main__":
    main()