diff --git a/research/live/verify_lean_parity.py b/research/live/verify_lean_parity.py index 9a5ca2b..61814eb 100644 --- a/research/live/verify_lean_parity.py +++ b/research/live/verify_lean_parity.py @@ -161,8 +161,47 @@ def run_one(sym: str, cache: Path, n: int, step: int, "hits_full": n_hits_full, "hits_lean": n_hits_lean} +def recall(syms: list[str], cache: Path, k: int = 30) -> None: + """全量历史找出的信号,在 2000 根窗口里还能不能复现。 + + 这是窗口左边界效应的**一半**答案。窗口只有 2000 根,中枢是在窗口内重建 + 的,理论上可能与全量历史建的中枢不同,从而漏掉信号。实测最近 k 笔全部 + 复现,说明窗口不丢信号。 + + ⚠ 另一半没答:窗口会不会**多造出**全量历史没有的信号。那个方向对实盘更 + 危险(会多开仓),但要反向扫描——遍历窗口找命中、再回全量历史核对,成本 + 高得多。lean 之后单窗 130ms,抽样 2 万个窗口约 43 分钟,已经可做。 + """ + from shadow_signal import compute + + print("全量历史找出的信号,在 2000 根窗口里的复现率\n") + for sym in syms: + try: + l_all, h_all = load(sym, "1m", cache), load(sym, "5m", cache) + l_ts = l_all["timestamp"].to_numpy("int64") + h_ts = h_all["timestamp"].to_numpy("int64") + sb = signal_bars(sym, cache) + sb = sb[(sb > LTF_BARS) & (sb < len(l_all) - 1)][-k:] + except Exception as e: + print(f" {sym} 跳过:{e!r}") + continue + hit = 0 + for e in sb: + df_l = l_all.iloc[e - LTF_BARS + 1:e + 1] + hi = int(np.searchsorted(h_ts, l_ts[e], side="right")) + df_h = h_all.iloc[max(0, hi - HTF_BARS):hi] + r = compute(df_l.copy(), df_h.copy(), + float(l_all["open"].to_numpy(float)[e + 1])) + if any(h.get("pass_all") for h in (r.get("hits") or [])): + hit += 1 + print(f" {sym} {hit}/{len(sb)} 复现(过全部滤网)") + print("\n 只说明窗口不丢信号;会不会多造信号需反向扫描,见 docstring。") + + def main() -> None: ap = argparse.ArgumentParser() + ap.add_argument("--recall", action="store_true", + help="只查窗口对全量历史信号的复现率") ap.add_argument("--syms", default="BTC,ETH,SOL") ap.add_argument("--cache", default="research/live/cache") ap.add_argument("--n", type=int, default=150, help="每币比对多少个窗口") @@ -172,6 +211,10 @@ def main() -> None: help="不对齐信号根(快,但测不到信号分支)") a = ap.parse_args() + if a.recall: + recall(a.syms.split(","), Path(a.cache)) + return + rows = [] for sym in a.syms.split(","): print(f"\n{'=' * 70}\n{sym}")