jackyu66gitandCursor
e137d92b50
step59 逐根回放：init_stream 预热后逐根 append_bar，每根重算笔中枢与 bsp， 记录信号首现根并按它入场。信号身份用「类型+极值KLC时刻」而非 sure_time， 后者正是会被重画的字段。必须逐根，分段重建等于多给信息。 结果把 §3.395 推翻了：召回 100%、幻影 0，即存在性是因果的；但首现根比全量 sure_time 晚中位 15 根、P90 31 根，0% 能准时拿到。按真实首现根入场， B1 反手 PF 2.35→0.92、S1 2.75→1.30，t 0.20/0.64 完全不显著。 教训：sure_time 是引擎事后标注的确认时刻，不等于可执行时刻。任何拿它当 entry 的回测都要先过逐根回放。 step60 用线段终点当标准答案验证识别本身。对照组取所有同向笔端点——B1 按构造 就长在笔低点上，不设这个基准任何绝对命中率都无法解读。5m 上 B1 命中 30.3% 对基准 15.0%，提升 2.02 倍，S1 1.81 倍；B3/S3 恰为 0%，符合三类长在趋势 中段的预期，两者互为标签有效性旁证。 所以识别是对的，但精度只有 30%，且那 70% 噪声 PF 只有 0.08。更关键的是即使 用未来函数把精度提到 100%，命中组也只有 PF 0.70~0.83，仍不赚钱——因为入场价 已在结构底上方 2.90 ATR。一类线三层逐层否定后到此为止。 Co-authored-by: Cursor <cursoragent@cursor.com>
237 lines
9.4 KiB
Python
237 lines
9.4 KiB
Python
"""因果性回放：一类反手的信号在当时真的发得出来吗？
|
||
|
||
§3.395 的 PF 2.3~3.0 建立在全量数据一次算完的 `bsp_list` 上。但增量模块的
|
||
文件头自己写着「笔必须整表重扫：**最后一笔 is_sure 允许收回**」，§5.41 也记了
|
||
中枢右边缘会重画。若信号是事后才浮现的，那个 PF 就是幻觉。
|
||
|
||
**做法**：用 `init_stream` 预热，随后逐根 `append_bar`，每根之后重算
|
||
`cal_bi_zs_list_pure` + `find_all_bsp`，记录每个信号**第一次出现**在哪一根。
|
||
入场用那一根（的次根开盘），而不是事后的 `sure_time` —— 实盘只能这样。
|
||
|
||
必须逐根，不能分段重建：在第 t 根用 `data[0:t+S]` 重算等于多给了 S 根的信息，
|
||
测出来的因果性是假的。
|
||
|
||
**三个要看的量**
|
||
召回 全量算出的信号，有多少在回放中真的出现过（没出现的是事后才浮现）
|
||
幻影 回放中出现、但全量里没有的（当时发了、后来被重画掉）
|
||
代价 回放首现根 vs 全量 sure_time 的滞后；以及按首现根入场的实际 PF
|
||
"""
|
||
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" / "step59_replay.feather"
|
||
SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48
|
||
WANT = {"B1": 1, "S1": -1}
|
||
|
||
|
||
def sig_key(b) -> tuple | None:
|
||
"""信号身份用「类型 + 极值 KLC 的结束时刻」。
|
||
|
||
不能用 sure_time 当身份：它正是会被重画的字段，用它做键会把同一个信号
|
||
在不同根上算成两个。极值点稳定得多。
|
||
"""
|
||
t = getattr(b.type, "name", str(b.type))
|
||
if t not in WANT:
|
||
return None
|
||
return (t, str(b.klc.end_time))
|
||
|
||
|
||
def replay(sym: str, tf: str, rows: int, warm: int, steps: int) -> pd.DataFrame | None:
|
||
from chanlun import TF_DF
|
||
from lib.data import fetch_ohlcv
|
||
|
||
df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows)
|
||
if df is None or len(df) < warm + steps + 100:
|
||
print(f" {sym} {tf} 数据不足 {0 if df is None else len(df)}", flush=True)
|
||
return None
|
||
df = df.iloc[-(warm + steps):].reset_index(drop=True)
|
||
|
||
# ---- 全量口径：一次算完，作为对照 ----
|
||
full = TF_DF(df, 1, tf, lean=False)
|
||
fz = full.cal_bi_zs_list_pure(full.bi_list)
|
||
full_sig = {}
|
||
for b in (full.find_all_bsp(full.bi_list, fz) or []):
|
||
k = sig_key(b)
|
||
if k and b.sure_time is not None:
|
||
full_sig[k] = str(b.sure_time)
|
||
|
||
# ---- 回放口径：逐根追加，记录首现根 ----
|
||
chan = TF_DF(df.iloc[:warm].copy(), 1, tf, lean=False)
|
||
chan.init_stream(df.iloc[:warm].copy(), 1, tf)
|
||
first_seen: dict[tuple, int] = {}
|
||
for i in range(warm, len(df)):
|
||
chan.append_bar(df.iloc[i])
|
||
try:
|
||
zs = chan.cal_bi_zs_list_pure(chan.bi_list)
|
||
bsp = chan.find_all_bsp(chan.bi_list, zs) if zs else []
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
for b in (bsp or []):
|
||
k = sig_key(b)
|
||
if k and k not in first_seen:
|
||
first_seen[k] = i
|
||
|
||
dser = pd.to_datetime(full.dataframe["date"])
|
||
if dser.dt.tz is not None:
|
||
dser = dser.dt.tz_localize(None)
|
||
didx = pd.DatetimeIndex(dser)
|
||
|
||
def to_i(ts) -> int:
|
||
t = pd.Timestamp(ts)
|
||
return int(didx.searchsorted(t.tz_localize(None) if t.tz else t))
|
||
|
||
rec = []
|
||
for k in set(full_sig) | set(first_seen):
|
||
t, ext_t = k
|
||
i_seen = first_seen.get(k)
|
||
# 只统计回放窗口内的：预热段的信号本来就不在考察范围
|
||
i_ext = to_i(ext_t)
|
||
if i_ext < warm - 200:
|
||
continue
|
||
rec.append({
|
||
"sym": sym, "tf": tf, "type": t, "dir": WANT[t],
|
||
"in_full": k in full_sig, "in_replay": i_seen is not None,
|
||
"i_ext": i_ext,
|
||
"i_seen": -1 if i_seen is None else i_seen,
|
||
"i_sure_full": to_i(full_sig[k]) if k in full_sig else -1,
|
||
})
|
||
r = pd.DataFrame(rec)
|
||
if r.empty:
|
||
return None
|
||
|
||
# 按回放首现根入场，跑与实盘一致的出场
|
||
from lib.exit_model import cfg_name, walk_exits
|
||
cdf = full.dataframe
|
||
atr = cdf["atr"].to_numpy(float)
|
||
cl = cdf["close"].to_numpy(float)
|
||
live = r[r.in_replay & (r.i_seen < len(cdf) - 2)].copy()
|
||
live = live[np.isfinite(atr[live.i_seen.values])
|
||
& (atr[live.i_seen.values] > 0)]
|
||
if not live.empty:
|
||
cfg = cfg_name(SL, RUNNER, MAXB, RSTOP)
|
||
# 反手：§3.395 判定该反着做
|
||
t_ = pd.DataFrame({"entry_idx": live.i_seen.values,
|
||
"direction": -live.dir.values})
|
||
res = walk_exits(cdf, t_, [SL], [RUNNER], [MAXB], scale_at=SCALE_AT,
|
||
runners=(RUNNER,), runner_stops=(RSTOP,))
|
||
if len(res) == len(live):
|
||
for c in ("g", "r", "c"):
|
||
live[c] = res[f"{cfg}_{c}"].to_numpy()
|
||
live["atr_pct"] = atr[live.i_seen.values] / cl[live.i_seen.values]
|
||
r = r.merge(live[["i_ext", "type", "g", "r", "c", "atr_pct"]],
|
||
on=["i_ext", "type"], how="left")
|
||
return r
|
||
|
||
|
||
def report(d: pd.DataFrame) -> None:
|
||
from lib.exit_model import fee_of, taker_notional
|
||
|
||
print("\n" + "=" * 92)
|
||
print("########## 一、召回与幻影 ##########")
|
||
rows = []
|
||
for (tf, t), x in d.groupby(["tf", "type"]):
|
||
full = x[x.in_full]
|
||
rep = x[x.in_replay]
|
||
both = x[x.in_full & x.in_replay]
|
||
rows.append({
|
||
"tf": tf, "类型": t,
|
||
"全量信号": len(full), "回放信号": len(rep),
|
||
"召回": f"{len(both)/max(len(full),1)*100:.1f}%",
|
||
"事后才浮现": len(full) - len(both),
|
||
"幻影(被重画掉)": len(rep) - len(both),
|
||
})
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
print("\n召回 = 全量算出的信号里，回放中真的出现过的比例。"
|
||
"\n幻影 = 回放中发过、全量里却没有的 —— 实盘会照做，回测却看不见它。")
|
||
|
||
print("\n" + "=" * 92)
|
||
print("########## 二、时点代价：回放首现 vs 全量 sure_time ##########")
|
||
b = d[d.in_full & d.in_replay].copy()
|
||
b["delay"] = b.i_seen - b.i_sure_full
|
||
for tf, x in b.groupby("tf"):
|
||
q = x.delay.quantile([.25, .5, .75, .9])
|
||
print(f" {tf} 中位 {q[.5]:+.0f} 根 P25 {q[.25]:+.0f} "
|
||
f"P75 {q[.75]:+.0f} P90 {q[.9]:+.0f} "
|
||
f"｜ 早于或等于全量的占比 {(x.delay <= 0).mean()*100:.0f}%")
|
||
print(" 正值 = 回放比全量晚知道，实盘要在更差的价位入场。")
|
||
|
||
if "g" not in d.columns:
|
||
return
|
||
print("\n" + "=" * 92)
|
||
print("########## 三、真正能落地的收益：按回放首现根入场（反手）##########")
|
||
x = d.dropna(subset=["g"]).copy()
|
||
rows = []
|
||
for (tf, t), g in x.groupby(["tf", "type"]):
|
||
if len(g) < 30:
|
||
continue
|
||
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()
|
||
rows.append({
|
||
"tf": tf, "类型(反手)": t, "笔数": 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,
|
||
"余量bp": round(net.mean() / tn.mean() * 1e4, 2),
|
||
"t值": round(gR.mean() / (gR.std(ddof=1) / np.sqrt(len(g))), 2),
|
||
})
|
||
if rows:
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
print("\n对照 · §3.395 全量口径 5m：B1反手 PF 2.35 / S1反手 2.75")
|
||
print("若这里明显掉下来，说明那个 PF 吃了右边缘重画的红利，不可落地。")
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--symbols", default="BTC,ETH,SOL")
|
||
ap.add_argument("--tf", default="5m")
|
||
ap.add_argument("--rows", type=int, default=60_000)
|
||
ap.add_argument("--warm", type=int, default=20_000)
|
||
ap.add_argument("--steps", type=int, default=10_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():
|
||
report(pd.read_feather(OUT))
|
||
return
|
||
syms = [x.strip() for x in args.symbols.split(",")]
|
||
print(f"[因果回放] {len(syms)} 币 × {args.steps} 根逐根重放"
|
||
f"（每根都要重算笔中枢与 bsp，慢是必然的）\n", flush=True)
|
||
parts = []
|
||
with ProcessPoolExecutor(max_workers=args.workers) as ex:
|
||
fut = {ex.submit(replay, s, args.tf, args.rows, args.warm,
|
||
args.steps): s for s in syms}
|
||
for i, f in enumerate(as_completed(fut), 1):
|
||
r = f.result()
|
||
print(f" [{i}/{len(syms)}] {fut[f]} "
|
||
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()
|