on_signal 校验字段，畸形记录只丢一条而不是停掉交易

on_signal 直接取 r["emit_ms"]，总线上一条字段不全的记录就会抛 KeyError
打死 poll 任务，进而整个执行器停止交易。一行坏数据换全面停摆，代价不对等。
搬运侧已挡半行，但挡不住字段级的不全。

现在缺 key/sym/emit_ms/direction/entry_px/atr_pct 任一项即丢弃该条，记日志
并推一条 Telegram（这类事应当可见，否则只是少做几笔，统计上看不出来）。

实测：两条字段不全 + 一条非法 JSON + 一条正常，前三条分别被丢弃/被
read_all 吞掉，正常那条照常下单，进程存活。

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

live/live_exec.py

@@ -341,7 +341,21 @@ class Exec:
             self.n_seen += 1
             await self.on_signal(r)
 
+    # 一条记录要能下单，这些字段一个都不能缺
+    NEEDED = ("key", "sym", "emit_ms", "direction", "entry_px", "atr_pct")
+
     async def on_signal(self, r: dict) -> None:
+        # 先校验再取值。缺了这一步，总线上一条字段不全的记录会抛 KeyError，
+        # 把 poll 任务打死，进而**整个执行器停止交易**——一行坏数据换全面停摆，
+        # 代价完全不对等。搬运侧已挡半行，但挡不住字段级的不全
+        miss = [k for k in self.NEEDED if r.get(k) is None]
+        if miss:
+            self.n_skip += 1
+            print(f"  ⊘ 丢弃畸形记录，缺字段 {miss}：{str(r)[:200]}", flush=True)
+            log_trade({"ev": "malformed", "missing": miss, "rec": str(r)[:500]})
+            await tg.error("总线上有畸形记录，已丢弃",
+                           f"缺字段 {miss}\n{str(r)[:300]}")
+            return
         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)