"""把 step66 的天花板换成实时可算的选样，看能兑现多少。 step66 测出上界：**未来函数选样 + 结构止损**在 5m 是 PF 2.29、15m 是 4.11。 但那个选样用了线段端点（事后才知道），不能交易。本脚本把它换成 step62 那个 **实时可算**的 `ext_run`（极值越过中枢边界几个 ATR，当根即可算），配上结构止损。 这是唯一同时处理两个已知约束的组合，也是 §3.3992 列出的第一个待跑实验： step62 有实时过滤器（精度 28.7% -> 47.9%）但配了 2 ATR 止损 -> PF 0.54 step58 有结构止损但没有过滤器 -> PF 0.36 step66 两个都有，但选样是未来函数 -> PF 2.29 / 4.11 **本脚本：实时过滤器 + 结构止损** -> ? 判读（对照 step66 的上界）： > 1.3 兑现了相当部分，一类线值得继续，下一步做因果回放（sure_time 仍是未来函数） ~ 1.0 过滤器精度不够，需要更好的实时特征 < 0.8 实时特征抓不到那 28.7%，天花板兑现不了，一类线仍关闭 ⚠️ 即使 > 1.3 也**不能算数**：入场仍在 `sure_time` 上，而 §3.396 证明实盘回放 还要再晚 15 根。这一步只决定「值不值得再花一次因果回放的机器时间」。 """ from __future__ import annotations import argparse import sys import warnings 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)) F62 = HERE / "out" / "step62_trend_end.feather" F64 = HERE / "out" / "step64_b2.feather" KEY = ["sym", "tf", "type", "i_ext", "i_sure"] def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--tf", default="5m") ap.add_argument("--rows", type=int, default=200_000) args = ap.parse_args() from chanlun import TF_DF from lib.data import fetch_ohlcv from step58_struct_stop import simulate, stats a = pd.read_feather(F62)[KEY + ["ext_run", "div", "ladder_n"]] b = pd.read_feather(F64)[KEY + ["entry_px", "own_ext", "atr_at_entry", "hit", "dir"]] d = a.merge(b, on=KEY, how="inner") d = d[d.tf == args.tf].reset_index(drop=True) d["to_ext"] = (d.entry_px - d.own_ext) * d.dir / d.atr_at_entry print(f"[实时过滤器 + 结构止损] {args.tf} · 一类 {len(d)} 笔 · " f"真线段端点占比 {d.hit.mean()*100:.1f}%") print(f"step66 上界（同止损、但用未来函数选样）：" f"{'PF 2.29' if args.tf == '5m' else 'PF 4.11'}\n") cache = {} for sym in sorted(d.sym.unique()): df = fetch_ohlcv(f"{sym}/USDT:USDT", args.tf, args.rows) cache[sym] = TF_DF(df, 1, args.tf).dataframe def run(x: pd.DataFrame, margin: float | None) -> dict: gs, rs, cs, ap_, sl_ = [], [], [], [], [] for sym, g in x.groupby("sym"): g = g.reset_index(drop=True) t = pd.DataFrame({"entry_idx": g.i_sure.values, "direction": g.dir.values, "atr_at_entry": g.atr_at_entry.values}) sl = (np.full(len(g), 2.0) if margin is None else (g.to_ext.values + margin)) res = simulate(cache[sym], t, sl, r_scale=False) ok = res.g.notna().values gs.append(res.g[ok]) rs.append(res.r[ok]) cs.append(res.c[ok]) ap_.append((g.atr_at_entry.values / g.entry_px.values)[ok]) sl_.append(sl[ok]) return stats(pd.concat(gs, ignore_index=True), pd.concat(rs, ignore_index=True), pd.concat(cs, ignore_index=True), np.concatenate(ap_), np.concatenate(sl_)) print("=" * 96) print("【一】ext_run 分档 × 止损口径 —— 过滤器的效果依赖止损吗") print("=" * 96) q = d.ext_run.quantile([.25, .5, .75]).values lab = ["Q1最短", "Q2", "Q3", "Q4最延伸"] d["bucket"] = pd.cut(d.ext_run, [-np.inf, *q, np.inf], labels=lab) rows = [] for bk, g in d.groupby("bucket", observed=True): for name, m in [("固定2ATR", None), ("结构+1.0", 1.0)]: rows.append({"ext_run档": bk, "止损": name, "真端点占比": f"{g.hit.mean()*100:.0f}%", **run(g, m)}) print(pd.DataFrame(rows).to_string(index=False)) print(""" §3.399 判过「过滤器符号随入场时点翻转 -> 拟合」。这里换的是**止损**而非入场， 若 Q4 在两种止损下都最好，说明 ext_run 是真信号；若又翻转，则仍是拟合。""") print("\n" + "=" * 96) print("【二】实时可交易组合 vs step66 上界") print("=" * 96) rows = [{"选样": "全体（无过滤）", "笔数": len(d), **run(d, 1.0)}] for p in (50, 75): thr = np.percentile(d.ext_run, p) g = d[d.ext_run >= thr] rows.append({"选样": f"ext_run ≥ P{p}（实时）", "笔数": len(g), **run(g, 1.0)}) g = d[(d.ext_run >= np.percentile(d.ext_run, 75)) & (d["div"] >= d["div"].median())] rows.append({"选样": "ext_run≥P75 且 div≥中位（实时）", "笔数": len(g), **run(g, 1.0)}) gh = d[d.hit] rows.append({"选样": "★真线段端点（未来函数=上界）", "笔数": len(gh), **run(gh, 1.0)}) print(pd.DataFrame(rows).to_string(index=False)) print(""" 判读：实时行若接近 ★ 那行，说明 ext_run 抓到了同一批信号，一类线值得继续； 若仍贴近「全体」，说明实时特征抓不到那 28.7%，天花板兑现不了。 ⚠️ 即便好也不算数——入场仍在 sure_time 上，必须再过一次因果回放。""") print("\n" + "=" * 96) print("【三】数字对不上，追一下：ext_run 挑出的真端点，质量还一样吗") print("=" * 96) print("Q4 的真端点浓度 44%（基线 28.7%），若真端点都值 PF 2.29，" "Q4 不该只有 0.51。\n拆开看 ext_run 在真端点**内部**是帮忙还是帮倒忙——" "这决定天花板是否可学：") rows = [] for k, g in d.groupby(d.hit): for bk, gg in g.groupby("bucket", observed=True): if len(gg) < 25: continue rows.append({"真端点": "是" if k else "否", "ext_run档": bk, "笔数": len(gg), **run(gg, 1.0)}) print(pd.DataFrame(rows).to_string(index=False)) print(""" 「是」组内若 Q4 明显低于 Q1 -> ext_run 在真端点里**反向选样**， 它提高浓度的同时挑走了最差的那些，两个效应抵消，这解释了 0.51 vs 2.29。 若「是」组内各档持平 -> 浓度提升是真的，缺的只是更强的实时特征。""") print("\n" + "=" * 96) print("【四】那需要多高的实时精度才能翻正") print("=" * 96) print("【三】显示只有「是不是真端点」这一个变量在起作用（组内各档持平）。" "\n于是 PF 只是两组按精度 p 的混合，可以直接解出盈亏平衡精度：") net = {} for k, g in d.groupby(d.hit): ns = [] for sym, x in g.groupby("sym"): x = x.reset_index(drop=True) t = pd.DataFrame({"entry_idx": x.i_sure.values, "direction": x.dir.values, "atr_at_entry": x.atr_at_entry.values}) res = simulate(cache[sym], t, x.to_ext.values + 1.0, r_scale=False) ok = res.g.notna().values from lib.exit_model import fee_of ns.append(res.g[ok].values - fee_of(res.r[ok].values, res.c[ok].values)) net[bool(k)] = np.concatenate(ns) def mix_pf(p: float) -> float: """精度 p 时的 PF：两组按 p 加权（组内分布不变，只变权重）。""" h, m = net[True], net[False] w = (p * h[h > 0].sum() / len(h) + (1 - p) * m[m > 0].sum() / len(m)) l = (p * -h[h <= 0].sum() / len(h) + (1 - p) * -m[m <= 0].sum() / len(m)) return w / l if l > 0 else np.inf grid = [0.287, 0.35, 0.44, 0.5, 0.6, 0.7, 0.8, 1.0] print(pd.DataFrame([{ "实时精度": f"{p*100:.1f}%", "PF": round(mix_pf(p), 2), "备注": {0.287: "← 当前基线", 0.44: "← ext_run Q4 已达到", 1.0: "← step66 上界"}.get(p, "")} for p in grid] ).to_string(index=False)) lo, hi = 0.287, 1.0 if mix_pf(hi) > 1 > mix_pf(lo): for _ in range(40): mid = (lo + hi) / 2 lo, hi = (mid, hi) if mix_pf(mid) < 1 else (lo, mid) print(f"\n **盈亏平衡精度 ≈ {(lo+hi)/2*100:.1f}%** " f"（当前 28.7%，ext_run Q4 已到 44.0%）") print(" 这就是「改识别规则」要够到的具体门槛，且它只是毛平衡；" "\n 还要再扣 §3.396 的因果滞后（实盘比 sure_time 再晚 15 根）。") if __name__ == "__main__": main()