jackyu66gitandCursor
7292ef5dd4
把 step62 的 ext_run 质量过滤接到 step56 的低滞后版上——两者各解决一半约束， 是唯一同时处理滞后和识别精度的路径。 失败，且失败方式本身是判据：过滤器符号反了。5m 上延伸度分档在滞后版是 Q1 0.12 / Q4 0.46（越延伸越好），在低滞后版是 Q1 0.58 / Q4 0.33（越短越好）。 叠加后 PF 从 0.37 掉到 0.30~0.33，比不过滤更差。div 同样翻转。 同一批信号、同一个特征，换个入场时点最优方向就反过来，说明这些过滤效果是 入场时点的交互产物而非信号的稳定属性，继续挑阈值就是拟合噪声。 一类线六次独立进攻全部止步 1.0 以下：引擎原生 0.24、反手 0.92（因果回放后）、 实时重写 0.41、结构止损 0.36、用未来函数选样 0.83、质量过滤 0.54、 低滞后+过滤 0.33。第五项尤其说明问题——即使选样做到完美也只到 0.83。 判定关闭。病根是入场价已在结构底上方 2.9 ATR、实盘再晚 15 根，来自「等笔确认」 机制本身而非参数。识别可以优化（精度能翻倍），但识别从来不是瓶颈。 fast_bsp1 顺带补 ext_atr 字段，用 ATR 归一才与 step62 的 ext_run 同口径。 Co-authored-by: Cursor <cursoragent@cursor.com>
225 lines
9.3 KiB
Python
225 lines
9.3 KiB
Python
"""实时版一类买卖点（fast B1/S1）能否翻正。
|
||
|
||
step55 已判定引擎的 B1/S1 不可用（5m PF 0.24/0.20、胜率 18.6%/14.8%、
|
||
t −17/−22），且病根是滞后 8~9 根带来的几何劣势，不是参数。B3 当年同病
|
||
（PF 0.66），靠重新定义成实时判据翻到 1.59（B4）。本脚本对一类做同样的尝试。
|
||
|
||
对照组三条，缺一不可：
|
||
引擎 B1/S1 step55 的数，说明「不改判据」是什么下场
|
||
fast B1/S1 本次
|
||
fast B3/S3 同一份数据、同一套出场跑 B4，确认管线本身能跑出正数
|
||
—— 少了它，fast B1 若为负就分不清是判据不行还是管线接错了
|
||
"""
|
||
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" / "step56_fast_bsp1.feather"
|
||
SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48
|
||
|
||
|
||
def collect(sym: str, rows: int, tf: str) -> 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.exit_model import cfg_name, walk_exits
|
||
from lib.fast_bsp1 import find_fast_bsp1, zones_with_enter_area
|
||
|
||
try:
|
||
df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows)
|
||
if df is None or len(df) < 5_000:
|
||
return None
|
||
chan = TF_DF(df, 1, tf)
|
||
cdf = ensure_timestamp(chan.dataframe)
|
||
zs_list = chan.cal_bi_zs_list_pure(chan.bi_list)
|
||
if not zs_list:
|
||
return None
|
||
|
||
atr = cdf["atr"].to_numpy(float)
|
||
cl = cdf["close"].to_numpy(float)
|
||
cfg = cfg_name(SL, RUNNER, MAXB, RSTOP)
|
||
parts = []
|
||
|
||
d1 = {}
|
||
s1 = find_fast_bsp1(cdf, zones_with_enter_area(zs_list, cdf), diag=d1)
|
||
if not s1.empty:
|
||
s1["kind"] = "fastB1"
|
||
parts.append(s1)
|
||
# 同数据同出场跑一遍 B4，作为「管线能出正数」的存在性证明
|
||
s3 = find_fast_bsp3(cdf, zones_from_zs_list(zs_list, cdf))
|
||
if not s3.empty:
|
||
s3["kind"] = "fastB3"
|
||
parts.append(s3)
|
||
if not parts:
|
||
return None
|
||
|
||
r = pd.concat(parts, ignore_index=True)
|
||
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, r[["entry_idx", "direction"]], [SL], [RUNNER],
|
||
[MAXB], scale_at=SCALE_AT, runners=(RUNNER,),
|
||
runner_stops=(RSTOP,))
|
||
if len(res) != len(r):
|
||
return None
|
||
for k in ("g", "r", "c", "b"):
|
||
r[k] = res[f"{cfg}_{k}"].to_numpy()
|
||
r["sym"], r["tf"] = sym, tf
|
||
r["atr_pct"] = atr[r.entry_idx.values] / cl[r.entry_idx.values]
|
||
r["date"] = cdf["date"].to_numpy()[r.entry_idx.values]
|
||
r["diag"] = str(d1)
|
||
return r
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" {sym} {tf} 失败: {type(e).__name__}: {e}", flush=True)
|
||
return None
|
||
|
||
|
||
def stats(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,
|
||
"R夏普": round(R.mean() / R.std(ddof=1), 3),
|
||
"余量bp": round(net.mean() / tn.mean() * 1e4, 2),
|
||
"中位持仓": int(np.median(g.b.values)),
|
||
"t值": round(gR.mean() / (gR.std(ddof=1) / np.sqrt(len(g))), 2),
|
||
}
|
||
|
||
|
||
def report(d: pd.DataFrame) -> None:
|
||
for tf, x in d.groupby("tf"):
|
||
span = (pd.to_datetime(x.date).max()
|
||
- pd.to_datetime(x.date).min()).total_seconds() / 86400
|
||
print("\n" + "#" * 92)
|
||
print(f"########## {tf} · {x.sym.nunique()} 币 · 跨 {span:.0f} 天"
|
||
f" ##########")
|
||
rows = []
|
||
for (kind, dirn), g in x.groupby(["kind", "direction"]):
|
||
if len(g) < 40:
|
||
continue
|
||
nm = {"fastB1": ("一买", "一卖"), "fastB3": ("三买", "三卖")}[kind]
|
||
rows.append({"信号": f"{kind} {nm[0 if dirn == 1 else 1]}",
|
||
"每币每天": round(len(g) / span / x.sym.nunique(), 3),
|
||
**stats(g)})
|
||
if rows:
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
print("\n对照 · step55 引擎原生（同周期同出场）：")
|
||
print(" 5m B1 PF 0.24 胜率 18.6% t −17.3 ｜ S1 PF 0.20 胜率 14.8% "
|
||
"t −22.3")
|
||
print(" 15m B1 PF 0.21 t −18.9 ｜ S1 PF 0.23 t −15.9")
|
||
|
||
f1 = x[x.kind == "fastB1"]
|
||
if len(f1) >= 120 and "ext_atr" in f1.columns:
|
||
y = f1.dropna(subset=["ext_atr"])
|
||
print("\n延伸度分档（ext_atr，step62 在滞后版上实测的唯一单调特征："
|
||
"\n命中线段顶点的比例 12%→44%，PF 0.12→0.46）：")
|
||
q = pd.qcut(y["ext_atr"], 4,
|
||
labels=["Q1最短", "Q2", "Q3", "Q4最延伸"],
|
||
duplicates="drop")
|
||
print(pd.DataFrame([{"档": k, **stats(v)}
|
||
for k, v in y.groupby(q, observed=True)
|
||
if len(v) >= 30]).to_string(index=False))
|
||
|
||
print("\n⭐ 组合：低滞后（本模块）+ 延伸过滤（step62）—— "
|
||
"唯一同时处理两个约束的路径")
|
||
p50, p75 = y["ext_atr"].quantile(.50), y["ext_atr"].quantile(.75)
|
||
dm = y["div"].median()
|
||
rows = [{"过滤器": "无", **stats(y)}]
|
||
for nm, g in [
|
||
(f"ext≥P50({p50:.1f})", y[y["ext_atr"] >= p50]),
|
||
(f"ext≥P75({p75:.1f})", y[y["ext_atr"] >= p75]),
|
||
(f"ext≥P75 且 div≥中位", y[(y["ext_atr"] >= p75)
|
||
& (y["div"] >= dm)]),
|
||
(f"ext≥P50 且 div≥中位", y[(y["ext_atr"] >= p50)
|
||
& (y["div"] >= dm)]),
|
||
]:
|
||
if len(g) >= 30:
|
||
rows.append({"过滤器": nm, **stats(g)})
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
print("对照 · 同过滤器在滞后版（step62，5m）：无 PF 0.22 → "
|
||
"ext≥P75且div≥中位 PF 0.54")
|
||
|
||
if len(f1) >= 60:
|
||
print("\n背驰强度分档（div = 离开段面积/进入段面积，越小背驰越强）：")
|
||
q = pd.qcut(f1["div"], 4, labels=["Q1最强", "Q2", "Q3", "Q4最弱"],
|
||
duplicates="drop")
|
||
print(pd.DataFrame([{"档": k, **stats(v)}
|
||
for k, v in f1.groupby(q, observed=True)
|
||
if len(v) >= 20]).to_string(index=False))
|
||
print("\n滞后分档（lag = 入场根 − 离开段极值根）：")
|
||
b = pd.cut(f1["lag"], [-1, 1, 3, 6, 12, 1e9],
|
||
labels=["≤1根", "2-3根", "4-6根", "7-12根", ">12根"])
|
||
print(pd.DataFrame([{"档": k, **stats(v)}
|
||
for k, v in f1.groupby(b, observed=True)
|
||
if len(v) >= 20]).to_string(index=False))
|
||
|
||
dg = d[d.kind == "fastB1"].diag.dropna()
|
||
if len(dg):
|
||
print("\n" + "=" * 92)
|
||
print("fast B1 漏斗（首个中枢样本）：", dg.iloc[0])
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE")
|
||
ap.add_argument("--tfs", default="5m,15m")
|
||
ap.add_argument("--rows", type=int, default=200_000)
|
||
ap.add_argument("--workers", type=int, default=3)
|
||
ap.add_argument("--reuse", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
if args.reuse and OUT.exists():
|
||
d = pd.read_feather(OUT)
|
||
else:
|
||
syms = [s.strip() for s in args.symbols.split(",")]
|
||
tfs = [t.strip() for t in args.tfs.split(",")]
|
||
print(f"[实时版一类] {len(syms)} 币 × {tfs} × {args.rows} 根\n",
|
||
flush=True)
|
||
parts = []
|
||
with ProcessPoolExecutor(max_workers=args.workers) as ex:
|
||
fut = {ex.submit(collect, s, args.rows, t): (s, t)
|
||
for s in syms for t in tfs}
|
||
for i, f in enumerate(as_completed(fut), 1):
|
||
r = f.result()
|
||
s, t = fut[f]
|
||
print(f" [{i}/{len(fut)}] {s} {t} "
|
||
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()
|