规则闸: 决策系统昨夜定性形态恶化时不新增买入, 拒绝落评审账本

app/core/rule_gate.py

@@ -22,6 +22,20 @@ from app.core.planner import check_all_caps
 
 BUY, SELL = "buy", "sell"
 
+# 决策系统昨夜结论的定性里算"形态恶化"的三个取值 (2026-08-05 补的第二道防线)。
+#
+# 这三个值一共有三处在用, 改一处必须三处一起改:
+#   桥 akg-factor-bridge/pool.py 的 BAD_SIGNALS —— 拿它把票移出候选池、记进回收站;
+#   决策系统 app/services/pms_advisor.py 的 BAD_Y_SIGNALS —— 拿它在择时应答里拦住买入;
+#   这里 —— 拿它在指令下发前终检时再拦一次, 并且把拒绝写进 pms_action_ledger。
+#
+# 为什么明明决策系统那侧已经拦了、这里还要再拦一道: 那一侧拦掉的东西只写在它自己的应答里,
+# PMS 的评审账本 (make t-gate) 一个字看不到; 而评审账本是"拒了的后来涨了多少"那份判分的
+# 事实源, 拒绝不落在这张表里就等于没发生过。要注意这道闸的数据只来自实现A的应答
+# (含它十个交易分钟的缓存): 退实现B时 PMS 手里根本没有决策系统的定性, 这道闸自然不生效 ——
+# 那是"拿不到意见"的正常降级, 不是漏拦。
+BAD_Y_SIGNALS = ("SELL", "AVOID", "DROPPED")
+
 
 def _num(v, default=0.0):
     try:
@@ -38,7 +52,8 @@ def check(*, side: str, action: str, qty: int, price: float, ctx: dict) -> dict:
       caps          check_all_caps 所需组合上下文 (仅买入用)
       day           {price, vwap, ma5, day_chg_from_open, halted, limit_up, limit_down}
       params        {no_chase_ma5, buy_halt_dayup, sector_source_ready}
-      flags         {buy_halt, exec_halt, brake_active, blacklisted, is_command}
+      flags         {buy_halt, exec_halt, brake_active, blacklisted, is_command, y_signal}
+                    y_signal 是决策系统昨夜结论对该股的定性, 由择时应答带回 (可能为 None)
     """
     pos = ctx.get("position") or {}
     day = ctx.get("day") or {}
@@ -102,6 +117,12 @@ def _check_buy(failed, warns, qty, price, ctx, pos, day, prm, flg, is_cmd, hard)
         failed.append("BUY_HALT: 全局暂停买入生效中")
     if flg.get("blacklisted"):
         failed.append("BLACKLIST: 该股在黑名单 (永不买入)")
+    # 决策系统昨夜定性为形态恶化 —— 无论有没有拿到定性, 都写进 hard_numbers, 这样
+    # 评审账本上"放行的那些当时定性是什么"同样查得到, 而不是只有被拒的才留下痕迹。
+    y_sig = str(flg.get("y_signal") or "").strip().upper()
+    hard["y_signal"] = y_sig or None
+    if y_sig in BAD_Y_SIGNALS:
+        failed.append(f"BAD_SIGNAL: 决策系统昨夜定性 {y_sig} (形态恶化), 不新增买入")
     if flg.get("brake_active"):
         if is_cmd:
             warns.append("BRAKE_ACTIVE: 组合刹车中 —— 命令驱动不受限, 仅提示")

app/services/exec_advisor.py

@@ -105,6 +105,9 @@ def decide(*, side: str, action: str, ts_code: str, now, day: dict, params: dict
         d = et.apply_advice(side=side, day=day, params=params, advice=advice, left=left)
         if d is not None:
             d["source"] = advice.get("_source") or "A"
+            # 把昨夜定性顺着决策一起带回去, 规则闸终检要用 (见 rule_gate.BAD_Y_SIGNALS)。
+            # 缓存命中的那条 advice 里也存着它, 所以十个交易分钟的缓存期内一样有值。
+            d["y_signal"] = advice.get("y_signal")
             return d
         note = f"研判动作无法识别: {advice.get('verdict')!r}"
 
@@ -179,6 +182,8 @@ def _advice(*, ts_code, side, action, now, day, pos, left, is_last_day, tdays_le
            "limit_price": data.get("limit_price"),
            "reason": str(data.get("reason") or "")[:200],
            "confidence": data.get("confidence"),
+           # 应答 observed 里的昨夜定性: 存进缓存, 让规则闸在缓存期内也拿得到
+           "y_signal": (data.get("observed") or {}).get("y_signal"),
            "valid_until_min": et.add_trade_minutes(now_min, valid_min),
            "consulted_at": et._fmt(now_min)}
     if prog is not None:

app/services/executor.py

@@ -198,7 +198,9 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict:
                      "flags": {"buy_halt": prm["buy_halt"], "exec_halt": prm["exec_halt"],
                                "brake_active": brake_active,
                                "blacklisted": bool(stock_params.get(code, {}).get("black")),
-                               "is_command": bool(prog.get("is_command"))}})
+                               "is_command": bool(prog.get("is_command")),
+                               # 决策系统昨夜对该股的定性, 由择时应答带回 (退实现B 时为 None)
+                               "y_signal": d.get("y_signal")}})
 
             if not gate["passed"]:
                 out["rejected"].append({"instruction_id": ins["instruction_id"], "code": code,

scripts/test_batch3_units.py

@@ -209,6 +209,28 @@ def _():
     assert any(x.startswith("LOT_INVALID") for x in r["failed"]), r
 
 
+@case("规则闸·昨夜定性形态恶化 → 买入拦截, 卖出不受影响, 定性一律留痕")
+def _():
+    # 2026-08-05 补这道闸的实机由来: 600841 昨夜定性 SELL、已被候选层请出池子, 而在途的
+    # 建仓指令照样在等买入区间 —— 候选层的结论当时对执行层完全不可见。
+    for sig in ("SELL", "AVOID", "DROPPED", " sell "):
+        r = rg.check(side="buy", action="OPEN", qty=1000, price=10.0,
+                     ctx=ctx(flags={"y_signal": sig}))
+        assert any(x.startswith("BAD_SIGNAL") for x in r["failed"]), (sig, r)
+        assert r["hard_numbers"]["y_signal"] == sig.strip().upper(), (sig, r)
+    # 正常定性不能误伤, 但定性本身照样要写进 hard_numbers (放行的也要查得到)
+    ok = rg.check(side="buy", action="OPEN", qty=1000, price=10.0,
+                  ctx=ctx(flags={"y_signal": "BUY"}))
+    assert ok["passed"] and ok["hard_numbers"]["y_signal"] == "BUY", ok
+    # 退实现B 时根本拿不到定性: 不拦, 留痕为 None (缺数据不算恶化, 与桥侧同口径)
+    nil = rg.check(side="buy", action="OPEN", qty=1000, price=10.0, ctx=ctx())
+    assert nil["passed"] and nil["hard_numbers"]["y_signal"] is None, nil
+    # 恶化只挡增持, 减持任何时候放行
+    sell = rg.check(side="sell", action="EXIT", qty=6000, price=10.0,
+                    ctx=ctx(flags={"y_signal": "SELL"}))
+    assert sell["passed"], sell
+
+
 @case("规则闸·刹车对自主是拦截, 对命令只是提示 (命令至上)")
 def _():
     auto = rg.check(side="buy", action="ADD", qty=1000, price=10.0,