# -*- coding: utf-8 -*- """ 上游选股计划接入 (HTTP /plan) —— 候选池的事实源 ================================================ 上游「选股系统」2026-07-30 起提供计划接口, 取代原先从 `trading_buy_plan` 捞 `is_active=7` 的路子 (那张表是 trading_service 时代的产物, 目标架构下没有明确写入方)。 GET {PMS_PLAN_API_BASE}{PMS_PLAN_API_PATH}?date=YYYY-MM-DD → {date, counts{main,observe,gate_covered}, theme_cap, heat_date, market_snapshot_days[], main:[...], observe:[...], changes, encoding} 主榜 main 每项: {rank, code("SH600418" 前缀式), name, score, heat, upside, tier, evidence{theme, n_sources, moved_ratio}}; 观察档 observe 同形但 无 tier、upside 为 null。 五条口径在此钉死 (口径本身的疑问见 UPSTREAM_PLAN_API.md 的待确认项): 1. **计划里没有价格、没有金额。** 上游只回答「买什么、排第几」, 买多少、什么价位是 PMS 自己的活 (sizer/planner)。价格一律走 `market.get_price`, 取不到就不进池 —— 绝不拿 score/upside 当价格用。 2. **拿不到 ≠ 今天没票可买。** 超时、报错、JSON 变形、日期过期一律抛 `PlanFeedError`, 由调用方记 ERROR 且让候选池为空, 绝不静默回退到旧表。沿用研判闸那条 「UNAVAILABLE ≠ PASS」的纪律: 事实源缺失要显式失败, 不能被当成"没有候选"。 3. **961 只主榜不是买入清单, 是排序池。** 按 score 降序取前 `PMS_PLAN_TOP_N` (默认 30) 进候选, 档位再由 `PMS_PLAN_TIERS` 白名单过滤。 4. **日期新鲜度硬校验。** 计划日期比今天旧超过 `PMS_PLAN_STALE_TDAYS` 个交易日即 过期 (抛错)。防的是节假日/上游停更时拿上周的榜当今天用 —— 这种错在盘中是静默的。 5. **upside = 相对现价的预期空间比例** (2.1203 → +212%; 2026-07-30 与上游 `format=md` 输出的「预期空间 +212%」对齐确认), 来源是券商目标价。噪音大 —— 榜首能 到 +212% —— 所以**永不参与排序** (排序始终是 score), 只提供下限过滤 `PMS_PLAN_MIN_UPSIDE`。 **`theme` 不是行业, 别拿它做行业约束** (2026-07-31 实测后定的, 见 UPSTREAM_PLAN_API.md §6): 主榜 300 条里 205 条根本没有 theme (没有传导的票靠"冷+便宜"排序), 词表也不规范 (整机制造×51 与 整机×9 并存; 锂电/动力电池/储能电池/消费类锂电池 四个分开), 而且它按定义 就是**事件驱动的传导主题**, 天天变。拿它当行业标签, 集中度硬拦截会跟着一起漂。 所以 `PMS_PLAN_THEME_SYNC` 默认关, 行业源走 `gp_stock_category`; theme 只用在候选阶段的 `PMS_PLAN_THEME_CAP_LOCAL` 上 —— 防单一传导主题刷屏, 那才是它擅长的事。 **榜单变化 PMS 自己算** (2026-07-31 加, 见 UPSTREAM_PLAN_API.md §9): 应答里的 `changes` 字段恒为 `null` (§4 Q4 已问未答), 而「谁新进榜、谁掉榜、谁降了档」是上游观点变化最直接的 信号。与其等对方补, 不如每次拉计划落一份名册快照 (`pms_plan_snapshot`), 差异由 `app.core.plan_diff` 纯逻辑算。顺带补上 §7.3 那个洞: `/plan` 是实时算的, 同一个 `date` 能 对应多个版本而应答里没有版本戳 —— 落了快照, 同一 `plan_date` 底下有几行就是重算过几版。 **候选卡的判决与理由 (2026-09-02 起上游随每行下发, 向下兼容)**: 每行多了五个键 —— `verdict` (候选 / 关注 / 仅展示 三个中文值)、`reasons` (理由列表, 这里只留前四条)、 `missing` (缺失项列表)、`risk` (风险列表)、`card_rank` (卡内序, 越小越靠前); 顶层多了 `plan_version` 与 `generated_at`。解析层只做「原样带出来, 缺就是 None」, 不在这里做任何 判断; 唯一的例外是 `select_candidates(route_by_verdict=True)` 会把判为「仅展示」的行在 top_n 截断**之前**剔掉 —— 那是为了不让它们白占候选池的名额, 剔掉的代码单独列在 `display_only` 里, 由自主建仓那条路写成跳过原因 (页面「系统在盯的候选」能看到)。 「关注」强制人工确认与「候选」自动流程的分流不在本模块, 见 action_engine.scan_open 与 proposal_service._route_one。旧版计划没有这些键时, 一切行为与 2026-09-02 之前一字不差。 **研究理由 (2026-09-03 起上游随每行下发, 同样向下兼容)**: 又多了两个键 —— `basis` (判决依据, 一句话说明这一行为什么落在这一档, 例如「三门槛全过、无硬风险、明确吸筹 且评分新鲜」) 与 `logic` (因果论断, 字符串列表, 每行是研报里那句话加上「——出处：《文档 标题》 披露日 · 论断编号」)。加这两个键要解决的是一件很具体的事: 在此之前 PMS 收下的 `reasons` 只有四条可交易口径的理由 (门槛过了几道、覆盖几家、当日涨了多少), 人在提议卡上 看不到研报究竟说了什么, 研究结论走到 PMS 门口就断了。 两个键的性质与那五个不同, 所以口径也不同: * **它们只给人看, 不参与任何判定。** 判决分流、自动执行开关、规则闸、名额与金额, 一处 都不读它们。`basis` 是上游对自己判决的解释, `logic` 是产业逻辑本身。 * **绝不送研判。** `judge.OPEN_JUDGE_KEYS` 不收这两个键 (那份名单里有一条守卫单测钉着)。 送过去等于请择时决策系统去读产业逻辑, 而它的活是判「此刻这只票买不买得下手」—— 四层分工里产业逻辑归数据基座与选股系统, 不归它。 * **`logic` 的条数在这里截断** (`LOGIC_KEEP`), 与 `reasons` 走 `REASONS_KEEP` 同一个 做法; `basis` 是一句话, 不截。 * **论断的日期就写在文字里。** 上游把出处与披露日拼进了每一行, 所以页面照原样显示就能 看出这条论断是哪年哪月的。上游那边正在给论断加陈旧度标注, 字段到了之前, 日期这件事 先靠这行文字兑现 —— 一条 2024 年的论断与一条上周的论断, 人必须一眼分得出。 模块级只依赖 stdlib + `app.core.command_spec` / `app.core.action_engine` (都是纯逻辑), 其余 (requests / param_store / pms_repo / plan_diff / tradedays) 一律函数内懒加载 —— 让解析与筛选这两段纯逻辑可以零依赖单测。 """ from __future__ import annotations import logging import threading import time from app.core.action_engine import VERDICT_DISPLAY from app.core.command_spec import normalize_code logger = logging.getLogger("pms.plan") BUCKET_MAIN, BUCKET_OBSERVE = "main", "observe" # 上游 GET /plan 的签名 (2026-07-30 拿到对方 api.py 确认): # get_plan(date=None, format="json", top=20, obs_top=10, theme_cap=5) # 三个都是**请求参数**, 不是上游的既定政策 —— 主榜给多少、观察档给多少、每主题限几只, # 全由调用方 (也就是 PMS) 决定。默认值写在这里, 用来判断"是不是被条数卡住了"。 UPSTREAM_DEFAULT_TOP, UPSTREAM_DEFAULT_OBS_TOP, UPSTREAM_DEFAULT_THEME_CAP = 20, 10, 5 # 候选池来源 (PMS_CANDIDATE_SOURCE) SRC_PLAN_API, SRC_BUY_PLAN, SRC_BOTH = "plan_api", "buy_plan", "both" FAIL_CACHE_SEC = 60.0 # 失败也缓存一会儿, 免得每分钟的调度位把 10 秒超时叠成雪崩 class PlanFeedError(RuntimeError): """上游计划取不到 / 变形 / 过期。**必须**冒泡, 不得吞成空计划 (见头部口径 2)。""" # ================================================================ 纯逻辑: 解析 def _int_or_none(v): try: if v is None or (isinstance(v, str) and not v.strip()): return None return int(float(v)) except (TypeError, ValueError): return None def _float_or_none(v): try: if v is None or (isinstance(v, str) and not v.strip()): return None return float(v) except (TypeError, ValueError): return None def _text_or_none(v): if v is None: return None s = str(v).strip() return s or None # 候选卡理由最多带几条进 PMS。上游一张卡的理由可以很长, 而这几条要进硬数字、进评审账本、 # 进研判请求体、显示在提议卡上 —— 四个去处都只需要前几条最要紧的。 REASONS_KEEP = 4 # 因果论断最多带几条进 PMS。取 3 有三条理由: # 一是上游 `card.logic_lines` 自己就按 3 条封顶, 取同一个数, 今天上游给了几条 PMS 就收 # 几条 —— 截断不会静默丢掉已经给出来的研究结论 (那是这两个键存在的全部意义); # 二是它仍然是一道硬上限: 上游哪天把自己那个 3 调大, 提议卡、评审账本、提议确认表不会 # 跟着一起被灌满, 这一层不用改码就挡得住; # 三是提议卡上放不下太多行 —— 论断行比理由长得多 (一句机制加一段出处), 卡片默认只显示 # 第一条, 其余折叠, 3 条正好是「折开也还看得完」的量。 LOGIC_KEEP = 3 # 逻辑状态四态 (2026-09-07 第三件): 上游每行带一个字典 —— state 是落定态 (下游按它分流), raw_state # 是当天原始态, why 是无法判断的子因 (证据不足 / 证据矛盾), settle_note 是抗抖动的说明, as_of 是 # 证据截止日, usable / missing 是四路里哪几路在、哪几路缺, reasons 是每路的一句话出处。 # 只收这几个键; 出处最多带三条 (与因果论断同一个量级, 提议卡与账本放得下)。 LOGIC_STATE_REASONS_KEEP = 3 def _logic_state_or_none(v): """逻辑状态归一: 不是字典或没有 state 一律 None (旧版计划没有这个字段)。""" if not isinstance(v, dict): return None state = _text_or_none(v.get("state")) if not state: return None return {"state": state, "raw_state": _text_or_none(v.get("raw_state")) or state, "why": _text_or_none(v.get("why")), "settle_note": _text_or_none(v.get("settle_note")), "as_of": _text_or_none(v.get("as_of")), "usable": _list_or_none(v.get("usable")) or [], "missing": _list_or_none(v.get("missing")) or [], "reasons": _list_or_none(v.get("reasons"), limit=LOGIC_STATE_REASONS_KEEP) or []} def _news_or_none(v, limit=5): """相关快讯 (2026-09-08 台账 050): 上游按财联社电报点名算的条目, 只留 count/latest 与最多 limit 条 {time, level, title, url}。只是上下文: 进硬数字给人看, 整句送研判; 不进任何判定。""" if not isinstance(v, dict) or not v.get("items"): return None items = [] for it in v.get("items") or []: if not isinstance(it, dict): continue items.append({"time": _text_or_none(it.get("time")), "level": _text_or_none(it.get("level")), "title": _text_or_none(it.get("title")), "url": _text_or_none(it.get("url"))}) if len(items) >= limit: break if not items: return None try: count = int(v.get("count") or len(items)) except (TypeError, ValueError): count = len(items) return {"count": count, "latest": _text_or_none(v.get("latest")), "items": items} def _industry_catalyst_or_none(v, limit=3): """行业催化 (2026-09-08 台账 051): 上游按环节标出的行业级催化, 只留最多 limit 条固定键。 产业逻辑不送研判 (judge.OPEN_JUDGE_KEYS 不收), 只进硬数字给人看与复盘。""" if not isinstance(v, list) or not v: return None out = [] for it in v: if not isinstance(it, dict) or not it.get("title"): continue out.append({k: _text_or_none(it.get(k)) for k in ("segment", "event_date", "title", "direction", "magnitude", "horizon", "confidence", "mechanism")}) if len(out) >= limit: break return out or None def _events_or_none(v): """催化事件归一 (2026-09-08): 上游给 {latest, count, events:[{date, types, orgs, title, n_reports, compound}]}。 只收约定的键, 事件最多五条; 不是字典或没有事件一律 None。""" if not isinstance(v, dict) or not isinstance(v.get("events"), list) or not v["events"]: return None evs = [] for e in v["events"][:5]: if not isinstance(e, dict) or not _text_or_none(e.get("date")): continue evs.append({"date": _text_or_none(e.get("date")), "types": _list_or_none(e.get("types")) or [], "orgs": _list_or_none(e.get("orgs")) or [], "title": _text_or_none(e.get("title")), "n_reports": _int_or_none(e.get("n_reports")), "compound": bool(e.get("compound"))}) if not evs: return None return {"latest": _text_or_none(v.get("latest")) or evs[0]["date"], "count": _int_or_none(v.get("count")) or len(evs), "events": evs} _PRICING_NUM_KEYS = ("pre5", "pre20", "gap", "intraday", "close_pos", "vol_ratio", "day_pct") def _pricing_or_none(v): """定价状态归一 (2026-09-08): state 是四情形之一或 None (算不出时 why 写原因), 六个数与涨停标记原样收。""" if not isinstance(v, dict) or not _text_or_none(v.get("event_date")): return None out = {"state": _text_or_none(v.get("state")), "why": _text_or_none(v.get("why")), "event_date": _text_or_none(v.get("event_date")), "has_event": bool(v.get("has_event")), "limit_up": (None if v.get("limit_up") is None else bool(v.get("limit_up")))} for k in _PRICING_NUM_KEYS: out[k] = _float_or_none(v.get(k)) return out def _list_or_none(v, limit: int = 0): """字符串列表归一: 列表逐项去空白、丢空项, 单个字符串当一项; 缺失或类型不对一律 None。 空列表**保留为空列表**而不是折成 None —— 「上游说没有风险」(risk=[]) 与「上游没给这个 字段」(缺键) 是两件事, 自动执行开关那条「风险列表为空」判据要分得开。 """ if v is None: return None if isinstance(v, str): s = v.strip() if not s: return None # 空串与缺键同义 (与 _text_or_none 一个口径) items = [s] elif isinstance(v, (list, tuple)): items = [str(x).strip() for x in v if x is not None and str(x).strip()] else: return None return items[:limit] if limit and limit > 0 else items def _rows(raw, bucket: str) -> list: """一档榜单归一。坏行跳过而不是整体抛错 —— 单条变形不该让整张计划失效。""" out, seen = [], set() for i, it in enumerate(raw or []): if not isinstance(it, dict): continue code = normalize_code(str(it.get("code") or it.get("ts_code") or "")) if not code or code in seen: continue seen.add(code) ev = it.get("evidence") if isinstance(it.get("evidence"), dict) else {} rank = _int_or_none(it.get("rank")) out.append({ "ts_code": code, "name": _text_or_none(it.get("name")), "rank": rank if rank is not None else i + 1, "score": _float_or_none(it.get("score")), "heat": _float_or_none(it.get("heat")), "upside": _float_or_none(it.get("upside")), # 相对现价的比例, 2.12=+212% "tier": _text_or_none(it.get("tier")), "theme": _text_or_none(ev.get("theme")), "n_sources": _int_or_none(ev.get("n_sources")), "moved_ratio": _float_or_none(ev.get("moved_ratio")), "bucket": bucket, # 候选卡的判决与理由 (2026-09-02 起上游下发; 旧版计划没有, 缺就是 None) "verdict": _text_or_none(it.get("verdict")), "reasons": _list_or_none(it.get("reasons"), limit=REASONS_KEEP), "missing": _list_or_none(it.get("missing")), "risk": _list_or_none(it.get("risk")), "card_rank": _int_or_none(it.get("card_rank")), # 研究理由 (2026-09-03): 只给人看, 不参与任何判定, 也不送研判。见模块说明。 "basis": _text_or_none(it.get("basis")), "logic": _list_or_none(it.get("logic"), limit=LOGIC_KEEP), # 逻辑状态四态 (2026-09-07 第三件): 跟着候选进硬数字, 不送研判 (judge.OPEN_JUDGE_KEYS 不收)。 "logic_state": _logic_state_or_none(it.get("logic_state")), # 安全边际三情景的整句 (2026-09-07 第四件): 上游按盈利预测算的悲观/中性/乐观估值与赔率, # 只给人看 (提议卡与账本), 不参与任何判定, 也不送研判。 "valuation_text": _text_or_none(it.get("valuation_text")), # 催化事件与定价状态 (2026-09-08《量价研判链吸收方案》3.5): 原值进硬数字给人看与复盘, # 两句整句送研判 (judge.OPEN_JUDGE_KEYS 收 events_text / pricing_text)。都不进任何判定。 "events": _events_or_none(it.get("events")), "events_text": _text_or_none(it.get("events_text")), "pricing_state": _pricing_or_none(it.get("pricing_state")), "pricing_text": _text_or_none(it.get("pricing_text")), # 相关快讯 (2026-09-08 台账 050): 财联社电报点名这只票的条目, 原值给人看, 整句送研判 # (judge.OPEN_JUDGE_KEYS 收 news_text)。只是上下文, 不判正负, 不进任何判定。 "news": _news_or_none(it.get("news")), "news_text": _text_or_none(it.get("news_text")), # 行业催化 (2026-09-08 台账 051): 环节级, 只给人看与复盘, 不送研判、不进判定。 "industry_catalyst": _industry_catalyst_or_none(it.get("industry_catalyst")), "industry_catalyst_text": _text_or_none(it.get("industry_catalyst_text")), }) return out def parse_plan(payload, *, requested=None) -> dict: """应答 → 内部结构。缺 date 或两档全空都算变形 (抛 PlanFeedError)。 requested: 本次实际发出的 {top, obs_top} (缺省按上游签名的默认值 20/10 算)。 判"还有没有更多"必须拿它跟返回条数比, **不能拿 counts 比** —— 见 _capped 的注释。 """ if not isinstance(payload, dict): raise PlanFeedError(f"应答不是 JSON 对象: {type(payload).__name__}") date = _text_or_none(payload.get("date")) if not date: raise PlanFeedError("应答缺 date —— 判不了新鲜度, 按取不到处理") main = _rows(payload.get("main"), BUCKET_MAIN) observe = _rows(payload.get("observe"), BUCKET_OBSERVE) if not main and not observe: raise PlanFeedError(f"计划 {date} 主榜与观察档都是空的") counts = payload.get("counts") if isinstance(payload.get("counts"), dict) else {} req = dict(requested or {}) req_top = _int_or_none(req.get("top")) req_top = UPSTREAM_DEFAULT_TOP if req_top is None else req_top req_obs = _int_or_none(req.get("obs_top")) req_obs = UPSTREAM_DEFAULT_OBS_TOP if req_obs is None else req_obs themes = {} for r in main + observe: # 主榜在前, 同码以主榜的 theme 为准 if r["theme"] and r["ts_code"] not in themes: themes[r["ts_code"]] = r["theme"] return { "date": date, # 计划版本与生成时刻 (2026-09-02 起上游下发)。同一个 date 的计划会被重算多版, # 这两项是上游自己的版本戳; 旧版应答没有, 缺就是 None (名册快照那套指纹照旧管用)。 "plan_version": _text_or_none(payload.get("plan_version")), "generated_at": _text_or_none(payload.get("generated_at")), "heat_date": _text_or_none(payload.get("heat_date")), "market_snapshot_days": [str(x) for x in (payload.get("market_snapshot_days") or [])], "theme_cap": _int_or_none(payload.get("theme_cap")), "encoding": _text_or_none(payload.get("encoding")), "counts": {"main": _int_or_none(counts.get("main")), "observe": _int_or_none(counts.get("observe")), "gate_covered": _int_or_none(counts.get("gate_covered"))}, "returned": {"main": len(main), "observe": len(observe)}, # 漏斗: counts 是上游的**打分池规模**, returned 是过完 # 「有券商预期 + 目标价不低于现价 + 每主题限额 + top」之后真给了几条。 # 两个数衡量的不是一回事, 相减没有意义 —— 只做展示。 "funnel": {"scored_main": _int_or_none(counts.get("main")), "returned_main": len(main), "scored_observe": _int_or_none(counts.get("observe")), "returned_observe": len(observe)}, "requested": {"top": req_top, "obs_top": req_obs, "theme_cap": _int_or_none(req.get("theme_cap"))}, "truncated": {"main": _capped(len(main), req_top), "observe": _capped(len(observe), req_obs)}, "main": main, "observe": observe, "themes": themes, } def _capped(got: int, requested) -> bool: """"是不是还有更多没拿到" = 返回条数吃满了我们要的条数。 **不能拿 counts 判。** counts.main=961 是打分池规模, 而 returned 是过完券商预期、 目标价不低于现价、每主题限额、top 之后的结果 —— 实测 top=1000 也只回 55 条 (被 theme_cap=5 卡住)。拿 961 跟 55 比会永远报"被截断", 变成一个天天喊狼来了的假警报。 吃满才说明是条数卡的, 没吃满就是上游确实只有这么多能给。 """ r = _int_or_none(requested) return bool(r is not None and r > 0 and got >= r) # ================================================================ 纯逻辑: 新鲜度 def plan_age_tdays(plan_date, today=None) -> int: """计划日期距今的交易日龄。当天=0, 上一个交易日=1; 日期在未来 (为下一交易日出的计划)=0。""" from app.core import tradedays as td left = td.trade_days_left(today, plan_date) # 含首尾; plan_date 晚于 today 时为 0 return max(0, left - 1) def assert_fresh(plan: dict, *, max_stale_tdays: int = 1, today=None) -> int: age = plan_age_tdays(plan["date"], today) if age > int(max_stale_tdays or 0): raise PlanFeedError( f"上游计划已过期: 日期 {plan['date']} 距今 {age} 个交易日 " f"(上限 {max_stale_tdays})。上游可能停更 —— 拿旧榜当今天用比没有候选更危险") return age # ================================================================ 纯逻辑: 筛选 def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=None, include_observe: bool = False, min_score=None, min_sources: int = 0, min_upside=None, theme_cap: int = 0, exclude_st: bool = True, route_by_verdict: bool = False) -> dict: """排序池 → 候选清单。 route_by_verdict: 按上游判决分流时, 判为「仅展示」的行在 top_n 截断**之前**剔掉并列进 `display_only` (代码列表), 让自主建仓那条路写成跳过原因。放在这一层而不是动作引擎, 是 因为 top_n 那一刀按纯 score 切 —— 仅展示的票分数未必低, 不先剔掉的话前 30 名可能大半 是仅展示, 真正的候选与关注排在第 31 名之后永远轮不上。默认 False = 与 2026-09-02 之前一字不差 (命令驱动建仓与探活脚本各自显式决定要不要开)。没有 verdict 的行永远 不受这条影响。 排序: score 降序, 同分按 rank 升序 (上游 rank 已是它自己的最终次序, 拿来当稳定次序)。 tier 白名单**只对带 tier 的行生效** —— 观察档没有 tier, 它的闸门是 include_observe。 min_upside 相反, **对所有行生效**: upside 缺失按 0 算一起挡掉 (观察档 upside 恒为 null, 所以设了下限等于把观察档全挡了)。方向选保守那边 —— 宁可少票。 exclude_st: 剔除 ST / *ST / S* / 退市整理。上游文档写明「名单当前**不剔除 ST/*ST**; 需要过滤请在下游做」—— 那这道闸就是 PMS 的责任, 默认开。判据只有 name 一个字段, 所以 name 缺失时**不敢放行**也不敢误杀: 保留但计入 dropped 之外的 st_unknown。 theme_cap: 同主题最多取几只, **在 top_n 截断之前**按 score 序生效 (0=不限)。 这一层存在的理由: 上游的 theme_cap 是请求参数, 我们可以让它别裁 (要个宽池子), 但 top_n 那一刀是按纯 score 切的 —— 宽池子里前 30 名可能全是储能, 切完再交给规则闸, 规则闸按 PMS_SECTOR_MAX_NAMES 一拦就剩 4 只, 白瞎 26 个名额且**日志上看不出来**。 在候选阶段先按主题摊开, top_n 切出来的才是能用的票。 """ held = {normalize_code(c) for c in (held or []) if c} black = {normalize_code(c) for c in (black or []) if c} tiers = {str(t).strip() for t in (tiers or []) if str(t).strip()} min_score = _float_or_none(min_score) min_upside = _float_or_none(min_upside) min_sources = int(min_sources or 0) pool = list(plan.get("main") or []) if include_observe: pool += list(plan.get("observe") or []) theme_cap = int(theme_cap or 0) dropped = {"held": 0, "black": 0, "tier": 0, "score": 0, "sources": 0, "upside": 0, "st": 0, "theme": 0, "dup": 0, "capped": 0, "display": 0} passed, seen, per_theme, st_unknown, display_only = [], set(), {}, [], [] for r in sorted(pool, key=lambda x: (-(x.get("score") or 0.0), x.get("rank") or 10 ** 9)): c = r["ts_code"] if c in seen: dropped["dup"] += 1 continue seen.add(c) if c in held: dropped["held"] += 1 continue if c in black: dropped["black"] += 1 continue if route_by_verdict and r.get("verdict") == VERDICT_DISPLAY: dropped["display"] += 1 # 上游判为仅展示: 不进池, 单独列出让人看得见 # 带上判决依据: 仅展示至少有三种来路 (差一个当日涨幅、吸筹没到确认线、 # 被昨夜坏信号一票否决), 三者处置完全不同。只记代码的话, 下游只能对所有 # 仅展示的票写同一句话, 人看不出哪只该继续盯。 display_only.append({"ts_code": c, "basis": r.get("basis")}) continue if exclude_st: nm = (r.get("name") or "").strip() if not nm: st_unknown.append(c) # 没名字判不了, 留着但要让人知道 elif is_st(nm): dropped["st"] += 1 continue if tiers and r.get("tier") is not None and r["tier"] not in tiers: dropped["tier"] += 1 continue if min_score is not None and (r.get("score") or 0.0) < min_score: dropped["score"] += 1 continue if min_sources and (r.get("n_sources") or 0) < min_sources: dropped["sources"] += 1 continue if min_upside is not None and (r.get("upside") or 0.0) < min_upside: dropped["upside"] += 1 continue if theme_cap > 0: t = r.get("theme") or "(无主题)" if per_theme.get(t, 0) >= theme_cap: dropped["theme"] += 1 continue per_theme[t] = per_theme.get(t, 0) + 1 passed.append(r) n = max(0, int(top_n or 0)) or len(passed) dropped["capped"] = max(0, len(passed) - n) items = [{"ts_code": r["ts_code"], "name": r["name"], "score": r.get("score") or 0.0, "sector": r.get("theme"), "theme": r.get("theme"), "tier": r.get("tier"), "heat": r.get("heat"), "upside": r.get("upside"), "rank": r.get("rank"), "bucket": r["bucket"], "src": "plan_api", # 候选卡五键原样透传 (缺就是 None), 动作引擎把它们放进硬数字 "verdict": r.get("verdict"), "reasons": r.get("reasons"), "missing": r.get("missing"), "risk": r.get("risk"), "card_rank": r.get("card_rank"), # 研究理由两键: 跟着候选一路带到提议卡与评审账本, 让人看得到研报说了什么。 # 它们不进送研判的那份名单 (judge.OPEN_JUDGE_KEYS), 产业逻辑不归择时决策系统判。 "basis": r.get("basis"), "logic": r.get("logic"), # 逻辑状态 (2026-09-07 第三件): 动作引擎放进硬数字; 判决候选而逻辑存疑的在 scan_open # 里强制交人。它不改判决、不改排序 —— 收敛规则是单调的 (选股系统 logic_state.apply_to_card)。 "logic_state": r.get("logic_state"), "valuation_text": r.get("valuation_text"), "events": r.get("events"), "events_text": r.get("events_text"), "pricing_state": r.get("pricing_state"), "pricing_text": r.get("pricing_text"), "news": r.get("news"), "news_text": r.get("news_text"), "industry_catalyst": r.get("industry_catalyst"), "industry_catalyst_text": r.get("industry_catalyst_text")} for r in passed[:n]] return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed), "items": items, "dropped": dropped, "st_unknown": st_unknown, "display_only": display_only, "tier_complete": _tier_complete(pool, tiers)} def is_st(name: str) -> bool: """ST / *ST / 退市整理 判定 (只看名字 —— 上游只给了这一个可判据)。 覆盖四种真实写法: 前缀 `ST` / `*ST` / `SST` / `S*ST`, 前缀 `退市`(退市整理期, 如「退市海润」), 后缀 `退`(如「海润退」)。A 股名称都是中文, 所以前缀按拉丁字母 `ST` 匹配不会误伤 —— 「圣农发展」「深赛格」这类 S 开头的中文名不受影响。 """ n = (name or "").replace(" ", "").upper() if not n: return False return (n.startswith(("*ST", "S*ST", "SST", "ST", "退市")) or n.endswith(("退", "退市"))) def _tier_complete(pool, tiers) -> bool: """白名单档位是不是已经取全了。 主榜分 = 200 + 传导档位×20 + 组内分, 所以**档位是分层的**: 按 score 降序排, 只要看见 了一条白名单之外的档位, 就说明白名单那几档已经全在前面拿到手了。 这条判据用来消掉"吃满 top 就报警"的假警报 —— 只要白名单齐了, 后面还有多少弱传导 根本不影响候选池。tiers 为空 (不按档过滤) 时无从判断, 返回 None。 """ if not tiers: return None return any((r.get("tier") is not None and r["tier"] not in tiers) for r in (pool or [])) # ================================================================ 取数 (带缓存) _cache = {"at": 0.0, "key": None, "plan": None, "error": None} _lock = threading.Lock() def _params() -> dict: from app.services import param_store as ps return { "base": (ps.get("PMS_PLAN_API_BASE", "") or "").strip().rstrip("/"), "path": (ps.get("PMS_PLAN_API_PATH", "/plan") or "/plan").strip(), "timeout": ps.get_int("PMS_PLAN_TIMEOUT", 10), "cache_sec": ps.get_int("PMS_PLAN_CACHE_SEC", 300), "top_n": ps.get_int("PMS_PLAN_TOP_N", 30), "tiers": ps.get_list("PMS_PLAN_TIERS", []), "include_observe": ps.get_bool("PMS_PLAN_INCLUDE_OBSERVE", False), "min_score": ps.get_float("PMS_PLAN_MIN_SCORE", 0.0), "min_sources": ps.get_int("PMS_PLAN_MIN_SOURCES", 0), "min_upside": ps.get_float("PMS_PLAN_MIN_UPSIDE", 0.0), "stale_tdays": ps.get_int("PMS_PLAN_STALE_TDAYS", 1), "theme_sync": ps.get_bool("PMS_PLAN_THEME_SYNC", False), "top": ps.get_int("PMS_PLAN_TOP", 300), "obs_top": ps.get_int("PMS_PLAN_OBS_TOP", 100), "theme_cap": ps.get_int("PMS_PLAN_THEME_CAP", 999), "theme_cap_local": ps.get_int("PMS_PLAN_THEME_CAP_LOCAL", 5), "exclude_st": ps.get_bool("PMS_PLAN_EXCLUDE_ST", True), "query_extra": parse_query_extra(ps.get("PMS_PLAN_QUERY_EXTRA", "")), "source": (ps.get("PMS_CANDIDATE_SOURCE", SRC_PLAN_API) or SRC_PLAN_API).strip(), "snapshot": ps.get_bool("PMS_PLAN_SNAPSHOT", True), "snapshot_keep": ps.get_int("PMS_PLAN_SNAPSHOT_KEEP", 200), "diff_rank_jump": ps.get_int("PMS_PLAN_DIFF_RANK_JUMP", 50), "diff_tail_guard": ps.get_float("PMS_PLAN_DIFF_TAIL_GUARD", 0.5), "route_by_verdict": ps.get_bool("PMS_PLAN_ROUTE_BY_VERDICT", True), } def build_query(p: dict) -> dict: """发给上游的查询参数。显式参数覆盖 PMS_PLAN_QUERY_EXTRA 里的同名键。 QUERY_EXTRA 保留是为了上游哪天加了新参数时不用改代码; 但 top/obs_top/theme_cap 这三个 已经知道签名了, 走各自的显式参数 —— 同一件事有两个入口的时候, 得有个明确的赢家。 """ q = dict(p.get("query_extra") or {}) for key, name in (("top", "top"), ("obs_top", "obs_top"), ("theme_cap", "theme_cap")): v = int(p.get(key) or 0) if v > 0: q[name] = str(v) # 0 = 不传该参数, 用上游默认 return q def effective_query() -> dict: """当前参数下实际会发出去的查询串。探活脚本要跟生产走同一条路 —— 上一版就是因为 没带这三个参数, 报出来的"请求参数"是上游默认值 20/10/5, 跟真实抓取行为对不上。""" return build_query(_params()) def enabled() -> bool: return bool(_params()["base"]) def parse_query_extra(text) -> dict: """`"limit=1000&offset=0"` → dict。解析不了就返回空 dict (不炸, 记 warning)。 存在的理由: 上游默认只回主榜 20 条 (见 parse_plan 的 truncated 注释)。取全量要带哪个 参数名还没确认 (limit? top? size?), 探出来以后**只改这个参数就能生效**, 不用改代码。 """ text = (text or "").strip().lstrip("?") if not text: return {} try: from urllib.parse import parse_qsl out = {k: v for k, v in parse_qsl(text, keep_blank_values=False) if k} if not out: raise ValueError("解析结果为空") return out except Exception as e: logger.warning("[上游计划] PMS_PLAN_QUERY_EXTRA 解析失败, 已忽略 (%r): %s", text, e) return {} # 接口地址没填时的那句话。环境变量名退到括号里做精确指认, 不做句子主干 —— # 原先那句以变量名开头, 读的人不知道该找谁、去哪儿填。 _NO_BASE_WHY = ("选股系统的接口地址没填，PMS 拿不到今天的股票名单。" "请运维在「运维 → 参数设置」里补上地址（PMS_PLAN_API_BASE），填完点右上角刷新。") def _fetch_fail_why(e, timeout_sec) -> str: """拉不到计划时给人看的那句话。按失败的类型分三种, 因为处置不同。 「过 1 分钟再试」这句不能省: 失败结果会被缓存住 FAIL_CACHE_SEC 秒 (默认 60), 早于这个点重读拿到的还是缓存里同一条失败, 人会以为重试没用。 """ name = type(e).__name__ wait = int(FAIL_CACHE_SEC // 60) or 1 if "Timeout" in name or "Connection" in name: return (f"选股系统没有回应，等了 {timeout_sec} 秒。现在拿不到今天的候选票。" f"过 {wait} 分钟，点右上角「选股计划」里的「只重读」再试一次；" f"还是不行就找运维。") code = getattr(getattr(e, "response", None), "status_code", None) if code: return f"选股系统返回了错误，状态码 {code}。详细原因已记进运行日志。" return "选股系统返回的内容不是正常的计划数据。详细原因已记进运行日志。" def fetch(*, date=None, base=None, path=None, timeout=None, extra_params=None) -> dict: """拉一次并解析 (不走缓存、不校新鲜度)。任何失败抛 PlanFeedError。 三个参数全给齐时**不读参数中心** —— 纯取数函数不该依赖 ParamStore, 否则连 「base 为空立即失败」这条守卫都没法脱库单测。 """ if base is None or path is None or timeout is None: p = _params() base = p["base"] if base is None else base path = p["path"] if path is None else path timeout = p["timeout"] if timeout is None else timeout base = (base or "").strip().rstrip("/") if not base: raise PlanFeedError(_NO_BASE_WHY) path = (path or "/plan").strip() or "/plan" if not path.startswith("/"): path = "/" + path url = base + path to = int(timeout or 10) q = dict(extra_params or {}) if date: q["date"] = date # date 是我们自己的语义, 不许被 extra 覆盖 try: import requests r = requests.get(url, params=(q or None), timeout=to) r.raise_for_status() payload = r.json() except PlanFeedError: raise except Exception as e: # 这条会渲染成页面顶部整条红色横幅, 交易员一进页面就看到。所以句子里不许出现 # 接口地址、异常类名、异常原文 —— 那些只进日志。取而代之要说清三件事: # 现在拿不到什么、多久之后重试有意义、还不行找谁。 logger.warning("拉取上游计划失败 %s: %s: %s", url, type(e).__name__, e) raise PlanFeedError(_fetch_fail_why(e, to)) from e plan = parse_plan(payload, requested=q) plan["url"] = url plan["fetched_at"] = time.time() plan["requested_date"] = date return plan def parse_logic_states(payload) -> dict: """按代码查询接口的应答 → {ts_code: 逻辑状态}。带 error 的行跳过 (代码形态认不出), 查不到的不在结果里。 上游按我们送去的原样 (input) 回, 代码用它映射回 PMS 的形态; 每条另带 source (daily = 早上落定的 当日行, computed = 此刻现算) 与日期, 页面要区分「早上的态」与「现算的态」。""" out = {} day = _text_or_none((payload or {}).get("date")) if isinstance(payload, dict) else None for it in ((payload or {}).get("states") or []) if isinstance(payload, dict) else []: if not isinstance(it, dict) or it.get("error"): continue code = normalize_code(str(it.get("input") or it.get("code") or "")) st = _logic_state_or_none(it) if not code or not st: continue st["source"] = _text_or_none(it.get("source")) st["date"] = day # 安全边际三情景 (第四件): 持仓页拿中性情景作参考目标价 (2026-09-07 拍板, 只显示不触发)。 v = it.get("valuation") if isinstance(it.get("valuation"), dict) else None st["valuation"] = ({"neut": _float_or_none(v.get("neut")), "pess": _float_or_none(v.get("pess")), "opt": _float_or_none(v.get("opt")), "price": _float_or_none(v.get("price")), "quarter": _text_or_none(v.get("quarter")), "firms": _int_or_none(v.get("firms")), "as_of": _text_or_none(v.get("as_of")), "na": _text_or_none(v.get("na"))} if v else None) st["valuation_text"] = _text_or_none(it.get("valuation_text")) out[code] = st return out def fetch_logic_states(codes, *, base=None, timeout=None, path: str = "/logic_state") -> dict: """按代码向选股系统查逐票逻辑状态 (2026-09-07 第三件, 给在持的票用)。任何失败抛 PlanFeedError, 由调用方决定留空 —— 这里不吞错, 因为「查不到」与「判为成立」在扫描层是两件事。""" codes = [normalize_code(str(c)) for c in (codes or []) if c] if not codes: return {} if base is None or timeout is None: p = _params() base = p["base"] if base is None else base timeout = p["timeout"] if timeout is None else timeout base = (base or "").strip().rstrip("/") if not base: raise PlanFeedError(_NO_BASE_WHY) url = base + (path if path.startswith("/") else "/" + path) to = int(timeout or 10) try: import requests r = requests.get(url, params={"codes": ",".join(codes)}, timeout=to) r.raise_for_status() payload = r.json() except Exception as e: logger.warning("查逻辑状态失败 %s: %s: %s", url, type(e).__name__, e) raise PlanFeedError(_fetch_fail_why(e, to)) from e return parse_logic_states(payload) def get_plan(*, force: bool = False, date=None) -> dict: """带缓存的当前计划。失败同样缓存 FAIL_CACHE_SEC, 但每次调用都照样抛。""" p = _params() q = build_query(p) key = (p["base"], p["path"], date or "", tuple(sorted(q.items()))) now = time.time() with _lock: fresh_hit = (not force and _cache["key"] == key and _cache["plan"] is not None and now - _cache["at"] < max(1, p["cache_sec"])) if fresh_hit: return _cache["plan"] if (not force and _cache["key"] == key and _cache["error"] and now - _cache["at"] < FAIL_CACHE_SEC): raise PlanFeedError(_cache["error"]) try: plan = fetch(date=date, extra_params=q) assert_fresh(plan, max_stale_tdays=p["stale_tdays"]) except PlanFeedError as e: with _lock: _cache.update({"at": time.time(), "key": key, "plan": None, "error": str(e)}) raise plan["age_tdays"] = plan_age_tdays(plan["date"]) if p["theme_sync"]: plan["theme_sync"] = _sync_themes_quiet(plan) if p["snapshot"]: plan["snapshot"] = _snapshot_quiet(plan, keep=p["snapshot_keep"]) with _lock: _cache.update({"at": time.time(), "key": key, "plan": plan, "error": None}) logger.info("[上游计划] %s 主榜 %d / 观察 %d (日龄 %d 交易日) ← %s", plan["date"], plan["returned"]["main"], plan["returned"]["observe"], plan["age_tdays"], plan["url"]) if plan["truncated"]["main"]: logger.warning("[上游计划] 主榜正好吃满 top=%s (打分池 %s) —— 可能还有更多没拿到, " "调大 PMS_PLAN_TOP 再看", plan["requested"]["top"], plan["funnel"]["scored_main"]) return plan def invalidate(): with _lock: _cache.update({"at": 0.0, "key": None, "plan": None, "error": None}) # ================================================================ theme → 行业映射 def sync_themes(plan: dict) -> dict: """把 code→theme upsert 进 pms_industry_map (行业源 custom_table 的灌数口径)。""" from app.repo import pms_repo from app.services import industry rows = [{"ts_code": c, "industry": t} for c, t in (plan.get("themes") or {}).items()] if not rows: return {"rows": 0, "affected": 0} n = pms_repo.upsert_industry(rows) industry.invalidate() return {"rows": len(rows), "affected": n} def _sync_themes_quiet(plan: dict) -> dict: """theme 落库失败不许阻断候选池 —— 行业约束停用是可接受的降级, 没候选不是。""" try: return sync_themes(plan) except Exception as e: logger.warning("[上游计划] theme 落 pms_industry_map 失败 (行业约束按未配置降级): %s", e) return {"rows": 0, "affected": 0, "error": f"{type(e).__name__}: {e}"} # ================================================================ 名册快照与榜单变化 def snapshot(plan: dict, *, keep: int = 200) -> dict: """把这一份计划的名册落进 pms_plan_snapshot。 同一份榜重复拉不重复落 (靠 uk_date_digest)。返回里带 `stored` —— 落进去了才是新版本, 没落进去说明这份榜跟库里最新那份一模一样。 """ from app.core import plan_diff as pdf from app.repo import pms_repo roster = pdf.roster_of(plan) digest = pdf.digest_of(roster) meta = {k: plan.get(k) for k in ("counts", "returned", "requested", "funnel", "theme_cap", "heat_date", "market_snapshot_days", "age_tdays", "url")} truncated = plan.get("truncated") or {} n = pms_repo.insert_plan_snapshot( plan_date=plan.get("date"), digest=digest, roster=pdf.roster_rows(roster), meta=meta, n_main=(plan.get("returned") or {}).get("main") or 0, n_observe=(plan.get("returned") or {}).get("observe") or 0, capped_main=bool(truncated.get("main")), capped_obs=bool(truncated.get("observe"))) out = {"digest": digest, "rows": len(roster), "stored": bool(n)} if n: out["pruned"] = pms_repo.prune_plan_snapshots(keep=keep) logger.info("[上游计划] 名册快照已落: %s digest=%s (%d 只)", plan.get("date"), digest[:10], len(roster)) return out def _snapshot_quiet(plan: dict, *, keep: int = 200) -> dict: """快照落库失败**不许阻断候选池** —— 没有变化提示是可接受的降级, 没候选不是。 同 `_sync_themes_quiet` 的口径: 这是个"锦上添花"的旁路, 它的故障不该传染主链路。 但也不静默: error 会一路带到页面上, 因为快照断了就意味着变化提示从此刻起是错的 (拿旧版本当上一版比)。 """ try: return snapshot(plan, keep=keep) except Exception as e: logger.warning("[上游计划] 名册快照落库失败 (榜单变化提示将不可用): %s", e) return {"stored": False, "error": f"{type(e).__name__}: {e}"} def _held_codes() -> list: """当前持仓代码。拿不到就当空 —— 持仓视图缺失只是少一段提示, 不该让整个接口失败。""" try: from app.repo import pms_repo return [p["ts_code"] for p in pms_repo.list_positions(only_open=True)] except Exception as e: logger.debug("[上游计划] 取持仓失败, 榜单变化的持仓视图留空: %s", e) return [] def _capped_of(row) -> dict: """一行快照的按档截断标志。两档是两个独立的请求参数 (`top` / `obs_top`), 各判各的 —— 拿主榜的截断去解释观察档的进出, 会把观察档一次真实的摘牌记成"截断噪音"。""" row = row or {} return {"main": bool(row.get("capped_main")), "observe": bool(row.get("capped_obs"))} def _prev_broken(prev) -> bool: """上一行"说自己有几百只、名册却解析成空" = 落库时坏了, 不是正常的"没有上一版"。 `pms_repo._loads` 在 JSON 解析失败时返回默认值 `[]`, 于是坏数据长得跟"首次"一模一样 —— 一条真实变化都不报, 页面还显示得很正常。这里把两者分开。 """ if not prev: return False claimed = int(prev.get("n_main") or 0) + int(prev.get("n_observe") or 0) return bool(claimed > 0 and not (prev.get("roster") or [])) def _diff_against(prev, curr_roster, *, curr_date, curr_capped, held, rank_jump, tail_guard): """比对的公共落点 —— `changes` (库里两份) 与 `preview_changes` (手上这份 vs 库里最新) 只在"谁是 curr"上不同, 比对口径必须完全一致, 所以共用这一段。""" from app.core import command_spec as cs from app.core import plan_diff as pdf p = _params() codes = held if held is not None else _held_codes() # 名册的键是点式 (parse_plan 已归一)。持仓来源万一给的是前缀式, 不归一的后果是 # **持仓视图静默为空** —— 不报错、不少数据, 就是一条都不命中。 codes = [cs.normalize_code(str(c)) for c in (codes or []) if c] d = pdf.diff(pdf.roster_from_rows(prev["roster"]) if prev else None, curr_roster, prev_date=(prev or {}).get("plan_date"), curr_date=curr_date, held=codes, rank_jump=(rank_jump if rank_jump is not None else p["diff_rank_jump"]), tail_guard=(tail_guard if tail_guard is not None else p["diff_tail_guard"]), prev_capped=_capped_of(prev), curr_capped=curr_capped, prev_broken=_prev_broken(prev)) d["prev_fetched_at"] = (prev or {}).get("fetched_at") return d def preview_changes(plan: dict, *, held=None, rank_jump=None, tail_guard=None) -> dict: """**只读**: 拿手上这份计划跟库里最新那份快照比, 一个字都不写。 探活脚本用 —— 那个脚本的契约是"只读, 不写任何表", 破了它比少个功能糟得多。 页面走 `changes()`: 页面本来就要落快照 (那是它的正常职责), 比的是库里最近两份。 """ from app.core import plan_diff as pdf out = {"ok": False, "readonly": True, "hint": "", "diff": None, "stored": None} try: from app.repo import pms_repo snaps = pms_repo.latest_plan_snapshots(limit=1) except Exception as e: out["hint"] = f"读名册快照失败: {type(e).__name__}: {e}" return out prev = snaps[0] if snaps else None curr_roster = pdf.roster_of(plan) out["stored"] = ({k: prev.get(k) for k in ("id", "plan_date", "digest", "fetched_at")} if prev else None) out["same_as_stored"] = bool(prev and prev["digest"] == pdf.digest_of(curr_roster)) try: tr = plan.get("truncated") or {} out["diff"] = _diff_against(prev, curr_roster, curr_date=plan.get("date"), curr_capped={"main": bool(tr.get("main")), "observe": bool(tr.get("observe"))}, held=held, rank_jump=rank_jump, tail_guard=tail_guard) except Exception as e: out["hint"] = f"比对失败: {type(e).__name__}: {e}" return out out["ok"] = True out["hint"] = out["diff"]["note"] return out def changes(*, held=None, rank_jump=None, tail_guard=None) -> dict: """榜单变化: 拿库里最近两份快照比。**绝不抛错** —— 这是页面提示, 不该有能力搞崩取数。 上游 `changes` 字段恒为 null (UPSTREAM_PLAN_API.md §4 Q4), 所以这段由 PMS 自己算。 比对语义与尾部闸见 `app/core/plan_diff.py` 的模块说明。 `in_sync`: 手上这份计划的指纹与库里最新快照**是不是同一个**。落库失败过 (DB 抖了一下) 的话这里会是 False —— 那说明比出来的是两个旧版本, 跟你现在用的候选池不是一回事。 宁可把这句话摆出来, 也不让页面显示一个看起来正常、实际过时的差异。 """ from app.core import plan_diff as pdf p = _params() out = {"ok": False, "enabled": bool(p["snapshot"]), "hint": "", "diff": None, "snapshots": [], "in_sync": None} if not p["snapshot"]: out["hint"] = "名册快照已关 (PMS_PLAN_SNAPSHOT=false) —— 榜单变化提示不可用" return out curr_digest = None try: # 先确保手上这份已落库 (缓存命中时是个空操作) plan = get_plan() curr_digest = pdf.digest_of(pdf.roster_of(plan)) except PlanFeedError as e: out["hint"] = f"上游计划取不到, 只能拿库里的历史快照比: {e}" except Exception as e: out["hint"] = f"{type(e).__name__}: {e}" try: from app.repo import pms_repo snaps = pms_repo.latest_plan_snapshots(limit=2) except Exception as e: out["hint"] = f"读名册快照失败: {type(e).__name__}: {e}" return out out["snapshots"] = [{k: s.get(k) for k in ("id", "plan_date", "digest", "fetched_at", "n_main", "n_observe", "capped_main", "capped_obs")} for s in snaps] if not snaps: out.update({"ok": True, "hint": out["hint"] or "还没有任何名册快照 —— 拉一次计划即有"}) return out if curr_digest is not None: out["in_sync"] = (snaps[0]["digest"] == curr_digest) curr, prev = snaps[0], (snaps[1] if len(snaps) > 1 else None) try: d = _diff_against(prev, pdf.roster_from_rows(curr["roster"]), curr_date=curr.get("plan_date"), curr_capped=_capped_of(curr), held=held, rank_jump=rank_jump, tail_guard=tail_guard) except Exception as e: # 兑现"绝不抛错": 顶层 ok 一为假页面就挂全局红条 logger.warning("[上游计划] 榜单变化比对失败: %s", e) out["hint"] = f"比对失败: {type(e).__name__}: {e}" return out d["curr_fetched_at"] = curr.get("fetched_at") out.update({"ok": True, "diff": d}) if out["in_sync"] is False: out["hint"] = ("手上这份计划**没落进快照** (上次落库失败?) —— 下面比的是两个旧版本, " "与当前候选池不是同一份。" + (out["hint"] or "")) elif not out["hint"]: out["hint"] = d["note"] return out def snapshot_log(*, plan_date=None, limit: int = 50) -> dict: """快照台账。同一 `plan_date` 有几行, 就是上游那天重算过几版 (§7.3)。""" try: from app.repo import pms_repo rows = pms_repo.list_plan_snapshots(plan_date=plan_date, limit=limit) by_date = {} for r in rows: by_date[r["plan_date"]] = by_date.get(r["plan_date"], 0) + 1 revised = {k: v for k, v in by_date.items() if v > 1} return {"ok": True, "rows": rows, "versions_per_date": by_date, "revised": revised} except Exception as e: return {"ok": False, "rows": [], "error": f"{type(e).__name__}: {e}"} # ================================================================ 对外: 候选与状态 def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict: """参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。 route_by_verdict 默认 False: 命令驱动建仓 (command_service._candidates) 不传, 于是用户 下的升仓命令照旧看全池, 与 2026-09-02 之前一字不差 —— 判决分流只管自主提议这条路 (方案 3.2 的措辞是「候选自动进方案、关注强制确认、仅展示不出**提议**」), 命令是用户 自己的决定, 不替他过滤。自主建仓 (proposal_service._scan_open) 按 PMS_PLAN_ROUTE_BY_VERDICT 显式传 True。要让命令驱动也按判决过滤, 改这里的调用方, 不改默认值。 """ p = _params() plan = get_plan() route = bool(route_by_verdict) out = select_candidates(plan, held=held, black=black, top_n=p["top_n"], tiers=p["tiers"], include_observe=p["include_observe"], min_score=(p["min_score"] or None), min_sources=p["min_sources"], min_upside=(p["min_upside"] or None), theme_cap=p["theme_cap_local"], exclude_st=p["exclude_st"], route_by_verdict=route) out["age_tdays"] = plan.get("age_tdays") out["theme_cap"] = plan.get("theme_cap") out["route_by_verdict"] = route return out def status() -> dict: """页面/运维用: 绝不抛错, 失败也要能显示出来。""" p = _params() st = {"source": p["source"], "base": p["base"], "path": p["path"], "top_n": p["top_n"], "tiers": p["tiers"], "min_upside": p["min_upside"], "theme_cap_local": p["theme_cap_local"], "query": build_query(p), "include_observe": p["include_observe"], "stale_tdays": p["stale_tdays"], "theme_sync": p["theme_sync"], "snapshot_on": p["snapshot"], "enabled": bool(p["base"])} if not p["base"]: st.update({"ok": False, "hint": "上游计划接口未配置 (PMS_PLAN_API_BASE 为空) —— " "候选池将为空, 升仓/建仓类命令无票可选"}) return st try: plan = get_plan() except PlanFeedError as e: st.update({"ok": False, "hint": str(e)}) return st except Exception as e: # 兜底: status 不许抛 st.update({"ok": False, "hint": f"{type(e).__name__}: {e}"}) return st st.update({"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"), "plan_version": plan.get("plan_version"), "generated_at": plan.get("generated_at"), "route_by_verdict": p["route_by_verdict"], "heat_date": plan.get("heat_date"), "market_snapshot_days": plan.get("market_snapshot_days"), "counts": plan["counts"], "returned": plan["returned"], "funnel": plan.get("funnel"), "requested": plan.get("requested"), "truncated": plan.get("truncated"), "theme_cap": plan.get("theme_cap"), "encoding": plan.get("encoding"), "theme_sync": plan.get("theme_sync"), "snapshot": plan.get("snapshot"), "fetched_at": plan.get("fetched_at"), "url": plan.get("url"), "hint": f"计划 {plan['date']} 已就绪 (日龄 {plan.get('age_tdays')} 交易日)"}) return st