"""Step 43:重画率——实时看到过的信号,有多少后来消失了。 step39 也报过「假阳性」,但那个测法是无效的:它从全序列随机抽非信号点, 看时点重建会不会凭空冒信号。1m 上信号密度约 474 根 1 个,180 个随机点里 本来就只期望撞上 0.38 个,测出 0% 几乎不含信息量。 重画不发生在随机点上,只发生在「差一点就成型」的结构附近。所以要 **逐根**做时点重建,把「当根确实出现了信号」的位置全收集起来, 再看它们在全量视角里还在不在。 实时信号 窗口只喂到第 T 根,重建后信号恰好落在第 T 根(实盘会下单的那些) 重画 该信号在全量重建里不存在(图上后来消失,但实盘已经开了仓) 漏看 全量有、实时当根没有(step39 已验证 ~0,这里顺带复核) 只有**深色信号**(h1_agree ∧ ladder_ok)才会真下单,所以分层报告: 浅色重画无所谓,深色重画才影响实盘。 """ from __future__ import annotations import argparse import os import sys import time 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) LTF, HTF = "1m", "5m" WINDOW = 2000 # step39 证明 1m 在 2000 根就饱和,实盘也用这个 HTF_WINDOW = 800 # 5m 侧窗口,同 step39 MAX_ROWS = 1_200_000 def _load(sym: str): from chanlun import TF_DF from lib.data import fetch_ohlcv from lib.fx_signal import extract_fx_signals, signals_to_frame from lib.nested_bsp import htf_fx_timeline pair = f"{sym}/USDT:USDT" df_l = fetch_ohlcv(pair, LTF, MAX_ROWS) df_h = fetch_ohlcv(pair, HTF, 10 ** 9) chan_l = TF_DF(df_l, 1, LTF) cdf = chan_l.dataframe chan_h = TF_DF(df_h, 1, HTF) hdf = chan_h.dataframe tl = htf_fx_timeline(signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf) return chan_l, cdf, hdf, tl def _full_signals(chan_l, cdf, tl): """全量视角的信号集(带两个过滤器)。""" from chanlun.analysis.fast_bsp import attach_zone_ladder from lib.fast_bsp3 import find_fast_bsp3 from lib.nested_bsp import attach_htf_context from lib.nested_level import build_htf_zones zones = build_htf_zones(cdf, LTF, chan=chan_l).reset_index(drop=True) if zones.empty: return pd.DataFrame() sig = find_fast_bsp3(cdf, zones) if sig.empty: return pd.DataFrame() sig = attach_htf_context(sig, cdf, tl, "h1") sig = attach_zone_ladder(sig, zones) sig["ts"] = cdf["timestamp"].to_numpy()[sig["entry_idx"].astype(int)] return sig def scan_chunk(task: tuple) -> dict: """逐根时点重建,只记录信号恰好落在当根的位置。""" 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 chanlun.analysis.fast_bsp import attach_zone_ladder 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, lo, hi = task try: chan_l, cdf, hdf, _tl_full = _load(sym) ltf_ts = cdf["timestamp"].to_numpy() htf_ts = hdf["timestamp"].to_numpy() hi = min(hi, len(cdf)) lo = max(lo, WINDOW) hits, t0 = [], time.perf_counter() for T in range(lo, hi): sl = cdf.iloc[T - WINDOW + 1: T + 1].reset_index(drop=True) z = build_htf_zones(sl, LTF) if z.empty: continue z = z.reset_index(drop=True) sig = find_fast_bsp3(sl, z) if sig.empty: continue last = len(sl) - 1 row = sig[sig["entry_idx"].astype(int) == last] if row.empty: continue # 只在真的出信号时才算过滤器——信号稀疏,这部分开销可忽略 h_end = int(np.searchsorted(htf_ts, ltf_ts[T], side="right")) agree = np.nan if h_end >= HTF_WINDOW: hsl = hdf.iloc[h_end - HTF_WINDOW: h_end].reset_index(drop=True) ch = TF_DF(hsl, 1, HTF) tl_p = htf_fx_timeline( signals_to_frame(extract_fx_signals(ch, ch.dataframe)), ch.dataframe) got = attach_htf_context(row.copy(), sl, tl_p, "h1") agree = float(got["h1_agree"].iloc[0] == 1) lad = attach_zone_ladder(row.copy(), z) hits.append({ "sym": sym, "T": int(T), "ts": int(ltf_ts[T]), "direction": int(row["direction"].iloc[0]), "h1_agree": agree, "ladder_ok": bool(lad["ladder_ok"].iloc[0]), }) return {"sym": sym, "lo": lo, "hi": hi, "n_bars": hi - lo, "hits": pd.DataFrame(hits), "secs": time.perf_counter() - t0} except Exception as e: return {"sym": sym, "error": repr(e)[:300]} def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--symbols", default="BTC,ETH,SOL,DOGE,XRP,LINK") ap.add_argument("--bars", type=int, default=30000, help="每币逐根扫描的根数") ap.add_argument("--workers", type=int, default=10) ap.add_argument("--chunk", type=int, default=2500) args = ap.parse_args() syms = [s.strip() for s in args.symbols.split(",")] out_dir = HERE / "out" out_dir.mkdir(exist_ok=True) # 先拿各币全量信号做基准,同时确定扫描区间 print(f"[重画审计] {len(syms)} 币 × 每币 {args.bars} 根逐根重建," f"窗口 {WINDOW}\n", flush=True) full_map, tasks = {}, [] for s in syms: try: chan_l, cdf, hdf, tl = _load(s) full = _full_signals(chan_l, cdf, tl) full_map[s] = full hi = len(cdf) - 1 lo = max(WINDOW, hi - args.bars) for a in range(lo, hi, args.chunk): tasks.append((s, a, min(a + args.chunk, hi))) print(f" {s}: 全量 {len(cdf)} 根,全量信号 {len(full)}," f"扫描 [{lo}, {hi})", flush=True) except Exception as e: print(f" {s}: 载入失败 {e!r}", flush=True) est = sum(t[2] - t[1] for t in tasks) * 0.202 / args.workers print(f"\n共 {len(tasks)} 个分块,预计 {est / 60:.0f} 分钟\n", flush=True) res, done = [], 0 with ProcessPoolExecutor(max_workers=args.workers) as ex: futs = {ex.submit(scan_chunk, t): t for t in tasks} for f in as_completed(futs): r = f.result() done += 1 if "error" in r: print(f" [{done}/{len(tasks)}] {r['sym']} 出错 {r['error']}", flush=True) continue res.append(r) print(f" [{done}/{len(tasks)}] {r['sym']} [{r['lo']},{r['hi']}) " f"实时信号 {len(r['hits'])} 个,{r['secs']:.0f}s", flush=True) if not res: print("无结果") return live = pd.concat([r["hits"] for r in res if len(r["hits"])], ignore_index=True) live.to_feather(out_dir / "step43_live_signals.feather") n_bars = sum(r["n_bars"] for r in res) # 对齐:实时信号的时间戳是否出现在全量信号集里 rows = [] for s, g in live.groupby("sym"): full = full_map.get(s) fts = set(full["ts"].astype(int).tolist()) if full is not None and len(full) else set() g = g.copy() g["survived"] = g["ts"].isin(fts) rows.append(g) live = pd.concat(rows, ignore_index=True) live["dark"] = (live["h1_agree"] == 1.0) & live["ladder_ok"] print("\n" + "=" * 100) print(f"########## 1. 总体(扫描 {n_bars} 根)##########") n, sv = len(live), int(live["survived"].sum()) print(f" 实时出现过的信号 {n} 个,全量视角仍在 {sv} 个," f"重画 {n - sv} 个 = {(n - sv) / max(n, 1) * 100:.2f}%") print("\n########## 2. 按过滤器分层(只有深色会真下单)##########") rows = [] for lab, m in (("深色(双过滤通过)", live["dark"]), ("浅色(未通过)", ~live["dark"])): gg = live[m] if not len(gg): continue k = int((~gg["survived"]).sum()) # Wilson 95% 上界,样本小的时候点估计没意义 from math import sqrt nn, p = len(gg), k / len(gg) z = 1.96 hi_b = (p + z * z / (2 * nn) + z * sqrt(p * (1 - p) / nn + z * z / (4 * nn * nn))) / (1 + z * z / nn) rows.append({"分层": lab, "实时信号": nn, "重画": k, "重画率": f"{p * 100:.2f}%", "95%上界": f"{hi_b * 100:.2f}%"}) print(pd.DataFrame(rows).to_string(index=False)) print("\n########## 3. 分币种 ##########") rows = [] for s, g in live.groupby("sym"): d = g[g["dark"]] rows.append({"币": s, "实时信号": len(g), "其中深色": len(d), "深色重画": int((~d["survived"]).sum()) if len(d) else 0, "全部重画": int((~g["survived"]).sum())}) print(pd.DataFrame(rows).to_string(index=False)) print("\n########## 4. 漏看(全量有、实时当根没有)复核 ##########") for s, g in live.groupby("sym"): full = full_map.get(s) if full is None or not len(full): continue lo = min(r["lo"] for r in res if r["sym"] == s) hi = max(r["hi"] for r in res if r["sym"] == s) inrange = full[(full["entry_idx"] >= lo) & (full["entry_idx"] < hi)] seen = set(g["ts"].astype(int).tolist()) miss = int((~inrange["ts"].astype(int).isin(seen)).sum()) print(f" {s}: 扫描区间内全量信号 {len(inrange)},实时当根未出现 {miss} 个" f"({miss / max(len(inrange), 1) * 100:.1f}%)") print("\n########## 结论 ##########") d = live[live["dark"]] if len(d): k = int((~d["survived"]).sum()) print(f" 深色信号 {len(d)} 个,重画 {k} 个。") print(" 重画的仓位是真实成交的,但出场(止损/止盈/超时)不依赖信号是否还在图上,") print(" 所以不会卡仓;影响仅限于「实盘比回测多开的这部分,质量不在回测统计里」。") if __name__ == "__main__": main()