research: 影子交易器落在 Hummingbot 上,并修掉 Bitget 连接器的换根延迟

1m 腿的滑点余量只有几个 bp,所以要测的必须是生产路径的滑点——换个运行时
测出来的数就不作数。框架因此从「滑点已知后再定」提前到测量阶段就定为
Hummingbot(Spot/Perp 连接器均 v2.0,Bitget 是 Foundation Partner)。

新增 research/live/。前置测量:

- bench_compute.py 本机算力,1m 单币 0.318s、三币串行 1.38s
- venue_parity.py Binance 与 Bitget 同根信号重合仅 14.6~42.6%
- signal_sensitivity.py 0.25bp 扰动就换掉一半信号
- aggregate_robustness.py 但总体期望不降——脆的是信号身份,不是 alpha
- bitget_baseline.py 因此改用 Bitget 原生基线定预算:余量 BTC -0.13bp、
  ETH +4.02bp、SOL +2.92bp。BTC 本就为负,只作延迟测量的参照物

运行时选型:

- parity_env.py 容器与本机信号逐一相同(下标、中枢数、checksum 全等),
  容器内 0.26s/币反而更快。故 chanlun 直接挂载进容器,不必另起信号服务。
  装进现有 .venv 那条路走不通:Hummingbot 要 numba>=0.61.2 与
  aiohttp<3.14,与本机 Python 3.14 冲突
- latency_ccxt.py / latency_hummingbot.py / latency_compare.py 初测显示
  Hummingbot 比 ccxt.pro 慢约 1030ms,90 根逐根配对里 80~97% 更慢
- probe_ws_action.py 否掉「丢弃 snapshot」的猜测:换根首条就是 update
- probe_hb_vs_raw.py 与 latency_attribute.py 四路归因——容器网络 2~18ms、
  Hummingbot 处理 -10~-30ms,1350~1480ms 全落在解析方式上
- probe_ws_payload.py 定位根因:Bitget 换根会推一条带两根的消息
  [上一根, 新一根],而上游取 data["data"][0] 拿到的是上一根,新一根要等
  下一条单元素消息

修复:

- patched_candles.py 处理消息里的全部元素。不能简单改成 [-1]——那样上一根
  的收盘价会永远停在换根前约 1 秒的那次推送上,而信号对 0.25bp 都敏感
- verify_patch.py 60 根配对验证:拿回 1060~1090ms,与原始 WS 只差 5~14ms
  已贴理论下限,19 根已收盘 K 线 OHLCV 逐根未变。折算 ETH 省 0.54bp、
  SOL 省 0.42bp。此 bug 值得向上游反馈

影子交易器:

- shadow_hb.py 不下单,读连接器真实盘口按仓位吃单深度算成交价,与次根开盘价
  (回测 entry_delay=1 的口径)相减,分解成延迟漂移、盘口价差、深度冲击。
  盘口 10Hz 滚动缓冲 30 秒,把延迟变成自变量:每个信号记 0.5/1/2/5s 与实际
  算完时刻各一个滑点值,本机算得慢也不影响能读出的曲线
- shadow_signal.py 信号计算隔离到子进程。0.26s 是纯 CPU 且 chanlun 受 GIL
  限制,放进 asyncio 循环会把行情处理一起卡住
- shadow_report.py 首日延迟门槛与滑点曲线报表

不用 paper trade 测滑点:它的成交由 Hummingbot 自己的撮合模型模拟,
测出来是模型行为而非市场行为。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Ubuntu
2026-08-27 23:51:57 +08:00
co-authored by Cursor
parent 66061f79a1
commit 7e339d2a54
44 changed files with 11946 additions and 0 deletions
+186
View File
@@ -0,0 +1,186 @@
"""最关键的一步:逐笔清单不可复现,总体期望还在不在。
signal_sensitivity 证明 0.25bp 的数据扰动就能换掉一半信号。这本身不判死刑——
趋势跟随策略允许「成交的具体是哪几笔」随机,只要总体期望稳定就仍可交易。
但如果扰动后 PF 与毛均收益也跟着塌,那回测测的就是噪声。
两种结果对应完全不同的下一步:
总体稳定 -> 改成「Bitget 原生信号测滑点 + Bitget 原生回测基线」,主线继续
总体也塌 -> 滑点根本不是瓶颈,1m 腿的问题在信号定义本身,影子交易器白写
口径与 step23 一致:SL/TP/MAX_BARS = 1.5/3.0/48ATR 取信号根,
入场为信号次根开盘价(entry_delay=1),成本 4bp 手续费 + 1bp 滑点。
3.91bp 的滑点预算就是从「毛均收益 +0.0991%」推出来的,所以毛均收益是主看指标。
输出 out/aggregate_robustness.csv。
"""
from __future__ import annotations
import argparse
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().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
sys.path.insert(0, str(HERE))
pd.set_option("display.width", 260)
from signal_sensitivity import TICK, perturb # noqa: E402
SYMS = ("BTC", "ETH", "SOL")
LEVELS = (0.0, 0.5, 1.0, 2.0)
SL, TP, MAX_BARS = 1.5, 3.0, 48
FEE, SLIP = 0.0004, 0.0001
def run_once(df_l: pd.DataFrame, df_h: pd.DataFrame) -> pd.DataFrame:
"""跑完整管线并逐笔模拟,返回交易表。"""
from chanlun import TF_DF
from lib.breakout import run_trades
from lib.fast_bsp3 import find_fast_bsp3
from lib.fx_signal import extract_fx_signals, signals_to_frame
from lib.nested_bsp import attach_htf_context, htf_fx_timeline
from lib.nested_level import build_htf_zones
chan_l = TF_DF(df_l, 1, "1m")
cdf = chan_l.dataframe
zones = build_htf_zones(cdf, "1m", chan=chan_l)
if zones.empty:
return pd.DataFrame()
sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
if sig.empty:
return pd.DataFrame()
chan_h = TF_DF(df_h, 1, "5m")
hdf = chan_h.dataframe
tl = htf_fx_timeline(signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
full = attach_htf_context(sig, cdf, tl, "h1")
fin = full[full["h1_agree"] == 1]
if fin.empty:
return pd.DataFrame()
entries = list(zip(fin["entry_idx"].astype(int), fin["direction"].astype(int)))
return run_trades(cdf, entries, SL, TP, MAX_BARS,
fee=FEE + SLIP, entry_delay=1)
def stats(tr: pd.DataFrame) -> dict:
if tr.empty:
return {"笔数": 0}
g = tr["gross"].to_numpy(dtype=float)
n = tr["ret"].to_numpy(dtype=float)
win, loss = n[n > 0].sum(), -n[n < 0].sum()
return {
"笔数": len(tr),
"胜率": f"{(n > 0).mean() * 100:.1f}%",
"毛均收益": f"{g.mean() * 100:+.4f}%",
"净均收益": f"{n.mean() * 100:+.4f}%",
"PF": round(win / loss, 2) if loss > 0 else np.inf,
"t值": round(n.mean() / n.std(ddof=1) * np.sqrt(len(n)), 2) if len(n) > 1 else np.nan,
"滑点余量bp": round(g.mean() * 1e4 - 6.0, 2),
}
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--symbols", default="BTC,ETH,SOL")
ap.add_argument("--bars", type=int, default=200_000, help="每币用多少根 1m")
ap.add_argument("--seeds", type=int, default=2)
ap.add_argument("--levels", default=None,
help="逗号分隔的噪声档(bp);只给 0 就是纯基线复现")
ap.add_argument("--tag", default="", help="产物文件名后缀")
args = ap.parse_args()
levels = (tuple(float(x) for x in args.levels.split(","))
if args.levels else LEVELS)
from lib.data import load_local
syms = [s.strip() for s in args.symbols.split(",")]
print(f"[总体稳健性] {syms} · 每币 {args.bars} 根 1m "
f"({args.bars / 1440:.0f} 天) · 噪声档 {levels} bp\n", flush=True)
rows = []
for sym in syms:
df_l = load_local(f"{sym}/USDT:USDT", "1m")
df_h = load_local(f"{sym}/USDT:USDT", "5m")
if df_l is None or df_h is None:
print(f"{sym}: 本地无数据,跳过")
continue
df_l = df_l.tail(args.bars).reset_index(drop=True)
lo = int(df_l["timestamp"].iloc[0])
df_h = df_h[df_h.timestamp >= lo].reset_index(drop=True)
print(f"── {sym} {len(df_l)} 根 1m / {len(df_h)} 根 5m "
f"{df_l['date'].iloc[0]:%Y-%m-%d} ~ {df_l['date'].iloc[-1]:%Y-%m-%d}",
flush=True)
for bp in levels:
for k in range(1 if bp == 0 else args.seeds):
t0 = time.perf_counter()
tr = run_once(perturb(df_l, bp, TICK[sym], 2000 + k), df_h)
s = stats(tr)
rows.append({"品种": sym, "噪声bp": bp, "种子": k, **s})
print(f" 噪声 {bp:>4.2f}bp 种子{k}: " +
" · ".join(f"{k2} {v}" for k2, v in s.items()) +
f" [{time.perf_counter() - t0:.0f}s]", flush=True)
if not rows:
print("无结果")
return
tb = pd.DataFrame(rows)
print("\n" + "=" * 130)
print("########## 1. 逐币 × 噪声档 ##########")
print(tb.to_string(index=False))
print("\n########## 2. 三币合并(同噪声档取均值)##########")
num = tb.copy()
num["毛均bp"] = num["毛均收益"].str.rstrip("%").astype(float) * 100
num["净均bp"] = num["净均收益"].str.rstrip("%").astype(float) * 100
num["胜率_"] = num["胜率"].str.rstrip("%").astype(float)
agg = num.groupby("噪声bp").agg(
笔数=("笔数", "mean"), 胜率=("胜率_", "mean"),
毛均bp=("毛均bp", "mean"), 净均bp=("净均bp", "mean"),
PF=("PF", "mean"), t值=("t值", "mean")).round(2)
agg["滑点余量bp"] = (agg["毛均bp"] - 6.0).round(2)
print(agg.to_string())
print("\n########## 结论 ##########")
print(" step23 的 1m 基线(3 币 3578 笔 / 2.28 年):毛均 9.91bp · PF 2.31 · "
"t 19.92 · 余量 3.91bp")
if 0.0 not in agg.index:
print(" 本次未跑无噪声档,无法给出相对基线的比例")
return
base = agg.loc[0.0]
print(f" 无噪声基线:毛均 {base['毛均bp']:.2f}bp · PF {base['PF']:.2f} · "
f"t {base['t值']:.2f} · 滑点余量 {base['滑点余量bp']:.2f}bp")
for bp in levels[1:]:
if bp not in agg.index:
continue
r = agg.loc[bp]
print(f" 噪声 {bp}bp:毛均 {r['毛均bp']:.2f}bp "
f"({r['毛均bp'] / base['毛均bp'] * 100:.0f}% of 基线) · "
f"PF {r['PF']:.2f} · t {r['t值']:.2f} · 余量 {r['滑点余量bp']:.2f}bp")
print("\n 毛均与 PF 若基本持平 → 逐笔身份随机但总体期望稳定,主线继续,")
print(" 但必须换成 Bitget 原生回测基线,Binance 的逐笔清单不可用于对照。")
print(" 若毛均随噪声单调下滑 → 回测吃的是数据噪声,滑点不是瓶颈。")
out = RESEARCH / "out" / f"aggregate_robustness{args.tag}.csv"
tb.to_csv(out, index=False)
print(f"\n产物写入 {out}")
if __name__ == "__main__":
main()
+204
View File
@@ -0,0 +1,204 @@
"""前置测量一:本机算力——影子交易器的计算延迟下限。
step39 在 Mac 上量到 2000 根窗口单次 0.20s。这台机器只有 2 vCPU 且单核更慢,
而计算延迟直接吃 1m 腿仅 3.9bp 的滑点预算,所以必须在写影子交易器之前实测。
量四件事:
1m 侧 TF_DF(2000根) + build_htf_zones + find_fast_bsp3
5m 侧 TF_DF(800根) + 分型 + 时间线(只在 5m 收盘那根变,可摊薄到 1/5)
串行 3 币 最后一个币要等多久 —— 这就是它的下单延迟
并行 3 币 2 核跑 3 进程会互相抢核,未必比串行快
输出 out/bench_compute.csv,判据是与 3.9bp 预算对应的漂移:本机 1m 波动
BTC 6.5bp/分钟,按 √t 折算,1 秒延迟约 0.8bp、5 秒约 1.9bp。
"""
from __future__ import annotations
import argparse
import os
import sys
import time
import warnings
from concurrent.futures import ProcessPoolExecutor
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().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
pd.set_option("display.width", 240)
LTF, HTF = "1m", "5m"
WIN_LTF = 2000
WIN_HTF = max(WIN_LTF // 5, 800)
SYMS = ("BTC", "ETH", "SOL")
def _signal_once(sl_ltf: pd.DataFrame) -> tuple[int, float]:
"""1m 侧:切片重建结构与信号。与 step39 的 _pit_once 同一套调用。"""
from chanlun import TF_DF
from lib.fast_bsp3 import find_fast_bsp3
from lib.nested_level import build_htf_zones
t0 = time.perf_counter()
chan = TF_DF(sl_ltf, 1, LTF)
cdf = chan.dataframe
zones = build_htf_zones(cdf, LTF, chan=chan)
n_sig = 0
if not zones.empty:
sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
n_sig = len(sig)
return n_sig, time.perf_counter() - t0
def _agree_once(sl_htf: pd.DataFrame) -> tuple[int, float]:
"""5m 侧:切片重建分型时间线。与 step39 的 _pit_agree 同一套调用。"""
from chanlun import TF_DF
from lib.fx_signal import extract_fx_signals, signals_to_frame
from lib.nested_bsp import htf_fx_timeline
t0 = time.perf_counter()
chan = TF_DF(sl_htf, 1, HTF)
tl = htf_fx_timeline(signals_to_frame(extract_fx_signals(chan, chan.dataframe)),
chan.dataframe)
return len(tl), time.perf_counter() - t0
def _load(sym: str) -> tuple[pd.DataFrame, pd.DataFrame]:
from lib.data import load_local
pair = f"{sym}/USDT:USDT"
df_l = load_local(pair, LTF)
df_h = load_local(pair, HTF)
if df_l is None or df_h is None:
raise SystemExit(f"{sym} 本地数据缺失,本机只有 BTC/ETH/SOL")
return df_l, df_h
def _slices(sym: str, reps: int, seed: int) -> list[tuple[pd.DataFrame, pd.DataFrame]]:
"""预先切好窗口。读 feather 是本机 IO,实盘数据来自 WS 缓冲,不计入延迟。"""
df_l, df_h = _load(sym)
rng = np.random.default_rng(seed)
# 从末段随机取窗口,避开数据头部(指标预热)与尾部(不足一窗)
lo, hi = max(WIN_LTF, len(df_l) - 200_000), len(df_l) - 1
ltf_ts = df_l["timestamp"].to_numpy()
htf_ts = df_h["timestamp"].to_numpy()
out = []
for i in rng.integers(lo, hi, size=reps):
i = int(i)
sl_l = df_l.iloc[i - WIN_LTF + 1: i + 1].reset_index(drop=True)
# 5m 只喂到不晚于该 1m 根的部分,与实盘一致
h_end = int(np.searchsorted(htf_ts, ltf_ts[i], side="right"))
sl_h = (df_h.iloc[h_end - WIN_HTF: h_end].reset_index(drop=True)
if h_end >= WIN_HTF else None)
out.append((sl_l, sl_h))
return out
def bench_pair(task: tuple) -> dict:
"""算一根:1m 信号 + 5m 时间线。在子进程里跑,import 都在函数内。"""
import warnings as _w
_w.filterwarnings("ignore")
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
sym, sl_l, sl_h = task
_, dt_l = _signal_once(sl_l)
dt_h = np.nan if sl_h is None else _agree_once(sl_h)[1]
return {"sym": sym, "ltf_s": dt_l, "htf_s": dt_h}
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--reps", type=int, default=15, help="每币采样窗口数")
ap.add_argument("--rounds", type=int, default=5, help="串行/并行各跑几轮")
args = ap.parse_args()
print(f"[本机算力] {os.cpu_count()} 逻辑核 · 1m 窗口 {WIN_LTF} 根 · "
f"5m 窗口 {WIN_HTF} 根 · 每币 {args.reps}\n", flush=True)
print("########## 1. 单币单次耗时 ##########", flush=True)
prepared = {s: _slices(s, args.reps, 7) for s in SYMS}
per = []
for s in SYMS:
d = pd.DataFrame([bench_pair((s, l, h)) for l, h in prepared[s]])
per.append(d)
print(f" {s}: 1m {d['ltf_s'].median():.3f}s (P95 {d['ltf_s'].quantile(.95):.3f}s)"
f" · 5m {d['htf_s'].median():.3f}s", flush=True)
allp = pd.concat(per, ignore_index=True)
m_ltf = allp["ltf_s"].median()
m_htf = allp["htf_s"].median()
# 5m 侧只在每 5 根 1m 里变一次,其余 4 根可复用缓存
amort = m_ltf + m_htf / 5
print(f"\n 三币合并中位:1m {m_ltf:.3f}s · 5m {m_htf:.3f}s")
print(f" 单币每根摊薄成本 {amort:.3f}s5m 每 5 根才重算一次)")
print(f" 对比 Mac 的 0.20s:本机慢 {m_ltf / 0.20:.1f}")
print("\n########## 2. 三币串行 vs 并行(最后一个币的下单延迟)##########",
flush=True)
print(" 只计算子耗时:读数据不计入,进程池常驻不计启动开销", flush=True)
ser, par2, par3 = [], [], []
pools = {w: ProcessPoolExecutor(max_workers=w) for w in (2, 3)}
try:
# 先各跑一次把子进程的 import 预热掉,否则首轮全是模块加载时间
for w, ex in pools.items():
list(ex.map(bench_pair, [(s, *prepared[s][0]) for s in SYMS]))
for k in range(args.rounds):
batch = [(s, *prepared[s][k % args.reps]) for s in SYMS]
t0 = time.perf_counter()
for t in batch:
bench_pair(t)
ser.append(time.perf_counter() - t0)
for w, bag in ((2, par2), (3, par3)):
t0 = time.perf_counter()
list(pools[w].map(bench_pair, batch))
bag.append(time.perf_counter() - t0)
print(f"{k + 1}: 串行 {ser[-1]:.2f}s · 并行2 {par2[-1]:.2f}s · "
f"并行3 {par3[-1]:.2f}s", flush=True)
finally:
for ex in pools.values():
ex.shutdown()
print("\n########## 3. 延迟折算成 bp3.9bp 预算的参照)##########")
# 各币 1m 收益标准差,用 √t 把延迟折成价格漂移的一个标准差
vol = {}
for s in SYMS:
df_l, _ = _load(s)
r = np.diff(np.log(df_l["close"].to_numpy(dtype=float)[-400_000:]))
vol[s] = float(np.std(r) * 1e4)
rows = []
for name, xs in (("串行", ser), ("并行2", par2), ("并行3", par3)):
d = float(np.median(xs))
rows.append({"方案": name, "三币总耗时": f"{d:.2f}s",
**{f"{s} 漂移1σ": f"{vol[s] * np.sqrt(d / 60):.2f}bp"
for s in SYMS}})
tb = pd.DataFrame(rows)
print(tb.to_string(index=False))
print(f"\n 1m 波动实测:" + " · ".join(f"{s} {vol[s]:.1f}bp/分钟" for s in SYMS))
print(" 漂移 1σ 是随机部分;入场时价格正朝信号方向跑,系统性追价另计。")
out = RESEARCH / "out" / "bench_compute.csv"
pd.DataFrame({
"指标": ["cpu核数", "1m窗口", "5m窗口", "1m中位s", "1m_P95s", "5m中位s",
"单币摊薄s", "串行3币s", "并行2_3币s", "并行3_3币s"],
"": [os.cpu_count(), WIN_LTF, WIN_HTF, round(m_ltf, 3),
round(allp["ltf_s"].quantile(.95), 3), round(m_htf, 3),
round(amort, 3), round(float(np.median(ser)), 2),
round(float(np.median(par2)), 2), round(float(np.median(par3)), 2)],
}).to_csv(out, index=False)
print(f"\n产物写入 {out}")
if __name__ == "__main__":
main()
+120
View File
@@ -0,0 +1,120 @@
"""Bitget 原生回测基线:影子交易器要对照的那个数。
aggregate_robustness 证明总体期望在数据扰动下稳定,但逐笔清单不可复现。
所以滑点不能逐笔对照 Binance 回测,只能对照「同一 venue 上的总体毛均收益」。
这一步就是把那个数算出来。
跑法与 Binance 侧完全对齐:同一时间窗、同样 300k 根 1m、同一套 lib/ 代码、
同样 SL/TP/MAX_BARS = 1.5/3.0/48 与 entry_delay=1,只换数据源。
特别关注 SOLBitget 的 tick 是 0.001、Binance 是 0.01,粗 10 倍会让
`close > high[j-1]` 大量平局不触发。Bitget 上信号多出 74%,多出来的是否
同样赚钱,只有跑一遍才知道——这不是随机扰动,是系统性差异。
输出 out/bitget_baseline.csv。
"""
from __future__ import annotations
import argparse
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().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
sys.path.insert(0, str(HERE))
pd.set_option("display.width", 260)
from aggregate_robustness import run_once, stats # noqa: E402
from venue_parity import fetch_bitget # noqa: E402
# Binance 侧在同一窗口的实测值,来自 out/aggregate_robustness_*.csv 的 0bp 档
BINANCE_REF = {"BTC": 6.11, "ETH": 12.31, "SOL": 10.01}
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--symbols", default="BTC,ETH,SOL")
ap.add_argument("--days", type=int, default=210,
help="与 Binance 侧的 300k 根(208 天)对齐")
ap.add_argument("--refresh", action="store_true")
ap.add_argument("--tag", default="", help="产物文件名后缀,避免多次调用互相覆盖")
args = ap.parse_args()
from lib.data import load_local
syms = [s.strip() for s in args.symbols.split(",")]
print(f"[Bitget 原生基线] {syms} · 近 {args.days} 天 1m\n", flush=True)
rows = []
for sym in syms:
t0 = time.perf_counter()
bg_l = fetch_bitget(sym, "1m", args.days, args.refresh)
bg_h = fetch_bitget(sym, "5m", args.days, args.refresh)
print(f"── {sym} {len(bg_l)} 根 1m / {len(bg_h)} 根 5m "
f"{bg_l['date'].iloc[0]:%Y-%m-%d} ~ {bg_l['date'].iloc[-1]:%Y-%m-%d}"
f" [拉取 {time.perf_counter() - t0:.0f}s]", flush=True)
t1 = time.perf_counter()
tr = run_once(bg_l, bg_h)
s = stats(tr)
rows.append({"品种": sym, "venue": "Bitget", **s})
print(f" Bitget : " + " · ".join(f"{k} {v}" for k, v in s.items())
+ f" [{time.perf_counter() - t1:.0f}s]", flush=True)
# 同窗口的 Binance 对照,直接重算一遍,避免口径漂移
lo, hi = int(bg_l.timestamp.min()), int(bg_l.timestamp.max())
bn_l = load_local(f"{sym}/USDT:USDT", "1m")
bn_h = load_local(f"{sym}/USDT:USDT", "5m")
if bn_l is None:
continue
bn_l = bn_l[(bn_l.timestamp >= lo) & (bn_l.timestamp <= hi)].reset_index(drop=True)
bn_h = bn_h[(bn_h.timestamp >= lo) & (bn_h.timestamp <= hi)].reset_index(drop=True)
t1 = time.perf_counter()
s2 = stats(run_once(bn_l, bn_h))
rows.append({"品种": sym, "venue": "Binance", **s2})
print(f" Binance: " + " · ".join(f"{k} {v}" for k, v in s2.items())
+ f" [{time.perf_counter() - t1:.0f}s]", flush=True)
if not rows:
print("无结果")
return
tb = pd.DataFrame(rows)
tb["毛均bp"] = tb["毛均收益"].str.rstrip("%").astype(float) * 100
print("\n" + "=" * 120)
print("########## 同窗口 · 同代码 · 只换数据源 ##########")
print(tb.to_string(index=False))
print("\n########## 毛均收益对照(bp##########")
piv = tb.pivot_table(index="品种", columns="venue", values="毛均bp")
piv["差额"] = (piv.get("Bitget", np.nan) - piv.get("Binance", np.nan)).round(2)
cnt = tb.pivot_table(index="品种", columns="venue", values="笔数")
piv["Bitget笔数"] = cnt.get("Bitget")
piv["Binance笔数"] = cnt.get("Binance")
piv["Bitget余量bp"] = (piv.get("Bitget", np.nan) - 6.0).round(2)
print(piv.round(2).to_string())
print("\n########## 结论 ##########")
print(" Bitget余量 = 该 venue 自己的毛均收益 6bp 双边费率(VIP1 taker + API 返50%)。")
print(" 这就是影子交易器测出的滑点要去比的那条线,逐笔清单不参与比较。")
print(" 两家毛均若接近 → 总体口径可迁移;若差很多 → 以 Bitget 的为准。")
out = RESEARCH / "out" / f"bitget_baseline{args.tag}.csv"
tb.to_csv(out, index=False)
print(f"\n产物写入 {out}")
if __name__ == "__main__":
main()
+116
View File
@@ -0,0 +1,116 @@
"""把 1030ms 归因到具体环节。
前面的测量已经排除两个可能:
· Bitget 换根首条推送就是 updateHummingbot 没有因丢弃 snapshot 而等待
· 容器内同进程里,Hummingbot 的 feed 只比原始 WS 慢 2~6ms,处理开销可忽略
剩下的变量是「位置」和「客户端」。本脚本对同一批 K 线三方对齐:
host_ccxt 宿主机 ccxt.pro watch_ohlcv latency_ccxt.csv
host_raw 宿主机 原始 aiohttp WS probe_raw_host.csv
cont_raw 容器内 原始 aiohttp WS probe_hb_vs_raw.csv
cont_hb 容器内 Hummingbot candles feed probe_hb_vs_raw.csv
据此可分离两件事:
cont_raw host_raw 容器网络的代价(同一份代码,只换位置)
host_ccxt host_raw ccxt.pro 客户端的差异(同一位置,只换客户端)
.venv/bin/python research/live/latency_attribute.py
"""
from __future__ import annotations
from pathlib import Path
import numpy as np
import pandas as pd
OUT = Path(__file__).resolve().parents[1] / "out"
SYMS = ("BTC", "ETH", "SOL")
def load() -> pd.DataFrame | None:
parts = []
f = OUT / "latency_ccxt.csv"
if f.exists():
d = pd.read_csv(f)
if not d.empty:
parts.append(d[["kline_ts", "sym", "t_data_ms"]]
.rename(columns={"t_data_ms": "host_ccxt"}))
f = OUT / "probe_raw_host.csv"
if f.exists():
d = pd.read_csv(f)
if not d.empty:
parts.append(d[["kline_ts", "sym", "raw_ms"]]
.rename(columns={"raw_ms": "host_raw"}))
f = OUT / "probe_hb_vs_raw.csv"
if f.exists():
d = pd.read_csv(f)
if not d.empty:
parts.append(d[["kline_ts", "sym", "raw_ms", "hb_ms"]]
.rename(columns={"raw_ms": "cont_raw",
"hb_ms": "cont_hb"}))
if not parts:
return None
m = parts[0]
for p in parts[1:]:
m = m.merge(p, on=["kline_ts", "sym"], how="outer")
return m
def main() -> None:
m = load()
if m is None or m.empty:
print("四路数据均缺失")
return
cols = [c for c in ("host_ccxt", "host_raw", "cont_raw", "cont_hb")
if c in m.columns]
print(f"各路样本数(重叠前):")
for c in cols:
print(f" {c}: {int(m[c].notna().sum())}")
print("\n########## 各路 t_data t_closems,中位)##########")
print(f"{'':<5}" + "".join(f"{c:>11}" for c in cols))
for s in SYMS:
g = m[m["sym"] == s]
line = f"{s:<5}"
for c in cols:
v = (g[c] - g["kline_ts"]).dropna()
line += f"{np.median(v):>11.0f}" if len(v) else f"{'':>11}"
print(line)
# 只在四路都有的 K 线上做差,避免不同子集的中位数互相错位
full = m.dropna(subset=cols)
print(f"\n########## 归因(仅四路齐全的 {len(full)} 根)##########")
if full.empty:
print(" 无四路齐全的 K 线;检查三个采集窗口是否重叠")
return
pairs = []
if "cont_raw" in cols and "host_raw" in cols:
pairs.append(("容器网络代价", "cont_raw", "host_raw"))
if "host_ccxt" in cols and "host_raw" in cols:
pairs.append(("ccxt.pro 客户端差异", "host_ccxt", "host_raw"))
if "cont_hb" in cols and "cont_raw" in cols:
pairs.append(("Hummingbot 处理开销", "cont_hb", "cont_raw"))
if "cont_hb" in cols and "host_ccxt" in cols:
pairs.append(("合计:容器 HB vs 宿主 ccxt", "cont_hb", "host_ccxt"))
print(f"{'环节':<26}" + "".join(f"{s:>9}" for s in SYMS))
for name, a, b in pairs:
line = f"{name:<26}"
for s in SYMS:
g = full[full["sym"] == s]
if g.empty:
line += f"{'':>9}"
continue
line += f"{np.median(g[a] - g[b]):>9.0f}"
print(line)
print("\n(单位 ms,正数表示前者更慢)")
if __name__ == "__main__":
main()
+120
View File
@@ -0,0 +1,120 @@
"""延迟基准 A:ccxt.pro 侧的数据到手时刻。
要测的事件只有一个:**我们在什么时候得知第 N 根 1m 已经收盘**。
它等价于 t_data − t_close,是下单延迟里最先发生、也往往最大的一段。
与 latency_hummingbot.py 测的是同一个事件,两边跑同一段时间才可比,
所以两个脚本都按整分钟对齐输出,事后按 K 线时间戳 join。
判据:ETH 的 Bitget 原生滑点余量只有 4.02bp、SOL 2.92bp,而本机 1m 波动
ETH 8.6bp/分钟、SOL 9.6bp/分钟。按 √t 折算,1 秒延迟就是 1.1~1.2bp 的
随机漂移,且入场方向上还有系统性追价。所以这个数值本身就可能决定生死。
输出 out/latency_ccxt.csv,每根一行。
"""
from __future__ import annotations
import argparse
import asyncio
import csv
import signal
import sys
import time
from pathlib import Path
HERE = Path(__file__).resolve().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
SYMS = ("BTC", "ETH", "SOL")
OUT = RESEARCH / "out" / "latency_ccxt.csv"
PERIOD_MS = 60_000
_stop = False
def _on_signal(*_):
global _stop
_stop = True
async def watch(ex, sym: str, writer, fh, stats: dict) -> None:
"""watch_ohlcv 每次推送都带整段最近 K 线,靠时间戳前进判断上一根已收盘。"""
pair = f"{sym}/USDT:USDT"
last_ts = None
while not _stop:
try:
o = await ex.watch_ohlcv(pair, "1m")
except Exception as e:
print(f" {sym} watch 异常 {type(e).__name__}: {e}", flush=True)
await asyncio.sleep(1)
continue
if not o:
continue
now_ms = int(time.time() * 1000)
newest = int(o[-1][0])
if last_ts is None:
last_ts = newest
continue
if newest > last_ts:
# newest 是刚开始的那根,故 last_ts 那根在 newest 时刻收盘
closed_ts = newest
lag_ms = now_ms - closed_ts
writer.writerow({"kline_ts": closed_ts, "sym": sym,
"t_data_ms": now_ms, "lag_ms": lag_ms})
fh.flush()
stats.setdefault(sym, []).append(lag_ms)
n = len(stats[sym])
if n % 5 == 1:
med = sorted(stats[sym])[n // 2]
print(f" {sym}: 第 {n} 根,本次 lag {lag_ms}ms,中位 {med}ms",
flush=True)
last_ts = newest
async def main_async(minutes: int) -> None:
import ccxt.pro as ccxtpro
ex = ccxtpro.bitget({"options": {"defaultType": "swap"},
"enableRateLimit": True})
OUT.parent.mkdir(parents=True, exist_ok=True)
fh = OUT.open("w", newline="")
writer = csv.DictWriter(fh, fieldnames=["kline_ts", "sym", "t_data_ms", "lag_ms"])
writer.writeheader()
stats: dict = {}
print(f"[ccxt.pro 延迟] {SYMS} · 计划跑 {minutes} 分钟 · 输出 {OUT.name}",
flush=True)
tasks = [asyncio.create_task(watch(ex, s, writer, fh, stats)) for s in SYMS]
deadline = time.time() + minutes * 60
while time.time() < deadline and not _stop:
await asyncio.sleep(1)
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
await ex.close()
fh.close()
print("\n########## t_data t_closems##########")
for s in SYMS:
v = sorted(stats.get(s, []))
if not v:
print(f" {s}: 无样本")
continue
print(f" {s}: n={len(v)} 中位 {v[len(v) // 2]}ms "
f"P90 {v[int(len(v) * .9)]}ms 最大 {v[-1]}ms")
print(f"\n产物写入 {OUT}")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--minutes", type=int, default=20)
args = ap.parse_args()
signal.signal(signal.SIGINT, _on_signal)
signal.signal(signal.SIGTERM, _on_signal)
asyncio.run(main_async(args.minutes))
if __name__ == "__main__":
main()
+122
View File
@@ -0,0 +1,122 @@
"""把两侧 t_data t_close 对齐,并折算成 bp,与滑点余量对照。
为什么要折算成 bp 才有意义:延迟本身不花钱,花钱的是延迟期间价格的漂移。
按随机游走,t 秒的价格标准差是 σ_1m · √(t/60),其中 σ_1m 是本币 1m 收益
的标准差。入场方向上还有系统性追价(信号触发往往伴随同向动量),所以随机
漂移只是下限,真实成本更高——这也是为什么最终仍要用真实盘口测滑点。
余量(bitget_baseline.py 得出,Bitget 原生基线减去手续费后剩下的空间):
BTC -0.13bp ETH +4.02bp SOL +2.92bp
BTC 本就为负,留着只作延迟测量的参照物,不作交易标的。
.venv/bin/python research/live/latency_compare.py
"""
from __future__ import annotations
import sys
from pathlib import Path
import numpy as np
import pandas as pd
HERE = Path(__file__).resolve().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
OUT = RESEARCH / "out"
BUDGET_BP = {"BTC": -0.13, "ETH": 4.02, "SOL": 2.92}
SYMS = ("BTC", "ETH", "SOL")
def vol_bp_per_min() -> dict[str, float]:
"""从 Bitget 缓存算 1m 收益标准差,单位 bp。"""
out = {}
for s in SYMS:
f = HERE / "cache" / f"bitget_{s}_1m_30d.feather"
if not f.exists():
f = HERE / "cache" / f"bitget_{s}_1m_210d.feather"
if not f.exists():
continue
df = pd.read_feather(f)
r = np.log(df["close"].to_numpy(dtype=float))
out[s] = float(np.nanstd(np.diff(r)) * 1e4)
return out
def drift_bp(lag_ms: float, vol: float) -> float:
"""随机游走下,lag 毫秒对应的价格漂移标准差(bp)。"""
return vol * np.sqrt(max(lag_ms, 0) / 60_000.0)
def load(tag: str) -> pd.DataFrame | None:
f = OUT / f"latency_{tag}.csv"
if not f.exists():
return None
df = pd.read_csv(f)
return df if not df.empty else None
def describe(df: pd.DataFrame, name: str, vols: dict) -> None:
print(f"\n########## {name}t_data t_close ##########")
print(f"{'':<5}{'n':>5}{'中位ms':>9}{'P90ms':>9}{'最大ms':>9}"
f"{'中位漂移bp':>12}{'余量bp':>9}{'占余量':>9}")
for s in SYMS:
v = df[df["sym"] == s]["lag_ms"].to_numpy(dtype=float)
if not len(v):
continue
med, p90 = float(np.median(v)), float(np.percentile(v, 90))
vol = vols.get(s)
d = drift_bp(med, vol) if vol else float("nan")
b = BUDGET_BP[s]
share = f"{d / b * 100:.0f}%" if b > 0 else "—(负)"
print(f"{s:<5}{len(v):>5}{med:>9.0f}{p90:>9.0f}{v.max():>9.0f}"
f"{d:>12.2f}{b:>9.2f}{share:>9}")
def main() -> None:
vols = vol_bp_per_min()
print("1m 收益标准差(bp/分钟,Bitget 缓存实测):")
for s, v in vols.items():
print(f" {s}: {v:.2f}")
a, b = load("ccxt"), load("hummingbot")
if a is None or b is None:
print(f"\n数据未就绪:ccxt={'' if a is not None else ''} "
f"hummingbot={'' if b is not None else ''}")
if a is not None:
describe(a, "ccxt.pro", vols)
if b is not None:
describe(b, "Hummingbot", vols)
return
describe(a, "ccxt.pro", vols)
describe(b, "Hummingbot", vols)
# 逐根配对才能消掉「不同分钟市场活跃度不同」的干扰
m = a.merge(b, on=["kline_ts", "sym"], suffixes=("_ccxt", "_hb"))
print(f"\n########## 逐根配对(重叠 {len(m)} 根)##########")
if m.empty:
print(" 两侧无重叠 K 线,无法配对;检查采集时间窗是否错开")
return
print(f"{'':<5}{'n':>5}{'ccxt中位':>10}{'HB中位':>10}"
f"{'差值中位':>10}{'HB更慢占比':>12}{'差值→bp':>10}")
for s in SYMS:
g = m[m["sym"] == s]
if g.empty:
continue
d = (g["lag_ms_hb"] - g["lag_ms_ccxt"]).to_numpy(dtype=float)
vol = vols.get(s)
# 差值转 bp:比较两条路径各自漂移的差,而非直接对差值开方
extra = (drift_bp(float(np.median(g["lag_ms_hb"])), vol)
- drift_bp(float(np.median(g["lag_ms_ccxt"])), vol)) if vol else float("nan")
print(f"{s:<5}{len(g):>5}{np.median(g['lag_ms_ccxt']):>10.0f}"
f"{np.median(g['lag_ms_hb']):>10.0f}{np.median(d):>10.0f}"
f"{(d > 0).mean() * 100:>11.0f}%{extra:>10.2f}")
print("\n判读:差值 → bp 若显著小于余量,说明选哪个运行时不影响结论,"
"可直接用 Hummingbot(生产路径一致);若接近或超过余量,"
"则运行时本身就是成本项,需要单独优化或放弃 1m。")
if __name__ == "__main__":
main()
+117
View File
@@ -0,0 +1,117 @@
"""延迟基准 B:Hummingbot 侧的数据到手时刻。
在 Hummingbot 容器内运行,测的事件与 latency_ccxt.py 完全相同:
**我们在什么时候得知第 N 根 1m 已经收盘**。
为什么必须在 Hummingbot 里测而不是复用 ccxt 的数字:如果最终执行走
Hummingbot,那决定成交价的是它的 WS 处理与事件循环延迟。用别的运行时测出
的滑点,换到 Hummingbot 上就不成立了。
用 max_records=20 让历史回填快速完成——这里只关心增量推送的时刻,不需要
2000 根窗口。
输出 out/latency_hummingbot.csv,字段与 A 侧一致,事后按 kline_ts join。
"""
from __future__ import annotations
import argparse
import asyncio
import csv
import time
from pathlib import Path
SYMS = ("BTC", "ETH", "SOL")
CONNECTOR = "bitget_perpetual"
OUT = Path("/out/latency_hummingbot.csv")
async def main_async(minutes: int) -> None:
from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory
from hummingbot.data_feed.candles_feed.data_types import CandlesConfig
feeds = {}
for s in SYMS:
cfg = CandlesConfig(connector=CONNECTOR, trading_pair=f"{s}-USDT",
interval="1m", max_records=20)
feeds[s] = CandlesFactory.get_candle(cfg)
print(f"[Hummingbot 延迟] {SYMS} · connector={CONNECTOR} · "
f"计划跑 {minutes} 分钟", flush=True)
for s, f in feeds.items():
if hasattr(f, "start"):
f.start()
else:
await f.start_network()
print(f" {s} 已启动订阅", flush=True)
# 等历史回填完成,否则 deque 尾部时间戳还在跳变
t0 = time.time()
while time.time() - t0 < 120:
if all(f.ready for f in feeds.values()):
break
await asyncio.sleep(0.5)
ready = {s: f.ready for s, f in feeds.items()}
print(f" 回填状态 {ready}{time.time() - t0:.1f}s", flush=True)
OUT.parent.mkdir(parents=True, exist_ok=True)
fh = OUT.open("w", newline="")
writer = csv.DictWriter(fh, fieldnames=["kline_ts", "sym", "t_data_ms", "lag_ms"])
writer.writeheader()
last = {}
for s, f in feeds.items():
c = f._candles
last[s] = int(c[-1][0]) if len(c) else None
stats: dict = {}
deadline = time.time() + minutes * 60
while time.time() < deadline:
for s, f in feeds.items():
c = f._candles
if not len(c):
continue
newest = int(c[-1][0])
if last[s] is None:
last[s] = newest
continue
if newest > last[s]:
now_ms = int(time.time() * 1000)
# Hummingbot 的时间戳是秒,统一成毫秒后再与本地钟相减
closed_ms = newest * 1000 if newest < 1e12 else newest
lag_ms = now_ms - closed_ms
writer.writerow({"kline_ts": closed_ms, "sym": s,
"t_data_ms": now_ms, "lag_ms": lag_ms})
fh.flush()
stats.setdefault(s, []).append(lag_ms)
n = len(stats[s])
if n % 5 == 1:
med = sorted(stats[s])[n // 2]
print(f" {s}: 第 {n} 根,本次 lag {lag_ms}ms,中位 {med}ms",
flush=True)
last[s] = newest
await asyncio.sleep(0.01)
for f in feeds.values():
f.stop()
fh.close()
print("\n########## t_data t_closems##########")
for s in SYMS:
v = sorted(stats.get(s, []))
if not v:
print(f" {s}: 无样本")
continue
print(f" {s}: n={len(v)} 中位 {v[len(v) // 2]}ms "
f"P90 {v[int(len(v) * .9)]}ms 最大 {v[-1]}ms")
print(f"\n产物写入 {OUT}")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--minutes", type=int, default=20)
args = ap.parse_args()
asyncio.run(main_async(args.minutes))
if __name__ == "__main__":
main()
+191
View File
@@ -0,0 +1,191 @@
"""环境等价性验证:同一份切片,在 .venv 与 Hummingbot 容器里必须算出同一组信号。
为什么要单独验这件事:容器里是 Python 3.13.14 + pandas 3.0.5 + numpy 2.4.6
本机 .venv 是 Python 3.14.4 + pandas 3.0.5 + numpy 2.5.2。pandas 同版本,
numpy 差一个小版本。信号已经被证明对 0.25bp 的数据扰动极度敏感(扰动会换掉
一半信号),所以浮点或 groupby 顺序上的任何细微差异都可能改变信号集合——
必须实测,不能推断。
用 --dump 先从 .venv 导出切片成 CSV,两个环境再读同一个 CSV,
这样数据来源差异为零,比出来的就是纯计算差异。
.venv/bin/python research/live/parity_env.py --dump # 导出切片
.venv/bin/python research/live/parity_env.py --run # 本机计算
docker run ... /app/parity_env.py --run --tag container # 容器计算
"""
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().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
LTF, HTF = "1m", "5m"
WINDOW = 2000 # step39 定下的窗口:命中率在此饱和
HTF_WINDOW = 800
SYMS = ("BTC", "ETH", "SOL")
def slice_dir() -> Path:
"""容器里挂在 /out,本机是 research/out。"""
p = Path("/out")
return p if p.is_dir() else RESEARCH / "out"
def dump() -> None:
"""从 Bitget 缓存导出末尾切片,供两个环境共用。
用 Bitget 而非 Binance 的数据:目标场地就是 Bitget,且这批缓存正是
bitget_baseline.py 算余量时用的同一份,口径可直接对上。
"""
d = slice_dir() / "parity_slices"
d.mkdir(parents=True, exist_ok=True)
cache = HERE / "cache"
for s in SYMS:
for tf, n in ((LTF, WINDOW), (HTF, HTF_WINDOW)):
src = cache / f"bitget_{s}_{tf}_30d.feather"
if not src.exists():
src = cache / f"bitget_{s}_{tf}_210d.feather"
if not src.exists():
print(f" {s} {tf}: 无缓存 {src.name},跳过")
continue
df = pd.read_feather(src)
if df is None or df.empty:
print(f" {s} {tf}: 缓存为空,跳过")
continue
out = df.tail(n).reset_index(drop=True)
f = d / f"{s}_{tf}.csv"
# 用 float 全精度写出,避免导出环节就引入舍入差异
out.to_csv(f, index=False, float_format="%.10f")
print(f" {s} {tf}: {len(out)} 根 -> {f.name}")
print(f"\n切片目录 {d}")
def signals(df_ltf: pd.DataFrame, df_htf: pd.DataFrame) -> dict:
"""复用 step39 的时点重建口径,返回信号下标与耗时。"""
from chanlun import TF_DF
from lib.fast_bsp3 import find_fast_bsp3
from lib.nested_level import build_htf_zones
t0 = time.perf_counter()
chan = TF_DF(df_ltf, 1, LTF)
cdf = chan.dataframe
zones = build_htf_zones(cdf, LTF, chan=chan)
raw: list[int] = []
if not zones.empty:
sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
if sig is not None and not sig.empty:
col = "idx" if "idx" in sig.columns else sig.columns[0]
raw = sorted(int(x) for x in sig[col].to_numpy())
dt = time.perf_counter() - t0
# 中枢边界是信号定义的核心中间量,一并指纹化:
# 若信号相同但中枢不同,说明差异只是被过滤掉了,仍是隐患
zsig = None
if not zones.empty:
num = zones.select_dtypes(include=[np.number])
zsig = float(np.nansum(num.to_numpy(dtype=float)))
return {"n_bars": int(len(df_ltf)), "n_signals": len(raw),
"signal_idx": raw, "zone_checksum": zsig,
"n_zones": int(len(zones)), "compute_s": round(dt, 4)}
def run(tag: str) -> None:
d = slice_dir() / "parity_slices"
res = {"tag": tag,
"python": sys.version.split()[0],
"pandas": pd.__version__,
"numpy": np.__version__}
per = {}
for s in SYMS:
f_ltf, f_htf = d / f"{s}_{LTF}.csv", d / f"{s}_{HTF}.csv"
if not f_ltf.exists():
print(f" {s}: 缺切片 {f_ltf}")
continue
df_ltf = pd.read_csv(f_ltf)
df_htf = pd.read_csv(f_htf) if f_htf.exists() else pd.DataFrame()
# date 存的是时间戳字符串,chanlun 的 kline builder 要真 datetime
# timestamp 是毫秒整数,保持数值不动
for df in (df_ltf, df_htf):
if not df.empty and "date" in df.columns:
df["date"] = pd.to_datetime(df["date"], utc=True)
r = signals(df_ltf, df_htf)
per[s] = r
print(f" {s}: {r['n_signals']} 信号 · {r['n_zones']} 中枢 · "
f"{r['compute_s']}s · zone_checksum={r['zone_checksum']}")
res["per_symbol"] = per
out = slice_dir() / f"parity_env_{tag}.json"
out.write_text(json.dumps(res, indent=2, ensure_ascii=False))
print(f"\n[{tag}] python {res['python']} pandas {res['pandas']} "
f"numpy {res['numpy']}")
print(f"产物写入 {out}")
def compare(a: str, b: str) -> None:
d = slice_dir()
ra = json.loads((d / f"parity_env_{a}.json").read_text())
rb = json.loads((d / f"parity_env_{b}.json").read_text())
print(f"{a}: python {ra['python']} pandas {ra['pandas']} numpy {ra['numpy']}")
print(f"{b}: python {rb['python']} pandas {rb['pandas']} numpy {rb['numpy']}")
print("\n########## 信号集合是否逐一相同 ##########")
ok = True
for s in SYMS:
pa, pb = ra["per_symbol"].get(s), rb["per_symbol"].get(s)
if not pa or not pb:
print(f" {s}: 缺结果,跳过")
continue
same_sig = pa["signal_idx"] == pb["signal_idx"]
same_zone = pa["n_zones"] == pb["n_zones"]
# checksum 是浮点求和,允许相对 1e-9 的差;超出即为真实分歧
za, zb = pa["zone_checksum"], pb["zone_checksum"]
same_cs = (za is None and zb is None) or (
za is not None and zb is not None
and abs(za - zb) <= 1e-9 * max(1.0, abs(za)))
ok = ok and same_sig and same_zone and same_cs
print(f" {s}: 信号 {'一致' if same_sig else '不一致'}"
f"{pa['n_signals']} vs {pb['n_signals']})· "
f"中枢 {'一致' if same_zone else '不一致'}"
f"{pa['n_zones']} vs {pb['n_zones']})· "
f"checksum {'一致' if same_cs else '不一致'}")
if not same_sig:
sa, sb = set(pa["signal_idx"]), set(pb["signal_idx"])
print(f"{a} 有: {sorted(sa - sb)[:10]}")
print(f"{b} 有: {sorted(sb - sa)[:10]}")
print(f" 耗时 {pa['compute_s']}s vs {pb['compute_s']}s")
print(f"\n结论:{'两环境等价,可直接在容器内算信号' if ok else '存在分歧,需把信号计算固定在单一环境'}")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--dump", action="store_true")
ap.add_argument("--run", action="store_true")
ap.add_argument("--tag", default="venv")
ap.add_argument("--compare", nargs=2, metavar=("A", "B"))
args = ap.parse_args()
if args.dump:
dump()
if args.run:
run(args.tag)
if args.compare:
compare(*args.compare)
if __name__ == "__main__":
main()
+65
View File
@@ -0,0 +1,65 @@
"""修正 Hummingbot bitget_perpetual candles feed 的换根延迟。
上游 _parse_websocket_message 里是:
candle = data["data"][0]
而 Bitget 在换根时会推一条带两根的消息 [上一根, 新一根]。取 [0] 拿到的是
上一根,其时间戳与 deque 尾部相同,于是只做了原地更新;新一根要等下一条
单元素消息才进入 deque——实测晚约 1.1 秒。
不能简单改成 [-1]:那样上一根的收盘价就永远停在换根前约 1 秒的那次推送上。
1m 信号对 0.25bp 的扰动都会换掉一半(见 signal_sensitivity.py),收盘价
偏一个 tick 是不能接受的。所以这里把**除最后一根外的元素就地写回 deque**,
再把最后一根交给基类走正常的 append 流程。
已向上游反馈前,本地用子类覆盖,不改动镜像。
"""
from __future__ import annotations
from typing import Any, Dict, Optional
import numpy as np
from hummingbot.data_feed.candles_feed.bitget_perpetual_candles import (
BitgetPerpetualCandles,
)
def _row_to_dict(row: list, ensure_s) -> Dict[str, Any]:
return {"timestamp": ensure_s(int(row[0])),
"open": float(row[1]), "high": float(row[2]),
"low": float(row[3]), "close": float(row[4]),
"volume": float(row[5]), "quote_asset_volume": float(row[6]),
"n_trades": 0., "taker_buy_base_volume": 0.,
"taker_buy_quote_volume": 0.}
class PatchedBitgetPerpetualCandles(BitgetPerpetualCandles):
"""与上游唯一的差别:一条消息里的多根 K 线全部处理,而非只取第一根。"""
def _parse_websocket_message(self, data: dict) -> Optional[Dict[str, Any]]:
if data == "pong":
return None
if not (data and data.get("data") and data.get("action") == "update"):
return None
rows = data["data"]
# 前面的元素都是已收盘 K 线的最终值:就地覆盖,保住真实收盘价
for row in rows[:-1]:
d = _row_to_dict(row, self.ensure_timestamp_in_seconds)
self._overwrite_existing(d)
# 最后一根交给基类:时间戳更大就 append,相同就原地更新
return _row_to_dict(rows[-1], self.ensure_timestamp_in_seconds)
def _overwrite_existing(self, d: Dict[str, Any]) -> None:
if not len(self._candles):
return
ts = int(d["timestamp"])
if int(self._candles[-1][0]) != ts:
return
self._candles[-1] = np.array(
[d["timestamp"], d["open"], d["high"], d["low"], d["close"],
d["volume"], d["quote_asset_volume"], d["n_trades"],
d["taker_buy_base_volume"], d["taker_buy_quote_volume"]]
).astype(float)
+186
View File
@@ -0,0 +1,186 @@
"""定位 Hummingbot 那 1030ms 究竟出在处理链路还是容器网络。
前一轮对照有两个变量同时在变:运行时(Hummingbot vs ccxt.pro)和位置
(容器内 vs 宿主机)。所以 1030ms 无法归因。原始 WS 探针已否掉「丢弃
snapshot」这个猜测——换根首条推送就是 update,Hummingbot 确实收到了。
本脚本在**容器内同一个进程**里并行跑两条路径,共用一个时钟、一条网络:
A. Hummingbot 的 BitgetPerpetualCandles,轮询 _candles 尾部时间戳
B. 一条原始 aiohttp WS,直接订阅 candle1m
两条路径对同一根 K 线各记一个到达时刻,差值即 Hummingbot 处理链路的净开销。
若 B 也慢,则是容器网络,与框架无关。
在容器内运行:
docker run --rm -v $PWD:/repo:ro -v $PWD/research/out:/out \
-w /home/hummingbot -e PYTHONPATH=/home/hummingbot \
--entrypoint /opt/conda/envs/hummingbot/bin/python \
hummingbot/hummingbot:latest /repo/research/live/probe_hb_vs_raw.py --minutes 20
"""
from __future__ import annotations
import argparse
import asyncio
import csv
import json
import time
from pathlib import Path
SYMS = ("BTC", "ETH", "SOL")
WSS = "wss://ws.bitget.com/v2/ws/public"
def out_dir() -> Path:
p = Path("/out")
return p if p.is_dir() else Path(__file__).resolve().parents[1] / "out"
async def raw_ws(rec: dict, stop: asyncio.Event) -> None:
"""B 路径:原始 WS,收到第一条带新时间戳的推送就记时刻。"""
import aiohttp
payload = {"op": "subscribe",
"args": [{"instType": "USDT-FUTURES", "channel": "candle1m",
"instId": f"{s}USDT"} for s in SYMS]}
while not stop.is_set():
try:
async with aiohttp.ClientSession() as sess, \
sess.ws_connect(WSS, heartbeat=20) as ws:
await ws.send_str(json.dumps(payload))
print(" [raw] 已订阅", flush=True)
last: dict[str, int] = {}
while not stop.is_set():
msg = await ws.receive()
if msg.type is not aiohttp.WSMsgType.TEXT:
# 关闭类消息必须跳出重连,否则 receive() 会立刻返回造成空转
print(f" [raw] 非文本消息 {msg.type},重连", flush=True)
break
if msg.data == "pong":
continue
d = json.loads(msg.data)
if "data" not in d or "arg" not in d:
continue
sym = d["arg"]["instId"].replace("USDT", "")
kts = int(d["data"][0][0])
if last.get(sym) is not None and kts > last[sym]:
rec.setdefault((sym, kts), {})["raw_ms"] = \
int(time.time() * 1000)
last[sym] = max(kts, last.get(sym, 0))
except asyncio.CancelledError:
raise
except Exception as e:
print(f" [raw] 异常 {type(e).__name__}: {e}", flush=True)
# 无论是正常跳出还是异常,重连前都歇一下,避免服务端持续拒绝时打成风暴
if not stop.is_set():
await asyncio.sleep(1)
async def hb_feed(rec: dict, stop: asyncio.Event) -> None:
"""A 路径:Hummingbot candles feed10ms 轮询 deque 尾部。"""
from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory
from hummingbot.data_feed.candles_feed.data_types import CandlesConfig
feeds = {}
for s in SYMS:
cfg = CandlesConfig(connector="bitget_perpetual",
trading_pair=f"{s}-USDT", interval="1m",
max_records=20)
f = CandlesFactory.get_candle(cfg)
f.start()
feeds[s] = f
print(" [hb] 已订阅", flush=True)
t0 = time.time()
while time.time() - t0 < 120 and not all(f.ready for f in feeds.values()):
await asyncio.sleep(0.5)
print(f" [hb] 回填完成 {time.time() - t0:.1f}s", flush=True)
last = {s: (int(f._candles[-1][0]) if len(f._candles) else None)
for s, f in feeds.items()}
while not stop.is_set():
for s, f in feeds.items():
if not len(f._candles):
continue
newest = int(f._candles[-1][0])
if last[s] is not None and newest > last[s]:
# HB 的时间戳是秒,统一成毫秒好与原始 WS 的 kline_ts 对齐
kts = newest * 1000 if newest < 1e12 else newest
rec.setdefault((s, kts), {})["hb_ms"] = int(time.time() * 1000)
last[s] = newest
await asyncio.sleep(0.01)
for f in feeds.values():
f.stop()
async def main_async(minutes: int, raw_only: bool = False) -> None:
rec: dict = {}
stop = asyncio.Event()
tasks = [asyncio.create_task(raw_ws(rec, stop))]
if not raw_only:
tasks.append(asyncio.create_task(hb_feed(rec, stop)))
where = "宿主机 raw 单跑" if raw_only else "容器内同进程对照"
print(f"[{where}] {SYMS} · 跑 {minutes} 分钟", flush=True)
deadline = time.time() + minutes * 60
reported = set()
while time.time() < deadline:
await asyncio.sleep(2)
for k, v in rec.items():
if k in reported or "raw_ms" not in v or "hb_ms" not in v:
continue
reported.add(k)
print(f" {k[0]} @{k[1]}: raw->hb 延后 {v['hb_ms'] - v['raw_ms']}ms",
flush=True)
stop.set()
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
if raw_only:
# 只落盘 raw 到达时刻,供与容器侧按 kline_ts 对齐
f = out_dir() / "probe_raw_host.csv"
with f.open("w", newline="") as fh:
w = csv.writer(fh)
w.writerow(["sym", "kline_ts", "raw_ms"])
for (s, k), v in sorted(rec.items(), key=lambda x: x[0][1]):
if "raw_ms" in v:
w.writerow([s, k, v["raw_ms"]])
print(f"\n宿主机 raw 样本 {sum('raw_ms' in v for v in rec.values())}")
print(f"产物写入 {f}")
return
f = out_dir() / "probe_hb_vs_raw.csv"
both = [(s, k, v) for (s, k), v in rec.items()
if "raw_ms" in v and "hb_ms" in v]
with f.open("w", newline="") as fh:
w = csv.writer(fh)
w.writerow(["sym", "kline_ts", "raw_ms", "hb_ms", "delta_ms"])
for s, k, v in sorted(both, key=lambda x: x[1]):
w.writerow([s, k, v["raw_ms"], v["hb_ms"],
v["hb_ms"] - v["raw_ms"]])
print(f"\n########## 同进程内 raw WS 与 Hummingbot 的到达差 ##########")
print(f"(正数 = Hummingbot 更慢;配对 {len(both)} 根)")
for s in SYMS:
d = sorted(v["hb_ms"] - v["raw_ms"] for ss, _, v in both if ss == s)
if not d:
print(f" {s}: 无配对样本")
continue
print(f" {s}: n={len(d)} 中位 {d[len(d) // 2]}ms "
f"P90 {d[int(len(d) * .9)]}ms 最小 {d[0]}ms 最大 {d[-1]}ms")
print(f"\n判读:中位接近 0 说明 1030ms 来自容器网络或宿主机对照本身;"
f"中位接近 1000ms 说明是 Hummingbot 处理链路的净开销。")
print(f"产物写入 {f}")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--minutes", type=int, default=20)
ap.add_argument("--raw-only", action="store_true",
help="不加载 hummingbot,只跑原始 WS(供宿主机对照)")
a = ap.parse_args()
asyncio.run(main_async(a.minutes, a.raw_only))
if __name__ == "__main__":
main()
+156
View File
@@ -0,0 +1,156 @@
"""验证 Hummingbot 那 1030ms 的来源:Bitget candle1m 的 snapshot / update 序列。
Hummingbot 的 bitget_perpetual candles feed 里有这么一句:
if data and data.get("data") and data["action"] == "update":
action == "snapshot" 的消息被整条丢弃。若新 K 线的首条推送恰是 snapshot,
Hummingbot 就必须等到下一条 update 才知道换根了,代价约等于一个推送间隔。
本脚本直接连原始 WS,对每条消息记录 action、K 线时间戳、到达时刻,
然后针对每次换根回答两件事:
1. 首条带新时间戳的消息,action 是什么
2. 若是 snapshot,到首条 update 之间隔了多久(= Hummingbot 白等的时间)
.venv/bin/python research/live/probe_ws_action.py --minutes 6
"""
from __future__ import annotations
import argparse
import asyncio
import json
import time
from collections import defaultdict
from pathlib import Path
WSS = "wss://ws.bitget.com/v2/ws/public"
SYMS = ("BTCUSDT", "ETHUSDT", "SOLUSDT")
OUT = Path(__file__).resolve().parents[1] / "out" / "probe_ws_action.csv"
async def run(minutes: int) -> None:
import csv
import aiohttp
payload = {"op": "subscribe",
"args": [{"instType": "USDT-FUTURES", "channel": "candle1m",
"instId": s} for s in SYMS]}
n_rows = 0
# 每个币记录:当前时间戳、该时间戳下已见过的 action 序列
cur: dict[str, int] = {}
seen: dict[str, list] = defaultdict(list)
rollovers: list[dict] = []
cnt: dict = defaultdict(lambda: defaultdict(int))
# 边收边写:进程若被杀,已采到的样本仍在盘上
OUT.parent.mkdir(parents=True, exist_ok=True)
fh = OUT.open("w", newline="")
w = csv.DictWriter(fh, fieldnames=["t_ms", "sym", "action", "kline_ts"])
w.writeheader()
print(f"[原始 WS 探针] {SYMS} · candle1m · 跑 {minutes} 分钟", flush=True)
deadline = time.time() + minutes * 60
try:
while time.time() < deadline:
try:
async with aiohttp.ClientSession() as sess, \
sess.ws_connect(WSS, heartbeat=20) as ws:
await ws.send_str(json.dumps(payload))
while time.time() < deadline:
msg = await ws.receive(timeout=30)
# 断开后 receive() 会立刻返回 CLOSED;这里若只 continue
# 就成了无等待空转,必须跳出去重连
if msg.type is not aiohttp.WSMsgType.TEXT:
print(f" 非文本消息 {msg.type},重连", flush=True)
break
raw = msg.data
if raw == "pong":
continue
try:
d = json.loads(raw)
except Exception:
continue
if "data" not in d or "arg" not in d:
if d.get("event"):
print(f" 事件: {d}", flush=True)
continue
now_ms = int(time.time() * 1000)
sym = d["arg"]["instId"]
action = d.get("action")
kts = int(d["data"][0][0])
w.writerow({"t_ms": now_ms, "sym": sym,
"action": action, "kline_ts": kts})
n_rows += 1
cnt[sym][action] += 1
if n_rows % 50 == 0:
fh.flush()
prev = cur.get(sym)
if prev is None:
cur[sym] = kts
seen[sym] = [(action, now_ms)]
continue
if kts > prev:
cur[sym] = kts
seen[sym] = [(action, now_ms)]
rollovers.append({"sym": sym, "kline_ts": kts,
"first_action": action,
"first_ms": now_ms})
print(f" {sym} 换根 -> 首条 action={action}",
flush=True)
else:
seen[sym].append((action, now_ms))
# 若首条是 snapshot,找该时间戳下首条 update 的延后量
for r in rollovers:
if (r["sym"] == sym and r["kline_ts"] == kts
and r["first_action"] == "snapshot"
and "gap_to_update_ms" not in r
and action == "update"):
r["gap_to_update_ms"] = now_ms - r["first_ms"]
print(f" {sym} snapshot->update 间隔 "
f"{r['gap_to_update_ms']}ms", flush=True)
except asyncio.CancelledError:
raise
except Exception as e:
print(f" 连接异常 {type(e).__name__}: {e}1s 后重连", flush=True)
# 重连前固定歇一下,避免服务端持续拒绝时打成重连风暴
if time.time() < deadline:
await asyncio.sleep(1)
finally:
fh.close()
print(f"\n########## 消息构成(共 {n_rows} 条)##########")
for s in SYMS:
print(f" {s}: {dict(cnt[s])}")
print(f"\n########## 换根时首条消息的 action{len(rollovers)} 次)##########")
by = defaultdict(lambda: defaultdict(int))
gaps = defaultdict(list)
for r in rollovers:
by[r["sym"]][r["first_action"]] += 1
if "gap_to_update_ms" in r:
gaps[r["sym"]].append(r["gap_to_update_ms"])
for s in SYMS:
g = sorted(gaps[s])
med = g[len(g) // 2] if g else None
print(f" {s}: 首条 action 分布 {dict(by[s])}"
+ (f" · snapshot->update 中位 {med}msn={len(g)}" if g else ""))
all_g = sorted(x for v in gaps.values() for x in v)
if all_g:
print(f"\n合计 snapshot->update 中位 {all_g[len(all_g) // 2]}ms "
f"n={len(all_g)})——这就是 Hummingbot 因丢弃 snapshot 白等的时间")
print(f"\n产物写入 {OUT}")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--minutes", type=int, default=6)
asyncio.run(run(ap.parse_args().minutes))
if __name__ == "__main__":
main()
+110
View File
@@ -0,0 +1,110 @@
"""查 Bitget candle1m 每条推送里到底带几根 K 线、最新的在哪一端。
归因结果显示 ccxt.pro 比原始 WS 快约 1.4 秒,而两者订阅的是同一条频道。
Hummingbot 的解析取 data["data"][0],我的探针也取 [0],两者一致地慢——
若 Bitget 一条消息里带多根、最新在末尾,取 [0] 就会系统性落后一个滑动窗口。
对每条消息记录:元素个数、首末元素的时间戳、以及首末之差。
若普遍 len>1 且末元素更新,则 [0] 就是那 1.4 秒的来源,且可用一行覆盖修好。
.venv/bin/python research/live/probe_ws_payload.py --minutes 4
"""
from __future__ import annotations
import argparse
import asyncio
import json
import time
from collections import Counter, defaultdict
WSS = "wss://ws.bitget.com/v2/ws/public"
SYMS = ("BTCUSDT", "ETHUSDT", "SOLUSDT")
async def run(minutes: int) -> None:
import aiohttp
payload = {"op": "subscribe",
"args": [{"instType": "USDT-FUTURES", "channel": "candle1m",
"instId": s} for s in SYMS]}
n_elems = Counter()
action_elems = Counter()
# 取 [0] 相对取 [-1] 的落后量:同一根 K 线,两种读法各自首次见到的时刻
first_seen_head: dict = {}
first_seen_tail: dict = {}
n_msg = 0
print(f"[载荷探针] {SYMS} · 跑 {minutes} 分钟", flush=True)
deadline = time.time() + minutes * 60
while time.time() < deadline:
try:
async with aiohttp.ClientSession() as sess, \
sess.ws_connect(WSS, heartbeat=20) as ws:
await ws.send_str(json.dumps(payload))
while time.time() < deadline:
msg = await ws.receive(timeout=30)
# 断开后 receive() 立刻返回 CLOSED,只 continue 会变成空转
if msg.type is not aiohttp.WSMsgType.TEXT:
print(f" 非文本消息 {msg.type},重连", flush=True)
break
if msg.data == "pong":
continue
d = json.loads(msg.data)
if "data" not in d or "arg" not in d:
continue
arr = d["data"]
if not arr:
continue
n_msg += 1
sym = d["arg"]["instId"].replace("USDT", "")
act = d.get("action")
n_elems[len(arr)] += 1
action_elems[(act, len(arr))] += 1
now_ms = int(time.time() * 1000)
head_ts, tail_ts = int(arr[0][0]), int(arr[-1][0])
first_seen_head.setdefault((sym, head_ts), now_ms)
first_seen_tail.setdefault((sym, tail_ts), now_ms)
if n_msg <= 6 or len(arr) > 1:
print(f" {sym} action={act} 元素数={len(arr)} "
f"首ts={head_ts} 末ts={tail_ts} "
f"跨度={(tail_ts - head_ts) // 1000}s", flush=True)
except asyncio.CancelledError:
raise
except Exception as e:
print(f" 连接异常 {type(e).__name__}: {e}1s 后重连", flush=True)
if time.time() < deadline:
await asyncio.sleep(1)
print(f"\n########## 每条消息的元素个数分布(共 {n_msg} 条)##########")
for k in sorted(n_elems):
print(f" {k} 根: {n_elems[k]}")
print("\n按 action 拆分:")
for k in sorted(action_elems, key=lambda x: (str(x[0]), x[1])):
print(f" action={k[0]} 元素数={k[1]}: {action_elems[k]}")
# 同一根 K 线,用 [-1] 读比用 [0] 读早多少
lead = defaultdict(list)
for (sym, ts), t_tail in first_seen_tail.items():
t_head = first_seen_head.get((sym, ts))
if t_head is not None:
lead[sym].append(t_head - t_tail)
print("\n########## 取 [-1] 比取 [0] 提前多少(ms##########")
for s in SYMS:
v = sorted(lead[s.replace("USDT", "")])
if not v:
print(f" {s}: 无配对")
continue
print(f" {s}: n={len(v)} 中位 {v[len(v) // 2]}ms 最大 {v[-1]}ms")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--minutes", type=int, default=4)
asyncio.run(run(ap.parse_args().minutes))
if __name__ == "__main__":
main()
+396
View File
@@ -0,0 +1,396 @@
"""影子交易器:在 Hummingbot 运行时上测 1m 腿的真实入场滑点。
不下单。用 Hummingbot 的 Bitget 连接器取真实盘口,按信号方向和仓位吃单深度
算出「若此刻市价单进场会成交在哪」,再与回测假设的成交价相减。
为什么必须跑在 Hummingbot 上而不是自写脚本:要测的是**生产路径**的滑点。
决定成交价的是实际执行链路的延迟,换个运行时测出来的数就不作数了。
(连接器的换根解析 bug 见 patched_candles.py,已修,拿回约 1.06 秒。)
为什么不能用 paper trade 的成交:那是 Hummingbot 自己的撮合模型模拟的,
测出来是模型行为不是市场行为。
口径对齐 aggregate_robustness.py:回测假设成交在**信号次根的开盘价**
(entry_delay=1),所以基准价就是换根后新一根的 open。滑点为正表示比回测差。
### 盘口滚动缓冲把延迟变成自变量
每 100ms 存一份盘口。信号触发后,不只记「我们实际算完时」的滑点,而是回查
t_close+0.5s / 1s / 2s / 5s 各一个。这样即使本机算得慢,也能读出「若延迟为
X 秒,滑点是多少」,决策不被自身实现拖累。
### 滑点分解
延迟漂移 中间价相对次根开盘价的偏移——主项,且入场方向上系统性追价
盘口价差 最优价相对中间价
深度冲击 吃单加权价相对最优价
Bitget 永续实测价差仅约 0.01bp、100 档深度,故预期延迟漂移占绝大部分。
docker run -d --name shadow -w /home/hummingbot \\
-e PYTHONPATH=/home/hummingbot:/repo/research:/repo/research/live:/repo \\
-v $PWD:/repo:ro -v $PWD/research/out:/out \\
--entrypoint /opt/conda/envs/hummingbot/bin/python \\
hummingbot/hummingbot:latest /repo/research/live/shadow_hb.py --hours 24
"""
from __future__ import annotations
import argparse
import asyncio
import csv
import time
from collections import deque
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
import numpy as np
import pandas as pd
SYMS = ("BTC", "ETH", "SOL")
# 多存一根:deque 尾部是尚未收盘的当前根,剔除后正好剩 step39 定下的窗口
LTF_BARS, HTF_BARS = 2001, 801 # 有效窗口 2000 / 800,命中率在此饱和
BOOK_HZ = 10 # 盘口采样 10Hz
BOOK_KEEP_S = 30 # 缓冲保留 30 秒,够回查到 +5s
BOOK_DEPTH = 25
DELAYS_S = (0.5, 1.0, 2.0, 5.0) # 回查点
NOTIONALS = (1_000.0, 5_000.0, 20_000.0)
NUM_COLS = ["timestamp", "open", "high", "low", "close", "volume"]
def out_dir() -> Path:
p = Path("/out")
return p if p.is_dir() else Path(__file__).resolve().parents[1] / "out"
def hb_to_research(cdf: pd.DataFrame) -> pd.DataFrame:
"""Hummingbot 的 candles_df 转成 research/lib/data.py 的列结构。
HB 的 timestamp 是秒且无 date 列;chanlun 的 kline builder 需要真 datetime,
时区跟 lib/data.py 取 Asia/Shanghai,保证与回测同一口径。
"""
ts_ms = (cdf["timestamp"].astype("int64") * 1000)
date = pd.to_datetime(ts_ms, unit="ms", utc=True).dt.tz_convert("Asia/Shanghai")
out = pd.DataFrame({"timestamp": ts_ms.astype("int64"), "date": date})
for c in ("open", "high", "low", "close", "volume"):
out[c] = pd.to_numeric(cdf[c], errors="coerce")
return out.dropna().drop_duplicates(subset=["timestamp"]) \
.sort_values("timestamp").reset_index(drop=True)
def walk_book(levels: list[tuple[float, float]], notional: float
) -> tuple[float, float]:
"""吃单到 notional(计价币)为止,返回 (加权成交价, 实际吃到的额度)。
深度不足时返回吃到的部分,由调用方按 filled < notional 判断是否可信。
"""
if not levels:
return float("nan"), 0.0
got = 0.0
cost = 0.0
qty = 0.0
for px, sz in levels:
avail = px * sz
take = min(avail, notional - got)
if take <= 0:
break
q = take / px
cost += q * px
qty += q
got += take
if got >= notional - 1e-9:
break
if qty <= 0:
return float("nan"), 0.0
return cost / qty, got
class BookBuffer:
"""每币一份滚动盘口。按时间戳回查,取第一个不早于目标时刻的快照。"""
def __init__(self) -> None:
self.buf: dict[str, deque] = {s: deque() for s in SYMS}
def push(self, sym: str, t_ms: int, bids: list, asks: list) -> None:
d = self.buf[sym]
d.append((t_ms, bids, asks))
cutoff = t_ms - BOOK_KEEP_S * 1000
while d and d[0][0] < cutoff:
d.popleft()
def at(self, sym: str, t_ms: int) -> tuple | None:
best = None
for snap in self.buf[sym]:
if snap[0] >= t_ms:
best = snap
break
return best
class Shadow:
def __init__(self, workers: int, hours: float) -> None:
self.workers = workers
self.deadline = time.time() + hours * 3600
self.books = BookBuffer()
self.pool: ProcessPoolExecutor | None = None
self.feeds_l: dict = {}
self.feeds_h: dict = {}
self.connector = None
self.stop = asyncio.Event()
self.n_signal = 0
self.n_bars = 0
d = out_dir()
# 追加模式:长跑期间若重启,已收集的样本不该被清掉
p_sig = d / "shadow_signals.csv"
new_sig = not p_sig.exists() or p_sig.stat().st_size == 0
self.f_sig = p_sig.open("a", newline="")
self.w_sig = csv.DictWriter(self.f_sig, fieldnames=[
"sym", "kline_ts", "direction", "h1_agree",
"t_close_ms", "t_data_ms", "t_signal_ms",
"lag_data_ms", "lag_signal_ms",
"delay_label", "delay_ms", "notional",
"baseline_px", "mid", "best_px", "fill_px", "filled",
"slip_bp", "drift_bp", "spread_bp", "impact_bp"])
if new_sig:
self.w_sig.writeheader()
p_lat = d / "shadow_latency.csv"
new_lat = not p_lat.exists() or p_lat.stat().st_size == 0
self.f_lat = p_lat.open("a", newline="")
self.w_lat = csv.DictWriter(self.f_lat, fieldnames=[
"sym", "kline_ts", "t_close_ms", "t_data_ms", "t_signal_ms",
"lag_data_ms", "lag_signal_ms", "compute_ms", "n_bars", "n_hits"])
if new_lat:
self.w_lat.writeheader()
# ---------- 启动 ----------
async def start(self) -> None:
from hummingbot.connector.derivative.bitget_perpetual.bitget_perpetual_derivative import (
BitgetPerpetualDerivative,
)
from patched_candles import PatchedBitgetPerpetualCandles
for s in SYMS:
self.feeds_l[s] = PatchedBitgetPerpetualCandles(
f"{s}-USDT", "1m", LTF_BARS)
self.feeds_h[s] = PatchedBitgetPerpetualCandles(
f"{s}-USDT", "5m", HTF_BARS)
self.feeds_l[s].start()
self.feeds_h[s].start()
print(f"[影子] {SYMS} · 1m×{LTF_BARS} + 5m×{HTF_BARS} · "
f"{self.workers} 个计算进程", flush=True)
# 只取公开数据:无密钥 + trading_required=False
self.connector = BitgetPerpetualDerivative(
bitget_perpetual_api_key="", bitget_perpetual_secret_key="",
bitget_perpetual_passphrase="",
trading_pairs=[f"{s}-USDT" for s in SYMS],
trading_required=False)
await self.connector.start_network()
print(" 连接器已启动,等盘口与历史回填", flush=True)
t0 = time.time()
while time.time() - t0 < 600:
ready = all(f.ready for f in
list(self.feeds_l.values()) + list(self.feeds_h.values()))
books = all(self._snapshot(s) is not None for s in SYMS)
if ready and books:
break
await asyncio.sleep(1)
print(f" 就绪 {time.time() - t0:.1f}s · "
f"1m {[len(self.feeds_l[s]._candles) for s in SYMS]} 根 · "
f"5m {[len(self.feeds_h[s]._candles) for s in SYMS]}", flush=True)
def _snapshot(self, sym: str):
try:
ob = self.connector.get_order_book(f"{sym}-USDT")
except Exception:
return None
if ob is None:
return None
bids = [(float(r.price), float(r.amount))
for r, _ in zip(ob.bid_entries(), range(BOOK_DEPTH))]
asks = [(float(r.price), float(r.amount))
for r, _ in zip(ob.ask_entries(), range(BOOK_DEPTH))]
if not bids or not asks:
return None
return bids, asks
# ---------- 三个循环 ----------
async def sample_books(self) -> None:
period = 1.0 / BOOK_HZ
while not self.stop.is_set():
t = int(time.time() * 1000)
for s in SYMS:
snap = self._snapshot(s)
if snap:
self.books.push(s, t, snap[0], snap[1])
await asyncio.sleep(period)
async def watch_bars(self) -> None:
last = {s: (int(self.feeds_l[s]._candles[-1][0])
if len(self.feeds_l[s]._candles) else None) for s in SYMS}
while not self.stop.is_set():
for s in SYMS:
c = self.feeds_l[s]._candles
if not len(c):
continue
newest = int(c[-1][0])
if last[s] is not None and newest > last[s]:
t_data = int(time.time() * 1000)
kts = newest * 1000 if newest < 1e12 else newest
asyncio.create_task(self.on_bar(s, kts, t_data))
last[s] = newest
await asyncio.sleep(0.01)
async def on_bar(self, sym: str, kline_ts: int, t_data: int) -> None:
"""kline_ts 是新一根的开盘时刻,也就是上一根的收盘时刻 t_close。"""
df_l = hb_to_research(self.feeds_l[sym].candles_df)
df_h = hb_to_research(self.feeds_h[sym].candles_df)
# 末行是刚开始的那根,未收盘,必须剔除,否则等于用未来数据
df_l = df_l[df_l["timestamp"] < kline_ts]
df_h = df_h[df_h["timestamp"] < kline_ts]
baseline = self._new_bar_open(sym, kline_ts)
t0 = time.perf_counter()
payload = (df_l[NUM_COLS].values.tolist(),
df_h[NUM_COLS].values.tolist())
loop = asyncio.get_running_loop()
from shadow_signal import compute_packed
res = await loop.run_in_executor(self.pool, compute_packed, payload)
compute_ms = int((time.perf_counter() - t0) * 1000)
t_signal = int(time.time() * 1000)
self.n_bars += 1
self.w_lat.writerow({
"sym": sym, "kline_ts": kline_ts, "t_close_ms": kline_ts,
"t_data_ms": t_data, "t_signal_ms": t_signal,
"lag_data_ms": t_data - kline_ts,
"lag_signal_ms": t_signal - kline_ts,
"compute_ms": compute_ms, "n_bars": res.get("n_bars", 0),
"n_hits": len(res.get("hits", []))})
self.f_lat.flush()
if res.get("error"):
print(f" [{sym}] 信号计算出错 {res['error']}", flush=True)
return
hits = res.get("hits", [])
if not hits:
return
if baseline is None or not np.isfinite(baseline):
print(f" [{sym}] 有信号但拿不到次根开盘价,跳过", flush=True)
return
for h in hits:
self.n_signal += 1
print(f" ★ [{sym}] {kline_ts} 方向 {h['direction']:+d} "
f"h1_agree={h['h1_agree']} · 数据 {t_data - kline_ts}ms "
f"信号 {t_signal - kline_ts}ms", flush=True)
# 最远的回查点在 t_close+5s,此刻尚未发生;等它过去再一次性落盘
asyncio.create_task(
self._record_later(sym, kline_ts, h, t_data, t_signal, baseline))
async def _record_later(self, sym: str, kline_ts: int, hit: dict,
t_data: int, t_signal: int, baseline: float) -> None:
target = kline_ts + int(max(DELAYS_S) * 1000) + 500
wait = target / 1000.0 - time.time()
if wait > 0:
await asyncio.sleep(wait)
self._record(sym, kline_ts, hit, t_data, t_signal, baseline)
def _new_bar_open(self, sym: str, kline_ts: int) -> float | None:
"""次根开盘价 = 回测假设的成交价。"""
c = self.feeds_l[sym]._candles
if not len(c):
return None
row = c[-1]
ts = int(row[0])
ts = ts * 1000 if ts < 1e12 else ts
return float(row[1]) if ts == kline_ts else None
def _record(self, sym: str, kline_ts: int, hit: dict,
t_data: int, t_signal: int, baseline: float) -> None:
points = [("actual", t_signal - kline_ts)]
points += [(f"{d}s", int(d * 1000)) for d in DELAYS_S]
d_sign = hit["direction"]
for label, delay_ms in points:
snap = self.books.at(sym, kline_ts + delay_ms)
if snap is None:
continue
_, bids, asks = snap
best_bid, best_ask = bids[0][0], asks[0][0]
mid = (best_bid + best_ask) / 2.0
# 多头吃卖盘,空头吃买盘
side = asks if d_sign > 0 else bids
best_px = best_ask if d_sign > 0 else best_bid
for notional in NOTIONALS:
fill, filled = walk_book(side, notional)
if not np.isfinite(fill):
continue
slip = d_sign * (fill - baseline) / baseline * 1e4
drift = d_sign * (mid - baseline) / baseline * 1e4
spread = d_sign * (best_px - mid) / mid * 1e4
impact = d_sign * (fill - best_px) / best_px * 1e4
self.w_sig.writerow({
"sym": sym, "kline_ts": kline_ts,
"direction": d_sign, "h1_agree": hit["h1_agree"],
"t_close_ms": kline_ts, "t_data_ms": t_data,
"t_signal_ms": t_signal,
"lag_data_ms": t_data - kline_ts,
"lag_signal_ms": t_signal - kline_ts,
"delay_label": label, "delay_ms": delay_ms,
"notional": notional, "baseline_px": baseline,
"mid": mid, "best_px": best_px, "fill_px": fill,
"filled": round(filled, 2),
"slip_bp": round(slip, 4), "drift_bp": round(drift, 4),
"spread_bp": round(spread, 4),
"impact_bp": round(impact, 4)})
self.f_sig.flush()
async def heartbeat(self) -> None:
while not self.stop.is_set():
await asyncio.sleep(300)
depth = {s: len(self.books.buf[s]) for s in SYMS}
print(f" [心跳] 已处理 {self.n_bars} 根 · 命中 {self.n_signal}"
f"· 盘口缓冲 {depth}", flush=True)
async def run(self) -> None:
await self.start()
tasks = [asyncio.create_task(self.sample_books()),
asyncio.create_task(self.watch_bars()),
asyncio.create_task(self.heartbeat())]
while time.time() < self.deadline:
await asyncio.sleep(5)
self.stop.set()
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
for f in list(self.feeds_l.values()) + list(self.feeds_h.values()):
f.stop()
await self.connector.stop_network()
self.f_sig.close()
self.f_lat.close()
print(f"\n收工:{self.n_bars} 根 · {self.n_signal} 个信号", flush=True)
async def main_async(workers: int, hours: float, pool) -> None:
sh = Shadow(workers, hours)
sh.pool = pool
await sh.run()
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--hours", type=float, default=24.0)
ap.add_argument("--workers", type=int, default=2)
a = ap.parse_args()
# 进程池必须在事件循环和任何 WS 连接之前建好:fork 一个已带活跃 socket
# 的进程会把连接状态一起复制过去,后果不可预测
with ProcessPoolExecutor(max_workers=a.workers) as pool:
asyncio.run(main_async(a.workers, a.hours, pool))
if __name__ == "__main__":
main()
+157
View File
@@ -0,0 +1,157 @@
"""影子交易器的报表:首日延迟门槛 + 滑点对延迟曲线。
两份产物对应计划里的两件事。
### 延迟门槛(提前止损用)
跑满 24 小时先看这个。若**总延迟已令预期漂移超过余量**,说明方案在这台机器
上就不成立,不必等两周样本再停。余量取 bitget_baseline.py 的实测值:
BTC 0.13bp(本就为负,只作参照)、ETH +4.02bp、SOL +2.92bp。
漂移按随机游走折算:σ_1m · √(t/60)。这是下限——入场条件是「收盘突破转强」,
那一刻价格正朝我们方向跑,延迟造成的是系统性追价,不会正负抵消。所以实测
滑点理应比这个折算值更差,两者对照本身就是个校验。
### 滑点对延迟曲线
把延迟当自变量:0.5s / 1s / 2s / 5s 各一个滑点值,外加「actual」= 本机实际
算完的时刻。这样即便本机算得慢,也能读出「若延迟压到 X 秒,滑点是多少」,
决策不被当前实现拖累。
.venv/bin/python research/live/shadow_report.py
"""
from __future__ import annotations
from pathlib import Path
import numpy as np
import pandas as pd
HERE = Path(__file__).resolve().parent
OUT = HERE.parent / "out"
SYMS = ("BTC", "ETH", "SOL")
BUDGET_BP = {"BTC": -0.13, "ETH": 4.02, "SOL": 2.92}
def vol_bp() -> dict[str, float]:
v = {}
for s in SYMS:
f = HERE / "cache" / f"bitget_{s}_1m_30d.feather"
if not f.exists():
f = HERE / "cache" / f"bitget_{s}_1m_210d.feather"
if not f.exists():
continue
c = np.log(pd.read_feather(f)["close"].to_numpy(float))
v[s] = float(np.nanstd(np.diff(c)) * 1e4)
return v
def latency_gate(lat: pd.DataFrame, vols: dict) -> None:
print("########## 一、延迟门槛 ##########")
span_h = (lat["t_close_ms"].max() - lat["t_close_ms"].min()) / 3.6e6
print(f"样本跨度 {span_h:.1f} 小时 · 共 {len(lat)}\n")
print(f"{'':<5}{'根数':>6}{'数据ms':>9}{'计算ms':>9}{'总延迟ms':>10}"
f"{'P90ms':>8}{'折算漂移bp':>12}{'余量bp':>9}{'占余量':>9}")
verdicts = {}
for s in SYMS:
g = lat[lat["sym"] == s]
if g.empty:
continue
d = float(np.median(g["lag_data_ms"]))
c = float(np.median(g["compute_ms"]))
t = float(np.median(g["lag_signal_ms"]))
p90 = float(np.percentile(g["lag_signal_ms"], 90))
vol = vols.get(s)
drift = vol * np.sqrt(t / 60_000) if vol else float("nan")
b = BUDGET_BP[s]
share = drift / b if b > 0 else float("nan")
verdicts[s] = (drift, b)
txt = f"{share * 100:.0f}%" if b > 0 else "—(负)"
print(f"{s:<5}{len(g):>6}{d:>9.0f}{c:>9.0f}{t:>10.0f}{p90:>8.0f}"
f"{drift:>12.2f}{b:>9.2f}{txt:>9}")
print("\n判读:")
for s, (drift, b) in verdicts.items():
if b <= 0:
print(f" {s}: 余量本就为负,不作交易标的,仅作延迟参照")
elif drift > b:
print(f" {s}: 折算漂移 {drift:.2f}bp 已超余量 {b:.2f}bp —— 停下改方案")
elif drift > b * 0.6:
print(f" {s}: 折算漂移 {drift:.2f}bp 吃掉余量 {b:.2f}bp 的六成以上,"
f"需要压延迟或放弃")
else:
print(f" {s}: 折算漂移 {drift:.2f}bp 对余量 {b:.2f}bp 尚有空间,继续收集")
def slippage_curve(sig: pd.DataFrame) -> None:
print("\n\n########## 二、滑点对延迟曲线 ##########")
if sig.empty:
print(" 尚无信号样本")
return
n_sig = sig.groupby(["sym", "kline_ts", "direction"]).ngroups
n_agree = sig[sig["h1_agree"] == 1].groupby(
["sym", "kline_ts", "direction"]).ngroups
print(f"信号总数 {n_sig}(其中 h1_agree=1 的 {n_agree} 个)\n")
order = ["0.5s", "1.0s", "2.0s", "5.0s", "actual"]
for scope, sub in (("全部信号", sig),
("仅 h1_agree=1(主口径)", sig[sig["h1_agree"] == 1])):
if sub.empty:
continue
print(f"--- {scope} ---")
print(f"{'延迟':<8}{'仓位':>9}{'n':>5}{'滑点均值bp':>12}"
f"{'中位bp':>9}{'漂移bp':>9}{'价差bp':>9}{'冲击bp':>9}")
for lb in order:
g0 = sub[sub["delay_label"] == lb]
if g0.empty:
continue
for nt in sorted(sub["notional"].unique()):
g = g0[g0["notional"] == nt]
if g.empty:
continue
print(f"{lb:<8}{int(nt):>9}{len(g):>5}"
f"{g['slip_bp'].mean():>12.2f}"
f"{g['slip_bp'].median():>9.2f}"
f"{g['drift_bp'].mean():>9.2f}"
f"{g['spread_bp'].mean():>9.2f}"
f"{g['impact_bp'].mean():>9.2f}")
print()
print("--- 分币种(仅 h1_agree=1,仓位 5000---")
m = sig[(sig["h1_agree"] == 1) & (sig["notional"] == 5000.0)]
if m.empty:
print(" 尚无样本")
return
print(f"{'':<5}{'延迟':<8}{'n':>5}{'滑点均值bp':>12}{'余量bp':>9}")
for s in SYMS:
for lb in order:
g = m[(m["sym"] == s) & (m["delay_label"] == lb)]
if g.empty:
continue
print(f"{s:<5}{lb:<8}{len(g):>5}{g['slip_bp'].mean():>12.2f}"
f"{BUDGET_BP[s]:>9.2f}")
def main() -> None:
vols = vol_bp()
print("1m 收益标准差(bp/分钟,Bitget 实测):"
+ " ".join(f"{s} {v:.2f}" for s, v in vols.items()) + "\n")
f_lat = OUT / "shadow_latency.csv"
if f_lat.exists() and f_lat.stat().st_size > 0:
lat = pd.read_csv(f_lat)
if not lat.empty:
latency_gate(lat, vols)
else:
print("尚无延迟数据")
f_sig = OUT / "shadow_signals.csv"
if f_sig.exists() and f_sig.stat().st_size > 0:
sig = pd.read_csv(f_sig)
slippage_curve(sig)
else:
print("\n尚无信号数据")
if __name__ == "__main__":
main()
+109
View File
@@ -0,0 +1,109 @@
"""影子交易器的信号函数——在子进程里跑,不碰事件循环。
单次调用约 0.26s 的纯 CPU,且 chanlun 是纯 Python 受 GIL 限制,放进
Hummingbot 的 asyncio 循环里会把行情处理一起卡住,所以必须隔离到独立进程。
口径与 [aggregate_robustness.run_once] 逐行对齐:同样的 build_htf_zones →
find_fast_bsp3 → attach_htf_context(h1) 链路,同样的 h1_agree 过滤。两边
必须一致,否则影子测出来的滑点没法和 3.9bp 预算对照。
与回测的唯一差别是这里只关心**最后一根已收盘 K 线**上有没有信号——
实盘只能在当下下单,历史信号无意义。
未过滤信号也一并返回:过滤后样本太稀,先用未过滤的当提前读数,
两者都记,靠 h1_agree 字段区分。
"""
from __future__ import annotations
import os
import warnings
warnings.filterwarnings("ignore")
for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
os.environ.setdefault(_v, "1")
def compute(df_l, df_h) -> dict:
"""在 df_l 的最后一根上找信号。df_l/df_h 都只含已收盘 K 线。
返回 dict
last_idx 最后一根在 chanlun 处理后 dataframe 里的下标
n_bars 实际参与计算的根数
hits 命中列表,每项 {direction, h1_agree}
error 出错时的说明,正常为 None
"""
import pandas as pd
try:
from chanlun import TF_DF
from lib.fast_bsp3 import find_fast_bsp3
from lib.fx_signal import extract_fx_signals, signals_to_frame
from lib.nested_bsp import attach_htf_context, htf_fx_timeline
from lib.nested_level import build_htf_zones
chan_l = TF_DF(df_l, 1, "1m")
cdf = chan_l.dataframe
last = len(cdf) - 1
base = {"last_idx": last, "n_bars": int(len(df_l)), "hits": [],
"error": None}
zones = build_htf_zones(cdf, "1m", chan=chan_l)
if zones.empty:
return base
sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
if sig is None or sig.empty:
return base
# 只留落在最后一根上的信号,其余是历史,实盘下不了
cur = sig[sig["entry_idx"].astype(int) == last]
if cur.empty:
return base
# 5m 同向过滤:算得出就标 h1_agree,算不出就当未过滤照记
agree_map: dict[int, int] = {}
if df_h is not None and len(df_h) > 0:
chan_h = TF_DF(df_h, 1, "5m")
hdf = chan_h.dataframe
tl = htf_fx_timeline(
signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
full = attach_htf_context(sig, cdf, tl, "h1")
f_cur = full[full["entry_idx"].astype(int) == last]
for _, r in f_cur.iterrows():
agree_map[int(r["direction"])] = int(r.get("h1_agree", 0))
base["hits"] = [{"direction": int(r["direction"]),
"h1_agree": agree_map.get(int(r["direction"]), 0)}
for _, r in cur.iterrows()]
return base
except Exception as e: # 子进程里异常必须带回主进程,否则只见超时不见原因
import traceback
return {"last_idx": -1, "n_bars": int(len(df_l)) if df_l is not None else 0,
"hits": [], "error": f"{type(e).__name__}: {e}",
"traceback": traceback.format_exc()}
NUM_COLS = ("timestamp", "open", "high", "low", "close", "volume")
def _rebuild(rows) -> "object":
"""只传数值列,date 在这里按 lib/data.py 的同一规则重建。
跨进程传 tz-aware 的 datetime 既慢又容易在字符串往返中丢时区,
而时区若与回测不一致,chanlun 的 K 线标签就会错位。
"""
import pandas as pd
df = pd.DataFrame(rows, columns=list(NUM_COLS))
df["timestamp"] = df["timestamp"].astype("int64")
date = pd.to_datetime(df["timestamp"], unit="ms", utc=True) \
.dt.tz_convert("Asia/Shanghai")
df.insert(1, "date", date)
return df
def compute_packed(payload: tuple) -> dict:
"""ProcessPoolExecutor 的入口:收 (l_rows, h_rows) 两组数值行。"""
l_rows, h_rows = payload
df_l = _rebuild(l_rows)
df_h = _rebuild(h_rows) if h_rows else None
return compute(df_l, df_h)
+150
View File
@@ -0,0 +1,150 @@
"""对照实验:信号集对微小数据差异有多敏感。
venue_parity 量到 Bitget 与 Binance 的 1m 信号同根重合率只有 16%~41%
而两家的 close 中位差仅 0.3bp、P95 约 2.5bp。在断言「换交易所会换掉一批信号」
之前,必须先排除另一种解释:**信号定义本身就对任何 2bp 级别的扰动极度敏感**。
这两种解释的后果完全不同:
venue 差异 -> 换成 Bitget 自己的回测基线即可归因
内在脆弱 -> Binance 回测的那份逐笔清单根本不可复现,滑点无从对照
做法:拿 Binance 原始数据当基线,注入不同幅度的 iid 噪声后重跑同一管线,
看重合率随噪声幅度的衰减曲线。噪声 0 必须给出 100%,否则说明管线不确定。
顺带单独测一档「tick 粗化」:把 Bitget SOL 的 0.001 精度四舍五入到 Binance
的 0.01,看重合率是否回升——若回升,SOL 的低重合就主要是精度差异造成的。
输出 out/signal_sensitivity.csv。
"""
from __future__ import annotations
import argparse
import os
import sys
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().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
pd.set_option("display.width", 240)
sys.path.insert(0, str(HERE))
from venue_parity import overlap, pipeline # noqa: E402
SYMS = ("BTC", "ETH", "SOL")
LEVELS = (0.0, 0.25, 0.5, 1.0, 2.0, 4.0) # bp
TICK = {"BTC": 0.1, "ETH": 0.01, "SOL": 0.01}
def perturb(df: pd.DataFrame, bp: float, tick: float, seed: int) -> pd.DataFrame:
"""给 OHLC 各自注入 iid 噪声,再修复 high/low 的包含关系并按 tick 归整。"""
if bp <= 0:
return df
out = df.copy()
rng = np.random.default_rng(seed)
sd = bp / 1e4
for c in ("open", "high", "low", "close"):
v = out[c].to_numpy(dtype=float)
out[c] = v * (1.0 + rng.normal(0.0, sd, size=len(v)))
o, h, l, c = (out[x].to_numpy(dtype=float) for x in ("open", "high", "low", "close"))
out["high"] = np.maximum.reduce([h, o, c])
out["low"] = np.minimum.reduce([l, o, c])
for x in ("open", "high", "low", "close"):
out[x] = np.round(out[x] / tick) * tick
return out
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--symbols", default="BTC,ETH,SOL")
ap.add_argument("--seeds", type=int, default=2)
ap.add_argument("--warmup", type=int, default=2000)
args = ap.parse_args()
from lib.data import load_local
syms = [s.strip() for s in args.symbols.split(",")]
period_ms = 60_000
print(f"[信号敏感性] {syms} · 噪声档 {LEVELS} bp · 每档 {args.seeds} 个种子\n",
flush=True)
rows = []
for sym in syms:
# 与 venue_parity 用同一段窗口,便于两组数字直接对照
cache = HERE / "cache" / f"bitget_{sym}_1m_30d.feather"
if not cache.exists():
print(f"{sym}: 缺 {cache.name},先跑 venue_parity.py")
continue
bg = pd.read_feather(cache)
lo, hi = int(bg.timestamp.min()), int(bg.timestamp.max())
bn_l = load_local(f"{sym}/USDT:USDT", "1m")
bn_h = load_local(f"{sym}/USDT:USDT", "5m")
bn_l = bn_l[(bn_l.timestamp >= lo) & (bn_l.timestamp <= hi)].reset_index(drop=True)
bn_h = bn_h[(bn_h.timestamp >= lo) & (bn_h.timestamp <= hi)].reset_index(drop=True)
base = pipeline(bn_l, bn_h)
cut = bn_l["timestamp"].to_numpy()[min(args.warmup, len(bn_l) - 1)]
base = base[base.entry_ts >= cut]
base_f = base[base["h1_agree"] == 1]
print(f"── {sym} 基线 原始 {len(base)} 笔 / 过滤后 {len(base_f)}",
flush=True)
for bp in LEVELS:
n_seeds = 1 if bp == 0 else args.seeds
acc = {"原始": [], "5m同向后": []}
cnt = {"原始": [], "5m同向后": []}
for k in range(n_seeds):
pert = pipeline(perturb(bn_l, bp, TICK[sym], 1000 + k), bn_h)
if pert.empty:
continue
pert = pert[pert.entry_ts >= cut]
pert_f = pert[pert["h1_agree"] == 1]
for tag, a, b in (("原始", base, pert), ("5m同向后", base_f, pert_f)):
o = overlap(a, b, period_ms)
acc[tag].append(o["同根"])
cnt[tag].append(o["b"])
for tag in ("原始", "5m同向后"):
if not acc[tag]:
continue
rows.append({"品种": sym, "口径": tag, "噪声bp": bp,
"基线笔数": len(base if tag == "原始" else base_f),
"扰动后笔数": round(float(np.mean(cnt[tag])), 1),
"同根重合": float(np.mean(acc[tag]))})
print(f" 噪声 {bp:>4.2f}bp: 原始 {np.mean(acc['原始']) * 100:5.1f}% · "
f"过滤后 {np.mean(acc['5m同向后']) * 100:5.1f}%", flush=True)
if not rows:
print("无结果")
return
tb = pd.DataFrame(rows)
print("\n" + "=" * 100)
print("########## 噪声幅度 → 同根重合率 ##########")
piv = tb[tb["口径"] == "5m同向后"].pivot_table(
index="噪声bp", columns="品种", values="同根重合")
print((piv * 100).round(1).to_string())
print(" 行是注入的 iid 噪声幅度(bp),值是与无噪声基线的同根重合率。")
print("\n########## 与 venue_parity 的实测对照 ##########")
print(" Bitget↔Binance 实测:close 中位差 0.05~0.33bp、P95 2.0~3.0bp")
print(" 过滤后同根重合 BTC 40.9% / ETH 25.0% / SOL 15.9%")
print(" 若上表在 1~2bp 档就掉到同一水平,说明主因是信号定义的内在脆弱,")
print(" 而不是 Bitget 这家交易所特殊。")
out = RESEARCH / "out" / "signal_sensitivity.csv"
tb.to_csv(out, index=False)
print(f"\n产物写入 {out}")
if __name__ == "__main__":
main()
+306
View File
@@ -0,0 +1,306 @@
"""前置测量二:venue 对齐——Bitget 与 Binance 的 1m 是不是同一批信号。
研究数据全部来自 Binance,影子交易器却跑在 Bitget。若两家的 1m K 线有差异,
信号集就会不同,而这个差异会被误记到滑点账上——那样收集一两周也不可归因。
所以先把两家同期的 1m 拉齐,跑同一套管线,比三件事:
K 线层 时间戳缺口、close 价差(bp)、high/low 差异
信号层 原始 fast_bsp3 的重合率
过滤后 加 5m 同向过滤后的重合率(这才是实际要交易的那批)
重合率高 → 后面测到的滑点可以直接对照 Binance 回测的 3.9bp 预算。
重合率低 → 必须先补 Bitget 自己的回测基线,否则实验不可归因。
输出 out/venue_parity.csv。Bitget 数据缓存在 live/cache/,重跑不必再拉。
"""
from __future__ import annotations
import argparse
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().parent
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
pd.set_option("display.width", 240)
CACHE = HERE / "cache"
LTF, HTF = "1m", "5m"
HTF_RATIO = 5
SYMS = ("BTC", "ETH", "SOL")
NUMERIC = ("open", "high", "low", "close", "volume")
def _exchange():
import ccxt
# 这台机器在新加坡,直连 Bitget 0.30s。绝不要照抄交接文档里 Mac 的代理配置,
# 代理会把延迟放大到秒级,测出来的滑点就是代理的账。
# rateLimit 默认 50ms,连拉上千页 history-candles 会被 429,放宽到 120ms
return ccxt.bitget({"options": {"defaultType": "swap"},
"enableRateLimit": True, "rateLimit": 120})
def fetch_bitget(sym: str, tf: str, days: int, refresh: bool = False) -> pd.DataFrame:
"""分页拉 Bitget 永续 K 线,落盘缓存,列结构对齐 lib/data.py。"""
CACHE.mkdir(parents=True, exist_ok=True)
path = CACHE / f"bitget_{sym}_{tf}_{days}d.feather"
if path.exists() and not refresh:
return pd.read_feather(path)
ex = _exchange()
pair = f"{sym}/USDT:USDT"
period_ms = ex.parse_timeframe(tf) * 1000
t0 = time.perf_counter()
calls = 0
# 远端 history-candles 每页硬上限 200 根,而 ccxt 会按 limit 推算 endTime
# 只把窗口末尾的 200 根还给你。若照 limit=1000 步进,每页就白丢 800 根——
# 210 天曾因此只拿到应有量的 31%。故分页一律按 200 走。
PAGE = 200
def one(since: int) -> list:
"""单次取数并退避重试。history-candles 连拉上千次会触发 429。"""
for attempt in range(6):
try:
return ex.fetch_ohlcv(pair, tf, since=since, limit=PAGE)
except Exception as e:
if attempt == 5:
raise
wait = 2 ** attempt
print(f" {sym} {tf}: {type(e).__name__}{wait}s 后重试",
flush=True)
time.sleep(wait)
return []
def page(start: int, stop: int) -> list:
"""向前分页。Bitget 把 since 当开区间,故每次从上一批最后一根重取,
边界少的那一根靠去重消化。"""
nonlocal calls
got, since = [], start
while since < stop:
batch = one(since)
calls += 1
if len(batch) < 2:
break
got.extend(batch)
if batch[-1][0] <= since:
break
since = batch[-1][0]
if calls % 200 == 0:
print(f" {sym} {tf}: {len(got)} 根 / {calls} 次请求", flush=True)
return got
now = ex.milliseconds()
rows = page(now - days * 86_400_000, now)
# 补缺口:远端接口一次只给 200 根,个别区段仍可能漏,逐个补到补不动为止
for _ in range(5):
ts = np.unique(np.array([r[0] for r in rows], dtype="int64"))
if len(ts) < 2:
break
holes = np.where(np.diff(ts) > period_ms)[0]
if not len(holes):
break
before = len(ts)
for i in holes:
rows.extend(page(int(ts[i]), int(ts[i + 1])))
if len(np.unique([r[0] for r in rows])) <= before:
break
df = pd.DataFrame(rows, columns=["timestamp", *NUMERIC])
df["timestamp"] = df["timestamp"].astype("int64")
for c in NUMERIC:
df[c] = pd.to_numeric(df[c], errors="coerce")
df = (df.dropna(subset=list(NUMERIC))
.drop_duplicates(subset=["timestamp"])
.sort_values("timestamp")
.reset_index(drop=True))
df["date"] = (pd.to_datetime(df["timestamp"], unit="ms", utc=True)
.dt.tz_convert("Asia/Shanghai"))
df = df[["timestamp", "date", *NUMERIC]]
gap = int(((np.diff(df["timestamp"].to_numpy()) // period_ms) - 1).clip(0).sum())
print(f" {sym} {tf}: {len(df)} 根,{calls} 次请求,"
f"{time.perf_counter() - t0:.1f}s,残余缺口 {gap}", flush=True)
df.to_feather(path)
return df
def pipeline(df_l: pd.DataFrame, df_h: pd.DataFrame) -> pd.DataFrame:
"""全量口径跑一遍:原始信号 + 5m 同向过滤,返回带时间戳的信号表。"""
from chanlun import TF_DF
from lib.fast_bsp3 import find_fast_bsp3
from lib.fx_signal import extract_fx_signals, signals_to_frame
from lib.nested_bsp import attach_htf_context, htf_fx_timeline
from lib.nested_level import build_htf_zones
chan_l = TF_DF(df_l, 1, LTF)
cdf = chan_l.dataframe
zones = build_htf_zones(cdf, LTF, chan=chan_l)
if zones.empty:
return pd.DataFrame()
sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
if sig.empty:
return pd.DataFrame()
chan_h = TF_DF(df_h, 1, HTF)
hdf = chan_h.dataframe
tl = htf_fx_timeline(signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
full = attach_htf_context(sig, cdf, tl, "h1")
full["entry_ts"] = cdf["timestamp"].to_numpy()[full["entry_idx"].astype(int)]
return full
def overlap(a: pd.DataFrame, b: pd.DataFrame, period_ms: int,
tol_bars: int = 1) -> dict:
"""按时间戳比对两个信号集。方向也必须一致才算命中。"""
if a.empty or b.empty:
return {"a": len(a), "b": len(b), "同根": np.nan, f"±{tol_bars}": np.nan}
bt = b["entry_ts"].to_numpy()
bd = b["direction"].to_numpy()
exact = near = 0
for ts, d in zip(a["entry_ts"].to_numpy(), a["direction"].to_numpy()):
hit = np.where((bt == ts) & (bd == d))[0]
if len(hit):
exact += 1
near += 1
continue
if np.any((np.abs(bt - ts) <= tol_bars * period_ms) & (bd == d)):
near += 1
return {"a": len(a), "b": len(b),
"同根": exact / len(a), f"±{tol_bars}": near / len(a)}
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--days", type=int, default=30)
ap.add_argument("--symbols", default="BTC,ETH,SOL")
ap.add_argument("--warmup", type=int, default=2000,
help="丢弃前若干根的信号,避开中枢左边界效应")
ap.add_argument("--refresh", action="store_true")
args = ap.parse_args()
from lib.data import load_local
syms = [s.strip() for s in args.symbols.split(",")]
period_ms = 60_000
print(f"[venue 对齐] {syms} · 近 {args.days} 天 1m · "
f"预热丢弃 {args.warmup}\n", flush=True)
bar_rows, sig_rows = [], []
for sym in syms:
print(f"── {sym}", flush=True)
bg_l = fetch_bitget(sym, LTF, args.days, args.refresh)
bg_h = fetch_bitget(sym, HTF, args.days, args.refresh)
pair = f"{sym}/USDT:USDT"
bn_l_all = load_local(pair, LTF)
bn_h_all = load_local(pair, HTF)
if bn_l_all is None or bn_h_all is None:
print(f" 跳过:本地无 Binance 数据")
continue
# 只比两家都有的那段时间
lo = max(bg_l["timestamp"].min(), bn_l_all["timestamp"].min())
hi = min(bg_l["timestamp"].max(), bn_l_all["timestamp"].max())
bg_l = bg_l[(bg_l.timestamp >= lo) & (bg_l.timestamp <= hi)].reset_index(drop=True)
bn_l = bn_l_all[(bn_l_all.timestamp >= lo) & (bn_l_all.timestamp <= hi)].reset_index(drop=True)
bg_h = bg_h[bg_h.timestamp <= hi].reset_index(drop=True)
bn_h = bn_h_all[(bn_h_all.timestamp >= bg_h["timestamp"].min())
& (bn_h_all.timestamp <= hi)].reset_index(drop=True)
span_d = (hi - lo) / 86_400_000
expect = int((hi - lo) / period_ms) + 1
# K 线层比对
m = bg_l.merge(bn_l, on="timestamp", suffixes=("_bg", "_bn"))
dc = (m["close_bg"] - m["close_bn"]) / m["close_bn"] * 1e4
dh = (m["high_bg"] - m["high_bn"]) / m["high_bn"] * 1e4
dl = (m["low_bg"] - m["low_bn"]) / m["low_bn"] * 1e4
bar_rows.append({
"品种": sym, "重叠天数": round(span_d, 1),
"Bitget根数": len(bg_l), "Binance根数": len(bn_l),
"应有根数": expect,
"Bitget缺口": expect - len(bg_l), "Binance缺口": expect - len(bn_l),
"共有根数": len(m),
"close中位差": f"{dc.median():+.2f}bp",
"close绝对差P95": f"{dc.abs().quantile(.95):.2f}bp",
"high绝对差P95": f"{dh.abs().quantile(.95):.2f}bp",
"low绝对差P95": f"{dl.abs().quantile(.95):.2f}bp",
})
print(f" K线:重叠 {span_d:.1f} 天,共有 {len(m)} 根,"
f"close 中位差 {dc.median():+.2f}bpP95 {dc.abs().quantile(.95):.2f}bp",
flush=True)
# 信号层比对
t0 = time.perf_counter()
s_bg = pipeline(bg_l, bg_h)
s_bn = pipeline(bn_l, bn_h)
print(f" 管线跑完 {time.perf_counter() - t0:.1f}s", flush=True)
if s_bg.empty or s_bn.empty:
print(" 信号为空,跳过信号层")
continue
cut_bg = bg_l["timestamp"].to_numpy()[min(args.warmup, len(bg_l) - 1)]
cut_bn = bn_l["timestamp"].to_numpy()[min(args.warmup, len(bn_l) - 1)]
cut = max(cut_bg, cut_bn)
s_bg = s_bg[s_bg.entry_ts >= cut]
s_bn = s_bn[s_bn.entry_ts >= cut]
f_bg = s_bg[s_bg["h1_agree"] == 1]
f_bn = s_bn[s_bn["h1_agree"] == 1]
for tag, x, y in (("原始", s_bg, s_bn), ("5m同向后", f_bg, f_bn)):
o1 = overlap(x, y, period_ms) # Bitget 的信号有多少在 Binance 也有
o2 = overlap(y, x, period_ms) # 反向
sig_rows.append({
"品种": sym, "口径": tag,
"Bitget信号": o1["a"], "Binance信号": o1["b"],
"BG→BN同根": f"{o1['同根'] * 100:.1f}%",
"BG→BN±1根": f"{o1['±1根'] * 100:.1f}%",
"BN→BG同根": f"{o2['同根'] * 100:.1f}%",
"BN→BG±1根": f"{o2['±1根'] * 100:.1f}%",
})
print(f" {tag}Bitget {o1['a']} 笔 / Binance {o1['b']} 笔,"
f"同根重合 {o1['同根'] * 100:.1f}%,±1根 {o1['±1根'] * 100:.1f}%",
flush=True)
if not bar_rows:
print("无结果")
return
print("\n" + "=" * 120)
print("########## 1. K 线层 ##########")
tb_bar = pd.DataFrame(bar_rows)
print(tb_bar.to_string(index=False))
print(" 缺口是「应有根数 − 实际根数」,永续在极端行情或维护时会漏推。")
print("\n########## 2. 信号层 ##########")
tb_sig = pd.DataFrame(sig_rows)
print(tb_sig.to_string(index=False))
print("\n########## 结论 ##########")
fin = tb_sig[tb_sig["口径"] == "5m同向后"]
if not fin.empty:
v = fin["BG→BN同根"].str.rstrip("%").astype(float)
print(f" 过滤后口径的同根重合率:{v.min():.1f}% ~ {v.max():.1f}%"
f"均值 {v.mean():.1f}%")
print(" 高 → 滑点可直接对照 Binance 回测的 3.9bp 预算;")
print(" 低 → 必须先补 Bitget 自己的 1m 回测基线,否则实验不可归因。")
out_dir = RESEARCH / "out"
tb_bar.to_csv(out_dir / "venue_parity_bars.csv", index=False)
tb_sig.to_csv(out_dir / "venue_parity_signals.csv", index=False)
print(f"\n产物写入 {out_dir}/venue_parity_bars.csv 与 venue_parity_signals.csv")
if __name__ == "__main__":
main()
+179
View File
@@ -0,0 +1,179 @@
"""验证补丁是否真把那 1.1 秒拿回来了。
容器内同一进程并行跑三条路径,共用时钟与网络:
stock 上游 BitgetPerpetualCandles(取 data[0]
patched PatchedBitgetPerpetualCandles(处理全部元素)
raw 原始 aiohttp WS,按 data[-1] 判断换根(理论最快)
预期:patched ≈ raw,且比 stock 早约 1.1 秒。
同时校验补丁没有破坏数据:两条 feed 的历史 K 线应逐根相等(补丁只影响
最新一根的到达时刻与收盘价更新时机,不该改动已收盘的历史)。
docker run --rm -v $PWD:/repo:ro -v $PWD/research/out:/out \
-w /home/hummingbot -e PYTHONPATH=/home/hummingbot:/repo/research/live \
--entrypoint /opt/conda/envs/hummingbot/bin/python \
hummingbot/hummingbot:latest /repo/research/live/verify_patch.py --minutes 20
"""
from __future__ import annotations
import argparse
import asyncio
import csv
import json
import time
from pathlib import Path
SYMS = ("BTC", "ETH", "SOL")
WSS = "wss://ws.bitget.com/v2/ws/public"
def out_dir() -> Path:
p = Path("/out")
return p if p.is_dir() else Path(__file__).resolve().parents[1] / "out"
async def raw_ws(rec: dict, stop: asyncio.Event) -> None:
import aiohttp
payload = {"op": "subscribe",
"args": [{"instType": "USDT-FUTURES", "channel": "candle1m",
"instId": f"{s}USDT"} for s in SYMS]}
while not stop.is_set():
try:
async with aiohttp.ClientSession() as sess, \
sess.ws_connect(WSS, heartbeat=20) as ws:
await ws.send_str(json.dumps(payload))
last: dict[str, int] = {}
while not stop.is_set():
msg = await ws.receive()
if msg.type is not aiohttp.WSMsgType.TEXT:
break
if msg.data == "pong":
continue
d = json.loads(msg.data)
if "data" not in d or "arg" not in d or not d["data"]:
continue
sym = d["arg"]["instId"].replace("USDT", "")
kts = int(d["data"][-1][0]) # 关键:取末元素
if last.get(sym) is not None and kts > last[sym]:
rec.setdefault((sym, kts), {})["raw"] = \
int(time.time() * 1000)
last[sym] = max(kts, last.get(sym, 0))
except asyncio.CancelledError:
raise
except Exception as e:
print(f" [raw] {type(e).__name__}: {e}", flush=True)
# 正常跳出与异常都要歇一下再重连,避免服务端持续拒绝时打成风暴
if not stop.is_set():
await asyncio.sleep(1)
async def poll_feeds(feeds: dict, key: str, rec: dict,
stop: asyncio.Event) -> None:
last = {s: (int(f._candles[-1][0]) if len(f._candles) else None)
for s, f in feeds.items()}
while not stop.is_set():
for s, f in feeds.items():
if not len(f._candles):
continue
newest = int(f._candles[-1][0])
if last[s] is not None and newest > last[s]:
kts = newest * 1000 if newest < 1e12 else newest
rec.setdefault((s, kts), {})[key] = int(time.time() * 1000)
last[s] = newest
await asyncio.sleep(0.01)
async def main_async(minutes: int) -> None:
from hummingbot.data_feed.candles_feed.bitget_perpetual_candles import (
BitgetPerpetualCandles,
)
from patched_candles import PatchedBitgetPerpetualCandles
stock, patched = {}, {}
for s in SYMS:
stock[s] = BitgetPerpetualCandles(f"{s}-USDT", "1m", 20)
patched[s] = PatchedBitgetPerpetualCandles(f"{s}-USDT", "1m", 20)
for d in (stock, patched):
for f in d.values():
f.start()
print(f"[补丁验证] {SYMS} · 跑 {minutes} 分钟", flush=True)
t0 = time.time()
while time.time() - t0 < 120:
if all(f.ready for d in (stock, patched) for f in d.values()):
break
await asyncio.sleep(0.5)
print(f" 回填完成 {time.time() - t0:.1f}s", flush=True)
rec: dict = {}
stop = asyncio.Event()
tasks = [asyncio.create_task(raw_ws(rec, stop)),
asyncio.create_task(poll_feeds(stock, "stock", rec, stop)),
asyncio.create_task(poll_feeds(patched, "patched", rec, stop))]
deadline = time.time() + minutes * 60
seen = set()
while time.time() < deadline:
await asyncio.sleep(2)
for k, v in rec.items():
if k in seen or not {"stock", "patched", "raw"} <= v.keys():
continue
seen.add(k)
print(f" {k[0]}: patched 比 stock 早 "
f"{v['stock'] - v['patched']}ms · 距 raw "
f"{v['patched'] - v['raw']}ms", flush=True)
stop.set()
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
# 数据一致性:两条 feed 的已收盘历史必须逐根相同
print("\n########## 补丁是否改动了已收盘 K 线 ##########")
for s in SYMS:
a = [list(map(float, r)) for r in list(stock[s]._candles)[:-1]]
b = [list(map(float, r)) for r in list(patched[s]._candles)[:-1]]
n = min(len(a), len(b))
ta = {int(r[0]): r for r in a[-n:]}
tb = {int(r[0]): r for r in b[-n:]}
common = sorted(set(ta) & set(tb))
diff = [t for t in common if ta[t][1:6] != tb[t][1:6]]
print(f" {s}: 共有 {len(common)} 根,OHLCV 不同 {len(diff)}"
+ (f"(示例 ts={diff[:3]}" if diff else ""))
for d in (stock, patched):
for f in d.values():
f.stop()
full = [(s, k, v) for (s, k), v in rec.items()
if {"stock", "patched", "raw"} <= v.keys()]
f = out_dir() / "verify_patch.csv"
with f.open("w", newline="") as fh:
w = csv.writer(fh)
w.writerow(["sym", "kline_ts", "raw", "patched", "stock",
"gain_ms", "patched_minus_raw_ms"])
for s, k, v in sorted(full, key=lambda x: x[1]):
w.writerow([s, k, v["raw"], v["patched"], v["stock"],
v["stock"] - v["patched"], v["patched"] - v["raw"]])
print(f"\n########## 补丁收益(配对 {len(full)} 根)##########")
print(f"{'':<5}{'n':>5}{'早于stock中位':>14}{'距raw中位':>12}")
for s in SYMS:
g = [v for ss, _, v in full if ss == s]
if not g:
print(f" {s}: 无样本")
continue
gain = sorted(v["stock"] - v["patched"] for v in g)
dr = sorted(v["patched"] - v["raw"] for v in g)
print(f"{s:<5}{len(g):>5}{gain[len(gain) // 2]:>14}"
f"{dr[len(dr) // 2]:>12}")
print(f"\n产物写入 {f}")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--minutes", type=int, default=20)
asyncio.run(main_async(ap.parse_args().minutes))
if __name__ == "__main__":
main()