From df966978ce0b6abda92c5670e9a945e2bef32f2b Mon Sep 17 00:00:00 2001 From: zlt Date: Fri, 31 Jul 2026 09:17:54 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9B=98=E5=89=8D=E6=98=A8=E6=94=B6=E5=85=9C?= =?UTF-8?q?=E5=BA=95(=E9=98=BB=E6=96=AD);=20theme=E4=B8=8D=E5=81=9A?= =?UTF-8?q?=E8=A1=8C=E4=B8=9A=E6=BA=90=E6=94=B9gp=5Fstock=5Fcategory;=20ST?= =?UTF-8?q?=E5=89=94=E9=99=A4;=20=E5=90=83=E6=BB=A1top=E6=94=B9tier=5Fcomp?= =?UTF-8?q?lete=E5=88=A4=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 4 +- UPSTREAM_PLAN_API.md | 78 +++++++++++++++++++++++++++++++++ app/services/command_service.py | 26 ++++++++--- app/services/market.py | 40 +++++++++++++++++ app/services/param_store.py | 3 +- app/services/plan_feed.py | 63 +++++++++++++++++++++----- app/web/static/index.html | 19 +++++--- config/settings.py | 5 ++- scripts/probe_plan_api.py | 25 ++++++++--- scripts/run_tests.py | 6 +-- scripts/test_batch7_units.py | 39 +++++++++++++++++ scripts/test_wiring.py | 34 +++++++++++++- 12 files changed, 304 insertions(+), 38 deletions(-) diff --git a/README.md b/README.md index a930019..64e61ac 100644 --- a/README.md +++ b/README.md @@ -60,8 +60,8 @@ scripts/ test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 11 例 test_batch5_units.py 决策系统信号流解析与消化口径 8 例 test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例 - test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 30 例 - test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 51 例 + test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 32 例 + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 52 例 init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检) check_db.py 实机连通性与表结构自检 (需真实 .env) gen_keys.py ws 通道密钥: 生成 / 只取公钥(--pubkey) / PEM(--pem) / 自检(--check) diff --git a/UPSTREAM_PLAN_API.md b/UPSTREAM_PLAN_API.md index 4a84bd7..31c38ad 100644 --- a/UPSTREAM_PLAN_API.md +++ b/UPSTREAM_PLAN_API.md @@ -269,3 +269,81 @@ PMS_PLAN_TOP_N=30 # 摊开之后 > 注意: 服务器上如果还留着 `PMS_PLAN_QUERY_EXTRA=top=30`, 清掉它。同名键以显式参数为准, > 留着不影响功能, 但两个地方写同一件事迟早看走眼。 + + +--- + +## 6. 拿到上游《选股说明·下游对接》之后的修正 (2026-07-31) + +对方给了系统说明 + `api.py`, 加上当天盘前的实机跑批, 推翻了两条我们先前的判断, 也补上了 +几条只有实盘才看得见的坑。 + +### 6.1 `theme` 不能当行业标签 (推翻 07-30 的决定) + +07-30 定的是「`theme` 灌进 `pms_industry_map`, 行业源走 `custom_table`」。当天的样例里每条 +都有 theme, 看着挺好。07-31 拉 300 条真实数据, 结论反了: + +``` +[5] 主榜主题分布 (共 10 个主题): (无)×205, 整机制造×51, 储能×23, 整机×9, 数据中心×4, + 整车×3, 锂电×2, 消费类锂电池×1, 动力电池×1, 储能电池×1 +``` + +三条硬伤: + +1. **覆盖率约三成。** 300 条里 205 条没有 theme —— 上游文档写得很清楚:「无板块传导的日子 + 主榜靠"冷+便宜"排序」, 没有传导就没有主题。持仓票不在传导链上就永远没有标签, + 它的行业硬拦截会**悄悄失效**。 +2. **词表不规范。** `整机制造×51` 与 `整机×9` 并存; `锂电 / 动力电池 / 储能电池 / + 消费类锂电池` 分成四个。当行业名做集中度累计, 同一门生意会被算成好几个行业。 +3. **语义就不是行业。** 它是**传导主题** —— 事件驱动, 昨天哪个板块异动就叫什么。同一只票 + 今天算「储能」明天算「整机制造」, 集中度约束跟着一起漂。 + +**改法**: `PMS_PLAN_THEME_SYNC` 默认关; 行业源用 `PMS_SECTOR_SOURCE=gp_stock_category` +(决策系统生态已有的行业表, 语义正确、覆盖全市场、不随事件漂)。theme 保留在候选阶段的 +`PMS_PLAN_THEME_CAP_LOCAL` 上 —— 防单一传导主题刷屏, 那才是它该干的活。两个维度各干各的。 + +### 6.2 盘前没有实时价 (阻断性, 已修) + +07-31 盘前跑批, 主榜前 30 只**全部**「无价」。原因: `market.get_price` 读 Redis db13 的 +**当日**分钟线, 盘前那张 key 根本不存在。而候选池的规矩是「取不到价就不进池」—— +盘前下的升仓命令会得到一个空候选池, 而且日志上只是一行 warning。 + +已加 `market.plan_price()`: 实时价优先, 取不到回落**昨收** (日线 `close_qfq`, 按日缓存), +两者都没有才剔除。候选带 `price_source` 字段, 日志把「用昨收定量」和「彻底无价剔除」分开报。 +规划价本来就只用来算批次数量, 真正的委托价由择时环节按实时行情现定, 用昨收完全够用。 + +### 6.3 ST 要 PMS 自己剔 (新增) + +上游文档: 「名单当前**不剔除 ST / \*ST**; 需要过滤请在下游做」。已加 +`PMS_PLAN_EXCLUDE_ST` (默认开), 判据覆盖 `ST` / `*ST` / `SST` / `S*ST` 前缀、`退市` 前缀、 +`退` 后缀。名称缺失时**不误杀也不静默**: 保留该行, 但计进 `st_unknown` 由页面和探活脚本报出来。 + +### 6.4 「吃满 top」不等于漏了票 (假警报已消) + +主榜分 = 200 + 传导档位×20 + 组内分, **档位是分层的**。我们只要「强传导」, 而强传导的分 +最高、全在榜首。所以吃满 `top=300` 并不意味着漏了强传导。 + +判据换成 `tier_complete`: 返回的池子里只要出现过一条白名单之外的档位 (比如「弱传导」), +就说明白名单那几档已经全拿到了。实测 07-31: 300 条里 226 条被 tier 白名单挡掉 → 已经看见 +大量弱传导 → 强传导取全。页面横幅和探活脚本都改看这个, 不再拿「吃满 top」报警。 + +### 6.5 文档回答掉的口径 (Q2/Q4/Q7/Q8/Q9 结案) + +| 问题 | 上游答复 | +|---|---| +| Q2 `date` 语义与出单时点 | **date = 数据日 (上一交易日)**; 每交易日早晨约 **07:15 前**就绪; 周二至周六晨跑, 周一盘前读周六那期。→ PMS 的 08:40 调度位与 `STALE_TDAYS=1` 都对得上 | +| Q4 `changes` | 升降档: 覆盖新增、估值闸翻转、传导链进出, 每日几十只。「首次覆盖 / 新进传导链」本身有信息量 —— 可以做成页面提示, 列进待办 | +| Q7 `tier` 枚举 | 强传导 > 弱传导 > 无传导 | +| Q8 查不到 | 返回 **404**; 局域网内部服务, v1 无鉴权 | +| Q9 观察档无 upside | 观察档=无券商覆盖, 没有估值锚, 所以 `upside` 恒 null。置信度本就低 | +| upside 用法 | 「一致预期取 90 天内全部研报的简单平均, 且研报报喜不报忧, **绝对值偏高, 请当相对排序用**」→ 印证 PMS 的做法: 永不参与排序 (上游已把"还便宜"折进 score 的组内分, 再排一次是重复计分), 只留 `PMS_PLAN_MIN_UPSIDE` 做下限 | + +### 6.6 还要盯着的两件事 + +1. **强传导规模随市场结构波动**: 「典型 1~5 个主题、合计几只到几十只」, 无板块传导的日子 + 可能一个都没有。所以候选池空**不一定是故障**。实测 07-31 合格只有 10 只 + (强传导 74 条 ∩ PMS 侧每主题 5 只 × 储能/整机制造 2 个主题)。要不要在没有强传导的日子 + 放宽到「弱传导」, 是个策略选择, 目前保持不放宽。 +2. **赛道硬门槛启用后主榜会收窄到约三分之一** (十五五前沿产业链: 核聚变、商业航天、通信、 + 算力、人工智能、低空经济)。上游说「已建成、暂未启用, 启用后另行通知」。到时候候选池 + 规模会跳变一次, `PMS_PLAN_TOP` 与 `THEME_CAP_LOCAL` 都要重新看。 diff --git a/app/services/command_service.py b/app/services/command_service.py index 689b459..14dd1ce 100644 --- a/app/services/command_service.py +++ b/app/services/command_service.py @@ -458,8 +458,10 @@ def _candidates(view: dict) -> list: 来源由 `PMS_CANDIDATE_SOURCE` 决定: plan_api (默认, 上游 /plan 接口) / buy_plan (旧 trading_buy_plan 表) / both (并集)。三条口径: - 1. **计划不带价格** —— /plan 只回答「买什么、排第几」, 价格一律 market.get_price - 现取, 取不到就不进池并记一条 warning。旧表的 target_price 若有则直接用。 + 1. **计划不带价格** —— /plan 只回答「买什么、排第几」, 价格由 market.plan_price + 现取: 实时价优先, 盘前回落昨收 (db13 存的是当日分钟线, 盘前那张 key 不存在, + 2026-07-31 盘前实测前 30 只全部无实时价)。两者都没有才剔除并记 warning。 + 旧表的 target_price 若有则直接用。 2. **上游拿不到不回退旧表** —— 候选池宁可为空。那张表在目标架构下没有明确写入方, 拿它当事实源比没有候选更危险 (沿用「拿不到 ≠ 通过」的纪律)。 3. **白名单必须压过计划票** —— planner 只认 score 一把尺子, 而计划的 score 是 200+ @@ -472,19 +474,25 @@ def _candidates(view: dict) -> list: black = {c for c, d in sp.items() if d.get("black")} src = (param_store.get("PMS_CANDIDATE_SOURCE", plan_feed.SRC_PLAN_API) or plan_feed.SRC_PLAN_API).strip() - out, seen, noprice = [], set(), [] + out, seen, noprice, fallback = [], set(), [], [] def _push(c, score, tag, *, price=None, theme=None): if not c or c in held or c in black or c in seen: return - px = float(price or 0) or (market.get_price(c) or 0) + px, psrc = float(price or 0), "given" + if px <= 0: + pp = market.plan_price(c) + px, psrc = float(pp["price"] or 0), pp["source"] if px <= 0: noprice.append(c) return + if psrc == "prev_close": + fallback.append(c) seen.add(c) # sector 取已配置的行业源 (与持仓侧的 sector_*_map 同一套词表); theme 只作展示, - # 不参与约束 —— 行业源若是 gp_stock_category, 混入上游主题会把累计口径搞乱。 - out.append({"ts_code": c, "price": px, "score": float(score or 0), + # 不参与约束 —— theme 是"传导主题"(事件驱动、天天变、覆盖率约三成), 拿它当行业标签 + # 会让集中度约束跟着漂, 已于 2026-07-31 明确不再灌行业表 (见 UPSTREAM_PLAN_API.md §6)。 + out.append({"ts_code": c, "price": px, "price_source": psrc, "score": float(score or 0), "sector": industry.get(c), "theme": theme, "src": tag}) if src in (plan_feed.SRC_PLAN_API, plan_feed.SRC_BOTH): @@ -513,8 +521,12 @@ def _candidates(view: dict) -> list: if d.get("white"): _push(c, white_score, "whitelist") + if fallback: + logger.info("[候选池] %d 只候选无实时价, 已用昨收定量 (盘前正常): %s", + len(fallback), fallback[:10]) if noprice: - logger.warning("[候选池] %d 只候选取不到价格已剔除 (前 10): %s", len(noprice), noprice[:10]) + logger.warning("[候选池] %d 只候选实时价与昨收都取不到, 已剔除 (前 10): %s", + len(noprice), noprice[:10]) return out diff --git a/app/services/market.py b/app/services/market.py index 32ac2e0..c69e799 100644 --- a/app/services/market.py +++ b/app/services/market.py @@ -170,6 +170,46 @@ def self_calc_refs(ts_code: str, base_cost=None) -> dict: _ma_cache = {"day": None, "data": {}} +def get_last_close(ts_code: str): + """上一交易日收盘 (日线 close_qfq)。**盘前定价的兜底** —— 按日缓存。 + + db13 里是**当日**分钟线, 盘前那张 key 根本不存在, get_price 必然返回 None。 + 2026-07-31 盘前实测: 上游计划前 30 只全部"无价", 候选池会整个空掉。计划价只用来 + 算批次数量 (真正的委托价由择时环节按实时行情现定), 用昨收完全够用。 + """ + today = datetime.now().strftime("%Y%m%d") + key = f"PC:{ts_code}" + if _ma_cache["day"] != today: + _ma_cache.update({"day": today, "data": {}}) + if key in _ma_cache["data"]: + return _ma_cache["data"][key] + val = None + try: + rows = _factor_rows(ts_code, days=3) + closes = [float(r["close_qfq"]) for r in rows if r.get("close_qfq")] + if closes: + val = round(closes[-1], 3) + except Exception as e: + logger.warning("昨收取数失败 [%s]: %s", ts_code, e) + _ma_cache["data"][key] = val + return val + + +def plan_price(ts_code: str) -> dict: + """规划用价: 实时价优先, 盘前/停更回落昨收。返回 {price, source}。 + + source: realtime / prev_close / none —— **一定要带出来**。拿昨收当现价去做"不追高" + 这类判断会出错, 所以只给规划期定量用, 判断类的检查仍走 day_snapshot。 + """ + px = get_price(ts_code) + if px: + return {"price": px, "source": "realtime"} + px = get_last_close(ts_code) + if px: + return {"price": px, "source": "prev_close"} + return {"price": None, "source": "none"} + + def get_ma5(ts_code: str): """MA5 (规则闸「不追高」用)。按日缓存 —— 因子分表是日频数据, 盘中重复查没意义。""" today = datetime.now().strftime("%Y%m%d") diff --git a/app/services/param_store.py b/app/services/param_store.py index 8673b57..017dadb 100644 --- a/app/services/param_store.py +++ b/app/services/param_store.py @@ -77,7 +77,8 @@ DESC = { "PMS_PLAN_MIN_SOURCES": "候选 evidence.n_sources 下限, 0=不设", "PMS_PLAN_MIN_UPSIDE": "候选预期空间下限 (相对现价, 0.5=+50%), 0=不设。只过滤不参与排序; 设了就会把 upside 缺失的行(含整个观察档)一起挡掉", "PMS_PLAN_STALE_TDAYS": "计划日龄超此交易日数即判过期拒用 (防上游停更时拿旧榜当今天)", - "PMS_PLAN_THEME_SYNC": "刷新计划时把 evidence.theme 灌进 pms_industry_map (行业源 custom_table 的数据来源)", + "PMS_PLAN_EXCLUDE_ST": "剔除 ST/*ST/退市整理 (上游名单明确不剔除, 这道闸在 PMS 侧)", + "PMS_PLAN_THEME_SYNC": "把 theme 灌进 pms_industry_map。默认关 —— theme 是事件驱动的传导主题(覆盖约三成、词表不规范), 当行业标签会让集中度约束天天漂; 行业源请用 gp_stock_category", "PMS_PLAN_QUERY_EXTRA": "计划接口附加查询串逃生口 (上游加了新参数时不用改代码); top/obs_top/theme_cap 请用各自的显式参数, 同名键以显式参数为准", "PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / custom_table (推荐, 由上游计划的 theme 灌数) / gp_stock_category", "PMS_SECTOR_MAX_NAMES": "同行业最大持仓只数 (硬拦截)", diff --git a/app/services/plan_feed.py b/app/services/plan_feed.py index 26d946b..bfa8bb1 100644 --- a/app/services/plan_feed.py +++ b/app/services/plan_feed.py @@ -30,10 +30,12 @@ 到 +212% —— 所以**永不参与排序** (排序始终是 score), 只提供下限过滤 `PMS_PLAN_MIN_UPSIDE`。 -行业约束的数据源也从这里来: `evidence.theme` 随每次刷新 upsert 进 `pms_industry_map`, -行业源仍是 `custom_table` (`PMS_SECTOR_SOURCE=custom_table`)。落库而不是即时查, 是因为 -即时查有个洞 —— 今天没上榜的持仓票查不到 theme, 行业硬拦截就对它悄悄失效了; 落库以后 -覆盖面随时间累积, 且页面可见可手改。 +**`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` 上 —— 防单一传导主题刷屏, 那才是它擅长的事。 模块级只依赖 stdlib + `app.core.command_spec` (纯逻辑), 其余 (requests / param_store / pms_repo / tradedays) 一律函数内懒加载 —— 让解析与筛选这两段纯逻辑可以零依赖单测。 @@ -201,7 +203,8 @@ def assert_fresh(plan: dict, *, max_stale_tdays: int = 1, today=None) -> int: # ================================================================ 纯逻辑: 筛选 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) -> dict: + min_sources: int = 0, min_upside=None, theme_cap: int = 0, + exclude_st: bool = True) -> dict: """排序池 → 候选清单。 排序: score 降序, 同分按 rank 升序 (上游 rank 已是它自己的最终次序, 拿来当稳定次序)。 @@ -209,6 +212,10 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N 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 名可能全是储能, 切完再交给规则闸, @@ -228,8 +235,8 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N theme_cap = int(theme_cap or 0) dropped = {"held": 0, "black": 0, "tier": 0, "score": 0, "sources": 0, "upside": 0, - "theme": 0, "dup": 0, "capped": 0} - passed, seen, per_theme = [], set(), {} + "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: @@ -242,6 +249,13 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N 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 @@ -270,7 +284,35 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N "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} + "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 [])) # ================================================================ 取数 (带缓存) @@ -292,11 +334,12 @@ def _params() -> dict: "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", True), + "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(), } @@ -457,7 +500,7 @@ def candidates(*, held=(), black=()) -> dict: 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"]) + 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 diff --git a/app/web/static/index.html b/app/web/static/index.html index a1a366d..6ef198e 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -562,15 +562,19 @@ →(券商预期 + 目标价≥现价 + 主题限额 + top)→ 实收 主榜 {{ (planStatus.funnel||{}).returned_main }} / 观察 {{ (planStatus.funnel||{}).returned_observe }} - +
按当前参数筛选: 排序池 {{ planCand.considered }} → 合格 {{ planCand.eligible }} → - 取 {{ (planCand.items||[]).length }} 只 · 丢弃 {{ JSON.stringify(planCand.dropped||{}) }} - {{ (planCand.items||[]).map(x=>x.ts_code).join(' ') }} + 取 {{ (planCand.items||[]).length }} 只 · + 白名单档位取全 {{ planCand.tier_complete===null ? '不适用' : (planCand.tier_complete ? '是':'否') }} · + 丢弃 {{ JSON.stringify(planCand.dropped||{}) }} + · + {{ planCand.st_unknown.length }} 只无名称, ST 判不了已放行 +
{{ (planCand.items||[]).map(x=>x.ts_code).join(' ') }}
@@ -596,6 +600,9 @@
发给上游的三个: PMS_PLAN_TOP / PMS_PLAN_OBS_TOP / PMS_PLAN_THEME_CAP (0=不传, 用上游默认 20/10/5)。PMS_PLAN_THEME_CAP_LOCAL 是 PMS 侧的同主题限额, 在 TOP_N 截断之前生效 —— 让上游别裁 (THEME_CAP 设大) + PMS 自己裁, 前 N 名才不会被单一主题垄断。 +
PMS_PLAN_EXCLUDE_ST 默认开: 上游名单明确不剔除 ST/*ST, 这道闸在 PMS 侧。 + 「主题」是传导主题(事件驱动、天天变、覆盖约三成), 不是行业 —— 行业硬拦截 + 请把 PMS_SECTOR_SOURCE 设成 gp_stock_category, 别用 PMS_PLAN_THEME_SYNC。 
{{ planResult }}
diff --git a/config/settings.py b/config/settings.py index 01d9bc0..a4985e8 100644 --- a/config/settings.py +++ b/config/settings.py @@ -97,7 +97,10 @@ class Settings(BaseSettings): PMS_PLAN_MIN_UPSIDE: float = 0.0 # 预期空间下限 (0.5=+50%), 0=不设。券商目标价 # 口径, 噪音大 (榜首能到 +212%), 只做下限过滤, **不参与排序** —— 排序始终是 score PMS_PLAN_STALE_TDAYS: int = 1 # 计划日龄超此交易日数即判过期并拒用 (防上游停更) - PMS_PLAN_THEME_SYNC: bool = True # 刷新时把 evidence.theme 灌进 pms_industry_map + PMS_PLAN_EXCLUDE_ST: bool = True # 剔除 ST/*ST/退市 (上游明确不剔除, 这道闸归 PMS) + PMS_PLAN_THEME_SYNC: bool = False # 把 theme 灌进 pms_industry_map。**默认关**: + # theme 是事件驱动的传导主题, 覆盖率约三成且词表不规范, 当行业标签会让集中度约束 + # 天天漂 —— 行业源请用 PMS_SECTOR_SOURCE=gp_stock_category PMS_PLAN_QUERY_EXTRA: str = "" # 附加查询串逃生口, 如 "foo=1"。上游哪天加了新参数 # 不用改代码即可透传; top/obs_top/theme_cap 已有显式参数, 同名键以显式参数为准 diff --git a/scripts/probe_plan_api.py b/scripts/probe_plan_api.py index 598f473..e9576af 100644 --- a/scripts/probe_plan_api.py +++ b/scripts/probe_plan_api.py @@ -198,8 +198,8 @@ def main(): f" →(券商预期 + 目标价≥现价 + 主题限额 + top)→" f" 实收 主榜 {f['returned_main']} / 观察 {f['returned_observe']}") if plan["truncated"]["main"]: - print(f" ! 主榜正好吃满 top={plan['requested']['top']} —— 可能还有更多, " - f"调大 PMS_PLAN_TOP 再看") + print(f" · 主榜正好吃满 top={plan['requested']['top']} —— 要不要紧看 [6] 的" + f"「白名单档位取全」那一行") else: print(f" 主榜没吃满 top={plan['requested']['top']}, 说明这就是上游能给的全部" f" (受主题限额 {plan['theme_cap']} 与价格筛限制)") @@ -221,7 +221,10 @@ def main(): from app.services import market for r in rows: try: - prices[r["ts_code"]] = market.get_price(r["ts_code"]) + pp = market.plan_price(r["ts_code"]) + prices[r["ts_code"]] = ((f"{pp['price']}" if pp["source"] == "realtime" + else f"{pp['price']} (昨收)") if pp["price"] + else "**无价(会被剔除)**") except Exception as e: prices[r["ts_code"]] = f"ERR {type(e).__name__}" for r in rows: @@ -229,8 +232,7 @@ def main(): f"{_fmt(r['score'], 2):>8} {(r['tier'] or '观察'):<6} " f"{(r['theme'] or '-'):<10} {_fmt(r['heat']):>7} {_pct(r['upside']):>9}") if args.with_price: - p = prices.get(r["ts_code"]) - line += f" {p if p else '**无价(会被剔除)**'}" + line += f" {prices.get(r['ts_code']) or '-'}" print(line) themes = {} @@ -246,12 +248,21 @@ def main(): min_score=(ps.get_float("PMS_PLAN_MIN_SCORE", 0.0) or None), min_sources=ps.get_int("PMS_PLAN_MIN_SOURCES", 0), min_upside=(ps.get_float("PMS_PLAN_MIN_UPSIDE", 0.0) or None), - theme_cap=ps.get_int("PMS_PLAN_THEME_CAP_LOCAL", 5)) + theme_cap=ps.get_int("PMS_PLAN_THEME_CAP_LOCAL", 5), + exclude_st=ps.get_bool("PMS_PLAN_EXCLUDE_ST", True)) print(f"[6] 按当前参数筛选 (top_n={ps.get_int('PMS_PLAN_TOP_N', 30)} " f"tiers={ps.get_list('PMS_PLAN_TIERS', [])} " f"observe={ps.get_bool('PMS_PLAN_INCLUDE_OBSERVE', False)} " - f"PMS侧主题限额={ps.get_int('PMS_PLAN_THEME_CAP_LOCAL', 5)})") + f"PMS侧主题限额={ps.get_int('PMS_PLAN_THEME_CAP_LOCAL', 5)} " + f"剔ST={ps.get_bool('PMS_PLAN_EXCLUDE_ST', True)})") + tc = sel.get("tier_complete") + tcs = ("是 —— 已看到白名单之外的档位, 后面再多也不影响候选池" if tc is True else + ("**否** —— 返回的全是白名单档位, 说明被 top 卡住了, 调大 PMS_PLAN_TOP" if tc is False + else "不适用 (没设档位白名单)")) + print(f" 白名单档位取全: {tcs}") print(f" 排序池 {sel['considered']} → 合格 {sel['eligible']} → 取 {len(sel['items'])} 只") + if sel.get("st_unknown"): + print(f" ! {len(sel['st_unknown'])} 只没有名称, ST 判不了已放行: {sel['st_unknown'][:8]}") print(f" 丢弃明细 {sel['dropped']} (此处未扣持仓/黑名单, 下命令时还会再扣)") st = {} for x in sel["items"]: diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 99755a1..31c82a9 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -11,9 +11,9 @@ test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 (11 例) test_batch5_units.py 决策系统信号流解析与消化口径 (8 例) test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 (65 例) - test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 (30 例) - test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (51 例) - 共 235 例 + test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 (32 例) + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (52 例) + 共 238 例 任一子集失败即整体失败 (退出码 1)。 """ import os diff --git a/scripts/test_batch7_units.py b/scripts/test_batch7_units.py index 225fb9f..68bd8f3 100644 --- a/scripts/test_batch7_units.py +++ b/scripts/test_batch7_units.py @@ -323,6 +323,45 @@ def _(): assert [x["ts_code"] for x in r["items"]] == ["600001.SH", "600002.SH"], r["items"] +@case("筛选·ST 剔除 (上游明确不剔, 这道闸归 PMS): ST/*ST/退市 都要挡, 名字缺失不误杀") +def _(): + assert pf.is_st("ST喜临门") and pf.is_st("*ST辉丰") and pf.is_st("S*ST前锋") + assert pf.is_st("SST华塑") and pf.is_st("*st 辉丰") # 大小写与空格都要吃掉 + assert pf.is_st("退市海润") and pf.is_st("海润退") # 退市整理期两种写法 + assert not pf.is_st("江淮汽车") and not pf.is_st("") and not pf.is_st(None) + # 「圣农发展」「深赛格」这类 S 开头的中文名不许误杀 (A 股名称都是中文, 拉丁 ST 前缀安全) + assert not pf.is_st("圣农发展") and not pf.is_st("深赛格") and not pf.is_st("三安光电") + d = {"date": "2026-07-29", "main": [ + _m(1, "SH600001", "ST喜临门", 240.0, "整机", 0.1, 1.0), + _m(2, "SH600002", "*ST辉丰", 239.0, "整机", 0.1, 1.0), + _m(3, "SH600003", "江淮汽车", 238.0, "整车", 0.1, 1.0), + {"rank": 4, "code": "SH600004", "score": 237.0, "tier": "强传导", + "evidence": {"theme": "整车"}}]} # 没有 name + p = pf.parse_plan(d) + r = pf.select_candidates(p, top_n=10, exclude_st=True) + assert [x["ts_code"] for x in r["items"]] == ["600003.SH", "600004.SH"], r["items"] + assert r["dropped"]["st"] == 2 + assert r["st_unknown"] == ["600004.SH"], r["st_unknown"] # 判不了要报出来, 不静默 + assert len(pf.select_candidates(p, top_n=10, exclude_st=False)["items"]) == 4 + + +@case("筛选·白名单档位取全判据: 看见白名单之外的档位才算齐 (吃满 top 不等于漏了)") +def _(): + strong = [_m(i, f"SH60{i:04d}", f"强{i}", 240 - i, "整车", 0.1, 1.0) for i in range(1, 4)] + weak = [_m(9, "SH609999", "弱", 220.0, "整车", 0.1, 1.0, tier="弱传导")] + # 只有强传导 → 判不出后面还有没有, tier_complete=False (被 top 卡住了) + only_strong = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}), + tiers=["强传导"], top_n=10) + assert only_strong["tier_complete"] is False + # 出现弱传导 → 强传导已经全在前面了 + both = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong + weak}), + tiers=["强传导"], top_n=10) + assert both["tier_complete"] is True + # 没设白名单时无从判断 + assert pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}), + top_n=10)["tier_complete"] is None + + @case("筛选·候选级主题限额在 top_n 截断之前生效 (否则前 N 名被单一主题垄断)") def _(): # 造一个"宽池子": 储能 6 只分最高, 传感器 3 只, 整车 2 只 diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index 29b5111..4a193ea 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -419,7 +419,7 @@ class FakeQmtRepo: return None -def install_fakes(prices=None, positions=None, params=None, high5=None): +def install_fakes(prices=None, positions=None, params=None, high5=None, prev_close=None): """把内存桩装到各模块上, 返回 FakeRepo 实例 (ws 通道桩挂在 .qmt 上)。""" from app.repo import downstream_repo, pms_repo, qmt_repo from app.services import industry, market, param_store, portfolio @@ -449,6 +449,13 @@ def install_fakes(prices=None, positions=None, params=None, high5=None): market.get_refs = lambda c, **kw: {"support": None, "pressure": None, "stop": None, "source": "none"} market.get_ma5 = lambda c: (prices or {}).get(c) + # plan_price 是规划期定价 (实时价 → 昨收兜底)。桩不接库, 一律按"有实时价"处理; + # 昨收那条路单独给 prev_close 参数的用例覆盖。 + market.get_last_close = lambda c: (prev_close or {}).get(c) + market.plan_price = lambda c: ( + {"price": (prices or {}).get(c), "source": "realtime"} if (prices or {}).get(c) + else ({"price": (prev_close or {}).get(c), "source": "prev_close"} + if (prev_close or {}).get(c) else {"price": None, "source": "none"})) market.get_high5 = lambda c: (high5 or prices or {}).get(c) market.day_snapshot = lambda c: ({} if not (prices or {}).get(c) else { "price": prices[c], "vwap": prices[c], "open": prices[c], "high": prices[c] * 1.02, @@ -1710,6 +1717,31 @@ def _(): assert by["600000.SH"]["score"] > plan_max, (by["600000.SH"]["score"], plan_max) +@case("候选池·盘前无实时价回落昨收 (db13 存的是当日分钟线, 盘前那张 key 不存在)") +def _(): + from app.services import command_service as csvc, plan_feed + install_fakes(prices={}, prev_close={"600418.SH": 12.5, "300952.SZ": 30.0}, + params={"PMS_CANDIDATE_SOURCE": "plan_api"}) + restore_sp = _stub_sp(csvc, {}) + orig = plan_feed.candidates + try: + plan_feed.candidates = lambda **kw: { + "date": "2026-07-30", "considered": 3, "eligible": 3, "dropped": {}, + "items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"}, + {"ts_code": "300952.SZ", "score": 241.91, "theme": "传感器"}, + {"ts_code": "688717.SH", "score": 242.05, "theme": "储能"}]} + pool = csvc._candidates({"held": []}) + finally: + plan_feed.candidates = orig + restore_sp() + by = {c["ts_code"]: c for c in pool} + # 有昨收的进池并标明价格来源; 实时价与昨收都没有的才剔除 + assert set(by) == {"600418.SH", "300952.SZ"}, by + assert by["600418.SH"]["price"] == 12.5 + assert by["600418.SH"]["price_source"] == "prev_close", by["600418.SH"] + assert "688717.SH" not in by + + @case("候选池·上游计划不可用: 池为空, 绝不静默回退旧 trading_buy_plan 表") def _(): from app.repo import downstream_repo