jackyu66gitandCursor
d2fcb27d01
§5.41 发现 available_ts 取「中枢最后一笔」是右边缘重画的根因，改取「第三笔」 能把重画率从 6.5% 压到 1.2%，当时据此判断它是「唯一可能同时改善收益与稳定性」 的改动。那个判断只测了稳定性，过早了。 8 个样本外币 × 30 万根 1m，两组共用同一个 TF_DF，只切 available_ts 的取法。 实盘口径（深色 ∧ ATR≥8bp，955 vs 989 笔）： 毛 R 0.933 → -0.000 净均 R 0.798 → -0.147 PF 3.22 → 0.80 滑点余量 15.07 → -2.20 bp 判决依据是毛 R 那一行：扣任何费用之前 edge 就没了，所以不是成本、门控或出场 参数的问题，是信号本身不再有预测力。逐币 8/8 全部变差。滞后确实降了 （2.16 → 2.01），但换来的是另一批交易——两组重合度只有约 30%。 原因是中枢没发育完就下注，支撑/压力还没立住。「等中枢最后一笔」那段等待不是 可以优化掉的延迟，它就是 alpha 本身。由此得一条一般规则：任何以「让信号更早 确定」为目标的改动，先测毛 R，不能只看重画率和滞后。 开关 AVAIL_BI_INDEX 保留只为可复现该 A/B，默认 -1 维持现行口径。环境变量在 调用时解析而非 import 时——fork 启动的子进程会继承已 import 的模块，import 时读会固化成父进程的值。 顺带交叉验证：现行口径本次算出滑点余量 15.07bp，与用优化前代码算的同组同期 15.19bp 吻合。 Co-authored-by: Cursor <cursoragent@cursor.com>
196 lines
7.8 KiB
Python
196 lines
7.8 KiB
Python
"""Step 47：中枢可用时刻取 bis[-1] 还是 bis[2] —— 按收益判，不按重画率判。
|
||
|
||
§5.41 发现 available_ts 取「中枢最后一笔」是右边缘重画的根因，改取「第三笔」
|
||
（中枢成立即固定）能把重画率从 6.5% 压到 1.2%。但那只测了稳定性。
|
||
|
||
小样本预检（60k 根 × 5 个币/周期）显示这不是一次「稳定性修补」：
|
||
信号数 +17% ~ +44%，而两组的**重合度只有约 30%**。
|
||
即 bis[2] 丢掉了原信号的多数，又换进来一批新的。
|
||
换句话说它是另一个策略，不是同一个策略的低延迟版。所以判据必须是扣费后的
|
||
净 R / PF / 滑点预算，重画率只能作为次要参考。
|
||
|
||
口径与 step44 一致（当前 1m 最优）：
|
||
出场 SL 2.0 / 3 ATR 减半 / runner 目标 8 ATR / runner 止损留原位 / 48 根超时
|
||
成本 taker 2bp、maker 0.8bp，滑点只加在 taker 腿
|
||
门控 ATR >= 8bp
|
||
过滤 深色（同向 ∧ 阶梯）——实盘只做这一档，见 §3.38
|
||
|
||
两组共用同一个 TF_DF，只切 available_ts 的取法，确保差异只来自这一处。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
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")
|
||
for v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
|
||
os.environ.setdefault(v, "1")
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(HERE))
|
||
sys.path.insert(0, str(HERE.parent))
|
||
pd.set_option("display.width", 340)
|
||
|
||
LTF, HTF = "1m", "5m"
|
||
SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48
|
||
GATE_BP = 8.0
|
||
VARIANTS = (("bis[-1] 现行", -1), ("bis[2] 中枢成立", 2))
|
||
|
||
|
||
def collect(sym: str, rows: int) -> pd.DataFrame | None:
|
||
import warnings as _w
|
||
_w.filterwarnings("ignore")
|
||
sys.path.insert(0, str(HERE))
|
||
sys.path.insert(0, str(HERE.parent))
|
||
|
||
from chanlun import TF_DF
|
||
from chanlun.analysis.fast_bsp import (
|
||
add_zone_ladder, attach_htf_agree, attach_zone_ladder,
|
||
build_htf_zones, find_fast_bsp3, htf_fx_timeline,
|
||
)
|
||
from lib.data import fetch_ohlcv
|
||
from lib.exit_model import cfg_name, walk_exits
|
||
|
||
try:
|
||
df = fetch_ohlcv(f"{sym}/USDT:USDT", LTF, rows)
|
||
if df is None or len(df) < 50_000:
|
||
return None
|
||
# lean：只需要笔/中枢/信号，跳过线段与 MACD 状态机（见 §5.6，已验证等价）
|
||
chan = TF_DF(df, 1, LTF, lean=True)
|
||
cdf = chan.dataframe
|
||
|
||
df_h = fetch_ohlcv(f"{sym}/USDT:USDT", HTF, 10 ** 9)
|
||
chan_h = TF_DF(df_h, 1, HTF, lean=True)
|
||
tl = htf_fx_timeline(chan_h, chan_h.dataframe)
|
||
|
||
idx_all, parts = None, []
|
||
for label, avail_bi in VARIANTS:
|
||
zones = build_htf_zones(cdf, LTF, chan=chan, avail_bi=avail_bi)
|
||
if zones.empty:
|
||
continue
|
||
zl = add_zone_ladder(zones.reset_index(drop=True))
|
||
sig = find_fast_bsp3(cdf, zl)
|
||
if sig.empty:
|
||
continue
|
||
sig = attach_zone_ladder(attach_htf_agree(sig, cdf, tl), zl)
|
||
|
||
res = walk_exits(cdf, sig, [SL], [RUNNER], [MAXB],
|
||
scale_at=SCALE_AT, runners=(RUNNER,), runner_stops=(RSTOP,))
|
||
cfg = cfg_name(SL, RUNNER, MAXB, RSTOP)
|
||
need = [f"{cfg}_g", f"{cfg}_r", f"{cfg}_c", f"{cfg}_b"]
|
||
if any(c not in res.columns for c in need):
|
||
continue
|
||
|
||
idx = sig["entry_idx"].to_numpy().astype(int)
|
||
atr = cdf["atr"].to_numpy(float)[idx]
|
||
close = cdf["close"].to_numpy(float)[idx]
|
||
out = res[need].copy()
|
||
out.columns = ["g", "r", "c", "b"]
|
||
out["sym"] = sym
|
||
out["variant"] = label
|
||
out["atr_pct"] = atr / close
|
||
out["lag"] = sig["lag"].to_numpy()
|
||
out["entry_idx"] = idx
|
||
out["htf_agree"] = sig["htf_agree"].to_numpy()
|
||
out["ladder_ok"] = sig["ladder_ok"].to_numpy()
|
||
parts.append(out)
|
||
return pd.concat(parts, ignore_index=True) if parts else None
|
||
except Exception as e:
|
||
print(f" {sym} 失败: {e!r}", flush=True)
|
||
return None
|
||
|
||
|
||
def describe(g: pd.DataFrame, label: str) -> dict:
|
||
from lib.exit_model import fee_of, taker_notional
|
||
|
||
if len(g) < 40:
|
||
return {"口径": label, "笔数": len(g), "备注": "样本不足"}
|
||
gross = g["g"].to_numpy()
|
||
reason, scaled = g["r"].to_numpy(), g["c"].to_numpy()
|
||
net = gross - fee_of(reason, scaled) # 未扣滑点：剩下的就是滑点余量
|
||
denom = SL * g["atr_pct"].to_numpy()
|
||
R, gR = net / denom, gross / denom
|
||
w, o = net[net > 0], -net[net <= 0].sum()
|
||
q = np.quantile(net[net > 0], 0.90) if (net > 0).any() else 0.0
|
||
t10 = net[(net > 0) & (net <= q)].sum()
|
||
tn = taker_notional(reason, scaled)
|
||
return {
|
||
"口径": label, "笔数": len(g),
|
||
"滞后": round(float(g["lag"].mean()), 2),
|
||
"胜率": f"{(net > 0).mean() * 100:.1f}%",
|
||
"毛R": round(gR.mean(), 3),
|
||
"净均R": round(R.mean(), 3),
|
||
"R夏普": round(R.mean() / R.std(ddof=1), 3),
|
||
"PF": round(w.sum() / o, 2) if o > 0 else np.inf,
|
||
"剔10%PF": round(t10 / o, 2) if o > 0 else np.inf,
|
||
"滑点余量bp": round(net.mean() / tn.mean() * 1e4, 2),
|
||
}
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--symbols", default="BTC,ETH,SOL,XRP,DOGE,LINK,ADA,LTC")
|
||
ap.add_argument("--rows", type=int, default=300_000)
|
||
ap.add_argument("--workers", type=int, default=3)
|
||
args = ap.parse_args()
|
||
|
||
syms = [s.strip() for s in args.symbols.split(",")]
|
||
print(f"[available_ts A/B] {len(syms)} 币 × {args.rows} 根 {LTF}\n"
|
||
f"出场 SL{SL}/减半{SCALE_AT}/runner{RUNNER}/rstop{RSTOP}/{MAXB}根，"
|
||
f"ATR 门控 {GATE_BP:g}bp\n", flush=True)
|
||
|
||
parts = []
|
||
with ProcessPoolExecutor(max_workers=args.workers) as ex:
|
||
futs = {ex.submit(collect, s, args.rows): s for s in syms}
|
||
for i, f in enumerate(as_completed(futs), 1):
|
||
r = f.result()
|
||
if r is None:
|
||
print(f" [{i}/{len(syms)}] {futs[f]} 跳过", flush=True)
|
||
continue
|
||
parts.append(r)
|
||
n = r.groupby("variant").size().to_dict()
|
||
print(f" [{i}/{len(syms)}] {futs[f]} {n}", flush=True)
|
||
if not parts:
|
||
print("无结果")
|
||
return
|
||
|
||
d = pd.concat(parts, ignore_index=True)
|
||
d["a_bp"] = d["atr_pct"] * 1e4
|
||
d.to_feather(HERE / "out" / "step47_avail_bi.feather")
|
||
|
||
dark = (d["htf_agree"] == 1.0) & d["ladder_ok"]
|
||
for lab, dd in (("全部信号", d), ("深色（实盘口径）", d[dark])):
|
||
for gate_lab, ddd in (("未门控", dd), (f"ATR>={GATE_BP:g}bp", dd[dd.a_bp >= GATE_BP])):
|
||
print("\n" + "=" * 118)
|
||
print(f"########## {lab} / {gate_lab} ##########")
|
||
print(pd.DataFrame([describe(ddd[ddd.variant == v], v)
|
||
for v, _ in [(a, b) for a, b in VARIANTS]]).to_string(index=False))
|
||
|
||
print("\n########## 逐币（深色 + 门控）##########")
|
||
dd = d[dark & (d.a_bp >= GATE_BP)]
|
||
rows = []
|
||
for s, g in dd.groupby("sym"):
|
||
r = {"币": s}
|
||
for v, _ in VARIANTS:
|
||
x = describe(g[g.variant == v], v)
|
||
tag = "现行" if "-1" in v else "新"
|
||
r[f"{tag}_笔数"] = x.get("笔数")
|
||
r[f"{tag}_净均R"] = x.get("净均R")
|
||
r[f"{tag}_余量bp"] = x.get("滑点余量bp")
|
||
rows.append(r)
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
|
||
print("\n判据：净均R 与滑点余量bp 同时不劣于现行，才值得换。"
|
||
"\n信号数变多本身不是好处——重合度只有约 30%，换的是另一批交易。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|