把 compute_ms 拆成排队与纯计算，据此否掉换机器这个方向

compute_ms 一直是「提交进程池到拿到结果」的墙钟时间，排队和纯计算混在一个
数里，所以「加核有没有用」只能靠猜——这也是原先打算在 AWS 开第二台比 CPU
的依据。

worker 内部自己计时，连同父进程传入的提交时刻一起回传，拆出 queue_ms 与
inner_ms。实测中位 3ms / 695ms：2 个 worker 跑 3 个币并不排队，因为三个币的
收盘消息错峰到达。瓶颈全在单线程，加核压不到。

顺带把 README 里的内存数据从臆测的 1.5GB 改成实测 410MiB，并注明跨站点比
CPU 收益有限。

Co-authored-by: Cursor <cursoragent@cursor.com>

research/live/deploy/README.md

@@ -117,10 +117,33 @@ python research/live/compare_sites.py \
 同理，`lag_data_ms` 两站也应当几乎相同（网络那段只有 2ms 空间）。真正该出现
 差异的是 `compute_ms`。如果 `lag_data_ms` 差很多，先查时钟——比查网络更可能。
 
+⚠ 但先读下面「资源占用」一节：`compute_ms` 的差异几乎全部来自单核性能，而
+现役机型之间单核差距很小。**跨站点比 CPU 这件事本身收益有限**，本节流程保留
+是为了比网络与时钟，不建议为了比 CPU 单独开机器。
+
 ## 资源占用
 
-本机实测：内存约 1.5GB（两个计算进程 + 盘口缓冲），CPU 单核不满。
-盘口与成交流落盘约 15MB/天（gzip）。一周 168 小时的量级在百 MB 内。
+本机实测（2 vCPU EPYC 9K65 / 3 个币 / 2 worker）：内存 **410MiB**，CPU 均值
+1~3%。盘口与成交流落盘约 15MB/天（gzip），一周在百 MB 内。
 
-`--workers 2` 是因为信号计算走独立进程池、不能阻塞事件循环。核数少的机型
-可以给 1，但要看心跳里的 `compute_ms`：若接近 60 秒就会开始堆积。
+内存和平均 CPU 都不是约束。约束是**单根 K 线的计算延迟**，而它是纯单线程的：
+
+```
+compute_ms  中位 764ms   父进程测的墙钟，含排队
+  queue_ms  中位   3ms   等空闲 worker
+  inner_ms  中位 695ms   进程内真正在算
+```
+
+`queue_ms` 只有 3ms，说明 **2 个 worker 跑 3 个币并不排队**——三个币的收盘消息
+错峰到达（SOL 最晚，排 66ms），没有真正的并发争抢。
+
+**这条结论直接否掉了「换更强机器」这个方向。** 加核只能压 queue_ms，而它已经
+是 3ms；695ms 全在单线程里，取决于单核性能。t3a.medium（Zen 1，2017）单核比
+本机 Zen 5 慢 1.8~2 倍，换过去 compute_ms 会涨到 1200ms 以上。c7a / c6a 这类
+现代机型单核与本机相当，也换不到东西。
+
+要压这 695ms 只有算法一条路：现在每分钟把 2001 根从头算一遍，其中 2000 根的
+结构与上一分钟完全相同。
+
+`--workers 2` 是因为信号计算走独立进程池、不能阻塞事件循环。币数超过 worker
+数才会看到 queue_ms 上来；届时加 worker 有效，加到与币数相等即可。

research/live/shadow_hb.py

@@ -301,6 +301,9 @@ class Shadow:
         self._tasks: set = set()
         self.n_broken = 0
         self._hb_last_bars = 0
+        # 排队 / 纯计算的滚动窗口，用来判断加核有没有用
+        self.q_hist: deque = deque(maxlen=90)
+        self.i_hist: deque = deque(maxlen=90)
         # 成交监听：已挂上的币，以及必须持有的 forwarder 强引用
         # （PubSub 只存弱引用，不持有的话监听会被 GC 静默摘掉）
         self._hooked: set[str] = set()
@@ -326,7 +329,10 @@ class Shadow:
             "slip_bp", "drift_bp", "spread_bp", "impact_bp"])
         self.f_lat, self.w_lat = _writer(d / "shadow_latency.csv", [
             "site", "sym", "kline_ts", "t_close_ms", "t_data_ms", "t_signal_ms",
-            "lag_data_ms", "lag_signal_ms", "compute_ms", "n_bars", "n_hits",
+            "lag_data_ms", "lag_signal_ms",
+            # compute_ms 含排队；queue_ms/inner_ms 把它拆开，用来判断加核有没有用
+            "compute_ms", "queue_ms", "inner_ms",
+            "n_bars", "n_hits",
             "n_pass", "atr_bp", "lag_med_ms", "lag_ok"])
         # 无条件漂移：每根都记，用来和信号根上的条件漂移对照
         self.f_drf, self.w_drf = _writer(d / "shadow_drift.csv", [
@@ -528,7 +534,7 @@ class Shadow:
 
         t0 = time.perf_counter()
         payload = (df_l[NUM_COLS].values.tolist(),
-                   df_h[NUM_COLS].values.tolist(), baseline)
+                   df_h[NUM_COLS].values.tolist(), baseline, time.time())
         loop = asyncio.get_running_loop()
         from shadow_signal import compute_packed
         try:
@@ -540,6 +546,10 @@ class Shadow:
             return
         compute_ms = int((time.perf_counter() - t0) * 1000)
         t_signal = int(time.time() * 1000)
+        if res.get("queue_ms") is not None:
+            self.q_hist.append(res["queue_ms"])
+        if res.get("inner_ms") is not None:
+            self.i_hist.append(res["inner_ms"])
 
         hits = res.get("hits", [])
         atr_pct = res.get("atr_pct")
@@ -552,7 +562,9 @@ class Shadow:
             "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),
+            "compute_ms": compute_ms,
+            "queue_ms": res.get("queue_ms"), "inner_ms": res.get("inner_ms"),
+            "n_bars": res.get("n_bars", 0),
             "n_hits": len(hits), "n_pass": n_pass, "atr_bp": atr_bp,
             "lag_med_ms": lag_med, "lag_ok": int(lag_ok)})
         self.f_lat.flush()
@@ -744,6 +756,14 @@ class Shadow:
                   f" · 盘口落盘 {self.blog.n} 份"
                   f" · 成交 {self.tape.n_trades} 笔{'' if self.tape.n_trades else ' ⚠监听未生效'}",
                   flush=True)
+            if self.q_hist and self.i_hist:
+                q, i = float(np.median(self.q_hist)), float(np.median(self.i_hist))
+                verdict = ("排队为主 → 加 worker/加核直接见效"
+                           if q > i else
+                           "纯计算为主 → 加核帮不上，需改增量计算")
+                print(f"  [计算] 排队中位 {q:.0f}ms · 纯计算中位 {i:.0f}ms"
+                      f" · {verdict}（worker {self.workers} 个 / 币 {len(SYMS)} 个）",
+                      flush=True)
             # 五分钟一根都没进来，说明管道断了。不喊一声就只能靠人翻日志
             if self.n_bars == self._hb_last_bars:
                 print(f"  ⚠ [停滞] 距上次心跳未处理任何 K 线"

research/live/shadow_signal.py

@@ -31,6 +31,7 @@ Hummingbot 的 asyncio 循环里会把行情处理一起卡住，所以必须隔
 from __future__ import annotations
 
 import os
+import time
 import warnings
 
 warnings.filterwarnings("ignore")
@@ -153,8 +154,27 @@ def _rebuild(rows) -> "object":
 
 
 def compute_packed(payload: tuple) -> dict:
-    """ProcessPoolExecutor 的入口：收 (l_rows, h_rows, entry_px)。"""
-    l_rows, h_rows, entry_px = payload
+    """ProcessPoolExecutor 的入口：收 (l_rows, h_rows, entry_px[, t_submit])。
+
+    返回里带上 `queue_ms` 与 `inner_ms`，把父进程看到的墙钟时间拆开：
+
+        compute_ms（父进程测）= queue_ms + 反序列化 + inner_ms + 回传
+
+    这个拆分决定「加核有没有用」。排队占大头说明 worker 数不够（币数多于
+    worker 数时，同一秒收盘的币只能排队），加核直接见效；纯计算占大头说明
+    单核性能受限，加核帮不上，得从算法上改成增量更新。
+    两者混在一个数里就只能靠猜。
+    """
+    t_start = time.time()
+    l_rows, h_rows, entry_px, *rest = payload
+    t_submit = rest[0] if rest else None
+
+    t0 = time.perf_counter()
     df_l = _rebuild(l_rows)
     df_h = _rebuild(h_rows) if h_rows else None
-    return compute(df_l, df_h, entry_px)
+    out = compute(df_l, df_h, entry_px)
+    out["inner_ms"] = int((time.perf_counter() - t0) * 1000)
+    # 同一台机器，父子进程时钟一致，可直接相减
+    out["queue_ms"] = int((t_start - t_submit) * 1000) \
+        if t_submit is not None else None
+    return out