影子路径开 lean 模式,实测完整链路 1410 → 683ms
拉到新引擎后在服务器侧直接实测,没有沿用 3.4 倍那个换算——那是在旧代码上量的。 先做等价性验证。不能直接引用 step46 的对拍结论:它固化的是 bsp_list 那条链的 哈希,而影子路径走 find_fast_bsp3 + build_htf_zones + htf_fx_timeline + attach_htf_context,两条链读的东西不一样。所以新建 verify_lean_parity.py 在 这条路径上逐根对拍 compute() 的每个返回字段。 其中一个坑:随机取窗口测不到信号分支。信号密度约 1/2000 根,头 12 个窗口命中 0 个,「一致」只覆盖了早退路径。改成一半窗口对齐到已知信号根,命中率才上来。 最终 180 窗口 / 两模式各 90 命中 / 零分歧。 生产实测(inner_ms,容器内同口径): compute_ms 646 → 132ms 4.89x inner_ms 612 → 128ms 4.80x lag_signal_ms 1410 → 683ms 完整链路,落回 800ms 线内 比 3.4 倍更好,因为是引擎 ~3.5x 叠 lean ~1.35x。 两点判读上的订正: - lag_data_ms 那 576→490ms 是噪声,不要记在引擎账上。均值 721±36 vs 740±127,重叠;而且引擎本来就影响不到交易所与网络那一段。 - 仍有 44% 的根超 800ms,但尾部现在完全由数据腿主导(lag_data P90 1482ms vs compute P90 237ms)。计算既不是瓶颈也不是尾部主因了,继续压计算换不到 尾部改善。800ms 那道闸取的是最近 30 根的中位数,683ms 已满足。 start.sh 加 SHADOW_LEAN(默认 1),设 0 可退回 full 复量两模式差异。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -13,6 +13,10 @@
|
|||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
|
|
||||||
NAME="${NAME:-shadow}"
|
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}"
|
IMAGE="${SHADOW_IMAGE:-hummingbot/hummingbot:latest}"
|
||||||
HOURS="${HOURS:-168}"
|
HOURS="${HOURS:-168}"
|
||||||
WORKERS="${WORKERS:-2}"
|
WORKERS="${WORKERS:-2}"
|
||||||
@@ -99,6 +103,7 @@ docker run -d --name "$NAME" -w /home/hummingbot \
|
|||||||
--restart unless-stopped \
|
--restart unless-stopped \
|
||||||
-e PYTHONPATH=/home/hummingbot:/repo/research:/repo/research/live:/repo \
|
-e PYTHONPATH=/home/hummingbot:/repo/research:/repo/research/live:/repo \
|
||||||
-e SHADOW_SITE="$SHADOW_SITE" \
|
-e SHADOW_SITE="$SHADOW_SITE" \
|
||||||
|
-e SHADOW_LEAN="$SHADOW_LEAN" \
|
||||||
-v "$REPO_ROOT:/repo:ro" \
|
-v "$REPO_ROOT:/repo:ro" \
|
||||||
-v "$OUT:/out" \
|
-v "$OUT:/out" \
|
||||||
--entrypoint /opt/conda/envs/hummingbot/bin/python \
|
--entrypoint /opt/conda/envs/hummingbot/bin/python \
|
||||||
|
|||||||
@@ -39,12 +39,22 @@ for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
|
|||||||
os.environ.setdefault(_v, "1")
|
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 线。
|
"""在 df_l 的最后一根上找信号。df_l/df_h 都只含已收盘 K 线。
|
||||||
|
|
||||||
entry_px 是次根开盘价(回测 entry_delay=1 的成交价),用作 atr_pct 的
|
entry_px 是次根开盘价(回测 entry_delay=1 的成交价),用作 atr_pct 的
|
||||||
分母。取不到时退回用信号根收盘价,并在返回里标 atr_ref="close"。
|
分母。取不到时退回用信号根收盘价,并在返回里标 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:
|
返回 dict:
|
||||||
last_idx 最后一根在 chanlun 处理后 dataframe 里的下标
|
last_idx 最后一根在 chanlun 处理后 dataframe 里的下标
|
||||||
n_bars 实际参与计算的根数
|
n_bars 实际参与计算的根数
|
||||||
@@ -55,6 +65,8 @@ def compute(df_l, df_h, entry_px: float | None = None) -> dict:
|
|||||||
import numpy as np
|
import numpy as np
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
|
|
||||||
|
lean = LEAN if lean is None else lean
|
||||||
|
|
||||||
try:
|
try:
|
||||||
from chanlun import TF_DF
|
from chanlun import TF_DF
|
||||||
from lib.fast_bsp3 import find_fast_bsp3
|
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.nested_level import build_htf_zones
|
||||||
from lib.shadow_budget import ATR_GATE_BP
|
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
|
cdf = chan_l.dataframe
|
||||||
last = len(cdf) - 1
|
last = len(cdf) - 1
|
||||||
base = {"last_idx": last, "n_bars": int(len(df_l)), "hits": [],
|
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
|
# 5m 同向。算不出时 h1_agree 记 0,该信号自然不会通过 pass_all
|
||||||
if df_h is not None and len(df_h) > 0:
|
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
|
hdf = chan_h.dataframe
|
||||||
tl = htf_fx_timeline(
|
tl = htf_fx_timeline(
|
||||||
signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
|
signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in New Issue
Block a user