research: 一类线门槛量化——全局只有一个变量，精度需 67~73%
把 §3.3992 的天花板换成实时可算的 ext_run 选样配结构止损：5m 0.51、15m 0.48， 对比上界 2.29/4.11，兑现不了。但数字对不上（Q4 真端点浓度已 44%），拆开后 得到本轮最有价值的一张表： ext_run 在真端点**内部**毫无反向选样（5m 各档 2.53/1.80/2.61/2.22 持平， 15m 单调递增到 6.10），在非真端点内部也毫无区分力（恒在 0.11~0.18）。 全局只有「是不是真端点」这一个变量在起作用，其余特征都是它的噪声代理。 这也修正了 §3.399：ext_run 的符号翻转不是拟合，它确实提纯（28.7%→44%）， 只是幅度远远不够。 于是 PF 退化为两组按精度 p 的混合，解出盈亏平衡精度：5m 72.6%、15m 67.1%。 即要求实时判「这是不是那个底」的准确率达七成，而现在是 28.7%。 顺带解释了为什么 B4 能做而一类不能：B4 是突破后的延续信号，不需要判断反转； 一类的全部难度集中在这一个二分类上。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,195 @@
|
||||
"""把 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()
|
||||
Reference in New Issue
Block a user