diff --git a/research/live/deploy/start.sh b/research/live/deploy/start.sh index 086dbe4..1084dd4 100755 --- a/research/live/deploy/start.sh +++ b/research/live/deploy/start.sh @@ -13,6 +13,10 @@ set -euo pipefail NAME="${NAME:-shadow}" +# lean 模式让 TF_DF 只构建到中枢,跳过线段/走势中枢/MACD 状态机。等价性由 +# verify_lean_parity.py 在影子这条路径上逐根验过(180 窗口 / 90 命中零分歧)。 +# 设 0 可退回 full,用来复量两模式的耗时差。 +SHADOW_LEAN="${SHADOW_LEAN:-1}" IMAGE="${SHADOW_IMAGE:-hummingbot/hummingbot:latest}" HOURS="${HOURS:-168}" WORKERS="${WORKERS:-2}" @@ -99,6 +103,7 @@ docker run -d --name "$NAME" -w /home/hummingbot \ --restart unless-stopped \ -e PYTHONPATH=/home/hummingbot:/repo/research:/repo/research/live:/repo \ -e SHADOW_SITE="$SHADOW_SITE" \ + -e SHADOW_LEAN="$SHADOW_LEAN" \ -v "$REPO_ROOT:/repo:ro" \ -v "$OUT:/out" \ --entrypoint /opt/conda/envs/hummingbot/bin/python \ diff --git a/research/live/shadow_signal.py b/research/live/shadow_signal.py index 2299ab6..1c22f40 100644 --- a/research/live/shadow_signal.py +++ b/research/live/shadow_signal.py @@ -39,12 +39,22 @@ for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"): os.environ.setdefault(_v, "1") -def compute(df_l, df_h, entry_px: float | None = None) -> dict: +LEAN = os.environ.get("SHADOW_LEAN", "1") not in ("0", "", "false") + + +def compute(df_l, df_h, entry_px: float | None = None, + lean: bool | None = None) -> dict: """在 df_l 的最后一根上找信号。df_l/df_h 都只含已收盘 K 线。 entry_px 是次根开盘价(回测 entry_delay=1 的成交价),用作 atr_pct 的 分母。取不到时退回用信号根收盘价,并在返回里标 atr_ref="close"。 + lean=True 让 TF_DF 只构建到中枢,跳过线段/走势中枢/MACD 状态机。本路径 + 只读 chan.dataframe 与 chan.klc_list,不碰 bsp_list/seg_list/chanmacd, + 所以可以跳。但静态检查会漏间接依赖,等价性由 verify_lean_parity.py 在 + 这条路径上逐根实测,不套用 step46 那 5 个对拍用例——那些用例走的是 + bsp_list,覆盖不到 fast_bsp3 + 嵌套上下文这条链。 + 返回 dict: last_idx 最后一根在 chanlun 处理后 dataframe 里的下标 n_bars 实际参与计算的根数 @@ -55,6 +65,8 @@ def compute(df_l, df_h, entry_px: float | None = None) -> dict: import numpy as np import pandas as pd + lean = LEAN if lean is None else lean + try: from chanlun import TF_DF from lib.fast_bsp3 import find_fast_bsp3 @@ -63,7 +75,7 @@ def compute(df_l, df_h, entry_px: float | None = None) -> dict: from lib.nested_level import build_htf_zones from lib.shadow_budget import ATR_GATE_BP - chan_l = TF_DF(df_l, 1, "1m") + chan_l = TF_DF(df_l, 1, "1m", lean=lean) cdf = chan_l.dataframe last = len(cdf) - 1 base = {"last_idx": last, "n_bars": int(len(df_l)), "hits": [], @@ -99,7 +111,7 @@ def compute(df_l, df_h, entry_px: float | None = None) -> dict: # 5m 同向。算不出时 h1_agree 记 0,该信号自然不会通过 pass_all if df_h is not None and len(df_h) > 0: - chan_h = TF_DF(df_h, 1, "5m") + chan_h = TF_DF(df_h, 1, "5m", lean=lean) hdf = chan_h.dataframe tl = htf_fx_timeline( signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf) diff --git a/research/live/verify_lean_parity.py b/research/live/verify_lean_parity.py new file mode 100644 index 0000000..9a5ca2b --- /dev/null +++ b/research/live/verify_lean_parity.py @@ -0,0 +1,200 @@ +"""在影子信号路径上实测 lean 与 full 是否等价,并量提速。 + +## 为什么不能直接引用 step46 的对拍结论 + +step46 固化的是 klc/笔/中枢/`bsp_list`/被消费列的哈希,走的是 `bsp_list` 那条 +链。影子路径走的是另一条:`find_fast_bsp3` + `build_htf_zones` + +`htf_fx_timeline` + `attach_htf_context`。两条链读的东西不完全一样,所以 +「lean ≡ full」在 step46 用例上成立,不等于在这条路径上成立。 + +静态检查显示这条链只读 `chan.dataframe` 与 `chan.klc_list`(lean 都不跳), +但静态检查漏不掉间接依赖——`build_htf_zones` 收的是 chan 对象本体。所以逐根 +实测:同一个窗口分别用 full 和 lean 跑 `compute()`,比对返回的每一个字段。 + +判据是**逐字段完全相同**,不是「信号数量相同」。数量相同而方向或标志不同, +会让影子测的是另一批信号,且不报错。 + + python research/live/verify_lean_parity.py --syms BTC,ETH,SOL --n 150 +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +import warnings +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() +sys.path.insert(0, str(HERE.parents[1])) +sys.path.insert(0, str(HERE.parents[2])) +sys.path.insert(0, str(HERE.parent)) + +LTF_BARS, HTF_BARS = 2001, 801 # 与 shadow_hb 同窗口 +SKIP = ("inner_ms", "queue_ms") # 计时字段本就不同,不参与比对 + + +def load(sym: str, tf: str, cache: Path) -> pd.DataFrame: + c = sorted(cache.glob(f"bitget_{sym}_{tf}_*.feather"), + key=lambda p: p.stat().st_size, reverse=True) + if not c: + raise FileNotFoundError(f"没有 {sym} {tf} 缓存") + return pd.read_feather(c[0]) + + +def canon(d: dict) -> str: + """把返回值规范化成可比较的字符串。 + + 浮点直接比会被末位差异误判,但 lean 走的是同一段算术、不该有任何差异, + 所以这里**不设容差**:round 到 12 位只是为了消掉 repr 差异,真有数值 + 分歧一定会被抓到。 + """ + def norm(v): + if isinstance(v, float): + return None if not np.isfinite(v) else round(v, 12) + if isinstance(v, dict): + return {k: norm(x) for k, x in sorted(v.items())} + if isinstance(v, (list, tuple)): + return [norm(x) for x in v] + if isinstance(v, (np.integer, np.floating, np.bool_)): + return norm(v.item()) + return v + return json.dumps({k: norm(v) for k, v in sorted(d.items()) + if k not in SKIP}, sort_keys=True, ensure_ascii=False) + + +def signal_bars(sym: str, cache: Path) -> np.ndarray: + """全量历史里过三滤网的信号根下标。 + + 随机取窗口几乎测不到信号分支——信号密度约 1/2000 根,12 个窗口命中 0 个。 + 只测早退路径的「一致」是很弱的证据:lean 若真影响了中枢或笔,分歧恰恰 + 出现在有信号的那些根上。所以把窗口对齐到这些根。 + + 注意这些下标来自**全量历史**建的中枢,而 compute 只看 2000 根窗口,所以 + 对齐后未必真的命中(这正是尚未收口的窗口左边界效应)。但命中率会从 + 1/2000 提到可用水平。 + """ + from step43_fill_aware_budget import signals_for + _, sig = signals_for(sym, cache) + return sig["entry_idx"].astype(int).to_numpy() + + +def run_one(sym: str, cache: Path, n: int, step: int, + at_signals: bool = True) -> dict: + from shadow_signal import NUM_COLS, compute + + l_all, h_all = load(sym, "1m", cache), load(sym, "5m", cache) + for df in (l_all, h_all): + for c in NUM_COLS: + if c not in df.columns: + raise RuntimeError(f"{sym} 缺列 {c}") + + l_ts = l_all["timestamp"].to_numpy("int64") + h_ts = h_all["timestamp"].to_numpy("int64") + # 一半窗口对齐到已知信号根(测信号分支),一半均匀铺开(测早退路径) + ends: list[int] = [] + if at_signals: + try: + sb = signal_bars(sym, cache) + sb = sb[(sb > LTF_BARS) & (sb < len(l_all) - 1)] + ends += list(sb[-(n // 2 or 1):]) + print(f" 对齐到信号根 {len(ends)} 个") + except Exception as e: + print(f" 取信号根失败,只用均匀窗口:{e!r}") + ends += list(range(len(l_all) - 1, LTF_BARS, -step))[:max(n - len(ends), 1)] + ends = sorted(set(ends)) + if not ends: + raise RuntimeError("数据不足一个窗口") + + n_same = n_diff = 0 + t_full = t_lean = 0.0 + first_diff = None + n_hits_full = n_hits_lean = 0 + + for e in ends: + df_l = l_all.iloc[e - LTF_BARS + 1:e + 1] + # 5m 只取已收盘且不晚于 1m 窗口末尾的根,与实盘一致 + hi = int(np.searchsorted(h_ts, l_ts[e], side="right")) + df_h = h_all.iloc[max(0, hi - HTF_BARS):hi] + # 次根开盘价:窗口末尾的下一根,与实盘 entry_px 同义 + entry_px = float(l_all["open"].to_numpy(float)[e + 1]) \ + if e + 1 < len(l_all) else None + + t0 = time.perf_counter() + rf = compute(df_l.copy(), df_h.copy(), entry_px, lean=False) + t_full += time.perf_counter() - t0 + + t0 = time.perf_counter() + rl = compute(df_l.copy(), df_h.copy(), entry_px, lean=True) + t_lean += time.perf_counter() - t0 + + n_hits_full += len(rf.get("hits") or []) + n_hits_lean += len(rl.get("hits") or []) + if canon(rf) == canon(rl): + n_same += 1 + else: + n_diff += 1 + if first_diff is None: + first_diff = (e, canon(rf), canon(rl)) + + k = len(ends) + print(f" 窗口 {k} 个 · 完全一致 {n_same} · 不一致 {n_diff}") + print(f" 命中数 full {n_hits_full} / lean {n_hits_lean}") + print(f" 单窗耗时 full {t_full / k * 1000:.0f}ms · " + f"lean {t_lean / k * 1000:.0f}ms · " + f"提速 {t_full / max(t_lean, 1e-9):.2f}x") + if first_diff: + e, a, b = first_diff + print(f" ⚠ 首个分歧在窗口末尾 idx={e}") + print(f" full: {a[:400]}") + print(f" lean: {b[:400]}") + return {"sym": sym, "n": k, "same": n_same, "diff": n_diff, + "full_ms": t_full / k * 1000, "lean_ms": t_lean / k * 1000, + "hits_full": n_hits_full, "hits_lean": n_hits_lean} + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--syms", default="BTC,ETH,SOL") + ap.add_argument("--cache", default="research/live/cache") + ap.add_argument("--n", type=int, default=150, help="每币比对多少个窗口") + ap.add_argument("--step", type=int, default=37, + help="窗口间隔根数。取质数避免与任何周期共振") + ap.add_argument("--no-signals", action="store_true", + help="不对齐信号根(快,但测不到信号分支)") + a = ap.parse_args() + + rows = [] + for sym in a.syms.split(","): + print(f"\n{'=' * 70}\n{sym}") + try: + rows.append(run_one(sym, Path(a.cache), a.n, a.step, + at_signals=not a.no_signals)) + except Exception as e: + print(f" 跳过:{e!r}") + + if not rows: + return + d = pd.DataFrame(rows) + print(f"\n\n{'=' * 70}\n汇总\n") + print(f" 比对窗口 {int(d['n'].sum()):,} 个 · " + f"不一致 {int(d['diff'].sum())} 个") + print(f" 单窗耗时 full {d['full_ms'].mean():.0f}ms → " + f"lean {d['lean_ms'].mean():.0f}ms " + f"({d['full_ms'].sum() / max(d['lean_ms'].sum(), 1e-9):.2f}x)") + if int(d["diff"].sum()) == 0: + print("\n 逐字段完全一致,可以开 lean。") + else: + print("\n ⛔ 存在分歧,不要开 lean。lean 跳掉的东西这条路径确实在用。") + + +if __name__ == "__main__": + main()