138 lines
7.0 KiB
Python
138 lines
7.0 KiB
Python
"""逻辑状态四态的覆盖率读数：只读，不写库，不改任何判决。
|
||
|
||
回答一个问题：四态这套东西今天在真实数据上能覆盖多少票、各落到哪个状态。
|
||
它是判断"这一路该接多大面"的依据，也是接进计划装配之前必须先看的读数。
|
||
|
||
三路各自的取数在这里，归一与合成都调 logic_state 里的纯函数——读数脚本绝不能自己
|
||
另写一套判据，否则读到的就不是系统真会给出的状态。
|
||
|
||
甲路 研报论断 基座 PG v_factor_logic（经 sources.logic_claims）
|
||
乙路 产业研判 平台 MySQL t_akg_judgement_snapshot（经 judgement.load_previous）
|
||
丙路 券商行动 平台 MySQL gp_report_rc：两个等长窗口的每股收益预测中位数与机构数
|
||
丁路 公司事件 无数据源，恒定缺失
|
||
|
||
丙路的取数口径（三条都要照做，否则读数是错的）：
|
||
一，两个窗口必须等长。不等长会让八成的票假显示覆盖收缩——实测前 135 天对近 45 天时
|
||
有 907 只票误报。
|
||
二，同一财年同一预测期才可比，按 quarter 精确匹配，跨财年比较没有意义。
|
||
三，同一家机构在窗口里可能发多篇，先按机构取最近一篇再算中位数，否则发得勤的
|
||
机构会被重复计入。
|
||
|
||
跑法（155 上）：docker exec akg_factor_bridge python3 logic_state_coverage.py [数据日] [计划日]
|
||
计划日不给就取数据日的次日。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
import statistics as st
|
||
import sys
|
||
from collections import Counter, defaultdict
|
||
|
||
import common
|
||
import config
|
||
import db
|
||
import judgement
|
||
import logic_state as ls
|
||
import plan
|
||
import sources
|
||
|
||
# 丙路两个窗口各自的长度（自然日）。等长是硬要求，见模块说明第一条。
|
||
BROKER_WINDOW_DAYS = 45
|
||
|
||
|
||
def main(ds: str | None = None, plan_day: str | None = None) -> None:
|
||
"""ds 是数据日（行情与论断按它取），plan_day 是计划日（行业观点快照按它取）。
|
||
|
||
这两个日期在生产里本来就差一天：计划日凌晨构建，用的是上一个交易日的数据，
|
||
而当天的行业观点快照按计划日落库。早先这个脚本用同一个日期取三样东西，
|
||
结果是要么读不到当天的快照，要么读不到行情。plan_day 不给就取数据日的次日。
|
||
"""
|
||
ds = ds or (dt.date.today() - dt.timedelta(days=1)).isoformat()
|
||
if not plan_day:
|
||
try:
|
||
plan_day = (dt.date.fromisoformat(ds) + dt.timedelta(days=1)).isoformat()
|
||
except ValueError:
|
||
plan_day = ds
|
||
print(f"=== 逻辑状态四态覆盖率读数 · 数据日 {ds} · 计划日 {plan_day} ===\n")
|
||
|
||
codes = db.read_pg("SELECT DISTINCT ts_code FROM v_factor_stock_daily "
|
||
"WHERE trade_date = %s", (ds,))["ts_code"].tolist()
|
||
codes = [common.to_prefix(str(c).strip()) for c in codes]
|
||
print(f"当日有行情的票 {len(codes)} 只\n")
|
||
|
||
claims = sources.logic_claims(codes, ds, per_stock=200)
|
||
brokers = sources.broker_actions(codes, ds)
|
||
# 要的是这个计划日当天那一版快照（带迁移与陈旧两列），不是它之前的那一版——
|
||
# load_previous 读的是严格早于计划日的，用它会在快照首日读到空。当天没有就退回上一版。
|
||
snaps = judgement.snapshot_of(plan_day) or judgement.load_previous(plan_day)
|
||
print(f"甲路取到 {len(claims)} 只票的论断；丙路算得出 {len(brokers)} 只票；"
|
||
f"乙路快照 {len(snaps)} 个簇\n")
|
||
|
||
# 乙路按环节名对上主题：候选卡按环节，产业研判按主题聚簇，两者不在一个命名空间，
|
||
# 这里只做同名匹配，对不上的票乙路就是缺失。这一路的天花板本来就低（实测 1.4%）。
|
||
by_seg = judgement.by_segment_name(snaps)
|
||
seg_of = {}
|
||
# 用完整的传导视图，不是只含已启动成员的那张。候选卡判"所在环节被指向"时认的就是
|
||
# 完整这张（plan.py 的 evd["pointed"] 是两张视图取或），读数脚本必须跟它一致，
|
||
# 否则算出来的乙路覆盖比卡上真实看到的低。
|
||
# 一只票可能挂在多个环节上，这里保留全部，只要有一个环节评析过就算对上。
|
||
try:
|
||
d = db.read_pg("SELECT ts_code, target FROM v_factor_transmission "
|
||
"WHERE scan_date = %s", (ds,))
|
||
for r in d.itertuples():
|
||
seg_of.setdefault(common.to_prefix(str(r.ts_code).strip()), []).append(str(r.target))
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" （传导视图读取失败，乙路整体缺席: {e!r}）")
|
||
|
||
per_path = {ls.PATH_CLAIM: Counter(), ls.PATH_JUDGE: Counter(),
|
||
ls.PATH_BROKER: Counter()}
|
||
states, whys = Counter(), Counter()
|
||
samples: dict = {}
|
||
for k in codes:
|
||
a = ls.from_claims(claims.get(k), ds, stale_days=config.LOGIC_STALE_DAYS)
|
||
# 一只票挂在多个环节上时，取第一个评析过的那个环节的研判。
|
||
row = next((by_seg[t] for t in seg_of.get(k, []) if t in by_seg), None)
|
||
b = ls.from_judgement(row)
|
||
c = brokers.get(k) or ls.signal(ls.PATH_BROKER, ls.SIG_NONE,
|
||
why="两个窗口里算不出可比的预测")
|
||
per_path[ls.PATH_CLAIM][a["signal"]] += 1
|
||
per_path[ls.PATH_JUDGE][b["signal"]] += 1
|
||
per_path[ls.PATH_BROKER][c["signal"]] += 1
|
||
r = ls.compose([a, b, c, ls.from_events()])
|
||
states[r["state"]] += 1
|
||
if r["why"]:
|
||
whys[f"{r['state']}·{r['why']}"] += 1
|
||
# 逻辑存疑与逻辑强化都是稀有且会触发动作的状态，全部列出来逐条看得见；
|
||
# 无法判断那两类只留一条样例，否则几千行没法读。
|
||
if r["state"] in (ls.STATE_STRONG, ls.STATE_DOUBT):
|
||
samples.setdefault(r["state"], []).append((k, r["reasons"][:3]))
|
||
elif r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT:
|
||
samples.setdefault("无法判断·证据矛盾", []).append((k, r["reasons"][:3]))
|
||
|
||
print("每路各自的信号分布")
|
||
for path, cnt in per_path.items():
|
||
tot = sum(cnt.values())
|
||
line = "、".join(f"{s} {n}（{n / tot:.1%}）" for s, n in cnt.most_common())
|
||
print(f" {path}：{line}")
|
||
print(f" {ls.PATH_EVENT}：缺失 {len(codes)}（100.0%，无数据源）\n")
|
||
|
||
print("合成后的四态分布")
|
||
for s, n in states.most_common():
|
||
print(f" {s} {n}（{n / len(codes):.1%}）")
|
||
if whys:
|
||
print("\n 无法判断的子因")
|
||
for w, n in whys.most_common():
|
||
print(f" {w} {n}")
|
||
for tag, rows in samples.items():
|
||
cap = len(rows) if tag in (ls.STATE_STRONG, ls.STATE_DOUBT) else 2
|
||
print(f"\n{tag}（{len(rows)} 只{'，全部列出' if cap == len(rows) else '，列前两只'}）")
|
||
for k, rs in rows[:cap]:
|
||
print(f" {k}")
|
||
for x in rs:
|
||
print(f" {x}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main(sys.argv[1] if len(sys.argv) > 1 else None,
|
||
sys.argv[2] if len(sys.argv) > 2 else None)
|