候选池发布给择时决策系统（台账 004）

只写代码列表到共享 Redis 的 pms:candidates 键，择时侧的盘中强势扫描拿它当覆盖面。
走 Redis 不走接口：两侧本来就连着同一台盘中库，不新增网络方向、不涉及鉴权；失败模式
也良性，键没了就是那边本轮不扫，绝不回退全市场。写失败只记日志，不影响本轮扫描。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/services/candidate_pub.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+"""把今天的候选池代码写进共享 Redis, 给择时决策系统的盘中强势扫描当覆盖面 (2026-09-09, 台账 004)。
+
+**只写一份, 只写代码。** 不写分数、不写理由、不写价格 —— 那些是 PMS 的内部判断,
+择时侧要的只是"今天该盯哪几十只"。
+
+为什么走 Redis 不走接口: 两侧本来就连着同一台行情信号 Redis 的盘中库 (PMS 的信号消化
+读它的盘中信号流, 择时侧写那条流), 不新增网络方向、不涉及鉴权; 失败模式也良性 ——
+键没了就是择时侧本轮不扫, 它绝不会因此回退到全市场。反过来让择时侧调 PMS 的计划接口,
+要新造一条"择时到 PMS"的调用方向, 还要处理会话票签。
+
+写失败只记日志, **绝不影响扫描主流程** —— 这个键是给别人用的方便, 不是 PMS 自己的依赖。
+"""
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime
+
+from app.services import signal_service
+
+logger = logging.getLogger("pms.candpub")
+
+KEY = "pms:candidates:{ymd}"
+TTL_SEC = 2 * 24 * 3600
+
+
+def publish(codes) -> bool:
+    """写今天的候选代码列表。返回是否写成功。"""
+    try:
+        arr = sorted({str(c).strip().upper() for c in (codes or []) if c})
+        if not arr:
+            return False
+        key = KEY.format(ymd=datetime.now().strftime("%Y%m%d"))
+        payload = json.dumps({"ymd": datetime.now().strftime("%Y%m%d"), "codes": arr,
+                              "at": datetime.now().isoformat(timespec="seconds"),
+                              "n": len(arr)}, ensure_ascii=False)
+        from config.settings import settings
+        signal_service._client(settings.SIGNAL_REDIS_DB_INTRADAY).set(key, payload, ex=TTL_SEC)
+        return True
+    except Exception as e:
+        logger.warning("[候选池发布] 写共享键失败 (不影响扫描): %s", e)
+        return False

app/services/proposal_service.py

@@ -39,8 +39,8 @@ from app.core import rule_gate
 from app.core import signal_rules as sr
 from app.core import tradedays as td
 from app.repo import pms_repo
-from app.services import (command_service, executor, industry, judge, market, param_store,
-                          reask_service,
+from app.services import (candidate_pub, command_service, executor, industry, judge, market,
+                          param_store, reask_service,
                           plan_feed, portfolio)
 
 logger = logging.getLogger("pms.proposal")
@@ -160,6 +160,14 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict:
             logger.exception("新建仓扫描失败")
             out["errors"].append(f"新建仓扫描失败: {type(e).__name__}: {e}")
     out["open_candidates"] = len(open_cands)
+    # 候选池发布给择时决策系统 (2026-09-09, 台账 004): 它的盘中强势扫描拿这份当覆盖面,
+    # 读不到就本轮不扫、绝不回退全市场。这里只写代码不写别的; 写失败只记日志, 不影响本轮。
+    # 落点选在候选产出之后、分流之前: 发布的是"今天该盯哪几只", 与它们最后买没买无关。
+    if not dry_run and open_cands:
+        try:
+            candidate_pub.publish([x.get("ts_code") for x in open_cands])
+        except Exception as e:
+            logger.warning("[新建仓] 候选池发布失败 (不影响本轮): %s", e)
 
     # 研判的时间预算从这一刻起算。**先跑已有持仓的四类, 再跑新建仓** —— 预算真用尽时,
     # 被推到下一跳的一定是新建仓, 已有持仓的动作行为与 2026-08-06 之前一致。