diff --git a/research/live/deploy/tg.env.example b/research/live/deploy/tg.env.example
index bb63692..9c38f9f 100644
--- a/research/live/deploy/tg.env.example
+++ b/research/live/deploy/tg.env.example
@@ -18,3 +18,21 @@ export TG_LEVERAGE="10"
 # 距「参考价成立」超过这么多秒就标为已失效。参考价是次根开盘价，
 # 过了就不是回测那个成交价了
 export TG_STALE_S="90"
+
+# ── 自动化实盘（live_exec.py）──────────────────────────────────
+# 只读+交易权限，**不要开提币权限**
+export BITGET_API_KEY=""
+export BITGET_API_SECRET=""
+export BITGET_PASSPHRASE=""
+# 名义额与杠杆。理由同 TG_NOTIONAL：抬名义额是为了压步长取整，不是为了赚更多
+export LIVE_NOTIONAL="500"
+export LIVE_LEVERAGE="10"
+# 硬约束。这三条封住的是「代价不随仓位缩小」的那几类故障：
+#   MAX_OPEN     失控下单（单笔小但笔数无界）
+#   MAX_DAY      同上，日维度
+#   MAX_DAY_LOSS 策略真的不行但没人盯着
+export LIVE_MAX_OPEN="3"
+export LIVE_MAX_DAY="15"
+export LIVE_MAX_DAY_LOSS="20"
+# 信号超过这么久就不做。参考成交价是次根开盘价，过期后跑的不是回测那个价
+export LIVE_STALE_S="20"
diff --git a/research/live/live_exec.py b/research/live/live_exec.py
new file mode 100644
index 0000000..9698dfb
--- /dev/null
+++ b/research/live/live_exec.py
@@ -0,0 +1,418 @@
+"""自动化小额实盘执行器。读信号总线，用 Hummingbot 的 PositionExecutor 下单。
+
+## 出场结构为什么能分解成两个半仓
+
+回测的结构是：2 ATR 止损 / 3 ATR 减半 / 8 ATR 目标 / 48 根超时，且**剩余半仓
+的止损保持在入场价的 2 ATR、不移动**。这一条已在 `lib/exit_model.py:151`
+核实——`runner_stops` 的 `ret` 是 `(entry - low[j]) / a`，从入场价算，且
+`RUNNER_STOP == SL == 2.0`。
+
+Hummingbot 的 `TripleBarrierConfig` 只有单级止盈，装不下两级。但因为两个半仓
+共用同一个不动的止损，结构可以**精确分解**：
+
+    半仓 A   市价入场 · TP 3 ATR · SL 2 ATR · 48 分钟超时
+    半仓 B   市价入场 · TP 8 ATR · SL 2 ATR · 48 分钟超时
+
+止损先到则两半都在 -2 ATR 出场；3 ATR 先到则 A 出场、B 继续持有且止损仍在
+2 ATR。与回测逐情形一致。若哪天把 RUNNER_STOP 改成不等于 SL（比如移到成本），
+这个分解就**不再成立**，必须改回单执行器加手工两级——`assert_decomposable()`
+会在启动时挡住这种情况。
+
+## 止损在机器人侧，不在交易所侧
+
+Bitget 连接器只支持 LIMIT / LIMIT_MAKER / MARKET，没有触发单。
+`PositionExecutor.control_stop_loss()` 是在本地控制循环里盯价、触发时才发市价
+单。所以**进程一死，仓位就是裸的**。10x 下强平需逆向 10%（约 100 个 ATR），
+48 分钟内极不可能，单次代价仍封在保证金内；但这一类靠交易所侧的兜底止损才能
+真正消掉，见 `backstop.py`。
+
+## 硬约束才是这个文件的重点
+
+一笔止损只亏约 1 USDT，所以"亏损可控"对单笔是成立的。但有三类故障的代价
+**不随仓位缩小**，必须显式封住：
+
+    失控下单   循环里的 bug 反复开仓，单笔小但笔数无界 → MAX_OPEN / MAX_DAY
+    裸仓       进程在"已入场、止损未挂"之间死掉      → backstop + 重启对账
+    亏损累积   策略真的不行，但没人盯着              → MAX_DAY_LOSS
+
+    python research/live/live_exec.py --dry-run      # 只打印不下单
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+import time
+from decimal import Decimal
+from pathlib import Path
+
+HERE = Path(__file__).resolve()
+sys.path.insert(0, str(HERE.parents[1]))
+sys.path.insert(0, str(HERE.parent))
+
+import signal_bus  # noqa: E402
+
+NOTIONAL = float(os.environ.get("LIVE_NOTIONAL", "500"))
+LEVERAGE = int(os.environ.get("LIVE_LEVERAGE", "10"))
+
+# ── 硬约束 ────────────────────────────────────────────────────────────
+# 并发仓位数。1 笔约占 50 USDT 保证金，3 笔 150 USDT。信号速率 5.3 笔/天、
+# 持仓 48 分钟，期望并发只有 0.18 笔，所以 3 已经很宽——超了说明有 bug
+MAX_OPEN = int(os.environ.get("LIVE_MAX_OPEN", "3"))
+# 日开仓上限。实测 5.3 笔/天，给 3 倍余量。这一条专门封"失控下单"
+MAX_DAY = int(os.environ.get("LIVE_MAX_DAY", "15"))
+# 日亏损上限（USDT）。一笔止损约 1 USDT，15 笔全亏 15 USDT
+MAX_DAY_LOSS = float(os.environ.get("LIVE_MAX_DAY_LOSS", "20"))
+# 信号超过这么久就不做了。参考成交价是次根开盘价，过期后跑的不是回测那个价
+STALE_S = float(os.environ.get("LIVE_STALE_S", "20"))
+
+STATE = Path(os.environ.get("LIVE_STATE", "research/out/live_state.json"))
+TRADES = Path(os.environ.get("LIVE_TRADES", "research/out/live_trades.jsonl"))
+
+SL_ATR, SCALE_ATR, RUNNER_ATR, MAXB = 2.0, 3.0, 8.0, 48
+
+
+def assert_decomposable() -> None:
+    """两个半仓的分解依赖 RUNNER_STOP == SL，不成立就必须停机。
+
+    若有人把剩余半仓的止损改成移到成本（RUNNER_STOP=0）或任何 != SL 的值，
+    这个分解就变成"两半共用同一止损"的错误近似，实盘跑的是另一个收益结构，
+    而且不会报错。所以在启动时硬挡。
+    """
+    from step43_fill_aware_budget import RUNNER_STOP, SL
+    if float(RUNNER_STOP) != float(SL):
+        raise SystemExit(
+            f"⛔ RUNNER_STOP({RUNNER_STOP}) != SL({SL})，两个半仓的分解不再\n"
+            f"   成立。live_exec 的出场结构会与回测不一致且不报错。\n"
+            f"   要改成单执行器 + 手工两级止盈，或把这两个值改回一致。")
+
+
+class Guard:
+    """硬约束与当日计数。状态落盘，重启后不清零。
+
+    不落盘的话，进程反复重启就等于反复重置日上限——"失控下单"这一类恰好常常
+    伴随反复重启，那时上限必须还记得。
+    """
+
+    def __init__(self, path: Path = STATE):
+        self.path = path
+        self.day = time.strftime("%Y-%m-%d")
+        self.n_day = 0
+        self.pnl_day = 0.0
+        self.done: set = set()
+        self._load()
+
+    def _load(self) -> None:
+        try:
+            d = json.loads(self.path.read_text())
+        except Exception:
+            return
+        # 跨日则计数归零，但已处理过的信号键要保留，否则会重开旧仓
+        if d.get("day") == self.day:
+            self.n_day = int(d.get("n_day", 0))
+            self.pnl_day = float(d.get("pnl_day", 0.0))
+        self.done = set(d.get("done", []))
+
+    def save(self) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = self.path.with_suffix(".tmp")
+        # 原子替换：直接覆写时若在写一半崩溃，状态文件会变成半个 JSON，
+        # 重启后读不出来 → 日计数归零 → 上限失效
+        tmp.write_text(json.dumps({
+            "day": self.day, "n_day": self.n_day, "pnl_day": self.pnl_day,
+            # 只留最近的，否则文件无界增长
+            "done": sorted(self.done)[-5000:]}))
+        tmp.replace(self.path)
+
+    def roll(self) -> None:
+        today = time.strftime("%Y-%m-%d")
+        if today != self.day:
+            print(f"  [guard] 跨日 {self.day} → {today}，"
+                  f"当日 {self.n_day} 笔 / PnL {self.pnl_day:+.2f} USDT",
+                  flush=True)
+            self.day, self.n_day, self.pnl_day = today, 0, 0.0
+            self.save()
+
+    def blocks(self, key: str, n_open: int) -> str | None:
+        """返回拒绝原因，None 表示放行。"""
+        self.roll()
+        if key in self.done:
+            return "已处理过（幂等）"
+        if n_open >= MAX_OPEN:
+            return f"并发仓位已达上限 {MAX_OPEN}"
+        if self.n_day >= MAX_DAY:
+            return f"当日开仓已达上限 {MAX_DAY}"
+        if self.pnl_day <= -MAX_DAY_LOSS:
+            return (f"当日亏损 {self.pnl_day:.2f} 已达上限 "
+                    f"-{MAX_DAY_LOSS}，停止开新仓")
+        return None
+
+    def took(self, key: str) -> None:
+        self.done.add(key)
+        self.n_day += 1
+        self.save()
+
+    def realized(self, pnl: float) -> None:
+        self.pnl_day += pnl
+        self.save()
+
+
+def legs(entry: float, atr_pct: float, direction: int) -> list[dict]:
+    """两个半仓的三重门参数。
+
+    止盈/止损用**比例**表达（Hummingbot 的 TripleBarrierConfig 就是比例），
+    所以直接用 ATR 的相对值，不必换成绝对价位。
+    """
+    a = atr_pct
+    return [
+        {"tag": "scale", "tp": SCALE_ATR * a, "sl": SL_ATR * a},
+        {"tag": "runner", "tp": RUNNER_ATR * a, "sl": SL_ATR * a},
+    ]
+
+
+def log_trade(rec: dict, path: Path = TRADES) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a") as f:
+        f.write(json.dumps(rec) + "\n")
+        f.flush()
+        os.fsync(f.fileno())
+
+
+class Exec:
+    def __init__(self, dry: bool, bus: Path):
+        self.dry = dry
+        self.bus = bus
+        self.guard = Guard()
+        self.conn = None
+        self.execs: dict = {}      # key → [PositionExecutor, ...]
+        self.offset = 0            # 已读到总线的哪一行
+        self.n_seen = self.n_took = self.n_skip = 0
+
+    # ── 启动 ──────────────────────────────────────────────────────
+    async def start(self) -> None:
+        assert_decomposable()
+        # 只从"现在"往后做。历史信号的参考成交价早就过期了，补做等于随机入场
+        self.offset = sum(1 for _ in signal_bus.read_all(self.bus))
+        print(f"  总线已有 {self.offset} 条历史信号，全部跳过（参考价已过期）",
+              flush=True)
+        if self.dry:
+            print("  ⚠ 空跑模式：不连交易所、不下真单", flush=True)
+            return
+
+        key = os.environ.get("BITGET_API_KEY", "")
+        sec = os.environ.get("BITGET_API_SECRET", "")
+        pas = os.environ.get("BITGET_PASSPHRASE", "")
+        if not (key and sec and pas):
+            raise SystemExit("⛔ 缺 BITGET_API_KEY / SECRET / PASSPHRASE。"
+                             "先跑 --dry-run 验链路。")
+        from hummingbot.connector.derivative.bitget_perpetual.bitget_perpetual_derivative import (  # noqa: E501
+            BitgetPerpetualDerivative,
+        )
+        syms = os.environ.get(
+            "SYMS", "BTC,ETH,SOL,BNB,XRP,DOGE,ADA,AVAX,LINK,LTC").split(",")
+        self.conn = BitgetPerpetualDerivative(
+            bitget_perpetual_api_key=key, bitget_perpetual_secret_key=sec,
+            bitget_perpetual_passphrase=pas,
+            trading_pairs=[f"{s}-USDT" for s in syms], trading_required=True)
+        await self.conn.start_network()
+        for _ in range(60):
+            await asyncio.sleep(1)
+            if self.conn.ready:
+                break
+        await self.reconcile()
+
+    async def reconcile(self) -> None:
+        """启动时把交易所的实际持仓对上。
+
+        进程崩溃重启后交易所可能还有仓位，而我们的 executor 全没了——那些仓位
+        没有任何止损在盯。**必须平掉而不是接管**：接管要重建入场价、ATR、剩余
+        半仓状态和已过的根数，任一项猜错都会让出场结构变成另一个东西，而平掉
+        的代价只是一笔小额亏损，且行为确定。
+        """
+        try:
+            pos = list(self.conn.account_positions.values())
+        except Exception as e:
+            print(f"  ⚠ 对账读持仓失败 {type(e).__name__}: {e}", flush=True)
+            return
+        pos = [p for p in pos if abs(float(p.amount)) > 0]
+        if not pos:
+            print("  对账：交易所无持仓，干净启动", flush=True)
+            return
+        print(f"  ⚠ 对账：发现 {len(pos)} 个遗留持仓，全部市价平掉", flush=True)
+        for p in pos:
+            print(f"     {p.trading_pair} {p.position_side} "
+                  f"{p.amount} @ {p.entry_price}", flush=True)
+            try:
+                await self.flatten(p)
+            except Exception as e:
+                print(f"     ⛔ 平仓失败 {type(e).__name__}: {e}，"
+                      f"需要人工介入", flush=True)
+
+    async def flatten(self, p) -> None:
+        from hummingbot.core.data_type.common import OrderType, PositionAction
+        amt = abs(Decimal(str(p.amount)))
+        is_long = float(p.amount) > 0
+        fn = self.conn.sell if is_long else self.conn.buy
+        fn(trading_pair=p.trading_pair, amount=amt,
+           order_type=OrderType.MARKET, price=Decimal("NaN"),
+           position_action=PositionAction.CLOSE)
+        log_trade({"ev": "reconcile_flatten", "pair": p.trading_pair,
+                   "amount": float(p.amount), "ts": int(time.time() * 1000)})
+
+    # ── 主循环 ────────────────────────────────────────────────────
+    async def poll(self) -> None:
+        while True:
+            try:
+                await self.step()
+            except Exception as e:
+                import traceback
+                print(f"  ⛔ 主循环异常 {type(e).__name__}: {e}", flush=True)
+                traceback.print_exc()
+            await asyncio.sleep(0.2)
+
+    async def step(self) -> None:
+        recs = list(signal_bus.read_all(self.bus))
+        if len(recs) <= self.offset:
+            return
+        new, self.offset = recs[self.offset:], len(recs)
+        for r in new:
+            self.n_seen += 1
+            await self.on_signal(r)
+
+    async def on_signal(self, r: dict) -> None:
+        age = time.time() - r["emit_ms"] / 1000.0
+        n_open = sum(1 for v in self.execs.values() if v)
+        why = self.guard.blocks(r["key"], n_open)
+        if why is None and age > STALE_S:
+            why = f"信号已过期 {age:.1f}s > {STALE_S:.0f}s"
+        if why:
+            self.n_skip += 1
+            print(f"  ⊘ {r['key']} 跳过：{why}", flush=True)
+            log_trade({"ev": "skip", "key": r["key"], "why": why,
+                       "age_s": round(age, 2)})
+            return
+
+        self.guard.took(r["key"])
+        self.n_took += 1
+        lg = legs(r["entry_px"], r["atr_pct"], r["direction"])
+        side = "LONG" if r["direction"] > 0 else "SHORT"
+        print(f"  ▶ {r['key']} {side} 名义 {NOTIONAL:.0f} {LEVERAGE}x "
+              f"· 延后 {age:.1f}s · ATR {r['atr_pct'] * 1e4:.1f}bp",
+              flush=True)
+        for x in lg:
+            print(f"      {x['tag']:<7}TP {x['tp'] * 1e4:6.1f}bp "
+                  f"SL {x['sl'] * 1e4:6.1f}bp 超时 {MAXB}min", flush=True)
+        log_trade({"ev": "entry", "key": r["key"], "side": side,
+                   "entry_px": r["entry_px"], "atr_pct": r["atr_pct"],
+                   "notional": NOTIONAL, "leverage": LEVERAGE,
+                   "age_s": round(age, 2), "legs": lg, "dry": self.dry})
+        if self.dry:
+            return
+        await self.open_position(r, lg)
+
+    async def open_position(self, r: dict, lg: list[dict]) -> None:
+        from hummingbot.core.data_type.common import OrderType, TradeType
+        from hummingbot.strategy_v2.executors.position_executor.data_types import (  # noqa: E501
+            PositionExecutorConfig, TripleBarrierConfig,
+        )
+        from hummingbot.strategy_v2.executors.position_executor.position_executor import (  # noqa: E501
+            PositionExecutor,
+        )
+
+        pair = f"{r['sym']}-USDT"
+        rule = self.conn.trading_rules.get(pair)
+        step = Decimal(str(rule.min_base_amount_increment)) if rule \
+            else Decimal("0")
+        px = Decimal(str(r["entry_px"]))
+        # 取到步长的偶数倍，两个半仓才各是精确一半。不这么做 SOL 的半仓会是
+        # 全仓的 43%（步长 0.1 币 ≈ 10.7 USDT），而回测假设 50/50
+        if step > 0:
+            grid = step * 2
+            n = max(Decimal("1"), (Decimal(str(NOTIONAL)) / px / grid)
+                    .quantize(Decimal("1")))
+            qty = n * grid
+        else:
+            qty = Decimal(str(NOTIONAL)) / px
+        half = qty / 2
+        side = TradeType.BUY if r["direction"] > 0 else TradeType.SELL
+
+        made = []
+        for x in lg:
+            cfg = PositionExecutorConfig(
+                id=f"{r['key']}:{x['tag']}", controller_id="chanlun_1m",
+                timestamp=time.time(), trading_pair=pair,
+                connector_name="bitget_perpetual", side=side,
+                amount=half, leverage=LEVERAGE,
+                triple_barrier_config=TripleBarrierConfig(
+                    stop_loss=Decimal(str(x["sl"])),
+                    take_profit=Decimal(str(x["tp"])),
+                    time_limit=MAXB * 60,
+                    open_order_type=OrderType.MARKET,
+                    # 止盈挂限价才是 maker，这是预算模型的一部分：
+                    # 止盈那 60% 不吃滑点、按 maker 费率计
+                    take_profit_order_type=OrderType.LIMIT_MAKER,
+                    stop_loss_order_type=OrderType.MARKET,
+                    time_limit_order_type=OrderType.MARKET))
+            ex = PositionExecutor(strategy=None, config=cfg)
+            ex.start()
+            made.append(ex)
+        self.execs[r["key"]] = made
+        print(f"      已起 {len(made)} 个 executor，各 {half} 币", flush=True)
+
+    async def sweep(self) -> None:
+        """收掉已结束的 executor，把已实现盈亏计入当日上限。"""
+        while True:
+            await asyncio.sleep(5)
+            for key, lst in list(self.execs.items()):
+                alive = [e for e in lst if not e.is_closed]
+                for e in lst:
+                    if e.is_closed and getattr(e, "_counted", False) is False:
+                        e._counted = True
+                        pnl = float(getattr(e, "net_pnl_quote", 0.0) or 0.0)
+                        self.guard.realized(pnl)
+                        log_trade({"ev": "close", "key": key,
+                                   "id": e.config.id,
+                                   "close_type": str(getattr(
+                                       e, "close_type", "")),
+                                   "pnl_quote": pnl,
+                                   "pnl_day": self.guard.pnl_day})
+                        print(f"  ◀ {e.config.id} 平仓 "
+                              f"{getattr(e, 'close_type', '')} "
+                              f"PnL {pnl:+.3f} · 当日 "
+                              f"{self.guard.pnl_day:+.2f}", flush=True)
+                if alive:
+                    self.execs[key] = alive
+                else:
+                    self.execs.pop(key, None)
+
+    async def heartbeat(self) -> None:
+        while True:
+            await asyncio.sleep(300)
+            n_open = sum(1 for v in self.execs.values() if v)
+            print(f"  [心跳] 见信号 {self.n_seen} · 已做 {self.n_took} · "
+                  f"跳过 {self.n_skip} · 在场 {n_open}/{MAX_OPEN} · "
+                  f"当日 {self.guard.n_day}/{MAX_DAY} 笔 · "
+                  f"当日 PnL {self.guard.pnl_day:+.2f}/-{MAX_DAY_LOSS}",
+                  flush=True)
+
+    async def run(self) -> None:
+        await self.start()
+        await asyncio.gather(self.poll(), self.sweep(), self.heartbeat())
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--dry-run", action="store_true",
+                    help="不连交易所、不下单，只验总线与约束逻辑")
+    ap.add_argument("--bus", default=str(signal_bus.BUS))
+    a = ap.parse_args()
+
+    print(f"实盘执行器 · 名义 {NOTIONAL:.0f} USDT · {LEVERAGE}x · "
+          f"并发≤{MAX_OPEN} · 日开仓≤{MAX_DAY} · 日亏损≤{MAX_DAY_LOSS}")
+    asyncio.run(Exec(a.dry_run, Path(a.bus)).run())
+
+
+if __name__ == "__main__":
+    main()
diff --git a/research/live/shadow_hb.py b/research/live/shadow_hb.py
index 122c13e..55e9cd7 100644
--- a/research/live/shadow_hb.py
+++ b/research/live/shadow_hb.py
@@ -71,6 +71,7 @@ import pandas as pd
 
 from lib.shadow_budget import LAG_ALARM_MS, LAG_WINDOW, lag_healthy
 
+import signal_bus
 import tg_notify
 
 # 站点标识。跨地对比时两台机器的 CSV 要能合起来读，没有这一列就分不清哪行
@@ -655,6 +656,11 @@ class Shadow:
                     print(f"  [TG] {sym} lag 退化，按停开新仓规则不推",
                           flush=True)
                 else:
+                    # 总线先写、推送后发。写盘是同步的且已 fsync，实盘据此
+                    # 下单；推送要走网络，不能让它的延迟挡在下单前面
+                    signal_bus.emit(sym, kline_ts, h["direction"],
+                                    float(baseline), float(atr_pct),
+                                    t_data - kline_ts)
                     self._spawn(
                         tg_notify.push_signal(
                             sym, h["direction"], float(baseline),
diff --git a/research/live/signal_bus.py b/research/live/signal_bus.py
new file mode 100644
index 0000000..1ec3150
--- /dev/null
+++ b/research/live/signal_bus.py
@@ -0,0 +1,75 @@
+"""影子把过滤网的信号写到这里，实盘执行器读这里。
+
+## 为什么不让实盘自己算信号
+
+三个理由，第三个最要紧：
+
+1. 2 核上再来一份十币计算，清空会从 247ms 推到 500ms+
+2. 实盘进程崩溃不该影响正在采的数据集
+3. **实盘交易的必须是影子测量的那一个信号。** 各算一份会让两边悄悄分叉，
+   之后就没法把实盘的实际成交和影子测的滑点曲线对照——而那个对照是整件事
+   的目的
+
+## 为什么用 append-only 文件而不是队列
+
+崩溃安全 + 留审计轨迹。实盘进程重启后能从文件里看到自己漏掉了哪些信号，
+而不是像内存队列那样直接消失。文件也让"影子在跑、实盘没在跑"这种状态成为
+可观测的（信号在攒着），而不是静默丢弃。
+
+每行一个 JSON，字段见 `emit`。`key` 是幂等键，实盘按它去重——同一根被重复
+处理（补根、进程池重建后重放）不该开两次仓。
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+
+BUS = Path(os.environ.get(
+    "SIGNAL_BUS", "research/out/signals_live.jsonl"))
+
+
+def key_of(sym: str, kline_ts: int, direction: int) -> str:
+    return f"{sym}:{int(kline_ts)}:{int(direction):+d}"
+
+
+def emit(sym: str, kline_ts: int, direction: int, entry_px: float,
+         atr_pct: float, lag_ms: float, path: Path | None = None) -> None:
+    """追写一条信号。任何失败只打日志——总线写不进去不能连坐采集。
+
+    `entry_px` 是次根开盘价，也就是回测口径的成交价。实盘据此算止损/止盈的
+    绝对价位，**不要**用实盘自己看到的现价，否则价位会随执行延迟漂移，跑的
+    就不是回测那个结构。
+    """
+    p = path or BUS
+    rec = {"key": key_of(sym, kline_ts, direction),
+           "sym": sym, "kline_ts": int(kline_ts),
+           "direction": int(direction),
+           "entry_px": float(entry_px), "atr_pct": float(atr_pct),
+           "lag_ms": float(lag_ms),
+           "emit_ms": int(time.time() * 1000)}
+    try:
+        p.parent.mkdir(parents=True, exist_ok=True)
+        with p.open("a") as f:
+            f.write(json.dumps(rec) + "\n")
+            f.flush()
+            # 实盘要在毫秒级看到，且进程被 SIGKILL 时不能丢——这两点都要求
+            # 落到磁盘，不能只停在 libc 缓冲里
+            os.fsync(f.fileno())
+    except Exception as e:
+        print(f"  [bus] 写信号失败 {type(e).__name__}: {e}", flush=True)
+
+
+def read_all(path: Path | None = None):
+    """读全部信号。坏行跳过——半行只可能出现在文件末尾的崩溃点。"""
+    p = path or BUS
+    if not p.exists():
+        return
+    for line in p.read_text().splitlines():
+        if not line.strip():
+            continue
+        try:
+            yield json.loads(line)
+        except json.JSONDecodeError:
+            continue