diff --git a/UPSTREAM_PLAN_API.md b/UPSTREAM_PLAN_API.md index 5223696..2cb85fb 100644 --- a/UPSTREAM_PLAN_API.md +++ b/UPSTREAM_PLAN_API.md @@ -662,3 +662,13 @@ make changes # 页面同款结论 隔一天再跑一次 `make changes`, 就能看到第一份真正的跨日比对 —— 那时候才验得到 `cross_day` 与持仓票视图。账本还空着, 所以持仓那段现在必然是空的, 等账本重建后才有内容。 + +## 10. 排序主轴换成公司质地与前缀效应 (2026-09-17, 台账 016) + +上游 2026-09-17 把选股计划的排序主轴从传导档位换成公司质地。对 PMS 的接入有三件事要知道。 + +**多了几个键。** 每行多 `quality_band` (质地带: 好 3 / 中 2 / 无评析 1 / 差 0)、`pointed` (是否被产业链指向)、`rank_axis` (quality 或 transmission)、`rank_old` (旧序名次, 供对照); 顶层多 `rank_axis` 与 `rank_rule` (一句话次序规则)。旧计划、或上游回退到传导轴时这些键缺, 即 None, `_rows` 与 `parse_plan` 按缺就是空处理。 + +**候选按上游名次排 (自门控)。** 计划顶层 `rank_axis` 为 quality 时, `select_candidates` 按上游名次升序排、关掉 PMS 自己的基本面立场分桶 (参数 `PMS_PLAN_RANK_BY_UPSTREAM`, 默认开); 上游没带这个键时无作用, 逐字回到按立场分桶排。这样选股序、入池、PMS 候选三处同一个序。本次用哪种序记在候选返回与名册快照元数据的 `rank_mode` 里 (upstream / quality_bucket / score)。 + +**前缀效应 —— 强传导票变少不是故障。** 质地序把好公司排到前面, 质地差但强传导的票会往后掉, 可能落到我们要的三百只主榜前缀之外, 于是这批票在候选里变少甚至消失。这与"质地差降关注"同向, 是预期结果, 不要当成上游漏发或接口出错。要看全量强传导票, 去上游的逐票对账接口 `/plan/verdict` (它的"主榜第 N 名"仍按原始分序, 应答带 `rank_axis: transmission` 说明), 或把 `PMS_PLAN_RANK_BY_UPSTREAM` 关掉临时回到按分桶排。 diff --git a/app/services/param_store.py b/app/services/param_store.py index a617e02..dbb949d 100644 --- a/app/services/param_store.py +++ b/app/services/param_store.py @@ -117,6 +117,8 @@ RUNTIME_EXTRA = { # ── 2026-09-11 三源合议工作包二·接入 (台账 006/009): 持仓增持门与候选按质地排序。 "PMS_TECH_GATE_INCREASE": (True, bool, "持仓增持门: 合议看空停回踩补足/盈利加仓/补仓, 技术面看空停回踩补足与盈利加仓, 基本面看空停补仓 (减持不受影响; 关掉即回到不设门)"), "PMS_PLAN_RANK_BY_QUALITY": (True, bool, "候选先按基本面立场分桶 (质地看多的桶排前) 再按分数, 在主题限额与截断之前生效; 技术面不参与排序 (关掉即回到只按分数排)"), + # ── 2026-09-17 上线协同阶段工作包乙 (台账 016): 上游选股计划排序主轴换成公司质地后, 候选按上游名次排。 + "PMS_PLAN_RANK_BY_UPSTREAM": (True, bool, "上游计划带 rank_axis=quality 时按上游名次升序排、关掉自己的立场分桶; 上游没带这个键 (旧计划或选股系统回退) 时无作用 (关掉即回到按 PMS_PLAN_RANK_BY_QUALITY 的分桶排)"), "PMS_PLAN_OBSERVE_IF_FUND_BULL": (False, bool, "把观察档里质地看多的票也纳入候选池 (默认关, 守宁缺毋滥; 开了它们与主榜一同参与分桶排序)"), # ── 2026-09-11 三源合议工作包三·离场纪律 (台账 010): SAR 转空自动离场。翻向新鲜天数复用 PMS_TECH_FLIP_FRESH_DAYS。 "PMS_TECH_EXIT_AUTONOMY": ("full", str, "技术面转空离场档位: off 不评 / propose_only 交人确认 / full 自动执行 (与保垫减仓同档, 按既定默认)"), diff --git a/app/services/plan_feed.py b/app/services/plan_feed.py index d4aa486..912b544 100644 --- a/app/services/plan_feed.py +++ b/app/services/plan_feed.py @@ -332,6 +332,11 @@ def _rows(raw, bucket: str) -> list: "company_review": _company_review_or_none(it.get("company_review")), "company_review_text": _text_or_none(it.get("company_review_text")), "company_gate": _text_or_none(it.get("company_gate")), + # 排序轴四键 (2026-09-17 工作包乙): 上游质地序带出来的键, 原样透传, 旧计划缺就是 None。 + "quality_band": _int_or_none(it.get("quality_band")), + "pointed": (bool(it.get("pointed")) if it.get("pointed") is not None else None), + "rank_axis": _text_or_none(it.get("rank_axis")), + "rank_old": _int_or_none(it.get("rank_old")), }) return out @@ -376,6 +381,10 @@ def parse_plan(payload, *, requested=None) -> dict: "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")), + # 排序轴 (2026-09-17 工作包乙): 上游主轴名与一句话次序规则。按上游名次排的自门控只看 rank_axis; + # 旧计划、或选股系统回退到 transmission 时缺这个键 (即 None), 自门控自动不生效。 + "rank_axis": _text_or_none(payload.get("rank_axis")), + "rank_rule": _text_or_none(payload.get("rank_rule")), "counts": {"main": _int_or_none(counts.get("main")), "observe": _int_or_none(counts.get("observe")), "gate_covered": _int_or_none(counts.get("gate_covered"))}, @@ -437,11 +446,22 @@ def _fund_stance_of(row) -> str: return "无读数" +def _rank_mode(plan: dict, rank_by_upstream: bool, rank_by_quality: bool) -> str: + """本次候选与名册用的排序模式（2026-09-17 工作包乙）： + upstream = 上游是质地轴（计划顶层 rank_axis=quality）且开了按上游名次排，此时按上游名次升序、立场桶不参与； + quality_bucket = 上游没给质地轴、但开了 PMS 自己的立场分桶；score = 都没开，纯原始分降序。 + 选股系统周报据名册元数据里这一键还原当天交付名单的序。""" + if rank_by_upstream and (plan or {}).get("rank_axis") == "quality": + return "upstream" + return "quality_bucket" if rank_by_quality else "score" + + 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, - rank_by_quality: bool = False, observe_if_fund_bull: bool = False) -> dict: + rank_by_quality: bool = False, rank_by_upstream: bool = False, + observe_if_fund_bull: bool = False) -> dict: """排序池 → 候选清单。 route_by_verdict: 按上游判决分流时, 判为「仅展示」的行在 top_n 截断**之前**剔掉并列进 @@ -487,7 +507,14 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N passed, seen, per_theme, st_unknown, display_only = [], set(), {}, [], [] # 排序: 默认按分数降序、同分按 rank 升序 (与 2026-09-11 之前逐字相同)。开了按质地排序后, 先按 # 基本面立场分桶 (质地看多的桶排在前) 再按分数, 在主题限额与截断之前生效; 技术面不参与排序。 - if rank_by_quality: + # 2026-09-17 工作包乙: 上游是质地轴且开了按上游名次排时, 直接按上游名次升序 (同名次原始分降序), + # 立场桶不参与 —— 选股序、入池、这里三处同一个序, 四周复核才说得清。上游没给质地轴 (旧计划或回退) + # 时自动落回下面两支, 逐字不动。 + mode = _rank_mode(plan, rank_by_upstream, rank_by_quality) + if mode == "upstream": + def _sort_key(x): + return (x.get("rank") or 10 ** 9, -(x.get("score") or 0.0)) + elif rank_by_quality: def _sort_key(x): return (0 if _fund_stance_of(x) == "看多" else 1, -(x.get("score") or 0.0), x.get("rank") or 10 ** 9) @@ -563,11 +590,14 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N "industry_catalyst": r.get("industry_catalyst"), "industry_catalyst_text": r.get("industry_catalyst_text"), "company_review": r.get("company_review"), "company_review_text": r.get("company_review_text"), - "company_gate": r.get("company_gate")} + "company_gate": r.get("company_gate"), + # 排序轴四键 (2026-09-17 工作包乙): 原样透传给动作引擎与提议卡, 缺就是 None。 + "quality_band": r.get("quality_band"), "pointed": r.get("pointed"), + "rank_axis": r.get("rank_axis"), "rank_old": r.get("rank_old")} 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, + "display_only": display_only, "rank_mode": mode, "tier_complete": _tier_complete(pool, tiers)} @@ -632,6 +662,9 @@ def _params() -> dict: "route_by_verdict": ps.get_bool("PMS_PLAN_ROUTE_BY_VERDICT", True), # 三源合议工作包二 (2026-09-11): 候选先按基本面立场分桶再按分数; 观察档里质地看多的行是否进池。 "rank_by_quality": ps.get_bool("PMS_PLAN_RANK_BY_QUALITY", True), + # 上游名次排 (2026-09-17 工作包乙): 上游计划 rank_axis=quality 时按上游名次排、关掉立场桶; + # 上游没带这个键时无作用。默认开, 靠上游的键自门控, 上游切质地序之前它什么都不改。 + "rank_by_upstream": ps.get_bool("PMS_PLAN_RANK_BY_UPSTREAM", True), "observe_if_fund_bull": ps.get_bool("PMS_PLAN_OBSERVE_IF_FUND_BULL", False), } @@ -828,7 +861,9 @@ def get_plan(*, force: bool = False, date=None) -> dict: if p["theme_sync"]: plan["theme_sync"] = _sync_themes_quiet(plan) if p["snapshot"]: - plan["snapshot"] = _snapshot_quiet(plan, keep=p["snapshot_keep"]) + plan["snapshot"] = _snapshot_quiet( + plan, keep=p["snapshot_keep"], + rank_mode=_rank_mode(plan, p["rank_by_upstream"], p["rank_by_quality"])) with _lock: _cache.update({"at": time.time(), "key": key, "plan": plan, "error": None}) logger.info("[上游计划] %s 主榜 %d / 观察 %d (日龄 %d 交易日) ← %s", @@ -869,7 +904,7 @@ def _sync_themes_quiet(plan: dict) -> dict: # ================================================================ 名册快照与榜单变化 -def snapshot(plan: dict, *, keep: int = 200) -> dict: +def snapshot(plan: dict, *, keep: int = 200, rank_mode=None) -> dict: """把这一份计划的名册落进 pms_plan_snapshot。 同一份榜重复拉不重复落 (靠 uk_date_digest)。返回里带 `stored` —— 落进去了才是新版本, @@ -883,6 +918,9 @@ def snapshot(plan: dict, *, keep: int = 200) -> dict: "theme_cap", "heat_date", "market_snapshot_days", "age_tdays", "url", "plan_source", "snapshot_generated_at", "gate_on")} + # 排序模式 (2026-09-17 工作包乙): 让选股系统周报按它还原当天交付名单的序。旧调用没传即不写。 + if rank_mode is not None: + meta["rank_mode"] = rank_mode 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, @@ -897,7 +935,7 @@ def snapshot(plan: dict, *, keep: int = 200) -> dict: return out -def _snapshot_quiet(plan: dict, *, keep: int = 200) -> dict: +def _snapshot_quiet(plan: dict, *, keep: int = 200, rank_mode=None) -> dict: """快照落库失败**不许阻断候选池** —— 没有变化提示是可接受的降级, 没候选不是。 同 `_sync_themes_quiet` 的口径: 这是个"锦上添花"的旁路, 它的故障不该传染主链路。 @@ -905,7 +943,7 @@ def _snapshot_quiet(plan: dict, *, keep: int = 200) -> dict: (拿旧版本当上一版比)。 """ try: - return snapshot(plan, keep=keep) + return snapshot(plan, keep=keep, rank_mode=rank_mode) except Exception as e: logger.warning("[上游计划] 名册快照落库失败 (榜单变化提示将不可用): %s", e) return {"stored": False, "error": f"{type(e).__name__}: {e}"} @@ -1091,6 +1129,7 @@ def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict: theme_cap=p["theme_cap_local"], exclude_st=p["exclude_st"], route_by_verdict=route, rank_by_quality=p["rank_by_quality"], + rank_by_upstream=p["rank_by_upstream"], observe_if_fund_bull=p["observe_if_fund_bull"]) out["age_tdays"] = plan.get("age_tdays") out["theme_cap"] = plan.get("theme_cap") diff --git a/docs/复盘决定台账.md b/docs/复盘决定台账.md index fad32e1..18cb228 100644 --- a/docs/复盘决定台账.md +++ b/docs/复盘决定台账.md @@ -233,3 +233,19 @@ **复核日期。** 部署后的下一个交易日盘中先只读试算一次，看 executed 里有没有 route 为 auto、why 含「系统采纳」的条目，或 rejected 里有没有评审方 system 的条目。收盘后跑 make consensus-review 看第五张表系统自决逐条、第六张表研判不可用占比。 **评审补记（2026-09-14 下午，同日评审后修订）。** 四处补齐，不改上面的决定。一，到价自动清仓前停买入侧改成与命令清仓同样的三件：撤方案、驳回该票待确认的买入提议、撤该票在途的买入指令；原来只撤方案，已下发的网格买单会与清仓单对倒。不再只在有活动方案时才做，一张在途补仓买单同样要撤。二，只有一方表态改按合议强弱打标记：强弱为弱一律按试探仓，不看路由；原来只在路由为交人时打，弱表态开关 PMS_CONSENSUS_WEAK_CONFIRM 关掉后会按标准仓采纳。三，自动采纳只认带应答体的真通过：动作不在研判范围时研判直接回通过且没有应答体，这种一律交人。四，decline 与 full 档下逻辑存疑的新建仓不送研判，直接放弃，省一次研判调用与时间预算；record 档照旧送研判。页面加两种处置词：系统已采纳、系统已放弃。 + +## 016 · 2026-09-17 · 候选按上游名次排：上游排序主轴换成公司质地后，选股序、入池、PMS 候选三处同序 + +**改动。** 上游选股系统 2026-09-17 把选股计划的排序主轴从传导档位换成公司质地，计划顶层带 rank_axis=quality、每行带 quality_band/pointed/rank_old。PMS 的候选选择在上游是质地轴时改按上游名次升序排、关掉自己的基本面立场分桶，让选股序、入池、PMS 候选三处是同一个序。上游没带 rank_axis（旧计划或选股系统回退到传导轴）时逐字回到按立场分桶排，一处不改。 + +**做了哪几件。** plan_feed 的 _rows 与 parse_plan 归一时带上四个新键与顶层 rank_axis、rank_rule（缺就是空）；select_candidates 加参数 rank_by_upstream，生效条件是它为真且计划 rank_axis=quality，生效时排序键为上游名次升序、同名次原始分降序，立场桶不参与；返回与名册快照元数据都写一键 rank_mode（upstream / quality_bucket / score），选股系统周报据它还原当天交付名单的序。参数 PMS_PLAN_RANK_BY_UPSTREAM 默认开、靠上游的键自门控。选股系统那侧 plan_review.pms_roster 读到 rank_mode 为 upstream 时按名次还原（另一仓库单独提交）。 + +**依据。** 入池已改成"新序里被指向的前五十"、选股系统候选单已是质地序；PMS 若仍按自己的立场桶排，三处里就它一处是另一个序，四周复核说不清它拿到的前三十为什么与选股系统不一样。立场桶本来就是没有上游质地序时的替代品，上游有了就退场。 + +**预期。** 09-22 上游切质地序后，PMS 候选序等于上游名次序（09-24 早上核 rank_mode 为 upstream、名册前十名次严格递增）。质地差的强传导票可能落到三百只前缀之外，与"质地差降关注"同向，是预期，不是故障。 + +**开关与回退。** PMS_PLAN_RANK_BY_UPSTREAM 关掉即回到按 PMS_PLAN_RANK_BY_QUALITY 的立场分桶排；上游一回退（rank_axis 不再是 quality）它自动回旧，不用动 PMS。off 逐字回旧由 test_batch29 的两支断言守。 + +**上线。** 09-19 交付、我复审后 09-23 收盘后重启 PMS 进程（重建镜像加 force-recreate，经用户同意）。不与 09-21 选股系统拉代码同一天，避免两处同时变化说不清。 + +**复核日期。** 上线次日 09-24 核候选序等于上游名次序；四周（10-16 前后）随选股系统排序轴一起复核。 diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 3a76e2f..583b172 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -81,7 +81,8 @@ test_batch29_units.py 三源合议·工作包二接入下单链路 (2026-09-11): scan_open 合议分流(跳过/观察带 wait_tech/交人强制确认/放行/缺合议退旧路/开关关掉逐字回旧)/定档改用 advise_v2 且六键进硬数字而不进送研判白名单/持仓增持门矩阵与 scan 端到端(减持不受影响、 - 开关关掉回旧)/装配服务持仓挂合议/候选按质地分桶排序与观察档进池开关 (24 例) + 开关关掉回旧)/装配服务持仓挂合议/候选按质地分桶排序与观察档进池开关/ + 上游名次排(rank_axis=quality 按上游名次、旧计划回分桶逐字不变、rank_mode 三态; 2026-09-17 工作包乙) (27 例) test_batch30_units.py 三源合议·工作包三离场纪律 (2026-09-11): eval_tech_exit 转空自动离场(确认清仓/ 未确认减三分之一/翻向超期不评/同一翻空一次/无读数弃权/propose_only 交人/可卖量 夹紧/开关关掉逐字回旧)/卖出优先级目标价<研究走弱<技术面转空<保垫减仓/scan 端到端/ diff --git a/scripts/test_batch29_units.py b/scripts/test_batch29_units.py index 83247d0..210a6ff 100644 --- a/scripts/test_batch29_units.py +++ b/scripts/test_batch29_units.py @@ -337,6 +337,42 @@ def _(): assert ps.RUNTIME_EXTRA["PMS_PLAN_OBSERVE_IF_FUND_BULL"][0] is False # 守宁缺毋滥, 默认关 +# ================================================================ G 上游名次排 (2026-09-17 工作包乙) +@case("G 上游名次排·rank_axis=quality 且开关开: 按上游名次升序、立场桶不参与、rank_mode=upstream") +def _(): + # 高分好票名次靠后(rank=3)、低分无评析名次靠前(rank=1): upstream 按名次排 → 无评析在前, 证明立场桶不参与 + plan = {"date": "2026-09-17", "rank_axis": "quality", + "main": [_row("600000.SH", 100, "好", rank=3), _row("000001.SZ", 60, rank=1)], "observe": []} + out = pf.select_candidates(plan, rank_by_upstream=True, rank_by_quality=True) + assert [x["ts_code"] for x in out["items"]] == ["000001.SZ", "600000.SH"] + assert out["rank_mode"] == "upstream" + + +@case("G 上游名次排·上游没带 rank_axis: 逐字回按分桶排 (rank_by_quality 生效, rank_mode=quality_bucket)") +def _(): + plan = {"date": "2026-09-17", # 无 rank_axis 键 (旧计划或选股系统回退) + "main": [_row("600000.SH", 100, rank=1), _row("000001.SZ", 90, "好", rank=2)], "observe": []} + out = pf.select_candidates(plan, rank_by_upstream=True, rank_by_quality=True) + assert [x["ts_code"] for x in out["items"]] == ["000001.SZ", "600000.SH"] # 看多分桶在前, 与既有用例同口径 + assert out["rank_mode"] == "quality_bucket" + + +@case("G 上游名次排·rank_mode 三态、四键透传、新开关登记") +def _(): + q = {"date": "2026-09-17", "rank_axis": "quality", "main": [_row("600000.SH", 100, rank=1)], "observe": []} + assert pf.select_candidates(q, rank_by_upstream=True, rank_by_quality=True)["rank_mode"] == "upstream" + assert pf.select_candidates(q, rank_by_upstream=False, rank_by_quality=True)["rank_mode"] == "quality_bucket" + assert pf.select_candidates(q, rank_by_upstream=False, rank_by_quality=False)["rank_mode"] == "score" + p2 = {"date": "2026-09-17", "rank_axis": "quality", "observe": [], + "main": [{**_row("600000.SH", 100, rank=1), + "quality_band": 3, "pointed": True, "rank_axis": "quality", "rank_old": 5}]} + it = pf.select_candidates(p2, rank_by_upstream=True)["items"][0] + assert (it["quality_band"], it["pointed"], it["rank_axis"], it["rank_old"]) == (3, True, "quality", 5) + assert "PMS_PLAN_RANK_BY_UPSTREAM" in ps.RUNTIME_EXTRA + assert ps.RUNTIME_EXTRA["PMS_PLAN_RANK_BY_UPSTREAM"][1] is bool + assert ps.RUNTIME_EXTRA["PMS_PLAN_RANK_BY_UPSTREAM"][0] is True + + def main(): ok = 0 for name, fn in RESULTS: