research: 订正额外信号的机制——不是重画，是中枢终点在实时不可知
我先前把 B4 的额外信号归因为「中枢重画」，用户两次指出后逐条查证，归因是错的： ① 中枢边界不重画（用户对）。zg/zd 由前三笔定死，verify_window_sens/step39 验过。 ② 中枢只由已确认的笔构成，这是结构性保证：cal_bi_zs_list_pure 要求 bi1/bi2/bi3.is_sure，延伸时要求 leave_bi/back_bi.is_sure。 step70 实测 789 个信号全部 z_sure=True，用户说的浅色 B4 在研究路径不存在。 因此我提的「补一道 is_sure 门」是空操作，实测 PF 0.73→0.73，已标记不要再提。 ③ 真机制是 available_ts 棘轮，§5.41 早写明：改动不是已确认的笔被推翻， 而是中枢又吸收了新笔、bis[-1] 换人。available_ts 取末笔 sure_time， 于是往后棘轮，find_fast_bsp3 的 200 根扫描窗口整体右移。 消失的不是笔，是中枢的终点。 顺带澄清一个伪问题：中枢跨度 > 200 根不会导致突破落不进窗口， 因为扫描起点是中枢结束（末笔确认）而非起点，此时价格已在离开中枢。 仍未对上的是数量级：中枢层面 6.5% vs 信号层面 75%。假设是 max_per_zone=1 的放大（起点棘轮越过旧入场点，同一中枢反复产出「第一个」）。 step69_mechanism.py 已写好判据但三次后台运行被中断，标为下一步前置项。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,240 @@
|
||||
"""那 72~77% 的「额外信号」到底怎么来的：中枢重画，还是扫描窗口错位？
|
||||
|
||||
我在 §3.3994 里把它归因为「中枢重画」，**这个归因没验证过，而且与已有结论矛盾**
|
||||
（step39 的假阳性 0%、`verify_window_sens` 的窗口 +200/+500/+1000 逐字段一致）。
|
||||
用户指出中枢不重画，代码注释也支持他：
|
||||
|
||||
# -1 = 最后一笔（历史默认）。中枢每吸收一根K线，最后一笔就可能后移，
|
||||
# 于是 available_ts 跟着漂——这是右边缘重画的根因
|
||||
—— chanlun/analysis/fast_bsp.py:47
|
||||
|
||||
漂的不是中枢**边界**（zg/zd），是它的**可用时刻**。而 `find_fast_bsp3` 只从
|
||||
`available_ts` 往后扫 `scan=200` 根。两种口径的窗口因此错位：
|
||||
|
||||
实时 中枢没吸收完，available_ts 偏早 -> 窗口开得早
|
||||
全量 中枢吸收完了，available_ts 偏晚（§5.41 实测中位晚 62 分钟）-> 窗口开得晚
|
||||
|
||||
落在「实时窗口内、全量窗口外」的信号，全量根本没扫到那个时段，于是显示为「额外」。
|
||||
|
||||
两种机制的修法完全不同，所以必须分清：
|
||||
|
||||
边界重画 结构本身不稳，只能用滞后换稳定性，代价大
|
||||
窗口错位 中枢是同一个真中枢，信号也是真信号，只是**开得太早、确认不足**
|
||||
—— 这正好解释它们为什么亏（PF 0.26~0.36），且修法是调 available_ts
|
||||
|
||||
判据：逐个额外信号，去全量中枢表里按 (zg, zd) 找它的中枢。
|
||||
找得到且边界一致 -> 窗口错位（用户是对的，我的归因错了）
|
||||
找不到或边界不同 -> 边界重画（我的归因成立）
|
||||
"""
|
||||
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" / "step69_mech.feather"
|
||||
WIN, MAX_GROW, SCAN = 2001, 500, 200
|
||||
TOL = 1e-6
|
||||
|
||||
|
||||
def replay(sym: str, ltf: str, rows: int, steps: int) -> 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
|
||||
|
||||
try:
|
||||
df = fetch_ohlcv(f"{sym}/USDT:USDT", ltf, rows)
|
||||
if df is None or len(df) < WIN + steps + 100:
|
||||
return None
|
||||
df = df.iloc[-(WIN + steps):].reset_index(drop=True)
|
||||
|
||||
full = TF_DF(df, 1, ltf)
|
||||
cdf = ensure_timestamp(full.dataframe)
|
||||
zf = zones_from_zs_list(full.cal_bi_zs_list_pure(full.bi_list), cdf)
|
||||
if zf is None or zf.empty:
|
||||
return None
|
||||
sig_full = find_fast_bsp3(cdf, zf)
|
||||
full_keys = {(int(r.entry_idx), int(r.direction))
|
||||
for r in sig_full.itertuples()} if not sig_full.empty \
|
||||
else set()
|
||||
fzg = zf["zg"].to_numpy(float)
|
||||
fzd = zf["zd"].to_numpy(float)
|
||||
fav = zf["available_ts"].to_numpy()
|
||||
ts_all = cdf["timestamp"].to_numpy()
|
||||
|
||||
rec: dict[tuple, dict] = {}
|
||||
chan, anchor = None, 0
|
||||
for i in range(WIN, len(df)):
|
||||
if chan is None or (i - anchor) >= MAX_GROW:
|
||||
w = df.iloc[i - WIN + 1:i + 1].copy()
|
||||
chan = TF_DF(w, 1, ltf)
|
||||
chan.init_stream(w, 1, ltf)
|
||||
anchor = i
|
||||
else:
|
||||
chan.append_bar(df.iloc[i])
|
||||
try:
|
||||
zl = chan.cal_bi_zs_list_pure(chan.bi_list)
|
||||
if not zl:
|
||||
continue
|
||||
sub = ensure_timestamp(chan.dataframe)
|
||||
z = zones_from_zs_list(zl, sub)
|
||||
if z is None or z.empty:
|
||||
continue
|
||||
s = find_fast_bsp3(sub, z)
|
||||
if s is None or s.empty:
|
||||
continue
|
||||
last = len(sub) - 1
|
||||
s = s[s["entry_idx"].astype(int) == last]
|
||||
if s.empty or "zone_i" not in s.columns:
|
||||
continue
|
||||
# find_fast_bsp3 的输出自带 zg/zd，直接 merge 会加后缀，
|
||||
# 所以中枢侧的列全部改名再接
|
||||
zc = z[["zg", "zd", "available_ts"]].rename(columns={
|
||||
"zg": "z_zg", "zd": "z_zd", "available_ts": "z_av"})
|
||||
zc["zone_i"] = np.arange(len(z))
|
||||
s = s.drop(columns=[c for c in ("z_zg", "z_zd", "z_av")
|
||||
if c in s.columns])
|
||||
s = s.merge(zc, on="zone_i", how="left")
|
||||
except Exception: # noqa: BLE001
|
||||
continue
|
||||
for r in s.itertuples():
|
||||
k = (i, int(r.direction))
|
||||
if k in rec:
|
||||
continue
|
||||
# 该中枢在全量表里是否存在（按边界匹配，边界是不该漂的量）
|
||||
zg_, zd_ = float(r.z_zg), float(r.z_zd)
|
||||
m = (np.abs(fzg - zg_) <= TOL * max(1.0, abs(zg_))) \
|
||||
& (np.abs(fzd - zd_) <= TOL * max(1.0, abs(zd_)))
|
||||
j = int(np.argmax(m)) if m.any() else -1
|
||||
rec[k] = {
|
||||
"sym": sym, "entry_idx": i, "direction": int(r.direction),
|
||||
"in_full": k in full_keys,
|
||||
"zone_found": bool(m.any()),
|
||||
"z_zg": zg_, "z_zd": zd_,
|
||||
"avail_rt": int(r.z_av),
|
||||
"avail_full": int(fav[j]) if j >= 0 else -1,
|
||||
"ts_entry": int(ts_all[i]) if i < len(ts_all) else -1,
|
||||
}
|
||||
return pd.DataFrame(list(rec.values())) if rec else None
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f" {sym} 失败: {type(e).__name__}: {e}", flush=True)
|
||||
return None
|
||||
|
||||
|
||||
def report(d: pd.DataFrame, ltf: str) -> None:
|
||||
ex = d[~d.in_full]
|
||||
print("\n" + "=" * 92)
|
||||
print("【一】判据：额外信号所在的中枢，在全量表里找得到吗")
|
||||
print("=" * 92)
|
||||
print(f"回放信号 {len(d)} · 额外 {len(ex)}")
|
||||
print(f"**额外信号中，其中枢按 (zg,zd) 在全量表里找得到的："
|
||||
f"{ex.zone_found.mean()*100:.1f}%**")
|
||||
print(f"（对照：非额外信号 {d[d.in_full].zone_found.mean()*100:.1f}%）")
|
||||
print("""
|
||||
≈100% -> 中枢边界没变，是**扫描窗口错位**，用户对、我的「重画」归因错
|
||||
明显偏低 -> 中枢确实消失或改边界，「重画」成立""")
|
||||
|
||||
print("\n" + "=" * 92)
|
||||
print("【一b】数量级对账：中枢层面只有 6.5% 被改，信号层面却 74% 是额外的")
|
||||
print("=" * 92)
|
||||
print("§5.41 的 A/B 实测：`bis[-1]` 口径下确认时刻被改 6.5%、中枢消失 9.0%。")
|
||||
print("若信号层面的 74% 成立，必然有放大机制。怀疑是 `max_per_zone=1`：")
|
||||
print(" 每个中枢只返回**第一个**入场点，而扫描起点随 available_ts 棘轮后移，")
|
||||
print(" 越过旧入场点后，同一中枢会重新产出一个「第一个」——全量只用最终值，")
|
||||
print(" 所以每中枢至多一个信号，实时却能反复触发。")
|
||||
for nm, x in [("额外信号", ex), ("非额外信号", d[d.in_full])]:
|
||||
if x.empty:
|
||||
continue
|
||||
nz = x.groupby(["sym", "z_zg", "z_zd"]).size() \
|
||||
if "z_zg" in x.columns else None
|
||||
if nz is None:
|
||||
print(" （缺 z_zg/z_zd 列，跳过）")
|
||||
break
|
||||
print(f"\n{nm}：{len(x)} 个信号，落在 {len(nz)} 个不同中枢上 "
|
||||
f"-> 每中枢 {len(x)/len(nz):.2f} 次")
|
||||
print(f" 同一中枢触发次数分布 中位 {nz.median():.0f} "
|
||||
f"P90 {nz.quantile(.9):.0f} 最大 {nz.max()}")
|
||||
print("""
|
||||
若「额外信号」的每中枢次数显著 > 1 而「非额外」≈ 1，放大机制坐实：
|
||||
6.5% 的中枢改动通过棘轮重扫，放大成信号层面的几百个。""")
|
||||
|
||||
g = ex[ex.zone_found & (ex.avail_full > 0)].copy()
|
||||
if g.empty:
|
||||
return
|
||||
bar_ms = {"1m": 60_000, "5m": 300_000, "15m": 900_000}.get(ltf, 300_000)
|
||||
g["drift_bars"] = (g.avail_full - g.avail_rt) / bar_ms
|
||||
g["from_rt"] = (g.ts_entry - g.avail_rt) / bar_ms
|
||||
g["from_full"] = (g.ts_entry - g.avail_full) / bar_ms
|
||||
|
||||
print("\n" + "=" * 92)
|
||||
print("【二】available_ts 漂了多少，以及入场落在谁的扫描窗口里")
|
||||
print("=" * 92)
|
||||
print(f"avail 漂移（全量 − 实时，根） 中位 {g.drift_bars.median():.0f} "
|
||||
f"P25 {g.drift_bars.quantile(.25):.0f} "
|
||||
f"P75 {g.drift_bars.quantile(.75):.0f}")
|
||||
print(f" §5.41 记的是中位晚 62 分钟，本表 {ltf} 下即 "
|
||||
f"{62*60_000/bar_ms:.0f} 根，可交叉验证")
|
||||
print(f"\n入场距实时 avail（根） 中位 {g.from_rt.median():.0f} "
|
||||
f"（应落在 0~{SCAN} 内，否则实时也扫不到）")
|
||||
print(f"入场距全量 avail（根） 中位 {g.from_full.median():.0f}")
|
||||
out = ((g.from_full < 0) | (g.from_full > SCAN)).mean()
|
||||
print(f"\n**入场落在全量扫描窗口 [0,{SCAN}] 之外的比例：{out*100:.1f}%**")
|
||||
print("""
|
||||
这是机制的直接证据：比例高 -> 全量根本没扫到那个时段，所以「没有」这个信号，
|
||||
与中枢是否重画无关。其中 from_full < 0 表示入场早于全量的可用时刻——
|
||||
即**实时抢跑了**，中枢还没吸收完就下单。""")
|
||||
early = (g.from_full < 0).mean()
|
||||
print(f" 其中抢跑（早于全量 avail）：{early*100:.1f}%")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE")
|
||||
ap.add_argument("--ltf", default="5m")
|
||||
ap.add_argument("--rows", type=int, default=45_000)
|
||||
ap.add_argument("--steps", type=int, default=20_000)
|
||||
ap.add_argument("--workers", type=int, default=5)
|
||||
ap.add_argument("--reuse", action="store_true")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.reuse and OUT.exists():
|
||||
report(pd.read_feather(OUT), args.ltf)
|
||||
return
|
||||
syms = [s.strip() for s in args.symbols.split(",")]
|
||||
print(f"[机制判定] {len(syms)} 币 × {args.steps} 根 · {args.ltf}\n",
|
||||
flush=True)
|
||||
parts = []
|
||||
with ProcessPoolExecutor(max_workers=args.workers) as ex_:
|
||||
fut = {ex_.submit(replay, s, args.ltf, args.rows, 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, args.ltf)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user