From 11fbc36119e92eb6cf6f4a53d5507e49b192efbe Mon Sep 17 00:00:00 2001 From: zlt Date: Wed, 9 Sep 2026 15:30:35 +0800 Subject: [PATCH] =?UTF-8?q?=E5=80=99=E9=80=89=E6=B1=A0=E5=8F=91=E5=B8=83?= =?UTF-8?q?=E7=BB=99=E6=8B=A9=E6=97=B6=E5=86=B3=E7=AD=96=E7=B3=BB=E7=BB=9F?= =?UTF-8?q?=EF=BC=88=E5=8F=B0=E8=B4=A6=20004=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 只写代码列表到共享 Redis 的 pms:candidates 键，择时侧的盘中强势扫描拿它当覆盖面。 走 Redis 不走接口：两侧本来就连着同一台盘中库，不新增网络方向、不涉及鉴权；失败模式 也良性，键没了就是那边本轮不扫，绝不回退全市场。写失败只记日志，不影响本轮扫描。 Co-Authored-By: Claude Opus 5 --- app/services/candidate_pub.py | 43 ++++++++++++++++++++++++++++++++ app/services/proposal_service.py | 12 +++++++-- 2 files changed, 53 insertions(+), 2 deletions(-) create mode 100644 app/services/candidate_pub.py diff --git a/app/services/candidate_pub.py b/app/services/candidate_pub.py new file mode 100644 index 0000000..5daa5f8 --- /dev/null +++ b/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 diff --git a/app/services/proposal_service.py b/app/services/proposal_service.py index 9ad5e08..a43b4cb 100644 --- a/app/services/proposal_service.py +++ b/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 之前一致。