"""因果性回放：一类反手的信号在当时真的发得出来吗？ §3.395 的 PF 2.3~3.0 建立在全量数据一次算完的 `bsp_list` 上。但增量模块的 文件头自己写着「笔必须整表重扫：**最后一笔 is_sure 允许收回**」，§5.41 也记了 中枢右边缘会重画。若信号是事后才浮现的，那个 PF 就是幻觉。 **做法**：用 `init_stream` 预热，随后逐根 `append_bar`，每根之后重算 `cal_bi_zs_list_pure` + `find_all_bsp`，记录每个信号**第一次出现**在哪一根。 入场用那一根（的次根开盘），而不是事后的 `sure_time` —— 实盘只能这样。 必须逐根，不能分段重建：在第 t 根用 `data[0:t+S]` 重算等于多给了 S 根的信息， 测出来的因果性是假的。 **三个要看的量** 召回 全量算出的信号，有多少在回放中真的出现过（没出现的是事后才浮现） 幻影 回放中出现、但全量里没有的（当时发了、后来被重画掉） 代价 回放首现根 vs 全量 sure_time 的滞后；以及按首现根入场的实际 PF """ from __future__ import annotations import argparse 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") HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) sys.path.insert(0, str(HERE.parent)) OUT = HERE / "out" / "step59_replay.feather" SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48 WANT = {"B1": 1, "S1": -1} def sig_key(b) -> tuple | None: """信号身份用「类型 + 极值 KLC 的结束时刻」。 不能用 sure_time 当身份：它正是会被重画的字段，用它做键会把同一个信号 在不同根上算成两个。极值点稳定得多。 """ t = getattr(b.type, "name", str(b.type)) if t not in WANT: return None return (t, str(b.klc.end_time)) def replay(sym: str, tf: str, rows: int, warm: int, steps: int) -> pd.DataFrame | None: from chanlun import TF_DF from lib.data import fetch_ohlcv df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows) if df is None or len(df) < warm + steps + 100: print(f" {sym} {tf} 数据不足 {0 if df is None else len(df)}", flush=True) return None df = df.iloc[-(warm + steps):].reset_index(drop=True) # ---- 全量口径：一次算完，作为对照 ---- full = TF_DF(df, 1, tf, lean=False) fz = full.cal_bi_zs_list_pure(full.bi_list) full_sig = {} for b in (full.find_all_bsp(full.bi_list, fz) or []): k = sig_key(b) if k and b.sure_time is not None: full_sig[k] = str(b.sure_time) # ---- 回放口径：逐根追加，记录首现根 ---- chan = TF_DF(df.iloc[:warm].copy(), 1, tf, lean=False) chan.init_stream(df.iloc[:warm].copy(), 1, tf) first_seen: dict[tuple, int] = {} for i in range(warm, len(df)): chan.append_bar(df.iloc[i]) try: zs = chan.cal_bi_zs_list_pure(chan.bi_list) bsp = chan.find_all_bsp(chan.bi_list, zs) if zs else [] except Exception: # noqa: BLE001 continue for b in (bsp or []): k = sig_key(b) if k and k not in first_seen: first_seen[k] = i dser = pd.to_datetime(full.dataframe["date"]) if dser.dt.tz is not None: dser = dser.dt.tz_localize(None) didx = pd.DatetimeIndex(dser) def to_i(ts) -> int: t = pd.Timestamp(ts) return int(didx.searchsorted(t.tz_localize(None) if t.tz else t)) rec = [] for k in set(full_sig) | set(first_seen): t, ext_t = k i_seen = first_seen.get(k) # 只统计回放窗口内的：预热段的信号本来就不在考察范围 i_ext = to_i(ext_t) if i_ext < warm - 200: continue rec.append({ "sym": sym, "tf": tf, "type": t, "dir": WANT[t], "in_full": k in full_sig, "in_replay": i_seen is not None, "i_ext": i_ext, "i_seen": -1 if i_seen is None else i_seen, "i_sure_full": to_i(full_sig[k]) if k in full_sig else -1, }) r = pd.DataFrame(rec) if r.empty: return None # 按回放首现根入场，跑与实盘一致的出场 from lib.exit_model import cfg_name, walk_exits cdf = full.dataframe atr = cdf["atr"].to_numpy(float) cl = cdf["close"].to_numpy(float) live = r[r.in_replay & (r.i_seen < len(cdf) - 2)].copy() live = live[np.isfinite(atr[live.i_seen.values]) & (atr[live.i_seen.values] > 0)] if not live.empty: cfg = cfg_name(SL, RUNNER, MAXB, RSTOP) # 反手：§3.395 判定该反着做 t_ = pd.DataFrame({"entry_idx": live.i_seen.values, "direction": -live.dir.values}) res = walk_exits(cdf, t_, [SL], [RUNNER], [MAXB], scale_at=SCALE_AT, runners=(RUNNER,), runner_stops=(RSTOP,)) if len(res) == len(live): for c in ("g", "r", "c"): live[c] = res[f"{cfg}_{c}"].to_numpy() live["atr_pct"] = atr[live.i_seen.values] / cl[live.i_seen.values] r = r.merge(live[["i_ext", "type", "g", "r", "c", "atr_pct"]], on=["i_ext", "type"], how="left") return r def report(d: pd.DataFrame) -> None: from lib.exit_model import fee_of, taker_notional print("\n" + "=" * 92) print("########## 一、召回与幻影 ##########") rows = [] for (tf, t), x in d.groupby(["tf", "type"]): full = x[x.in_full] rep = x[x.in_replay] both = x[x.in_full & x.in_replay] rows.append({ "tf": tf, "类型": t, "全量信号": len(full), "回放信号": len(rep), "召回": f"{len(both)/max(len(full),1)*100:.1f}%", "事后才浮现": len(full) - len(both), "幻影(被重画掉)": len(rep) - len(both), }) print(pd.DataFrame(rows).to_string(index=False)) print("\n召回 = 全量算出的信号里，回放中真的出现过的比例。" "\n幻影 = 回放中发过、全量里却没有的 —— 实盘会照做，回测却看不见它。") print("\n" + "=" * 92) print("########## 二、时点代价：回放首现 vs 全量 sure_time ##########") b = d[d.in_full & d.in_replay].copy() b["delay"] = b.i_seen - b.i_sure_full for tf, x in b.groupby("tf"): q = x.delay.quantile([.25, .5, .75, .9]) print(f" {tf} 中位 {q[.5]:+.0f} 根 P25 {q[.25]:+.0f} " f"P75 {q[.75]:+.0f} P90 {q[.9]:+.0f} " f"｜ 早于或等于全量的占比 {(x.delay <= 0).mean()*100:.0f}%") print(" 正值 = 回放比全量晚知道，实盘要在更差的价位入场。") if "g" not in d.columns: return print("\n" + "=" * 92) print("########## 三、真正能落地的收益：按回放首现根入场（反手）##########") x = d.dropna(subset=["g"]).copy() rows = [] for (tf, t), g in x.groupby(["tf", "type"]): if len(g) < 30: continue net = g.g.values - fee_of(g.r.values, g.c.values) gR = g.g.values / (SL * g.atr_pct.values) R = net / (SL * g.atr_pct.values) tn = taker_notional(g.r.values, g.c.values) w, o = net[net > 0].sum(), -net[net <= 0].sum() rows.append({ "tf": tf, "类型(反手)": t, "笔数": len(g), "胜率": f"{(net > 0).mean()*100:.1f}%", "毛R": round(gR.mean(), 3), "净均R": round(R.mean(), 3), "PF": round(w / o, 2) if o > 0 else np.inf, "余量bp": round(net.mean() / tn.mean() * 1e4, 2), "t值": round(gR.mean() / (gR.std(ddof=1) / np.sqrt(len(g))), 2), }) if rows: print(pd.DataFrame(rows).to_string(index=False)) print("\n对照 · §3.395 全量口径 5m：B1反手 PF 2.35 / S1反手 2.75") print("若这里明显掉下来，说明那个 PF 吃了右边缘重画的红利，不可落地。") def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--symbols", default="BTC,ETH,SOL") ap.add_argument("--tf", default="5m") ap.add_argument("--rows", type=int, default=60_000) ap.add_argument("--warm", type=int, default=20_000) ap.add_argument("--steps", type=int, default=10_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(): report(pd.read_feather(OUT)) return syms = [x.strip() for x in args.symbols.split(",")] print(f"[因果回放] {len(syms)} 币 × {args.steps} 根逐根重放" f"（每根都要重算笔中枢与 bsp，慢是必然的）\n", flush=True) parts = [] with ProcessPoolExecutor(max_workers=args.workers) as ex: fut = {ex.submit(replay, s, args.tf, args.rows, args.warm, args.steps): s for s in syms} for i, f in enumerate(as_completed(fut), 1): r = f.result() print(f" [{i}/{len(syms)}] {fut[f]} " 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) OUT.parent.mkdir(exist_ok=True) d.to_feather(OUT) report(d) if __name__ == "__main__": main()