research: 一类线重开——病根是选样×止损的交互，不是没机会
用户问「修改 B1/B2 的识别规则呢」，查下来推翻了 §3.399 的关闭结论。 step65 先量机会本身：直接取 seg_list（不经过检测器），线段中位幅度 11.4 ATR， 扣掉 2.9 ATR 确认成本还剩 8.5，是止损的 4.26 倍，93% 的线段装得下。 五币四级别一致。**滞后不是瓶颈**，与原先预期相反。 但这与 §3.397 矛盾（空间在却拿不到），step66 给出解答：之前六次进攻每次只动 一半。完美选样配 2ATR 止损是 0.70/0.83，无选样配结构止损是 0.36， **两个一起是 2.29(5m)/4.11(15m)**，胜率 36%→63% / 44%→70%。是交互不是叠加。 机制来自 MFE/MAE：真底那批逆向行程中位仅 2.24 ATR，2 ATR 止损打掉了 51~58% 的好单；非真底那批逆向行程中位 4.76，放宽只是亏更多。固定 2 ATR 同时做错两件事。 ⚠️ 上界含两个未来函数（线段端点选样、sure_time 入场），是靶子不是策略。 顺带证伪一个我自己的猜测：step58 首轮用 r_scale=True 把 3.9 ATR 止损对应的 runner 目标推到 15.6 ATR（比整段行情还长）。加 --abs-targets 改绝对目标重跑， PF 仍是 0.36，与 r_scale 版完全相同。首轮结论正确，只是理由错了。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,152 @@
|
||||
"""改识别规则到底有没有用：把天花板一次性测到顶。
|
||||
|
||||
用户问「改 B1/B2 的识别规则呢」。step65 让这个问题变得可判定，但也制造了一个
|
||||
必须解释的矛盾：
|
||||
|
||||
线段中位 11.4 ATR，进场花掉 2.9，剩 8.5 —— 是 2 ATR 止损的 4.26 倍。
|
||||
可 step60 用未来函数选出真底的那批 B1，PF 仍只有 0.70~0.83。
|
||||
|
||||
空间明明在，就是拿不到。两种解释对用户的问题给出**相反**的答案：
|
||||
|
||||
选样问题：只有 30% 的 B1 落在真线段底，那 8.5 ATR 只存在于这 30% 里，
|
||||
其余 70% 是趋势中途，后面根本没有行情 -> **改识别规则有用**
|
||||
路径问题：就算落在真底，11.4 ATR 是净幅度，路上的回撤照样打掉止损
|
||||
-> **改识别规则没用**
|
||||
|
||||
区分只需要一个从没跑过的组合：**未来函数选样 + 结构止损 + 绝对目标**。
|
||||
三样单独都失败过（step60 的 0.83 / step58 的 0.36 / 本轮 abs 的 0.36），
|
||||
组合起来没试过。它测的不是某条规则，是**所有识别规则的上界**：
|
||||
|
||||
上界 < 1.0 -> 路径问题，一类线彻底关闭，别再改规则
|
||||
上界 > 1.5 -> 选样问题，改识别规则有用，且这里就是要够到的目标
|
||||
|
||||
同时量 MFE 做交叉验证：真底那批到底跑没跑出 8.5 ATR。
|
||||
跑出来了却还是亏 -> 铁证是止损/路径
|
||||
根本没跑出来 -> 说明 11.4 ATR 那个测量传导不到这批信号上，step65 要重解释
|
||||
"""
|
||||
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))
|
||||
|
||||
SRC = HERE / "out" / "step64_b2.feather"
|
||||
MAXB = 48
|
||||
|
||||
|
||||
def mfe_mae(cdf: pd.DataFrame, idx: np.ndarray, dirs: np.ndarray,
|
||||
atr: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""入场后 MAXB 根内最远的顺向/逆向行程（ATR）。"""
|
||||
hi, lo = cdf["high"].to_numpy(float), cdf["low"].to_numpy(float)
|
||||
op = cdf["open"].to_numpy(float)
|
||||
n = len(cdf)
|
||||
fe = np.full(len(idx), np.nan)
|
||||
ae = np.full(len(idx), np.nan)
|
||||
for k, (i, d, a) in enumerate(zip(idx, dirs, atr)):
|
||||
e = int(i) + 1
|
||||
if e >= n or not np.isfinite(a) or a <= 0:
|
||||
continue
|
||||
j = min(e + MAXB, n)
|
||||
px = op[e]
|
||||
if d == 1:
|
||||
fe[k] = (hi[e:j].max() - px) / a
|
||||
ae[k] = (px - lo[e:j].min()) / a
|
||||
else:
|
||||
fe[k] = (px - lo[e:j].min()) / a
|
||||
ae[k] = (hi[e:j].max() - px) / a
|
||||
return fe, ae
|
||||
|
||||
|
||||
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
|
||||
|
||||
d = pd.read_feather(SRC)
|
||||
d = d[(d.tf == args.tf) & d.type.isin(["B1", "S1"])].copy()
|
||||
d["to_ext"] = (d.entry_px - d.own_ext) * d.dir / d.atr_at_entry
|
||||
print(f"[识别规则天花板] {args.tf} · 一类 {len(d)} 笔 · "
|
||||
f"其中落在真线段端点 {d.hit.sum()} 笔 ({d.hit.mean()*100:.1f}%)\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
|
||||
|
||||
print("=" * 96)
|
||||
print("【一】MFE 交叉验证：真底那批到底跑没跑出 step65 说的 8.5 ATR")
|
||||
print("=" * 96)
|
||||
rows = []
|
||||
for k, g in d.groupby(d.hit):
|
||||
fe, ae = [], []
|
||||
for sym, x in g.groupby("sym"):
|
||||
f_, a_ = mfe_mae(cache[sym], x.i_sure.values, x.dir.values,
|
||||
x.atr_at_entry.values)
|
||||
fe.append(f_)
|
||||
ae.append(a_)
|
||||
fe, ae = np.concatenate(fe), np.concatenate(ae)
|
||||
ok = np.isfinite(fe)
|
||||
rows.append({
|
||||
"落在真线段端点": "是" if k else "否", "笔数": int(ok.sum()),
|
||||
"MFE中位": round(float(np.median(fe[ok])), 2),
|
||||
"MFE≥3ATR": f"{(fe[ok] >= 3).mean()*100:.0f}%",
|
||||
"MFE≥8ATR": f"{(fe[ok] >= 8).mean()*100:.0f}%",
|
||||
"MAE中位": round(float(np.median(ae[ok])), 2),
|
||||
"MAE≥2ATR(会被现止损打掉)": f"{(ae[ok] >= 2).mean()*100:.0f}%",
|
||||
})
|
||||
print(pd.DataFrame(rows).to_string(index=False))
|
||||
print("""
|
||||
判读：若「是」那行 MFE 中位远低于 8.5，说明 step65 的线段幅度传导不到
|
||||
信号上（进场时行情已经走掉了大半）；若 MFE 够大而 MAE 也大，则是路径问题。""")
|
||||
|
||||
print("\n" + "=" * 96)
|
||||
print("【二】天花板：未来函数选样 + 结构止损 + 绝对目标（三者首次组合）")
|
||||
print("=" * 96)
|
||||
rows = []
|
||||
for k, g in d.groupby(d.hit):
|
||||
for label, margin in [("固定2ATR", None), ("结构+0.5", 0.5),
|
||||
("结构+1.0", 1.0), ("结构+1.5", 1.5)]:
|
||||
gs, rs, cs, ap_, sl_ = [], [], [], [], []
|
||||
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})
|
||||
sl = (np.full(len(x), 2.0) if margin is None
|
||||
else (x.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((x.atr_at_entry.values / x.entry_px.values)[ok])
|
||||
sl_.append(sl[ok])
|
||||
rows.append({
|
||||
"落在真线段端点": "是" if k else "否", "止损": label,
|
||||
**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(pd.DataFrame(rows).to_string(index=False))
|
||||
print("""
|
||||
「是」那几行就是**任何识别规则的上界**（因为选样已经用了未来函数）：
|
||||
< 1.0 路径问题，一类线关闭，改规则无解
|
||||
> 1.5 选样问题，改识别规则有用，且这就是要够到的目标""")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user