755 lines
37 KiB
Python
755 lines
37 KiB
Python
# -*- 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` 底下有几行就是重算过几版。
|
||
|
||
模块级只依赖 stdlib + `app.core.command_spec` (纯逻辑), 其余 (requests / param_store /
|
||
pms_repo / plan_diff / tradedays) 一律函数内懒加载 —— 让解析与筛选这两段纯逻辑可以零依赖
|
||
单测。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import threading
|
||
import time
|
||
|
||
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
|
||
|
||
|
||
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,
|
||
})
|
||
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,
|
||
"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) -> dict:
|
||
"""排序池 → 候选清单。
|
||
|
||
排序: 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}
|
||
passed, seen, per_theme, st_unknown = [], 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 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"}
|
||
for r in passed[:n]]
|
||
return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed),
|
||
"items": items, "dropped": dropped, "st_unknown": st_unknown,
|
||
"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),
|
||
}
|
||
|
||
|
||
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 {}
|
||
|
||
|
||
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("上游计划接口未配置 (PMS_PLAN_API_BASE 为空)")
|
||
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:
|
||
raise PlanFeedError(f"拉取上游计划失败 {url}: {type(e).__name__}: {e}") from e
|
||
plan = parse_plan(payload, requested=q)
|
||
plan["url"] = url
|
||
plan["fetched_at"] = time.time()
|
||
plan["requested_date"] = date
|
||
return plan
|
||
|
||
|
||
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=()) -> dict:
|
||
"""参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。"""
|
||
p = _params()
|
||
plan = get_plan()
|
||
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"])
|
||
out["age_tdays"] = plan.get("age_tdays")
|
||
out["theme_cap"] = plan.get("theme_cap")
|
||
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"),
|
||
"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
|