204 lines
11 KiB
Python
204 lines
11 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"}
|
||
|
||
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) -> 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 因果论断整理成的带出处文字行，与判决无关
|
||
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")),
|
||
"started": 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"决策系统昨夜信号 {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%} 低于负容忍线" 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("无传导")
|
||
|
||
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"]:
|
||
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]
|
||
if risk:
|
||
verdict = VERDICT_SHOW
|
||
basis = "硬风险否决：" + "；".join(risk)
|
||
elif all_gates and confirm:
|
||
verdict = VERDICT_CANDIDATE
|
||
basis = "三门槛全过、无硬风险、明确吸筹且评分新鲜"
|
||
elif all_gates and (confirm_missing or confirm_stale):
|
||
verdict = VERDICT_WATCH
|
||
basis = ("三门槛全过、无硬风险，但确认线"
|
||
+ ("缺失（无吸筹评分）" if confirm_missing else "陈旧（明确吸筹但评分日龄超限或未知）")
|
||
+ "——系统无法判断，交人裁决")
|
||
elif all_gates:
|
||
verdict = VERDICT_SHOW
|
||
failed.append("confirm")
|
||
basis = f"三门槛全过，但吸筹评分为「{accum_state.split('·')[0].strip()}」，未达确认线，不升格关注"
|
||
else:
|
||
verdict = VERDICT_SHOW
|
||
basis = "门槛未过：" + "、".join(failed)
|
||
|
||
return {"verdict": verdict, "reasons": reasons, "missing": missing, "risk": risk,
|
||
"gates": gates, "confirm": confirm, "failed_gates": failed, "basis": basis,
|
||
"logic": logic_lines(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))
|