"""逐笔追「额外信号」：全量那一遍到底为什么没产出它。 用户说「原来的计算是没有问题的，是你算错了」。之前几轮都是统计口径， 容易把自己的 bug 说成市场现象。这次不做统计，挑具体信号逐个对账。 对每个额外信号，把两边的中间量全摆出来： 回放侧 中枢 (zg,zd)、available_ts、扫描起点、入场根 全量侧 同一个中枢是否存在、它的 available_ts、扫描区间 [start, start+200] 入场根落不落在这个区间里、`diag` 记的拒绝原因 四种可能的结论，指向完全不同的处理： A 全量里那个中枢的扫描区间**不覆盖**入场根 -> available_ts 棘轮，机制成立，不是 bug B 中枢在全量里**不存在** -> 中枢集合本身有差异，要查是不是我窗口用错了 C 区间覆盖了、全量却仍没出信号 -> 两边输入不同（我传错了 df/zones），**是我的 bug** D 入场根索引对不上（差几根） -> 索引口径错，`chan.dataframe` 与原始 df 不是 1:1，**是我的 bug** D 尤其要查：回放里我用原始 df 的下标 i 当 key，全量用的是 `cdf` 的 entry_idx， 两者只有在 `TF_DF.dataframe` 与输入逐行对齐时才等价。 """ 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)) WIN, MAX_GROW, SCAN = 2001, 500, 200 def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--sym", default="BTC") ap.add_argument("--tf", default="5m") ap.add_argument("--rows", type=int, default=45_000) ap.add_argument("--steps", type=int, default=3_000) ap.add_argument("--show", type=int, default=8) args = ap.parse_args() 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 df = fetch_ohlcv(f"{args.sym}/USDT:USDT", args.tf, args.rows) df = df.iloc[-(WIN + args.steps):].reset_index(drop=True) full = TF_DF(df, 1, args.tf) cdf = ensure_timestamp(full.dataframe) print("=" * 92) print("【0】先查 D：索引口径是否 1:1") print("=" * 92) print(f"原始 df 行数 {len(df)} · TF_DF.dataframe 行数 {len(cdf)} " f"-> {'一致' if len(df) == len(cdf) else '**不一致，索引口径有问题**'}") if len(df) == len(cdf): t_df = pd.to_datetime(df["date"]) t_cd = pd.to_datetime(cdf["date"]) if t_df.dt.tz is not None: t_df = t_df.dt.tz_localize(None) if t_cd.dt.tz is not None: t_cd = t_cd.dt.tz_localize(None) same = int((t_df.values == t_cd.values).sum()) print(f"逐行时间戳相同 {same}/{len(df)} " f"-> {'逐行对齐' if same == len(df) else '**有错位**'}") zsf = full.cal_bi_zs_list_pure(full.bi_list) zf = zones_from_zs_list(zsf, cdf) diag_full: dict = {} sig_full = find_fast_bsp3(cdf, zf, diag=diag_full) fkeys = {(int(r.entry_idx), int(r.direction)) for r in sig_full.itertuples()} ts = cdf["timestamp"].to_numpy() print(f"\n全量：中枢 {len(zf)} 个，信号 {len(sig_full)} 个") print(f"全量 diag：{diag_full}") # ---- 回放 ---- rec = [] 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, args.tf) chan.init_stream(w, 1, args.tf) 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: continue except Exception: # noqa: BLE001 continue for r in s.itertuples(): zi = int(r.zone_i) rec.append({ "i": i, "d": int(r.direction), "in_full": (i, int(r.direction)) in fkeys, "zg": float(z.zg.iloc[zi]), "zd": float(z.zd.iloc[zi]), "avail_rt": int(z.available_ts.iloc[zi]), "win_len": len(sub), "bo": int(r.bo_idx), "last": last, }) rp = pd.DataFrame(rec) if rp.empty: print("回放无信号") return print(f"\n回放：信号 {len(rp)} 个 · 其中全量也有 " f"{int(rp.in_full.sum())} · 额外 {int((~rp.in_full).sum())}") # ---- 逐笔对账 ---- print("\n" + "=" * 92) print(f"【1】抽 {args.show} 个额外信号逐笔对账") print("=" * 92) fzg, fzd = zf.zg.to_numpy(float), zf.zd.to_numpy(float) fav = zf.available_ts.to_numpy() n = len(cdf) rows = [] for r in rp[~rp.in_full].head(args.show).itertuples(): m = (np.abs(fzg - r.zg) < 1e-9) & (np.abs(fzd - r.zd) < 1e-9) if not m.any(): rows.append({"入场根": r.i, "中枢在全量": "**不存在**", "结论": "B 中枢集合有差异"}) continue j = int(np.argmax(m)) start = int(np.searchsorted(ts, fav[j], side="left")) end = min(start + SCAN, n) cover = start <= r.i < end rows.append({ "入场根": r.i, "中枢在全量": "存在", "回放avail根": int(np.searchsorted(ts, r.avail_rt, side="left")), "全量avail根": start, "棘轮(根)": start - int(np.searchsorted(ts, r.avail_rt, side="left")), "全量扫描区间": f"[{start},{end})", "覆盖入场根": "是" if cover else "否", "结论": "C **是我的bug**" if cover else "A 棘轮，机制成立", }) print(pd.DataFrame(rows).to_string(index=False)) print(""" A = 全量扫描区间不覆盖入场根（available_ts 棘轮后移），机制成立 B = 中枢在全量里不存在 -> 中枢集合有差异，要查窗口 C = 区间覆盖了全量却没出信号 -> 两边输入不同，是我的 bug""") print("\n" + "=" * 92) print("【2】全体额外信号按结论归类") print("=" * 92) cnt = {"A 棘轮": 0, "B 中枢不存在": 0, "C 我的bug": 0} for r in rp[~rp.in_full].itertuples(): m = (np.abs(fzg - r.zg) < 1e-9) & (np.abs(fzd - r.zd) < 1e-9) if not m.any(): cnt["B 中枢不存在"] += 1 continue j = int(np.argmax(m)) start = int(np.searchsorted(ts, fav[j], side="left")) cnt["C 我的bug" if start <= r.i < min(start + SCAN, n) else "A 棘轮"] += 1 tot = max(sum(cnt.values()), 1) print(pd.DataFrame([{"结论": k, "个数": v, "占比": f"{v/tot*100:.1f}%"} for k, v in cnt.items()]).to_string(index=False)) print("\n" + "=" * 92) print("【3】按信号条数统计是否被同一中枢的重复触发放大了") print("=" * 92) print("上表可见 2691/2707/2750 的全量 avail 同为 2767，是**同一个中枢**触发三次。") print("`max_per_zone=1` 只保证「每次扫描返回一个」，但扫描起点随棘轮后移，") print("越过旧入场点后同一中枢会重新产出「第一个」——于是按条数统计被放大。\n") rows = [] for nm, x in [("额外", rp[~rp.in_full]), ("全量也有", rp[rp.in_full])]: if x.empty: continue nz = x.groupby(["zg", "zd"]).size() rows.append({"分组": nm, "信号条数": len(x), "不同中枢数": len(nz), "每中枢触发": round(len(x) / len(nz), 2), "最多触发": int(nz.max())}) print(pd.DataFrame(rows).to_string(index=False)) print(""" 若「额外」组每中枢触发 >> 1 而「全量也有」组 ≈ 1，则先前那个「额外信号占 74%」 是**按条数**统计的放大结果，不等于实盘会多开 74% 的仓。 真实多开多少，取决于执行层对同一中枢是否去重/冷却 —— 那一层仍未审计。""") if __name__ == "__main__": main()