diff --git a/research/HANDOFF.md b/research/HANDOFF.md index 1785fe6..1e2df75 100644 --- a/research/HANDOFF.md +++ b/research/HANDOFF.md @@ -1603,8 +1603,44 @@ B4 错在**存在性**（信号当根就有，但其中 74% 事后会消失） 还要看执行层是否有额外去重/冷却，**这一层未审计** **这不否定 B4 的 alpha**——回测口径那 51 笔 PF 4.21、t 5.90 是真实的，说明 -「真中枢上的 B4」确实赚钱。问题是实时分不出真假中枢。可能的方向（均未测）： -要求中枢的构成笔 `is_sure`（用滞后换稳定性）、或等 N 根看信号是否仍然存在。 +「真中枢上的 B4」确实赚钱。问题是实时分不出真假中枢。 + +#### 3.3994a 机制订正：不是重画，是中枢终点在实时不可知 + +我最初把它归因为「中枢重画」，**这个归因是错的**，用户两次指出后逐条查证： + +**① 中枢边界不重画（用户对）。** `zg/zd` 由前三笔定死，`verify_window_sens` +与 step39 都验过。 + +**② 中枢只由已确认的笔构成（结构性保证）。** `cal_bi_zs_list_pure`： + +```python +if not (bi1.is_sure and bi2.is_sure and bi3.is_sure): continue +if not (leave_bi.is_sure and back_bi.is_sure): break # 延伸时同样要求 +``` + +step70 实测：789 个信号**全部** `z_sure=True`。所以用户说的「浅色 B4 会出现 +然后消失」在研究路径上不存在（那来自 web 展示用的非 `_pure` 中枢）。 +⛔ **「补一道 is_sure 门」是空操作**，实测 PF 0.73 → 0.73 分毫未动，不要再提。 + +**③ 真正的机制是 `available_ts` 棘轮，§5.41 早已写明：** + +> 6.5% 的改动**不是已确认的笔被推翻**，而是中枢又吸收了新笔、`bis[-1]` 变成了 +> 另一根笔。**原来那根笔本身没变。** + +`available_ts` = 中枢**最后一笔**的 `sure_time`（`AVAIL_BI_INDEX = -1`）， +而中枢会持续吸收新确认的笔 → 「最后一笔」换人 → `available_ts` 往后棘轮 → +`find_fast_bsp3` 的 200 根扫描窗口整体右移。消失的不是笔，是**中枢的终点**。 + +顺带澄清一个我一度担心的伪问题：**中枢跨度 > 200 根不会导致突破落不进窗口**， +因为扫描起点是中枢**结束**（末笔确认）而非起点，此时价格已在离开中枢。 + +**⚠️ 唯一还没对上的是数量级：中枢层面 6.5%，信号层面 75%。** +假设是 `max_per_zone=1` 的放大——每个中枢只返回**第一个**入场点，而 +`start = searchsorted(ts, available_ts)` 每棘轮一次就越过旧入场点， +同一中枢于是反复产出新的「第一个」；全量只用最终 `available_ts`，每中枢至多一个。 +**这条未测**（`step69_mechanism.py` 已写好，判据是「额外信号落在几个不同中枢上」， +三次后台运行都被中断，需重跑）。 ### 3.4 alpha 的来源（step32 消融） @@ -2824,8 +2860,15 @@ API 限流风险隔离三个好处。 1m 的关键分组只有 34 笔。扩到 8~11 币或 5 万根再看是否稳定 - [ ] **①c 审计执行层是否有去重/冷却**。「额外信号」是实时算得出、事后消失的， 但实盘是否真的每个都下单，取决于执行层，这一层还没看过 -- [ ] **② 试「中枢构成笔 is_sure」与「等 N 根看信号是否仍在」**（均未测）。 - 前者用滞后换稳定性，后者直接过滤会消失的信号。目标是把那 74% 压下去 +- [x] ⛔ **「中枢构成笔 is_sure」已证是空操作**（step70，§3.3994a）。 + `cal_bi_zs_list_pure` 本就强制全部构成笔 is_sure，789 个信号全是已确认， + 加这道门 PF 0.73 → 0.73。**不要再提这条** +- [ ] **② 先测数量级放大：那 592 个额外信号落在几个不同中枢上**（§3.3994a）。 + 中枢层面只改 6.5%，信号层面却 75%，怀疑 `max_per_zone=1` + 起点棘轮 + 让同一中枢反复产出「第一个」入场点。脚本已写好 `step69_mechanism.py`， + **三次后台运行都被中断，需在前台重跑**。这是修法的前置—— + 若确是重复触发，修法是「每中枢只认一次」，代价远小于动 available_ts +- [ ] **③ 若非重复触发，再试「等 N 根看信号是否仍在」**（用滞后换稳定性） - [ ] **③ 通知服务器侧**：当前实盘在做一批回测里不存在的单。在 ① 出结果前 不建议改参数，但应知晓风险 - [x] **窗口左边界效应判为否**（§3.3994）。2001 根滚动窗口与 2万→4万根增长窗口 diff --git a/research/step69_mechanism.py b/research/step69_mechanism.py new file mode 100644 index 0000000..9a34da3 --- /dev/null +++ b/research/step69_mechanism.py @@ -0,0 +1,240 @@ +"""那 72~77% 的「额外信号」到底怎么来的：中枢重画，还是扫描窗口错位？ + +我在 §3.3994 里把它归因为「中枢重画」，**这个归因没验证过，而且与已有结论矛盾** +（step39 的假阳性 0%、`verify_window_sens` 的窗口 +200/+500/+1000 逐字段一致）。 +用户指出中枢不重画，代码注释也支持他： + + # -1 = 最后一笔（历史默认）。中枢每吸收一根K线，最后一笔就可能后移， + # 于是 available_ts 跟着漂——这是右边缘重画的根因 + —— chanlun/analysis/fast_bsp.py:47 + +漂的不是中枢**边界**（zg/zd），是它的**可用时刻**。而 `find_fast_bsp3` 只从 +`available_ts` 往后扫 `scan=200` 根。两种口径的窗口因此错位： + + 实时 中枢没吸收完，available_ts 偏早 -> 窗口开得早 + 全量 中枢吸收完了，available_ts 偏晚（§5.41 实测中位晚 62 分钟）-> 窗口开得晚 + +落在「实时窗口内、全量窗口外」的信号，全量根本没扫到那个时段，于是显示为「额外」。 + +两种机制的修法完全不同，所以必须分清： + + 边界重画 结构本身不稳，只能用滞后换稳定性，代价大 + 窗口错位 中枢是同一个真中枢，信号也是真信号，只是**开得太早、确认不足** + —— 这正好解释它们为什么亏（PF 0.26~0.36），且修法是调 available_ts + +判据：逐个额外信号，去全量中枢表里按 (zg, zd) 找它的中枢。 + 找得到且边界一致 -> 窗口错位（用户是对的，我的归因错了） + 找不到或边界不同 -> 边界重画（我的归因成立） +""" +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" / "step69_mech.feather" +WIN, MAX_GROW, SCAN = 2001, 500, 200 +TOL = 1e-6 + + +def replay(sym: str, ltf: 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 + + 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) + zf = zones_from_zs_list(full.cal_bi_zs_list_pure(full.bi_list), cdf) + if zf is None or zf.empty: + return None + sig_full = find_fast_bsp3(cdf, zf) + full_keys = {(int(r.entry_idx), int(r.direction)) + for r in sig_full.itertuples()} if not sig_full.empty \ + else set() + fzg = zf["zg"].to_numpy(float) + fzd = zf["zd"].to_numpy(float) + fav = zf["available_ts"].to_numpy() + ts_all = cdf["timestamp"].to_numpy() + + rec: dict[tuple, dict] = {} + chan, anchor = None, 0 + for i in range(WIN, len(df)): + 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 = zones_from_zs_list(zl, sub) + if z is None or 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 or "zone_i" not in s.columns: + continue + # find_fast_bsp3 的输出自带 zg/zd，直接 merge 会加后缀， + # 所以中枢侧的列全部改名再接 + zc = z[["zg", "zd", "available_ts"]].rename(columns={ + "zg": "z_zg", "zd": "z_zd", "available_ts": "z_av"}) + zc["zone_i"] = np.arange(len(z)) + s = s.drop(columns=[c for c in ("z_zg", "z_zd", "z_av") + if c in s.columns]) + s = s.merge(zc, on="zone_i", how="left") + except Exception: # noqa: BLE001 + continue + for r in s.itertuples(): + k = (i, int(r.direction)) + if k in rec: + continue + # 该中枢在全量表里是否存在（按边界匹配，边界是不该漂的量） + zg_, zd_ = float(r.z_zg), float(r.z_zd) + m = (np.abs(fzg - zg_) <= TOL * max(1.0, abs(zg_))) \ + & (np.abs(fzd - zd_) <= TOL * max(1.0, abs(zd_))) + j = int(np.argmax(m)) if m.any() else -1 + rec[k] = { + "sym": sym, "entry_idx": i, "direction": int(r.direction), + "in_full": k in full_keys, + "zone_found": bool(m.any()), + "z_zg": zg_, "z_zd": zd_, + "avail_rt": int(r.z_av), + "avail_full": int(fav[j]) if j >= 0 else -1, + "ts_entry": int(ts_all[i]) if i < len(ts_all) else -1, + } + return pd.DataFrame(list(rec.values())) if rec else None + except Exception as e: # noqa: BLE001 + print(f" {sym} 失败: {type(e).__name__}: {e}", flush=True) + return None + + +def report(d: pd.DataFrame, ltf: str) -> None: + ex = d[~d.in_full] + print("\n" + "=" * 92) + print("【一】判据：额外信号所在的中枢，在全量表里找得到吗") + print("=" * 92) + print(f"回放信号 {len(d)} · 额外 {len(ex)}") + print(f"**额外信号中，其中枢按 (zg,zd) 在全量表里找得到的：" + f"{ex.zone_found.mean()*100:.1f}%**") + print(f"（对照：非额外信号 {d[d.in_full].zone_found.mean()*100:.1f}%）") + print(""" + ≈100% -> 中枢边界没变，是**扫描窗口错位**，用户对、我的「重画」归因错 + 明显偏低 -> 中枢确实消失或改边界，「重画」成立""") + + print("\n" + "=" * 92) + print("【一b】数量级对账：中枢层面只有 6.5% 被改，信号层面却 74% 是额外的") + print("=" * 92) + print("§5.41 的 A/B 实测：`bis[-1]` 口径下确认时刻被改 6.5%、中枢消失 9.0%。") + print("若信号层面的 74% 成立，必然有放大机制。怀疑是 `max_per_zone=1`：") + print(" 每个中枢只返回**第一个**入场点，而扫描起点随 available_ts 棘轮后移，") + print(" 越过旧入场点后，同一中枢会重新产出一个「第一个」——全量只用最终值，") + print(" 所以每中枢至多一个信号，实时却能反复触发。") + for nm, x in [("额外信号", ex), ("非额外信号", d[d.in_full])]: + if x.empty: + continue + nz = x.groupby(["sym", "z_zg", "z_zd"]).size() \ + if "z_zg" in x.columns else None + if nz is None: + print(" （缺 z_zg/z_zd 列，跳过）") + break + print(f"\n{nm}：{len(x)} 个信号，落在 {len(nz)} 个不同中枢上 " + f"-> 每中枢 {len(x)/len(nz):.2f} 次") + print(f" 同一中枢触发次数分布 中位 {nz.median():.0f} " + f"P90 {nz.quantile(.9):.0f} 最大 {nz.max()}") + print(""" + 若「额外信号」的每中枢次数显著 > 1 而「非额外」≈ 1，放大机制坐实： + 6.5% 的中枢改动通过棘轮重扫，放大成信号层面的几百个。""") + + g = ex[ex.zone_found & (ex.avail_full > 0)].copy() + if g.empty: + return + bar_ms = {"1m": 60_000, "5m": 300_000, "15m": 900_000}.get(ltf, 300_000) + g["drift_bars"] = (g.avail_full - g.avail_rt) / bar_ms + g["from_rt"] = (g.ts_entry - g.avail_rt) / bar_ms + g["from_full"] = (g.ts_entry - g.avail_full) / bar_ms + + print("\n" + "=" * 92) + print("【二】available_ts 漂了多少，以及入场落在谁的扫描窗口里") + print("=" * 92) + print(f"avail 漂移（全量 − 实时，根） 中位 {g.drift_bars.median():.0f} " + f"P25 {g.drift_bars.quantile(.25):.0f} " + f"P75 {g.drift_bars.quantile(.75):.0f}") + print(f" §5.41 记的是中位晚 62 分钟，本表 {ltf} 下即 " + f"{62*60_000/bar_ms:.0f} 根，可交叉验证") + print(f"\n入场距实时 avail（根） 中位 {g.from_rt.median():.0f} " + f"（应落在 0~{SCAN} 内，否则实时也扫不到）") + print(f"入场距全量 avail（根） 中位 {g.from_full.median():.0f}") + out = ((g.from_full < 0) | (g.from_full > SCAN)).mean() + print(f"\n**入场落在全量扫描窗口 [0,{SCAN}] 之外的比例：{out*100:.1f}%**") + print(""" + 这是机制的直接证据：比例高 -> 全量根本没扫到那个时段，所以「没有」这个信号， + 与中枢是否重画无关。其中 from_full < 0 表示入场早于全量的可用时刻—— + 即**实时抢跑了**，中枢还没吸收完就下单。""") + early = (g.from_full < 0).mean() + print(f" 其中抢跑（早于全量 avail）：{early*100:.1f}%") + + +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("--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), args.ltf) + return + syms = [s.strip() for s in args.symbols.split(",")] + print(f"[机制判定] {len(syms)} 币 × {args.steps} 根 · {args.ltf}\n", + flush=True) + parts = [] + with ProcessPoolExecutor(max_workers=args.workers) as ex_: + fut = {ex_.submit(replay, s, args.ltf, 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, args.ltf) + + +if __name__ == "__main__": + main() diff --git a/research/step70_sure_split.py b/research/step70_sure_split.py new file mode 100644 index 0000000..22cdecc --- /dev/null +++ b/research/step70_sure_split.py @@ -0,0 +1,280 @@ +"""把 B4 按「中枢末笔是否已确认」拆开——之前的统计把两种混在一起了。 + +用户指出：B4 有两种，**没确认的会出现然后消失，确认的不会**。而我在 step63/68 +里统计的是 `find_fast_bsp3` 的全部输出，它的返回列里根本没有确认标志 +（entry_idx/direction/bo_idx/pb_idx/lag/depth/zg/zd/width_pct/occ/zone_i）， +**两种被混在一起了**，所以「额外信号 74%」这个数字不能直接拿来说实盘。 + +确认状态在更上游，`zones_from_zs_list`： + + sure_key = str(getattr(key_bi, "sure_time", "") or "") + end_key = str(getattr(key_bi, "end_time", "") or "") + avail = ts_of.get(sure_key) or ts_of.get(end_key) # ← 静默退回 end_time + +`ChanBI` 初始 `is_sure=False / sure_time=None`，确认时才 `set_is_sure(True, ...)`。 +所以**末笔未确认时 available_ts 退回 end_time，而 end_time 随笔延伸而移动**， +中枢的可用时刻跟着漂 —— 这正是「出现然后消失」的那一种。笔一旦确认， +`sure_time` 固定，中枢不再动。 + +这也解释了 §5.41 的数量级：中枢层面确认时刻只被改 6.5%，而我在信号层面看到 74%。 + +本脚本按 `zs.bi_list[-1].is_sure` 把信号拆成两组，分别看： + 额外率 未确认组应显著高（会出现然后消失），已确认组应接近 0 + 收益 若亏损集中在未确认组，那么修法就是**信号侧加一道 is_sure 门**， + 而不是动出场参数或放弃 B4 + +⚠️ 同时要查的第二件事：**实盘路径到底交易哪一种。** `shadow_signal.py` 的三道 +滤网是同向 + 阶梯 + ATR 门控，**没有 is_sure 这一道**。若未确认组确实是亏损源， +且实盘没有挡它，那这道门就是要补的东西。 +""" +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" / "step70_sure.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 _sure_map(zs_list) -> dict: + """(zg, zd) -> 末笔是否已确认。zones_from_zs_list 会按 available_ts 重排， + 索引对不上，所以用中枢边界当键接回去。""" + m = {} + for zs in zs_list: + bis = getattr(zs, "bi_list", []) or [] + if not bis: + continue + m[(round(float(zs.zg), 10), round(float(zs.zd), 10))] = \ + bool(getattr(bis[-1], "is_sure", False)) + return m + + +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) + zsf = full.cal_bi_zs_list_pure(full.bi_list) + if not zsf: + return None + sig_full = find_fast_bsp3(cdf, zones_from_zs_list(zsf, cdf)) + full_keys = {(int(r.entry_idx), int(r.direction)) + for r in sig_full.itertuples()} if not sig_full.empty \ + else set() + + 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, anchor = None, 0 + for i in range(WIN, len(df)): + 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 = zones_from_zs_list(zl, sub) + if z is None or z.empty: + continue + sm = _sure_map(zl) + 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)) + z["z_sure"] = [ + sm.get((round(float(a), 10), round(float(b), 10)), False) + for a, b in zip(z["zg"], z["zd"])] + 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 or "zone_i" not in s.columns: + continue + s = s.merge(z[["zone_i", "z_above", "z_below", "z_sure"]], + 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": k in full_keys, + "z_sure": bool(getattr(r, "z_sure", False)), + "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) < 15: + return None + net = g.g.values - fee_of(g.r.values, g.c.values) + gR = g.g.values / (SL * g.atr_pct.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() + / taker_notional(g.r.values, g.c.values).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) + rows = [] + for k, g in d.groupby(d.z_sure): + rows.append({"中枢末笔": "已确认" if k else "未确认", "信号数": len(g), + "额外(全量没有)": int((~g.in_full).sum()), + "额外率": f"{(~g.in_full).mean()*100:.1f}%"}) + print(pd.DataFrame(rows).to_string(index=False)) + print(""" + 用户的判断：「没确认的会出现然后消失，确认的不会」。 + 若已确认组额外率接近 0 -> 判断成立，之前 74% 是把两种混在一起统计的结果。""") + + print("\n" + "=" * 92) + print("【二】亏损是不是也集中在未确认组") + print("=" * 92) + rows = [] + for k, g in d.groupby(d.z_sure): + nm = "已确认" if k else "未确认" + for lab, x in [("全部", g), ("三道滤网后", g[g.pass_all == 1])]: + s = perf(x) + if s: + rows.append({"中枢末笔": nm, "口径": lab, **s}) + print(pd.DataFrame(rows).to_string(index=False)) + + print("\n" + "=" * 92) + print("【三】若补一道 is_sure 门，实盘口径会变成什么样") + print("=" * 92) + rows = [] + for nm, x in [ + ("现状：三道滤网", d[d.pass_all == 1]), + ("**加 is_sure 门**", d[(d.pass_all == 1) & d.z_sure]), + (" 对照：仅未确认", d[(d.pass_all == 1) & ~d.z_sure]), + ]: + s = perf(x) + if s: + rows.append({"口径": nm, **s}) + print(pd.DataFrame(rows).to_string(index=False)) + print(""" + ⚠️ `shadow_signal.py` 的三道滤网是同向 + 阶梯 + ATR 门控，**没有 is_sure**。 + 若加上这道门 PF 明显回升，那它就是要补进信号路径的东西。""") + + +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} 根 · {args.ltf}\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()