"""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()