303 lines
17 KiB
Python
303 lines
17 KiB
Python
"""候选卡：把一只票的各条证据线装配成判决与理由（纯函数，零 IO）。
|
||
|
||
## 为什么要有它
|
||
|
||
桥原来的产物是一个分数：主榜分 = 200 + 传导档位×20 + 组内分。分数排在最前的
|
||
强传导档，实证里过去一个月是四档中最差的（五日超额 −2.04，见
|
||
docs/主观选股改进方案_2026-09-02.md 第 1.8b 节）。方案把桥从"打分排序器"改成
|
||
"候选卡装配器"：分数与档位一律不动，另出一份带理由的候选单。规则只有三条硬门槛、
|
||
一条确认、三条硬风险，全部是一个主观选股的人会用的判断，不算总分。
|
||
|
||
门槛一 所在环节当日被传导指向（只认环节类目标）
|
||
门槛二 成员当日已启动：数据日涨幅达到 CARD_START_PCT（默认 3%，与基座热点扫描同口径）
|
||
门槛三 有券商覆盖、预期空间不低于负容忍线、非 ST 与退市族
|
||
确认 决策系统吸筹评分为"明确吸筹"，且评分日龄不超过 CARD_ACCUM_MAX_AGE 个交易日
|
||
硬风险 决策系统昨夜给出卖出、回避或剔除信号；传导快照日与计划日不符；吸筹评分为高位派发
|
||
|
||
判决 候选 = 三门槛全过、硬风险为空、确认成立
|
||
关注 = 三门槛全过、硬风险为空，但确认线缺失（没有吸筹评分）或陈旧（明确吸筹但评分
|
||
日龄超过上限或日龄未知）——这是"系统无法判断"，交人裁决
|
||
仅展示 = 其余，卡上标明未过项。只差券商覆盖、吸筹评分为潜在吸筹或其他非明确状态，
|
||
都归这一档（2026-09-03 台账 013：潜在吸筹市值中性后为负，不升格）
|
||
|
||
依据：方案 1.8d——"环节被指向、当日已涨 3% 以上、且明确吸筹"是纯可交易口径下唯一
|
||
合并样本为正的规则（五日 +2.18、十日 +1.94），但它是顺风策略、不免疫环境；两个旋钮
|
||
保持默认，等样本外复盘读数再定，不在历史样本上挑参数（2026-09-02 决定）。
|
||
关注定义的细化出自《主观量化系统方案_2026-09-03》第 3.2 节"无法判断"的三个出口：
|
||
候选自动进建仓方案，关注强制人工确认，仅展示不出提议。
|
||
|
||
## 因果论断证据线（2026-09-03）
|
||
|
||
证据字典可带 logic：数据基座因果论断视图给出的论断列表（方向、机制、时效、出处文档标题与
|
||
披露日、论断编号）。judge 把它整理成带出处的一行行文字放进输出的 logic 键，只展示、
|
||
不作门槛、不进判决——分析结论到选股的断裂点先接通，要不要当门槛等复盘案例再定。
|
||
|
||
## 边界
|
||
|
||
本模块不读库、不读配置，只吃调用方装配好的证据字典，输出判决字典。这样它能
|
||
离线单测（test_card.py），也保证同一段规则在计划装配、复盘脚本、对账工具里只有一份。
|
||
坏信号集合 BAD_SIGNALS 是"三处同源"纪律里选股系统的那一份（另两处在择时决策系统
|
||
pms_advisor.py 与 PMS rule_gate.py），改一处必须三处同改；pool.py 从这里引用，本仓库只此一处。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
# 决策系统夜间结论表 signal_type 里的坏信号（三处同源，见模块说明）
|
||
BAD_SIGNALS = {"SELL", "AVOID", "DROPPED"}
|
||
|
||
# 昨夜信号的代号翻译成中文。这些代号是决策系统那边的枚举值，只在系统之间传递；
|
||
# 一旦要写进给人看的句子里，就必须先翻成人话——页面上出现一个 SELL，读的人得先
|
||
# 知道这是个枚举值才看得懂。翻不出来的原样带上，总比吞掉强。
|
||
SIGNAL_CN = {"SELL": "卖出", "AVOID": "回避", "DROPPED": "已剔除",
|
||
"BUY": "买入", "WATCH": "观察", "HOLD": "持有"}
|
||
|
||
|
||
VERDICT_CANDIDATE = "候选"
|
||
VERDICT_WATCH = "关注"
|
||
VERDICT_SHOW = "仅展示"
|
||
_VERDICT_ORDER = {VERDICT_CANDIDATE: 0, VERDICT_WATCH: 1, VERDICT_SHOW: 2}
|
||
|
||
|
||
def _num(v) -> float | None:
|
||
if v is None:
|
||
return None
|
||
try:
|
||
x = float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if x != x: # NaN
|
||
return None
|
||
return x
|
||
|
||
|
||
def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
|
||
accum_max_age: int = 30, neg_tol: float = 0.0,
|
||
logic_stale_days: int = 90, require_started: bool = True) -> dict[str, Any]:
|
||
"""一只票的证据字典 -> 判决字典。
|
||
|
||
ev 里认的键（缺键按缺失处理，不报错）：
|
||
pointed bool 所在环节当日被传导指向（只认 Segment 目标）
|
||
theme str 目标环节名；n_sources int 源数；chain_fit 链符（展示用）
|
||
pct0 float 数据日涨幅（百分数，3.2 表示 +3.2%）
|
||
covered bool 有券商覆盖；upside float 预期空间（比例，0.25 表示 +25%）
|
||
risk_name bool 证券简称命中 ST / *ST / 退市族
|
||
accum_state str 吸筹状态文案（决策系统 fund_flow.state）；accum_score float；
|
||
accum_age int 评分日龄（交易日）
|
||
y_signal str 决策系统昨夜 signal_type
|
||
stale_snapshot bool 传导快照日与计划日不符
|
||
logic list 数据基座因果论断列表（每条是字典：direction / mechanism / horizon /
|
||
doc_title / disclosure_date / claim_id 等），只展示不进判决
|
||
返回：verdict / reasons / missing / risk / gates / confirm / failed_gates / basis / logic。
|
||
basis 一句话判决依据，说明落到这一档的原因（关注即"系统无法判断"，交人裁决）
|
||
logic 因果论断整理成的带出处文字行，与判决无关
|
||
logic_quality 论断的质量画像（条数、最新披露日、距今天数、几份来源、疑似误抽与
|
||
分歧条数、按环节展开的条数），取数层在截断前算好带上来，与判决无关
|
||
failed_gates 是未过的门槛名；三门槛全过但吸筹评分为非明确状态而落仅展示时，附加 "confirm"
|
||
"""
|
||
pct0 = _num(ev.get("pct0"))
|
||
upside = _num(ev.get("upside"))
|
||
accum_score = _num(ev.get("accum_score"))
|
||
accum_age = ev.get("accum_age")
|
||
accum_state = str(ev.get("accum_state") or "")
|
||
y_signal = str(ev.get("y_signal") or "").strip().upper()
|
||
|
||
gates = {
|
||
"pointed": bool(ev.get("pointed")),
|
||
# require_started 关掉时这一条恒过：涨跌多少都不拦，没有当日行情也不拦。
|
||
# 这是整条门槛的开关，不是把阈值调成一个很小的负数——那样"没有行情"的票
|
||
# 仍然会被拦下，而关掉门槛的本意是这一条根本不参与判断。
|
||
"started": (not require_started) or (pct0 is not None and pct0 >= float(start_pct)),
|
||
"covered": bool(ev.get("covered")) and upside is not None
|
||
and upside >= -float(neg_tol or 0.0),
|
||
"clean_name": not bool(ev.get("risk_name")),
|
||
}
|
||
accum_fresh = (isinstance(accum_age, int) and 0 <= accum_age <= int(accum_max_age))
|
||
confirm = accum_state.startswith("明确") and accum_fresh
|
||
# 确认线的三种"不成立"要分开：缺失（没有评分）与陈旧（明确吸筹但日龄超限或未知）是
|
||
# "系统无法判断"；评分为潜在吸筹、信号不明、无吸筹等非明确状态，是系统已经判断过、
|
||
# 只是没有达到确认线，不升格（台账 013）。
|
||
confirm_missing = not accum_state
|
||
confirm_stale = accum_state.startswith("明确") and not accum_fresh
|
||
confirm_negative = bool(accum_state) and not accum_state.startswith("明确")
|
||
|
||
risk: list[str] = []
|
||
if y_signal in BAD_SIGNALS:
|
||
risk.append(f"择时决策系统昨夜给出{SIGNAL_CN.get(y_signal, y_signal)}信号")
|
||
if ev.get("stale_snapshot"):
|
||
risk.append("用的传导快照不是计划当天的")
|
||
if "派发" in accum_state:
|
||
risk.append("资金在高位派发")
|
||
|
||
missing: list[str] = []
|
||
if pct0 is None:
|
||
missing.append("没有当日行情")
|
||
if not ev.get("covered"):
|
||
missing.append("没有券商覆盖")
|
||
elif not gates["covered"]:
|
||
missing.append(f"券商预期空间 {upside:+.0%}，低于 {neg_tol:+.0%} 的容忍线"
|
||
if upside is not None else "算不出券商预期空间")
|
||
if confirm_missing:
|
||
missing.append("还没有吸筹评分（这只票没进过夜间扫描）")
|
||
elif confirm_stale:
|
||
missing.append(f"吸筹评分是 {accum_age} 个交易日前的，太旧了"
|
||
if isinstance(accum_age, int) else "吸筹评分有多旧算不出来")
|
||
elif confirm_negative and "派发" not in accum_state:
|
||
state_short = accum_state.split("·")[0].strip()
|
||
missing.append(f"吸筹只到「{state_short}」，没到明确吸筹这条线")
|
||
if not gates["pointed"]:
|
||
missing.append("所在环节当日没有被传导指向")
|
||
# 研报论断的质量提示。放进缺失项是因为它和"吸筹评分陈旧"是同一类事：证据还在，
|
||
# 但已经旧到不该当新证据用。两者此前一个会过期一个不会，卡上看着一样新，
|
||
# 这个不对称本身会误导人。这里只提示不作门槛——门槛仍是那三条可交易口径。
|
||
missing.extend(logic_notes(ev.get("logic"), logic_stale_days))
|
||
|
||
reasons: list[str] = []
|
||
if gates["pointed"]:
|
||
theme = ev.get("theme") or "环节"
|
||
n = ev.get("n_sources")
|
||
reasons.append(f"所在环节「{theme}」被{n}路指向" if n else f"所在环节「{theme}」被指向")
|
||
if gates["started"] and require_started:
|
||
reasons.append(f"当日已启动 {pct0:+.1f}%")
|
||
elif not require_started and pct0 is not None:
|
||
# 门槛关着时它不再是一条理由，但当日涨跌仍是人要看的信息，照样写出来。
|
||
reasons.append(f"当日涨幅 {pct0:+.1f}%（启动线已关闭，不作门槛）")
|
||
if confirm:
|
||
reasons.append(f"明确吸筹（评分 {accum_score:.0f}，{accum_age} 日前）"
|
||
if accum_score is not None else f"明确吸筹（{accum_age} 日前）")
|
||
if gates["covered"]:
|
||
reasons.append(f"券商覆盖，预期空间 {upside:+.0%}")
|
||
|
||
all_gates = all(gates.values())
|
||
failed = [k for k, ok in gates.items() if not ok]
|
||
# 判决依据是这张卡上唯一一句"为什么"，PMS 原样显示在页面上给人读。所以它必须是
|
||
# 一句完整的中文，带上具体数字，不许出现门槛的变量名（started 这类）或信号的
|
||
# 枚举值（SELL 这类）——那些是系统之间传递用的，不是给人看的。
|
||
if risk:
|
||
verdict = VERDICT_SHOW
|
||
basis = "；".join(risk)
|
||
elif all_gates and confirm:
|
||
verdict = VERDICT_CANDIDATE
|
||
basis = (f"三条门槛都过了，没有风险，且明确吸筹（{accum_age} 个交易日前评的）"
|
||
if isinstance(accum_age, int) else "三条门槛都过了，没有风险，且明确吸筹")
|
||
elif all_gates and (confirm_missing or confirm_stale):
|
||
verdict = VERDICT_WATCH
|
||
why = ("这只票还没有吸筹评分" if confirm_missing
|
||
else (f"吸筹评分是 {accum_age} 个交易日前的，太旧了"
|
||
if isinstance(accum_age, int) else "吸筹评分有多旧算不出来"))
|
||
basis = f"三条门槛都过了，也没有风险，但{why}，系统判断不了，交人裁决"
|
||
elif all_gates:
|
||
verdict = VERDICT_SHOW
|
||
failed.append("confirm")
|
||
basis = (f"三条门槛都过了，但吸筹只到「{accum_state.split('·')[0].strip()}」，"
|
||
f"没到明确吸筹这条线")
|
||
else:
|
||
verdict = VERDICT_SHOW
|
||
basis = "；".join(gate_reasons(failed, pct0=pct0, start_pct=start_pct,
|
||
covered=bool(ev.get("covered")), upside=upside,
|
||
neg_tol=neg_tol))
|
||
|
||
return {"verdict": verdict, "reasons": reasons, "missing": missing, "risk": risk,
|
||
"gates": gates, "confirm": confirm, "failed_gates": failed, "basis": basis,
|
||
"logic": logic_lines(ev.get("logic")),
|
||
"logic_quality": logic_quality(ev.get("logic"))}
|
||
|
||
|
||
def logic_lines(claims, limit: int = 3) -> list[str]:
|
||
"""把因果论断列表整理成带出处的文字行：方向、机制、时效，加"《文档标题》披露日·论断编号"。
|
||
只展示，不进判决；缺字段的部分省略，不报错。"""
|
||
out: list[str] = []
|
||
for c in (claims or [])[:limit]:
|
||
if not isinstance(c, dict):
|
||
continue
|
||
head = "：".join(x for x in (c.get("direction"), c.get("mechanism")) if x)
|
||
if c.get("horizon"):
|
||
head = f"{head}（{c['horizon']}）" if head else f"（{c['horizon']}）"
|
||
if c.get("condition"):
|
||
head = f"{head}，条件：{c['condition']}"
|
||
src = "".join(x for x in (
|
||
f"《{c['doc_title']}》" if c.get("doc_title") else "",
|
||
f" {c['disclosure_date']}" if c.get("disclosure_date") else "",
|
||
f" · {c['claim_id']}" if c.get("claim_id") else "") if x)
|
||
line = head or "因果论断"
|
||
if src:
|
||
line = f"{line}——出处：{src.strip()}"
|
||
out.append(line)
|
||
return out
|
||
|
||
|
||
def sort_key(row: dict[str, Any]) -> tuple:
|
||
"""卡内序：先判决（候选 < 关注 < 仅展示），再按数据日涨幅降序。不算总分。"""
|
||
v = _VERDICT_ORDER.get(row.get("verdict"), 9)
|
||
pct0 = _num(row.get("pct0"))
|
||
return (v, -(pct0 if pct0 is not None else -1e9))
|
||
|
||
|
||
def logic_quality(claims) -> dict | None:
|
||
"""把取数层算好的质量画像取出来。取数层在截断之前算，所以它统计的是这票的
|
||
全部论断，而卡上只展示前几条——两个数不一样是对的。论断为空时返回 None，
|
||
与"上游没给这个字段"同形（卡上都不显示这一节）。"""
|
||
for c in (claims or []):
|
||
if isinstance(c, dict) and isinstance(c.get("quality"), dict):
|
||
return c["quality"]
|
||
return None
|
||
|
||
|
||
def logic_notes(claims, stale_days: int = 90) -> list[str]:
|
||
"""论断质量的提示文字，只提示不作门槛。四条，都以实际读数为准，凑不齐就少写几条：
|
||
|
||
陈旧：最新一条论断距数据日超过 stale_days 天。
|
||
单一来源：这票有多条论断但全部出自同一份研报——那份研报的观点一旦过时或本身
|
||
有偏，这票的整条研究证据一起失效，而分开看每一条都像是独立的证据。
|
||
疑似误抽：逻辑评析读过之后标出来的，模型的判断不是确定性事实，所以只提示。
|
||
未结分歧：同期方向相反、还没有结论的，表示市场没有共识。
|
||
"""
|
||
q = logic_quality(claims)
|
||
if not q or not q.get("n"):
|
||
return []
|
||
out: list[str] = []
|
||
age = q.get("age_days")
|
||
if isinstance(age, int) and age > int(stale_days):
|
||
latest = q.get("latest_date")
|
||
out.append(f"研报论断已过 {age} 天（最新一条 {latest}）" if latest
|
||
else f"研报论断已过 {age} 天")
|
||
if q["n"] > 1 and q.get("n_docs") == 1:
|
||
out.append(f"{q['n']} 条论断全部出自同一份研报")
|
||
if q.get("n_flagged"):
|
||
out.append(f"其中 {q['n_flagged']} 条被标为疑似误抽")
|
||
if q.get("n_disputed"):
|
||
out.append(f"其中 {q['n_disputed']} 条处于未结的方向分歧")
|
||
return out
|
||
|
||
|
||
def gate_reasons(failed, *, pct0=None, start_pct: float = 3.0, covered: bool = False,
|
||
upside=None, neg_tol: float = 0.0) -> list[str]:
|
||
"""把没过的门槛翻成一句句人话，带上当时的实际数字。
|
||
|
||
门槛在代码里的名字是 pointed、started、covered，那是给程序读的。写进给人看的
|
||
句子必须换成人话，并且带上数字——「当日跌 3.2%，没到 3% 的启动线」比「涨幅门槛
|
||
未过」有用得多，因为前者能让人自己判断差多少。
|
||
|
||
认不出的门槛名原样带上而不是吞掉：真出现了新门槛，宁可页面上难看一次，
|
||
也好过一句话里少了一条原因。
|
||
"""
|
||
out: list[str] = []
|
||
for k in failed or []:
|
||
if k == "pointed":
|
||
out.append("所在环节当日没有被传导指向")
|
||
elif k == "started":
|
||
out.append(f"当日涨幅 {pct0:+.1f}%，没到 {start_pct:g}% 的启动线"
|
||
if pct0 is not None else "没有当日行情，算不出涨幅")
|
||
elif k == "covered":
|
||
if not covered:
|
||
out.append("没有券商覆盖")
|
||
elif upside is not None:
|
||
out.append(f"券商预期空间 {upside:+.0%}，低于 {neg_tol:+.0%} 的容忍线")
|
||
else:
|
||
out.append("算不出券商预期空间")
|
||
elif k == "confirm":
|
||
out.append("吸筹没到明确吸筹这条线")
|
||
else:
|
||
out.append(f"门槛 {k} 未过")
|
||
return out
|