diff --git a/research/HANDOFF.md b/research/HANDOFF.md index a26ec68..5149f20 100644 --- a/research/HANDOFF.md +++ b/research/HANDOFF.md @@ -1531,6 +1531,61 @@ MFE/MAE 给出机制：真底那批**逆向行程中位 2.24 ATR（5m）/ 2.08 > **这解释了为什么 B4 能做而一类不能**：B4 是突破后的**延续**信号， > 不需要判断反转；一类的全部难度集中在「这是不是那个底」这一个二分类上。 +### 3.3994 🚨 B4 因果回放：时点干净，但实盘做的不是回测那批单（step63/68） + +**这条影响正在跑的实盘，优先级高于本文档其余全部内容。** + +**好消息：B4 的时点完全干净，一类那个坑它没有** + +| | B1（§3.396） | **B4** | +|---|---|---| +| 召回 | 100% | 100% | +| **准时率（首现==entry_idx）** | **0%** | **100%** | +| 迟到中位 | +15 根 | **0 根** | + +**坏消息：实时回放会多产出一批全量口径里不存在的信号，且它们显著亏钱** + +按**实盘口径**（2001 根滚动窗口、每 500 根 `init_stream` 重建、只做当根收盘， +逐行对齐 `shadow_signal.py`）重放 5 币 × 20000 根： + +| 过滤 | 留下 | **额外占比** | 全部 PF | 额外 PF | 回测口径 PF | +|---|---|---|---|---|---| +| 无过滤 | 789 | 75% | 0.64 | 0.46 | 1.37 | +| 仅同向 | 402 | 77% | 0.65 | 0.35 | 2.65 | +| 仅阶梯 | 341 | 68% | 0.71 | 0.31 | 2.14 | +| 仅 ATR 门控 | 721 | 76% | 0.64 | 0.46 | 1.35 | +| **三道全开** | **195** | **74%** | **0.73** | **0.26** | **4.21** | + +**三道滤网挡不住这批信号。** 它们把成交量砍掉 75%（789 → 195），但额外信号占比 +从 75% 只降到 74%——按同样比例刷掉好的和坏的。实盘口径拆开： + +| 分组 | 笔数 | 胜率 | PF | 余量bp | t值 | +|---|---|---|---|---|---| +| 回测里也有的 | 51 | 74.5% | **4.21** | +30.15 | +5.90 | +| **回测里没有的** | **144** | 22.9% | **0.26** | −14.25 | **−6.21** | +| **实盘实际会做的** | 195 | 36.4% | **0.73** | −4.65 | −0.59 | + +**结论：回测报的 4.21 是拿不到的。** 那 51 笔只有事后用全量历史重算才能识别； +实盘在当下拿到 195 个信号，**没有任何办法分辨哪 51 个是真的**。机制是重画—— +实时算出的中枢，数据变多后被修正掉，信号随之消失。 + +这与 §3.396 是同一类错误的两种表现：一类错在**时点**（信号存在但晚 15 根）， +B4 错在**存在性**（信号当根就有，但其中 74% 事后会消失）。 + +**顺带闭掉一条挂了很久的待办：窗口长度不是问题。** 2001 根滚动窗口与 +2 万→4 万根增长窗口跑出**完全相同**的 789/197/592，中枢是局部结构，截断无影响。 +§5.41 那条「9.0% 中枢消失是不是窗口左边界效应」可以判为**否**。 + +⚠️ **两点限定**： +- 本轮测的是 **5m/30m**，实盘跑的是 **1m**。机制相同但数字必须单独验， + 这是下一步第一件事 +- 同向过滤器被我**开了未来函数的后门**（大级别分型时间线用全量历史算）， + 真实盘的 HTF 同样会重画，只会**更差**不会更好 + +**这不否定 B4 的 alpha**——回测口径那 51 笔 PF 4.21、t 5.90 是真实的，说明 +「真中枢上的 B4」确实赚钱。问题是实时分不出真假中枢。可能的方向（均未测）： +要求中枢的构成笔 `is_sure`（用滞后换稳定性）、或等 N 根看信号是否仍然存在。 + ### 3.4 alpha 的来源（step32 消融） 逐条拆掉 `fast_bsp3` 的条件后发现：**alpha 完全来自缠论中枢的上下文定位， @@ -2738,10 +2793,18 @@ API 限流风险隔离三个好处。 ## 10. 待办清单 -- [ ] ⭐ **B4 因果回放（step63，跑着）——当前唯一的关键未决项**。 - §3.396 证明 `sure_time` 对一类是后视产物（PF 2.35 → 0.92）。B4 用的是 - 同一个 `sure_time` 字段，**在这个回放出结果前，所有 B4 的历史 PF 都待定**。 - 若 B4 也栽，整条线要重估；若过了，B4 是目前唯一活着的信号 +- [ ] 🚨 **B4 实盘口径 PF 是 0.73，回测报 4.21 —— 最高优先级**（§3.3994）。 + 时点干净（100% 准时，一类那个坑没有），但实时会多产出 74% 的信号， + 它们 PF 0.26 / t −6.21，**三道滤网按同比例刷除、完全挡不住**。 + 回测那 51 笔要事后全量重算才能识别，实盘当下分不出来。**要做三件事**： +- [ ] **① 在 1m 上复验**（本轮测的是 5m/30m，实盘跑 1m）。机制相同但数字要单独出， + `step68_live_window.py --ltf 1m --htf 5m`，这是决定是否要动实盘的依据 +- [ ] **② 试「中枢构成笔 is_sure」与「等 N 根看信号是否仍在」**（均未测）。 + 前者用滞后换稳定性，后者直接过滤会消失的信号。目标是把那 74% 压下去 +- [ ] **③ 通知服务器侧**：当前实盘在做一批回测里不存在的单。在 ① 出结果前 + 不建议改参数，但应知晓风险 +- [x] **窗口左边界效应判为否**（§3.3994）。2001 根滚动窗口与 2万→4万根增长窗口 + 跑出完全相同的 789/197/592，中枢是局部结构，截断无影响 - [ ] **一类线：问题已完全定性，等一个决策**（§3.3992 / §3.3993）。 全局只有一个变量——**「是不是真线段端点」**：是则 PF 2.29(5m)/4.11(15m)， 否则恒为 0.13，其余特征都只是它的噪声代理。**盈亏平衡精度 5m 72.6%、 diff --git a/research/step63_b4_replay.py b/research/step63_b4_replay.py new file mode 100644 index 0000000..62c1128 --- /dev/null +++ b/research/step63_b4_replay.py @@ -0,0 +1,246 @@ +"""B4 主线的因果回放：实盘信号的时点是不是干净的。 + +step59 在一类上抓到一个会骗人的坑：`sure_time` 是引擎**事后**标注的确认时刻， +不等于可执行时刻，用它当 entry 的回测 PF 虚高一倍以上。B4 正在跑真钱， +必须过同一关。 + +**先验比一类好，但方向要说清**（§5.41 的代码审计）： + + 一类 `sure_time` 直接当**入场时刻** -> 早了就是虚高，回测被高估 + B4 `available_ts` 只是**扫描起点** -> 晚了只会漏信号，回测偏保守 + +§5.41 实测全量的 `available_ts` 系统性**更晚**（`bis[-1]` 取的是中枢结束而非 +形成，中位晚 62 分钟）。所以预期是「回测保守」而非「回测虚高」。但那是 300 +时点抽样 + 代码审计，不是逐根验证，而且留了个未知： +**实盘会产出更多、更早的信号，那部分的质量不在回测统计里。** + +本脚本逐根重放，同时量两边： + + 准时率 全量信号在其 entry_idx 当根就能算出来的比例 + 迟到 首现晚于 entry_idx 的，实盘只能在更差的价位追 + 额外信号 回放发得出、全量却没有的 —— §5.41 预言存在，但没人统计过它们赚不赚 + +⚠️ 性能：每根扫全部中枢跑不完。一个中枢只能在其 available_ts 之后 scan 根内 +出信号，所以每根只需把窗口内的中枢喂给 `find_fast_bsp3`。这是等价裁剪， +不改变结果。 +""" +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" / "step63_b4_replay.feather" +SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48 +SCAN = 200 + + +def replay(sym: str, tf: str, rows: int, warm: int, + steps: int) -> pd.DataFrame | None: + from chanlun import TF_DF + from chanlun.analysis.fast_bsp import ( + ensure_timestamp, + find_fast_bsp3, + zones_from_zs_list, + ) + from lib.data import fetch_ohlcv + + try: + df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows) + if df is None or len(df) < warm + steps + 100: + return None + df = df.iloc[-(warm + steps):].reset_index(drop=True) + + # ---- 全量口径：回测就是这么算的 ---- + full = TF_DF(df, 1, tf) + cdf = ensure_timestamp(full.dataframe) + zs_full = full.cal_bi_zs_list_pure(full.bi_list) + if not zs_full: + return None + sig_full = find_fast_bsp3(cdf, zones_from_zs_list(zs_full, cdf)) + full_keys = {(int(r.entry_idx), int(r.direction)) + for r in sig_full.itertuples()} if not sig_full.empty \ + else set() + + # ---- 回放口径：逐根重算 zones 再扫 ---- + chan = TF_DF(df.iloc[:warm].copy(), 1, tf) + 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: + zl = chan.cal_bi_zs_list_pure(chan.bi_list) + if not zl: + continue + sub = ensure_timestamp(chan.dataframe) + z = zones_from_zs_list(zl, sub) + if z.empty: + continue + # 等价裁剪：available_ts 早于 scan 根之前的中枢，其扫描窗口 + # 已经过去，不可能在本根产出新信号 + lo = sub["timestamp"].to_numpy()[max(0, len(sub) - SCAN - 2)] + z = z[z.available_ts >= lo] + if z.empty: + continue + s = find_fast_bsp3(sub, z, scan=SCAN) + except Exception: # noqa: BLE001 + continue + if s is None or s.empty: + continue + for r in s.itertuples(): + k = (int(r.entry_idx), int(r.direction)) + if k not in first_seen: + first_seen[k] = i + + rec = [] + for k in set(full_keys) | set(first_seen): + e, d = k + if e < warm: # 预热段不计入 + continue + seen = first_seen.get(k) + rec.append({ + "sym": sym, "tf": tf, "entry_idx": e, "direction": d, + "in_full": k in full_keys, "in_replay": seen is not None, + "i_seen": -1 if seen is None else seen, + "late": (np.nan if seen is None else seen - e), + }) + if not rec: + return None + r = pd.DataFrame(rec) + + # 实盘真正会做的：首现根入场（首现==entry_idx 即准时） + from lib.exit_model import cfg_name, walk_exits + atr = cdf["atr"].to_numpy(float) + cl = cdf["close"].to_numpy(float) + n = len(cdf) + cfg = cfg_name(SL, RUNNER, MAXB, RSTOP) + live = r[r.in_replay & (r.i_seen < n - 2)].copy() + live = live[np.isfinite(atr[live.i_seen.values]) + & (atr[live.i_seen.values] > 0)] + if not live.empty: + res = walk_exits(cdf, pd.DataFrame({ + "entry_idx": live.i_seen.values, + "direction": live.direction.values}), [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[["entry_idx", "direction", "g", "r", "c", + "atr_pct"]], + on=["entry_idx", "direction"], how="left") + return r + except Exception as e: # noqa: BLE001 + print(f" {sym} {tf} 失败: {type(e).__name__}: {e}", flush=True) + return None + + +def perf(g: pd.DataFrame) -> dict: + from lib.exit_model import fee_of, taker_notional + 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() + return { + "笔数": 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), + } + + +def report(d: pd.DataFrame) -> None: + print("\n" + "=" * 92) + print("########## 一、准时率与额外信号 ##########") + rows = [] + for tf, x in d.groupby("tf"): + both = x[x.in_full & x.in_replay] + rows.append({ + "tf": tf, + "全量信号": int(x.in_full.sum()), + "回放信号": int(x.in_replay.sum()), + "召回": f"{len(both)/max(int(x.in_full.sum()),1)*100:.1f}%", + "准时(首现==entry)": f"{(both.late == 0).mean()*100:.1f}%", + "迟到中位": (f"{both.late[both.late > 0].median():.0f} 根" + if (both.late > 0).any() else "—"), + "额外信号": int((x.in_replay & ~x.in_full).sum()), + }) + print(pd.DataFrame(rows).to_string(index=False)) + print("\n额外信号 = 回放发得出、全量没有的。§5.41 预言它们存在" + "（实盘 available_ts 更早 -> 信号更多更早），本表给出数量。") + + if "g" not in d.columns: + return + print("\n" + "=" * 92) + print("########## 二、分组收益：回测口径 vs 实盘口径 ##########") + for tf, x in d.groupby("tf"): + y = x.dropna(subset=["g"]) + if len(y) < 30: + continue + print(f"\n--- {tf} ---") + rows = [] + for nm, g in [ + ("全部回放信号（=实盘会做的）", y[y.in_replay]), + ("其中 准时的", y[y.in_replay & (y.late == 0)]), + ("其中 迟到的", y[y.in_replay & (y.late > 0)]), + ("其中 额外的（全量没有）", y[y.in_replay & ~y.in_full]), + ("回测口径（全量∩回放）", y[y.in_full & y.in_replay]), + ]: + if len(g) >= 30: + rows.append({"分组": nm, **perf(g)}) + print(pd.DataFrame(rows).to_string(index=False)) + print("\n判读：若「全部回放信号」的 PF 不低于「回测口径」，说明 B4 的时点" + "是干净的，\n且 §5.41 说的『回测偏保守』成立 —— 实盘拿到的反而更多。" + "\n若额外信号那组显著更差，那就是回测没统计到的隐性成本。") + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE") + ap.add_argument("--tf", default="5m") + ap.add_argument("--rows", type=int, default=45_000) + ap.add_argument("--warm", type=int, default=20_000) + ap.add_argument("--steps", type=int, default=20_000) + ap.add_argument("--workers", type=int, default=5) + ap.add_argument("--reuse", action="store_true") + args = ap.parse_args() + + if args.reuse and OUT.exists(): + report(pd.read_feather(OUT)) + return + syms = [s.strip() for s in args.symbols.split(",")] + print(f"[B4 因果回放] {len(syms)} 币 × {args.steps} 根逐根重放\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() diff --git a/research/step68_live_window.py b/research/step68_live_window.py new file mode 100644 index 0000000..6393cc1 --- /dev/null +++ b/research/step68_live_window.py @@ -0,0 +1,265 @@ +"""按**实盘口径**重放 B4，并测三道过滤能否刷掉那批亏钱的额外信号。 + +step63 的结论是一好一坏： + 好 时点干净 —— 100% 召回、100% 准时、零滞后（一类是 0% 准时、+15 根） + 坏 回放多出 592 个全量口径没有的信号，PF 0.46 / t −5.03，混合后 0.64 < 1 + +但 step63 有两个口径问题，本脚本一并修掉： + + ① 窗口不对。回测用全量 45000 根一次算完，step63 用 2 万涨到 4 万根的增长窗口， + **而实盘用 2001 根滚动窗口、每 500 根 init_stream 拉回**（`shadow_signal.py` + 的 MAX_GROW）。三种口径的中枢结构都不一样。这也是 HANDOFF 里挂着的 + 「回测用全量历史建中枢、实盘用 2000 根窗口」那条待办。 + 顺带：窗口封顶后单步成本恒定，不再是 step63 那个平方级（143ms@2万根 -> + 292ms@4万根），所以本脚本快得多。 + + ② 没测过滤。step63 跑的是裸信号，而实盘有三道滤网。ATR 门控已单独测过—— + 它刷掉 7.9% 的额外信号却刷掉 10.7% 的好信号，PF 纹丝不动。剩下两道要测。 + +**一处刻意的简化，方向是保守的**：大级别分型时间线用全量历史算（真实盘的 HTF +也会重画）。这等于**给同向过滤器开了未来函数的后门**。若连这样都刷不掉额外信号， +结论只会更强。 +""" +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" / "step68_live_window.feather" +SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48 +WIN, MAX_GROW, GATE_BP = 2001, 500, 8.0 + + +def _ladder(zones: pd.DataFrame) -> pd.DataFrame: + 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)) + return z + + +def replay(sym: str, ltf: str, htf: str, rows: int, + steps: int) -> pd.DataFrame | None: + from chanlun import TF_DF + from chanlun.analysis.fast_bsp import ( + ensure_timestamp, + find_fast_bsp3, + zones_from_zs_list, + ) + from lib.data import fetch_ohlcv + from lib.fx_signal import extract_fx_signals, signals_to_frame + from lib.nested_bsp import attach_htf_context, htf_fx_timeline + + try: + df = fetch_ohlcv(f"{sym}/USDT:USDT", ltf, rows) + if df is None or len(df) < WIN + steps + 100: + return None + df = df.iloc[-(WIN + steps):].reset_index(drop=True) + + full = TF_DF(df, 1, ltf) + cdf = ensure_timestamp(full.dataframe) + zs_full = full.cal_bi_zs_list_pure(full.bi_list) + if not zs_full: + return None + sig_full = find_fast_bsp3(cdf, zones_from_zs_list(zs_full, cdf)) + full_keys = {(int(r.entry_idx), int(r.direction)) + for r in sig_full.itertuples()} if not sig_full.empty \ + else set() + + # 大级别分型时间线：全量算（见模块 docstring 的「刻意简化」） + dh = fetch_ohlcv(f"{sym}/USDT:USDT", htf, rows) + tl = None + if dh is not None and len(dh) > 500: + ch = TF_DF(dh, 1, htf) + tl = htf_fx_timeline( + signals_to_frame(extract_fx_signals(ch, ch.dataframe)), + ch.dataframe) + + rec: dict[tuple, dict] = {} + chan = None + anchor = 0 + for i in range(WIN, len(df)): + # 实盘的窗口纪律：2001 根起，长过 MAX_GROW 就 init_stream 拉回 + if chan is None or (i - anchor) >= MAX_GROW: + w = df.iloc[i - WIN + 1:i + 1].copy() + chan = TF_DF(w, 1, ltf) + chan.init_stream(w, 1, ltf) + anchor = i + else: + chan.append_bar(df.iloc[i]) + try: + zl = chan.cal_bi_zs_list_pure(chan.bi_list) + if not zl: + continue + sub = ensure_timestamp(chan.dataframe) + z = _ladder(zones_from_zs_list(zl, sub)) + if z.empty: + continue + s = find_fast_bsp3(sub, z) + if s is None or s.empty: + continue + last = len(sub) - 1 + s = s[s["entry_idx"].astype(int) == last] # 实盘只做当根 + if s.empty: + continue + if "zone_i" in s.columns: + s = s.merge(z[["zone_i", "z_above", "z_below"]], + on="zone_i", how="left") + if tl is not None: + s = attach_htf_context(s, sub, tl, "h1") + except Exception: # noqa: BLE001 + continue + for r in s.itertuples(): + k = (i, int(r.direction)) + if k in rec: + continue + push = getattr(r, "z_above" if r.direction == 1 + else "z_below", None) + ag = getattr(r, "h1_agree", 0) + rec[k] = { + "sym": sym, "entry_idx": i, "direction": int(r.direction), + "in_full": (i, int(r.direction)) in full_keys, + "ladder_ok": int(bool(pd.notna(push) and bool(push))), + "h1_agree": int(ag) if pd.notna(ag) else 0, + } + if not rec: + return None + r = pd.DataFrame(list(rec.values())) + + from lib.exit_model import cfg_name, walk_exits + atr = cdf["atr"].to_numpy(float) + cl = cdf["close"].to_numpy(float) + r = r[(r.entry_idx < len(cdf) - 2) + & np.isfinite(atr[r.entry_idx.values]) + & (atr[r.entry_idx.values] > 0)].reset_index(drop=True) + if r.empty: + return None + res = walk_exits(cdf, pd.DataFrame({ + "entry_idx": r.entry_idx.values, + "direction": r.direction.values}), [SL], [RUNNER], [MAXB], + scale_at=SCALE_AT, runners=(RUNNER,), runner_stops=(RSTOP,)) + cfg = cfg_name(SL, RUNNER, MAXB, RSTOP) + if len(res) != len(r): + return None + for c in ("g", "r", "c"): + r[c] = res[f"{cfg}_{c}"].to_numpy() + r["atr_pct"] = atr[r.entry_idx.values] / cl[r.entry_idx.values] + r["gate_ok"] = (r.atr_pct * 1e4 >= GATE_BP).astype(int) + r["pass_all"] = ((r.h1_agree == 1) & (r.ladder_ok == 1) + & (r.gate_ok == 1)).astype(int) + return r + except Exception as e: # noqa: BLE001 + print(f" {sym} 失败: {type(e).__name__}: {e}", flush=True) + return None + + +def perf(g: pd.DataFrame) -> dict | None: + from lib.exit_model import fee_of, taker_notional + if len(g) < 20: + return None + net = g.g.values - fee_of(g.r.values, g.c.values) + gR = g.g.values / (SL * g.atr_pct.values) + tn = taker_notional(g.r.values, g.c.values) + w, o = net[net > 0].sum(), -net[net <= 0].sum() + return { + "笔数": len(g), "胜率": f"{(net > 0).mean()*100:.1f}%", + "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), + } + + +def report(d: pd.DataFrame) -> None: + print("\n" + "=" * 92) + print("【一】实盘窗口下还有多少额外信号") + print("=" * 92) + print(f"回放信号 {len(d)} · 其中全量口径也有 {int(d.in_full.sum())} · " + f"**额外 {int((~d.in_full).sum())}**") + print(f"（step63 的增长窗口口径：197 / 592）") + + print("\n" + "=" * 92) + print("【二】三道过滤能不能刷掉额外信号 —— 这决定实盘是否在亏钱") + print("=" * 92) + rows = [] + for nm, m in [("① 无过滤", None), ("② 仅同向", d.h1_agree == 1), + ("③ 仅阶梯", d.ladder_ok == 1), + ("④ 仅ATR门控", d.gate_ok == 1), + ("⑤ 三道全开(实盘口径)", d.pass_all == 1)]: + x = d if m is None else d[m] + if x.empty: + continue + ex, bt = x[~x.in_full], x[x.in_full] + row = {"过滤": nm, "留下": len(x), + "额外占比": f"{(~x.in_full).mean()*100:.0f}%"} + for lab, g in [("全部", x), ("额外", ex), ("回测口径", bt)]: + s = perf(g) + row[f"{lab}PF"] = "—" if s is None else s["PF"] + row[f"{lab}n"] = len(g) + rows.append(row) + print(pd.DataFrame(rows).to_string(index=False)) + + print("\n" + "=" * 92) + print("【三】实盘口径（三道全开）的完整表现") + print("=" * 92) + rows = [] + for nm, g in [("实盘会做的全部", d[d.pass_all == 1]), + (" 其中额外的", d[(d.pass_all == 1) & ~d.in_full]), + (" 其中回测也有的", d[(d.pass_all == 1) & d.in_full])]: + s = perf(g) + if s: + rows.append({"分组": nm, **s}) + print(pd.DataFrame(rows).to_string(index=False)) + print(""" + 判读：「实盘会做的全部」PF > 1 -> 实盘安全，额外信号被滤网挡住了 + PF < 1 -> **实盘在做一批回测里不存在、且亏钱的信号**，要立刻处理""") + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE") + ap.add_argument("--ltf", default="5m") + ap.add_argument("--htf", default="30m") + ap.add_argument("--rows", type=int, default=45_000) + ap.add_argument("--steps", type=int, default=20_000) + ap.add_argument("--workers", type=int, default=5) + ap.add_argument("--reuse", action="store_true") + args = ap.parse_args() + + if args.reuse and OUT.exists(): + report(pd.read_feather(OUT)) + return + syms = [s.strip() for s in args.symbols.split(",")] + print(f"[实盘口径回放] {len(syms)} 币 × {args.steps} 根 · " + f"{WIN} 根滚动窗口 / 每 {MAX_GROW} 根重建\n", flush=True) + parts = [] + with ProcessPoolExecutor(max_workers=args.workers) as ex: + fut = {ex.submit(replay, s, args.ltf, args.htf, args.rows, + 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()