候选卡加论断质量标注：陈旧度、来源集中、疑似误抽、未结分歧
这四项都只在卡上提示，不参与三门槛判决。 起因是同一张卡上的双标：吸筹评分超过三十个交易日就判陈旧，而研报论断此前 只有"披露日不晚于数据日"这一个上界，永不过期。09-03 实测 442 只带论断的票， 论断披露日中位 05-05，超过 90 天的 166 只（37.6%），超过一年的 22 只——这些 论断会原样打印成理由，读的人看不出它已经很旧了。 另外三项同样来自当日实测：267 只票带多条论断，其中 77 只（29%）的论断全部 出自同一份研报，那份研报一旦过时或本身有偏，这票的整条研究证据一起失效； 213 条被逻辑评析标为疑似误抽；未结的方向分歧当日是 0 条但视图里有这一列。 改动三处： - sources.py 多取 disputed 与 review_flag 两列，并在按票截断之前算质量画像。 必须在截断前算——截断后只剩最近三条，"这票一共几份来源"就统计不出来了。 - card.py 把画像转成缺失项里的提示文字，另出一个 logic_quality 键供下游用。 - config.py 加陈旧线 LOGIC_STALE_DAYS，默认 90 天，写明与吸筹日龄上限的可比关系。 测试：候选卡 14 例、取数层 9 例，本地全套通过（test_xxl_trigger 缺 fastapi 是 本机环境所致，与本次改动无关）。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
zlt
parent
848c6ced67
commit
3fb25fe1ad
49
card.py
49
card.py
|
|
@ -65,7 +65,8 @@ def _num(v) -> float | None:
|
|||
|
||||
|
||||
def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
|
||||
accum_max_age: int = 30, neg_tol: float = 0.0) -> dict[str, Any]:
|
||||
accum_max_age: int = 30, neg_tol: float = 0.0,
|
||||
logic_stale_days: int = 90) -> dict[str, Any]:
|
||||
"""一只票的证据字典 -> 判决字典。
|
||||
|
||||
ev 里认的键（缺键按缺失处理，不报错）：
|
||||
|
|
@ -83,6 +84,8 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
|
|||
返回：verdict / reasons / missing / risk / gates / confirm / failed_gates / basis / logic。
|
||||
basis 一句话判决依据，说明落到这一档的原因（关注即"系统无法判断"，交人裁决）
|
||||
logic 因果论断整理成的带出处文字行，与判决无关
|
||||
logic_quality 论断的质量画像（条数、最新披露日、距今天数、几份来源、疑似误抽与
|
||||
分歧条数、按环节展开的条数），取数层在截断前算好带上来，与判决无关
|
||||
failed_gates 是未过的门槛名；三门槛全过但吸筹评分为非明确状态而落仅展示时，附加 "confirm"
|
||||
"""
|
||||
pct0 = _num(ev.get("pct0"))
|
||||
|
|
@ -133,6 +136,10 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
|
|||
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"]:
|
||||
|
|
@ -170,7 +177,8 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
|
|||
|
||||
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": logic_lines(ev.get("logic")),
|
||||
"logic_quality": logic_quality(ev.get("logic"))}
|
||||
|
||||
|
||||
def logic_lines(claims, limit: int = 3) -> list[str]:
|
||||
|
|
@ -201,3 +209,40 @@ 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
|
||||
|
|
|
|||
|
|
@ -229,6 +229,13 @@ MARKET_MYSQL_SOURCE = os.environ.get("MARKET_MYSQL_SOURCE", "price").strip().low
|
|||
# 只展示、不作门槛、不进判决；设 0 表示不读该视图（视图未建时也可用它关掉那一行告警）。
|
||||
LOGIC_CLAIMS_PER_STOCK = int(os.environ.get("LOGIC_CLAIMS_PER_STOCK", "3"))
|
||||
|
||||
# 论断陈旧线（自然日）。超过它就在候选卡的缺失项里写明"研报论断已过 N 天"，只提示不作门槛。
|
||||
# 取 90 天的理由是与吸筹评分的日龄上限可比：那一条是三十个交易日，约合六周多的自然日，
|
||||
# 而研报的更新节奏本来就比资金结构评分慢得多（行业深度按季度、跟踪报告按事件），
|
||||
# 所以放到大约两个季度这个量级。这个数一次定死，不按复盘读数回调。
|
||||
# 2026-09-03 实测：当日 442 只带论断的票里，超过 90 天的有 166 只（37.6%），超过一年的 22 只。
|
||||
LOGIC_STALE_DAYS = int(os.environ.get("LOGIC_STALE_DAYS", "90"))
|
||||
|
||||
# --- 日频行业观点快照（2026-09-03 方案第四之五之三节，落地顺序第一步；模块见 judgement.py）------
|
||||
# 数据基座的研判结论按簇键唯一，重评时整行覆盖，库里永远只有最新一版，查不到"上一次是什么样、
|
||||
# 什么时候变的"。选股系统每个计划日把最新一版抄一行存下来，自己攒版本史，只写不判。
|
||||
|
|
|
|||
3
plan.py
3
plan.py
|
|
@ -148,7 +148,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
|
|||
"stale_snapshot": stale, "logic": logic.get(k) or []}
|
||||
j = card.judge(evd, start_pct=config.CARD_START_PCT,
|
||||
accum_max_age=config.CARD_ACCUM_MAX_AGE,
|
||||
neg_tol=config.UPSIDE_NEG_TOLERANCE)
|
||||
neg_tol=config.UPSIDE_NEG_TOLERANCE,
|
||||
logic_stale_days=config.LOGIC_STALE_DAYS)
|
||||
cards[k] = {
|
||||
**j,
|
||||
"theme": theme, "n_sources": n_sources, "chain_fit": evd["chain_fit"],
|
||||
|
|
|
|||
45
sources.py
45
sources.py
|
|
@ -227,7 +227,11 @@ def _to_dot(code: str) -> str:
|
|||
# ============================================================================
|
||||
_LOGIC_COLS = ("ts_code", "subject_name", "object_name", "direction", "mechanism", "condition",
|
||||
"horizon", "strength", "tier", "confidence", "disclosure_date", "doc_id",
|
||||
"doc_title", "source_span", "claim_id", "via_segment")
|
||||
"doc_title", "source_span", "claim_id", "via_segment",
|
||||
# 质量两列（2026-09-03 起取）：disputed 是这条论断还在多空分歧里没结论，
|
||||
# review_flag 是逻辑评析读过之后标的"疑似误抽"。两者都只作提示，不过滤——
|
||||
# 那是模型的判断不是确定性事实，过滤会让人看不到系统曾经抽到过什么。
|
||||
"disputed", "review_flag")
|
||||
|
||||
|
||||
def logic_claims(codes, ds: str, per_stock: int | None = None, read_pg=None) -> dict[str, list[dict]]:
|
||||
|
|
@ -278,14 +282,51 @@ def logic_claims(codes, ds: str, per_stock: int | None = None, read_pg=None) ->
|
|||
"source_span": (_s(r.get("source_span")) or "")[:200] or None,
|
||||
"claim_id": _s(r.get("claim_id")), "via_segment": _s(r.get("via_segment")),
|
||||
"subject": _s(r.get("subject_name")), "object": _s(r.get("object_name")),
|
||||
"disputed": bool(r.get("disputed")), "review_flag": _s(r.get("review_flag")),
|
||||
})
|
||||
out = {}
|
||||
for k, items in by_code.items():
|
||||
items.sort(key=lambda c: (c["disclosure_date"] or "", c["confidence"] or -1.0), reverse=True)
|
||||
out[k] = items[:n_per]
|
||||
# 质量指标必须在截断之前算：截断后只剩最近几条，"这票一共有几份研报来源"就看不出来了。
|
||||
# 算完挂在每条论断上（每条都带同一份，卡上取第一条即可），不另开一个返回值，
|
||||
# 免得上层要同时接两个字典、少接一个就静默丢掉质量提示。
|
||||
quality = _logic_quality(items, ds)
|
||||
kept = items[:n_per]
|
||||
for c in kept:
|
||||
c["quality"] = quality
|
||||
out[k] = kept
|
||||
return out
|
||||
|
||||
|
||||
def _logic_quality(items: list, ds: str) -> dict:
|
||||
"""一票全部论断的质量画像，在截断之前算。四项：
|
||||
|
||||
age_days 最新一条论断距数据日多少天。同一张卡上吸筹评分超过三十个交易日就判陈旧，
|
||||
而论断此前永不过期，这是双标。这里给出天数，阈值判定在 card.py。
|
||||
n_docs 论断来自几份不同研报。实测近三成带多条论断的票，全部论断出自同一份研报——
|
||||
那份研报一旦过时或本身有偏，这票的整条研究证据一起失效，而卡上看不出来。
|
||||
n_flagged 被逻辑评析标为疑似误抽的条数；n_disputed 还在多空分歧里没结论的条数。
|
||||
n_via_segment 客体是环节、按成员展开挂上来的条数。那类不是直接讲这家公司，
|
||||
与直接讲这家公司的论断分量不同，卡上要分得开。
|
||||
"""
|
||||
days = None
|
||||
dates = [c["disclosure_date"] for c in items if c.get("disclosure_date")]
|
||||
if dates and ds:
|
||||
try:
|
||||
days = (dt.date.fromisoformat(ds) - dt.date.fromisoformat(max(dates))).days
|
||||
except ValueError:
|
||||
days = None
|
||||
return {
|
||||
"n": len(items),
|
||||
"age_days": days,
|
||||
"latest_date": max(dates) if dates else None,
|
||||
"n_docs": len({c["doc_id"] for c in items if c.get("doc_id")}),
|
||||
"n_flagged": sum(1 for c in items if c.get("review_flag")),
|
||||
"n_disputed": sum(1 for c in items if c.get("disputed")),
|
||||
"n_via_segment": sum(1 for c in items if c.get("via_segment")),
|
||||
}
|
||||
|
||||
|
||||
def _s(v) -> str | None:
|
||||
if v is None or (isinstance(v, float) and v != v):
|
||||
return None
|
||||
|
|
|
|||
54
test_card.py
54
test_card.py
|
|
@ -2,7 +2,8 @@
|
|||
|
||||
覆盖（2026-09-03 台账 013 的关注定义细化）：候选 / 关注只剩"确认线缺失或陈旧"两种 /
|
||||
只差覆盖与潜在吸筹归仅展示 / 仅展示（未启动、硬风险否决）/ 缺失文案与判决依据 /
|
||||
因果论断证据线只展示不进判决 / 旋钮 / 卡内序 / 坏信号集合与 pool 同源。
|
||||
因果论断证据线只展示不进判决 / 论断质量标注（陈旧、单一来源、疑似误抽、未结分歧，
|
||||
四项都只提示不作门槛）/ 旋钮 / 卡内序 / 坏信号集合与 pool 同源。
|
||||
|
||||
跑法：python3 test_card.py 或 pytest test_card.py
|
||||
"""
|
||||
|
|
@ -103,9 +104,58 @@ def main():
|
|||
t("卡内序：候选在前、同判决涨幅降序、缺行情垫底",
|
||||
[x["pct0"] for x in rows] == [7.2, 3.5, 9.0, None])
|
||||
|
||||
# ---- 论断质量标注（只展示、只提示，不进判决）----
|
||||
# 质量画像由取数层在截断之前算好挂在每条论断上，这里按那个形状造数据。
|
||||
def q(**kw):
|
||||
base = dict(n=1, age_days=10, latest_date="2026-08-25", n_docs=1,
|
||||
n_flagged=0, n_disputed=0, n_via_segment=0)
|
||||
base.update(kw)
|
||||
return base
|
||||
|
||||
def claims(n, quality):
|
||||
return [{**CLAIM, "quality": quality} for _ in range(n)]
|
||||
|
||||
r = card.judge({**BASE, "logic": claims(1, q())})
|
||||
t("论断新鲜且单条：不产生任何质量提示",
|
||||
r["verdict"] == "候选" and not [m for m in r["missing"] if "论断" in m])
|
||||
t("质量画像随判决一起出", r["logic_quality"]["age_days"] == 10)
|
||||
|
||||
r = card.judge({**BASE, "logic": claims(1, q(age_days=200, latest_date="2026-02-16"))})
|
||||
t("论断超陈旧线：写明过了多少天与最新一条的日期",
|
||||
"研报论断已过 200 天（最新一条 2026-02-16）" in r["missing"])
|
||||
t("论断陈旧不改判决（只提示不作门槛）", r["verdict"] == "候选")
|
||||
|
||||
r = card.judge({**BASE, "logic": claims(1, q(age_days=91))}, logic_stale_days=120)
|
||||
t("陈旧线旋钮生效：放宽到 120 天后 91 天不再提示",
|
||||
not [m for m in r["missing"] if "研报论断已过" in m])
|
||||
r = card.judge({**BASE, "logic": claims(1, q(age_days=90))})
|
||||
t("边界：正好等于陈旧线不算陈旧（超过才提示）",
|
||||
not [m for m in r["missing"] if "研报论断已过" in m])
|
||||
|
||||
r = card.judge({**BASE, "logic": claims(3, q(n=3, n_docs=1))})
|
||||
t("多条论断出自同一份研报：单一来源提示",
|
||||
"3 条论断全部出自同一份研报" in r["missing"])
|
||||
r = card.judge({**BASE, "logic": claims(3, q(n=3, n_docs=2))})
|
||||
t("来源多于一份就不提示", not [m for m in r["missing"] if "同一份研报" in m])
|
||||
r = card.judge({**BASE, "logic": claims(1, q(n=1, n_docs=1))})
|
||||
t("只有一条论断时不算单一来源（那是条数少不是来源集中）",
|
||||
not [m for m in r["missing"] if "同一份研报" in m])
|
||||
|
||||
r = card.judge({**BASE, "logic": claims(4, q(n=4, n_docs=2, n_flagged=2, n_disputed=1))})
|
||||
t("疑似误抽与未结分歧各写一条",
|
||||
"其中 2 条被标为疑似误抽" in r["missing"] and "其中 1 条处于未结的方向分歧" in r["missing"])
|
||||
|
||||
r = card.judge(dict(BASE))
|
||||
t("没有论断时不产生质量提示、画像为空",
|
||||
r["logic_quality"] is None and not [m for m in r["missing"] if "论断" in m])
|
||||
r = card.judge({**BASE, "logic": [CLAIM]})
|
||||
t("上游没给质量画像时不报错、也不瞎提示",
|
||||
r["logic_quality"] is None and not [m for m in r["missing"] if "论断" in m])
|
||||
t("论断为空列表时不产生提示", card.logic_notes([]) == [] and card.logic_quality([]) is None)
|
||||
|
||||
t("坏信号集合口径", card.BAD_SIGNALS == {"SELL", "AVOID", "DROPPED"})
|
||||
print("ALL OK — 候选卡判决 / 关注两种（缺失、陈旧）/ 只差覆盖与潜在吸筹归仅展示 / 硬风险三种 / "
|
||||
"因果论断只展示 / 缺失文案与依据 / 旋钮 / 卡内序 全部通过")
|
||||
"因果论断只展示 / 论断质量标注四项 / 缺失文案与依据 / 旋钮 / 卡内序 全部通过")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
|
|
@ -82,7 +82,7 @@ def _pg_ok(sql, params=None):
|
|||
{"ts_code": "600000.SH", "direction": "利好", "mechanism": "机制甲", "condition": None, "horizon": "一年",
|
||||
"strength": "强", "tier": "T1", "confidence": 0.8, "disclosure_date": dt.date(2026, 8, 20),
|
||||
"doc_id": "d1", "doc_title": "文档一", "source_span": "x" * 300, "claim_id": "c1", "via_segment": "环节甲",
|
||||
"subject_name": "甲", "object_name": "乙"},
|
||||
"subject_name": "甲", "object_name": "乙", "review_flag": "疑似误抽"},
|
||||
{"ts_code": "600000.SH", "direction": "利好", "mechanism": "机制乙", "condition": "条件乙", "horizon": "半年",
|
||||
"strength": "中", "tier": "T2", "confidence": 0.6, "disclosure_date": "2026-08-30",
|
||||
"doc_id": "d2", "doc_title": "文档二", "source_span": None, "claim_id": "c2", "via_segment": None,
|
||||
|
|
@ -94,7 +94,7 @@ def _pg_ok(sql, params=None):
|
|||
{"ts_code": "600000.SH", "direction": "利好", "mechanism": "机制丁", "condition": None, "horizon": None,
|
||||
"strength": None, "tier": None, "confidence": 0.5, "disclosure_date": "2026-07-01",
|
||||
"doc_id": "d4", "doc_title": "文档四", "source_span": None, "claim_id": "c4", "via_segment": None,
|
||||
"subject_name": None, "object_name": None},
|
||||
"subject_name": None, "object_name": None, "disputed": True},
|
||||
{"ts_code": "000001.SZ", "direction": "利好", "mechanism": "机制戊", "condition": None, "horizon": None,
|
||||
"strength": None, "tier": None, "confidence": None, "disclosure_date": 20260815,
|
||||
"doc_id": "d5", "doc_title": "文档五", "source_span": None, "claim_id": "c5", "via_segment": None,
|
||||
|
|
@ -170,6 +170,25 @@ def test_logic_claims():
|
|||
t("读失败返回空字典不抛错", sources.logic_claims(["SH600000"], DS, read_pg=_boom) == {})
|
||||
t("空代码集不查库", sources.logic_claims([], DS, read_pg=_boom) == {})
|
||||
t("整数日期也能归一", sources._ymd(20260815) == "2026-08-15" and sources._ymd("2026-08-15 10:00:00") == "2026-08-15")
|
||||
|
||||
# ---- 质量画像：必须在截断之前算 ----
|
||||
# 假数据这票共 4 条论断（c1 到 c4，四份不同文档），每票上限 3 条。
|
||||
# 如果在截断之后统计，条数会变成 3、疑似误抽那条（c1，披露日最老被截掉）会漏掉——
|
||||
# 那正是"这票的研究证据集中在一份研报上"这类判断会失真的地方。
|
||||
got = sources.logic_claims(["SH600000"], DS, read_pg=_pg_ok)
|
||||
items = got["SH600000"]
|
||||
q = items[0]["quality"]
|
||||
t("质量画像统计的是全部 4 条而不是留下的 3 条", q["n"] == 4 and len(items) == 3)
|
||||
t("来源份数按文档去重", q["n_docs"] == 4)
|
||||
t("陈旧度按最新一条算（2026-08-30 距 2026-09-02 三天）",
|
||||
q["age_days"] == 3 and q["latest_date"] == "2026-08-30")
|
||||
t("疑似误抽那条被截掉了仍统计得到", q["n_flagged"] == 1)
|
||||
t("未结分歧同样在截断前统计", q["n_disputed"] == 1)
|
||||
t("按环节展开的条数单独计", q["n_via_segment"] == 1)
|
||||
t("留下的每条都带同一份画像", all(c["quality"] is q for c in items))
|
||||
|
||||
got = sources.logic_claims(["SH600000"], DS, per_stock=1, read_pg=_pg_ok)
|
||||
t("上限收到 1 条时画像仍是全量口径", got["SH600000"][0]["quality"]["n"] == 4)
|
||||
t("前缀式转点后缀式", sources._to_dot("SH600000") == "600000.SH" and sources._to_dot("600000.SH") == "600000.SH")
|
||||
|
||||
|
||||
|
|
@ -187,7 +206,7 @@ def main():
|
|||
test_market_context()
|
||||
test_logic_claims()
|
||||
test_decision_ledger_titles()
|
||||
print("ALL OK — 市场四项 / 单项失败不阻断 / 因果论断索引与上限 / 台账标题解析 全部通过")
|
||||
print("ALL OK — 市场四项 / 单项失败不阻断 / 因果论断索引与上限 / 论断质量画像（截断前统计）/ 台账标题解析 全部通过")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
Loading…
Reference in New Issue