From 8978b9dfe58cb36d68ebfac5fc4e75b803ce7b2b Mon Sep 17 00:00:00 2001 From: zlt Date: Thu, 3 Sep 2026 11:44:16 +0800 Subject: [PATCH] =?UTF-8?q?=E9=80=89=E8=82=A1=E7=B3=BB=E7=BB=9F=E6=8E=A5?= =?UTF-8?q?=E9=80=9A=E5=88=86=E6=9E=90=E7=BB=93=E8=AE=BA=EF=BC=9A=E5=80=99?= =?UTF-8?q?=E9=80=89=E5=8D=A1=E8=AF=BB=E5=9B=A0=E6=9E=9C=E8=AE=BA=E6=96=AD?= =?UTF-8?q?=E4=BD=9C=E8=AF=81=E6=8D=AE=E7=BA=BF=E3=80=81=E5=85=B3=E6=B3=A8?= =?UTF-8?q?=E5=88=A4=E5=86=B3=E7=BB=86=E5=8C=96=E4=B8=BA=E7=B3=BB=E7=BB=9F?= =?UTF-8?q?=E6=97=A0=E6=B3=95=E5=88=A4=E6=96=AD=E3=80=81=E8=AE=A1=E5=88=92?= =?UTF-8?q?=E7=8E=AF=E5=A2=83=E6=AE=B5=E5=8A=A0=E5=B8=82=E5=9C=BA=E5=9B=9B?= =?UTF-8?q?=E9=A1=B9=E3=80=81=E5=85=A5=E6=B1=A0=E4=B8=8A=E4=B8=8B=E6=96=87?= =?UTF-8?q?=E8=A1=A5=E8=AF=81=E6=8D=AE=E5=AD=97=E6=AE=B5=E3=80=81=E5=A4=8D?= =?UTF-8?q?=E7=9B=98=E5=8A=A0=E5=9B=9B=E4=BB=BD=E5=90=8D=E5=8D=95=E4=B8=8E?= =?UTF-8?q?=E4=B8=89=E5=A5=97=E5=AF=B9=E7=85=A7=E5=8F=B0=E8=B4=A6=E5=AF=B9?= =?UTF-8?q?=E8=A1=A8=E4=B8=A4=E8=8A=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 判决定义按台账 013：关注只保留三门槛全过无硬风险但确认线缺失或陈旧；只差覆盖与潜在吸筹归仅展示。人工裁决说明同步修订。 Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 12 +- README.md | 4 +- api.py | 12 +- card.py | 86 +++++++++-- common.py | 2 + config.py | 17 ++ docs/人工裁决看什么_2026-09-02.md | 57 ++++--- docs/复盘决定台账.md | 9 +- freeze.py | 14 +- plan.py | 96 ++++++++++-- plan_review.py | 149 +++++++++++++++++- pool.py | 57 +++++-- regime.py | 12 +- sources.py | 247 +++++++++++++++++++++++++++++- test_card.py | 71 ++++++--- test_market_context.py | 189 +++++++++++++++++++++++ test_pool_logic.py | 45 +++++- 17 files changed, 980 insertions(+), 99 deletions(-) create mode 100644 test_market_context.py diff --git a/CLAUDE.md b/CLAUDE.md index 310fa02..6866386 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -33,15 +33,17 @@ ## 测试（改完必跑） -纯逻辑单测三个，不连库。test_plan_verdict.py 全离线，开发机系统 Python 直接跑；另两个需要 pandas 与 fastapi 实体，在有 Docker 的机器用一次性容器跑全套： +纯逻辑单测六个，都不连库。开发机系统 Python（3.9，无 pandas）能直接跑五个：test_plan_verdict.py、 +test_card.py、test_regime.py、test_market_context.py、test_pool_logic.py（后两个只给缺席的依赖装最小桩， +依赖存在时不覆盖）；test_xxl_trigger.py 需要 fastapi 实体，在有 Docker 的机器用一次性容器跑全套： # 开发机（零依赖，秒出） - python3 test_plan_verdict.py + for t in test_plan_verdict.py test_card.py test_regime.py test_market_context.py test_pool_logic.py; do echo == $t ==; python3 $t || break; done - # 基座机 tlai4090（全套三个，一次性容器，跑完即弃） - rm -rf ~/akg_tmp_bridge_test && git clone -q git@192.168.18.24:zlt/akg-factor-bridge.git ~/akg_tmp_bridge_test && docker run --rm -v ~/akg_tmp_bridge_test:/w -w /w python:3.12-slim bash -c "pip install -q -r requirements.txt; for t in test_plan_verdict.py test_pool_logic.py test_xxl_trigger.py; do echo == \$t ==; python \$t || exit 1; done"; docker run --rm -v ~/akg_tmp_bridge_test:/w python:3.12-slim rm -rf /w/data /w/__pycache__; rm -rf ~/akg_tmp_bridge_test + # 基座机 tlai4090（全套六个，一次性容器，跑完即弃） + rm -rf ~/akg_tmp_bridge_test && git clone -q git@192.168.18.24:zlt/akg-factor-bridge.git ~/akg_tmp_bridge_test && docker run --rm -v ~/akg_tmp_bridge_test:/w -w /w python:3.12-slim bash -c "pip install -q -r requirements.txt; for t in test_plan_verdict.py test_card.py test_regime.py test_market_context.py test_pool_logic.py test_xxl_trigger.py; do echo == \$t ==; python \$t || exit 1; done"; docker run --rm -v ~/akg_tmp_bridge_test:/w python:3.12-slim rm -rf /w/data /w/__pycache__; rm -rf ~/akg_tmp_bridge_test -预期：三个测试分别打出 ALL OK 与 ALL PASS。2026-09-01 实测全过。 +预期：各测试分别打出 ALL OK 或 ALL PASS。2026-09-01 实测三个全过；2026-09-03 开发机五个全过（容器全套待跑）。 ## 代码地图 diff --git a/README.md b/README.md index 5bacf6a..b676627 100644 --- a/README.md +++ b/README.md @@ -166,9 +166,9 @@ docker compose exec akg-factor-bridge python run.py freeze --date 2026-07-24 | 端点 | 作用 | |---|---| -| `GET /health` | 存活探针 | +| `GET /health` | 存活探针；2026-09-03 起带 `pool_top`（POOL_TOP）与 `pool_max`（POOL_MAX），供 PMS 池深探针比对 | | `GET /plan/dates?limit=30` | 有计划文件的日期清单 | -| `GET /plan?date=&format=json\|md&top=&obs_top=` | 当日选股计划（基座今日页 `/plan/today` 与 PMS 下单候选都读它） | +| `GET /plan?date=&format=json\|md&top=&obs_top=` | 当日选股计划（基座今日页 `/plan/today` 与 PMS 下单候选都读它）。顶层 `regime`（区制）与 `market`（两市成交额、广度、融资、恐贪四项）都只从当日快照读，快照缺失为空；每行的判决类字段 2026-09-03 起多 `basis`（判决依据）与 `logic`（因果论断带出处） | | `POST /plan/refresh?date=` | 重生成当日计划文件 | | `GET /plan/verdict?codes=300750,SH600438&date=` | 逐票『计划判决』（2026-08-18 PMS 对接加）：decision（main/observe/reject/absent）+ 与命令行 `plan_reconcile` 逐字同口径的 verdict_text + score/rank/tier/赛道/图谱证据。单票数据异常或代码形态认不出只报该票 error，不崩整批 | | `GET\|POST /api/v1/xxl/daily-build?key=&steps=&date=` | XXL-JOB 平台触发盘前链（build→plan→push-pool，白名单步骤、单实例互斥、回调结案）；`.env` 不配 `XXL_TRIGGER_KEY` 则整组禁用 | diff --git a/api.py b/api.py index 53f5d60..e6b83f0 100644 --- a/api.py +++ b/api.py @@ -20,6 +20,7 @@ import pandas as pd from fastapi import FastAPI, HTTPException, Request from fastapi.responses import PlainTextResponse +import config import db import plan import plan_reconcile @@ -34,11 +35,14 @@ app.include_router(xxl_router) @app.get("/health") def health(): + """存活探针。2026-09-03 起带 pool_top 与 pool_max：PMS 的池深探针拿它与自身的计划深度 + 比较（池深不变式，台账 008），库连不上时也照样返回这两项。""" + depth = {"pool_top": config.POOL_TOP, "pool_max": config.POOL_MAX} try: d = plan._latest_date("t_factor_akg_score") # noqa: SLF001 —— 桥内自用 except Exception as e: # noqa: BLE001 —— 库连不上也要能回答"我还活着" - return {"ok": False, "error": repr(e)} - return {"ok": True, "latest_plan_date": d} + return {"ok": False, "error": repr(e), **depth} + return {"ok": True, "latest_plan_date": d, **depth} @app.get("/plan/dates") @@ -57,7 +61,8 @@ def get_plan(request: Request, date: str | None = None, format: str = "json", top: int = 20, obs_top: int = 10, theme_cap: int = 5): """向下兼容承诺（2026-09-02 方案 2.7）：main / observe 的装配、排序、裁剪与既有字段 一字不动，每行只多联入判决类字段；顶层只新增 generated_at、plan_version、regime、 - card_counts、candidates、watch、segments_pointed、snapshot。PMS 按字段名取值、忽略未知键。""" + card_counts、candidates、watch、segments_pointed、snapshot，2026-09-03 再加 market + （环境段市场四项，与 regime 一样只从当日快照读，快照缺失为空字典）。PMS 按字段名取值、忽略未知键。""" try: data = plan.collect(date, top, obs_top, theme_cap) except RuntimeError as e: @@ -68,6 +73,7 @@ def get_plan(request: Request, date: str | None = None, format: str = "json", data["snapshot"] = "present" if os.path.exists(regime.snapshot_path(ds)) else "missing" data["regime"] = reg or {"status": regime.UNKNOWN, "weak_day": None, "source": "当日快照无环境段（08:45 追加未跑或快照缺失）"} + data["market"] = regime.read_section(ds, "market") or {} _access.info("plan client=%s date=%s regime=%s generated_at=%s version=%s " "top=%s obs_top=%s theme_cap=%s", request.client.host if request.client else "-", ds, diff --git a/card.py b/card.py index 33bb38d..b6672ab 100644 --- a/card.py +++ b/card.py @@ -15,19 +15,29 @@ docs/主观选股改进方案_2026-09-02.md 第 1.8b 节）。方案把桥从" 硬风险 决策系统昨夜给出卖出、回避或剔除信号；传导快照日与计划日不符；吸筹评分为高位派发 判决 候选 = 三门槛全过、硬风险为空、确认成立 - 关注 = 硬风险为空，且（只差覆盖一项）或（门槛全过但无确认） - 仅展示 = 其余，卡上标明未过项 + 关注 = 三门槛全过、硬风险为空，但确认线缺失（没有吸筹评分）或陈旧（明确吸筹但评分 + 日龄超过上限或日龄未知）——这是"系统无法判断"，交人裁决 + 仅展示 = 其余，卡上标明未过项。只差券商覆盖、吸筹评分为潜在吸筹或其他非明确状态， + 都归这一档（2026-09-03 台账 013：潜在吸筹市值中性后为负，不升格） 依据：方案 1.8d——"环节被指向、当日已涨 3% 以上、且明确吸筹"是纯可交易口径下唯一 合并样本为正的规则（五日 +2.18、十日 +1.94），但它是顺风策略、不免疫环境；两个旋钮 -保持默认，等样本外复盘读数再拍，不在历史样本上挑参数（2026-09-02 拍板）。 +保持默认，等样本外复盘读数再定，不在历史样本上挑参数（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 从这里引用，桥内只此一处。 +坏信号集合 BAD_SIGNALS 是"三处同源"纪律里选股系统的那一份（另两处在择时决策系统 +pms_advisor.py 与 PMS rule_gate.py），改一处必须三处同改；pool.py 从这里引用，本仓库只此一处。 """ from __future__ import annotations @@ -68,7 +78,12 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0, accum_age int 评分日龄（交易日） y_signal str 决策系统昨夜 signal_type stale_snapshot bool 传导快照日与计划日不符 - 返回：verdict / reasons / missing / risk / gates / confirm。 + 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")) @@ -86,6 +101,12 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0, } 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: @@ -99,12 +120,17 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0, if pct0 is None: missing.append("无行情") if not ev.get("covered"): - missing.append("无券商覆盖") - if not accum_state: + missing.append("无券商覆盖（不升格关注，归仅展示）") + elif not gates["covered"]: + missing.append(f"预期空间 {upside:+.0%} 低于负容忍线" if upside is not None else "预期空间缺失") + if confirm_missing: missing.append("无吸筹评分（未入池）") - elif accum_state.startswith("明确") and not accum_fresh: + 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("无传导") @@ -122,20 +148,52 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0, reasons.append(f"券商覆盖，预期空间 {upside:+.0%}") all_gates = all(gates.values()) - only_missing_coverage = (gates["pointed"] and gates["started"] and gates["clean_name"] - and not gates["covered"]) + 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 - elif only_missing_coverage or (all_gates and not confirm): + 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) - failed = [k for k, ok in gates.items() if not ok] return {"verdict": verdict, "reasons": reasons, "missing": missing, "risk": risk, - "gates": gates, "confirm": confirm, "failed_gates": failed} + "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: diff --git a/common.py b/common.py index 743d62d..4147a63 100644 --- a/common.py +++ b/common.py @@ -1,4 +1,6 @@ """共用：股票码规范化、覆盖池、交易日历、幂等写因子表、注册 factor_metadata（自适应列）。""" +from __future__ import annotations # 注解不在定义时求值：开发机的 Python 3.9 也能导入本模块跑离线单测 + import json import pandas as pd diff --git a/config.py b/config.py index c082602..0dc2fc3 100644 --- a/config.py +++ b/config.py @@ -211,3 +211,20 @@ POOL_SOURCE = os.environ.get("POOL_SOURCE", "tier") # 低优先入池切片：让"门槛全过但缺吸筹评分"的关注票入池、当晚获得评分，否则复盘缺数据是环状依赖。 # 0 = 不开（默认）；受 POOL_MAX 约束，只改 Mongo 池成分，PMS 不读 Mongo 池。 POOL_WATCH_SLICE = int(os.environ.get("POOL_WATCH_SLICE", "0")) +# 股票池分组文档的 pool_type 字段。以前写死为 "core"，现在改为可配置，默认值不变， +# 写入的分组文档与择时决策系统读到的形状完全一样；只有在择时决策系统按 pool_type 区分 +# 池子用途时才需要改它（2026-09-03 方案第 3.3 节"入池上下文补证据字段"）。 +POOL_TYPE = os.environ.get("POOL_TYPE", "core") + +# --- 计划环境段的市场四项（2026-09-03 方案第 1.4 节与第 3.3 节"环境段扩展"）------------------ +# 两市成交额读平台行情库的指数日线表 zs_day_data，融资余额读 eastmoney_rzrq_data， +# 恐贪指数读 fear_greed_index。这三张表默认与个股日线 gp_day_data 在同一个 MySQL 实例 +# （PRICE_MYSQL_*，默认复用平台因子库），所以默认取 "price"；若它们实际落在 153 代理库， +# 把这个值改成 "heat" 或 "pms" 即可，不必改代码。可选值就是 db.read_mysql 认的四个名字： +# price / factor / heat / pms。每一项读失败都只是环境段里该项为空，不阻断出计划。 +MARKET_MYSQL_SOURCE = os.environ.get("MARKET_MYSQL_SOURCE", "price").strip().lower() + +# --- 候选卡的因果论断证据线（2026-09-03 方案第 3.3 节"候选卡读因果论断"）---------------------- +# 每只票从数据基座的因果论断视图 v_factor_logic 取最近披露日的最多几条论断挂在卡上。 +# 只展示、不作门槛、不进判决；设 0 表示不读该视图（视图未建时也可用它关掉那一行告警）。 +LOGIC_CLAIMS_PER_STOCK = int(os.environ.get("LOGIC_CLAIMS_PER_STOCK", "3")) diff --git a/docs/人工裁决看什么_2026-09-02.md b/docs/人工裁决看什么_2026-09-02.md index 1a84075..2354002 100644 --- a/docs/人工裁决看什么_2026-09-02.md +++ b/docs/人工裁决看什么_2026-09-02.md @@ -1,35 +1,52 @@ -# 人工裁决看什么（过渡期使用说明，2026-09-02） +# 人工裁决看什么（2026-09-02 首版，2026-09-03 按新方案修订） -这份说明给裁决 PMS 新建仓提议的人看。PMS 自 09-02 起新建仓只提议不自动执行；它送来的提议全部来自桥的强传导档，而这一档过去一个月五日超额约负 2 个百分点、跑赢比例三成三。所以裁决的默认答案是"不进"，只有桥当日计划里判为"候选"的票才考虑批。台账 011 记了这条原则。 +这份说明给裁决 PMS 新建仓提议的人看。 -## 一、在哪里看 +2026-09-03 的修订改了两件事。第一，称呼统一：选股系统指 akg-factor-bridge，择时决策系统指 bionic_trader，数据基座指 astock-kg。第二，也是更重要的一条，"关注"判决的含义变了：它现在专指"系统无法判断"，要交人裁决，而不是一律不批。依据是《主观量化系统方案_2026-09-03》第 3.2 节与复盘决定台账第 013 条。 -桥的当日计划有两种形式，都在 155 的桥接口上。 +## 一、三种判决各自意味着什么 -- Markdown：浏览器打开 `http://192.168.16.155:8300/plan?format=md`。看"候选单""关注环节""关注单"三节。09-03 起可用。 -- JSON：`http://192.168.16.155:8300/plan?top=300&obs_top=100&theme_cap=999`，让上游不裁剪。每行带判决字段。这一形式要等桥容器重启加载新接口代码后才带新字段。 +选股系统每天对每只票给一个判决，只有三种。 -PMS 的面板不显示判决，它只读排名、代码、分数、档位、预期空间。所以人必须另开桥的计划看。 +**候选**：三条门槛全过、无硬风险、且明确吸筹评分新鲜。这是系统给出的正面判断，进入建仓方案与研判闸。 -## 二、逐票看五样东西 +**关注**：三条门槛全过、无硬风险，但确认线缺失或陈旧——要么这只票不在择时决策系统的夜间分析池里、没有资金结构评分，要么评分是明确吸筹但已经超过三十个交易日或日龄不明。**这就是系统在说"我无法判断"，交人裁决。** 人在这里做的是系统做不了的那件事：补上催化剂、预期、失效条件的判断，或者直接说今天不做。 -第一，判决（verdict）。三个值：候选、关注、仅展示。只有"候选"进入考虑范围。强传导档里大多数票会是"仅展示"，未过项通常是"未启动"，这本身就是拒的理由。 +**仅展示**：其余全部。包括门槛没过的，也包括三条门槛全过但吸筹评分为潜在吸筹、信号不明等非明确状态的票。后一种不是系统不知道，是系统已经判断过、只是没达到确认线，所以不升格为关注（潜在吸筹在市值中性后为负，见台账 013）。仅展示不出提议。 -第二，理由（reasons）。候选的四条理由是：所在环节被几路传导指向；数据日涨幅是多少（已启动）；明确吸筹的评分与评分日龄；券商覆盖与预期空间。预期空间超过 100% 按噪音看。 +## 二、在哪里看 -第三，硬风险（risk）。三条里有任一条就拒：决策系统昨夜信号是 SELL、AVOID 或 DROPPED；传导快照日与计划日不符；吸筹评分为高位派发。 +选股系统的当日计划有两种形式，都在 155 的接口上。 -第四，缺失（missing）。"无吸筹评分"表示这只票不在决策系统夜间分析池里，没有资金结构读数，不当候选看。"评分陈旧 N 日"表示评分超过三十个交易日，同样不当候选看。 +- Markdown：浏览器打开 `http://192.168.16.155:8300/plan?format=md`。看"候选单""关注单""关注环节"三节。 +- JSON：`http://192.168.16.155:8300/plan?top=300&obs_top=100&theme_cap=999`，让上游不裁剪。每行带判决字段。 -第五，关注环节一节。看这只票所在环节的已启动成员数、领涨者与涨幅。环节里已启动的比例很高时，说明这一轮可能已到尾声，即使是候选也要更谨慎。这一节回答的是"刚启动还是尾声"。 +PMS 的提议卡会显示判决与理由（09-03 方案的改动落地后）。判决字段在 PMS 面板上看得到之前，仍需另开选股系统的计划核对。 -## 三、不看什么，不做什么 +## 三、逐票看六样东西 -- 环境标签（八个指数里弱势的个数）只当背景，不当规则。它还没有验证过预测力。 -- "关注"判决的票不批。关注桶里混着只差覆盖、已动但缺确认、潜在吸筹三种情况，没有一种有稳定的正证据。 -- 决策系统昨夜的 BUY 信号可以看，但它不改变资格。 -- 不用桥的读数定持有期、止损或仓位，那些留给 PMS 自己的旋钮。 +第一，判决（verdict）与判决依据（basis）。依据是一句话，直接说清这只票为什么落在这一档。 -## 四、留痕 +第二，理由（reasons）。候选的四条理由是：所在环节被几路传导指向；数据日涨幅是多少；明确吸筹的评分与评分日龄；券商覆盖与预期空间。预期空间超过 100% 按噪音看。 -每次拒绝都在 PMS 里写明原因。复盘脚本会把被拒绝的票单列成一份名单，与其他名单同口径算收益，回答"拒了的后来涨了多少"。人批的票与人拒的票分开算，这是过渡期唯一能读出"人在环路有没有价值"的方式。 +第三，硬风险（risk）。三条里有任一条就拒：择时决策系统昨夜信号是 SELL、AVOID 或 DROPPED；传导快照日与计划日不符；吸筹评分为高位派发。 + +第四，缺失（missing）。它说明关注这一档具体缺什么："无吸筹评分"是这只票不在夜间分析池里，"评分陈旧 N 日"是评分过期。这两种正是要人裁决的情形，不是拒绝的理由。 + +第五，因果论断（logic）。数据基座从研报里抽出的论断：谁利好或利空这家公司、机制是什么、多长时效、出处是哪份文档。只作展示，不进判决。这是判断"论点还成立吗"的材料。 + +第六，关注环节一节。看这只票所在环节的已启动成员数、领涨者与涨幅。环节里已启动的比例很高时，说明这一轮可能已到尾声，即使是候选也要更谨慎。这一节回答的是"刚启动还是尾声"。 + +## 四、裁决原则 + +候选：系统的正面判断，人可以否决，否决要写明理由。 + +关注：人独立裁决。可用的判断依据是因果论断、关注环节的时序、以及系统里没有的信息——催化剂时点、政策与监管、盘面直觉。 + +仅展示：不出提议，不需要裁决。 + +不看什么：环境标签（八个指数里弱势的个数）只当背景，不当规则，它还没有验证过预测力。择时决策系统昨夜的 BUY 信号可以看，但它不改变资格。不用选股系统的读数定持有期、止损或仓位，那些留给 PMS 自己的旋钮。 + +## 五、留痕 + +采纳与驳回都必须写明理由。复盘脚本按裁决者与判决把票分成机器通过、人批、人拒三份名单，与其他名单同口径算收益，回答"拒了的后来涨了多少"。人批与人拒分开算，这是读出"人在环路有没有价值"的唯一方式。 diff --git a/docs/复盘决定台账.md b/docs/复盘决定台账.md index acb1343..1337032 100644 --- a/docs/复盘决定台账.md +++ b/docs/复盘决定台账.md @@ -1,7 +1,7 @@ # 复盘决定台账 这份台账记每一条影响候选单的规则改动：改了什么、依据是哪份读数、预期看到什么、什么时候复核。 -复盘周报第五节"台账对表"逐条核对这里的预期有没有兑现。只记决定，不记讨论过程。 +复盘周报第八节"台账对表"逐条核对这里的预期有没有兑现。只记决定，不记讨论过程。 新条目追加在最后，不改旧条目；旧条目被推翻时在它下面加一行"撤销于某日，见某条"。 格式：日期、改动、依据、预期、复核日期。 @@ -98,3 +98,10 @@ - 依据：方案 3.2；用户要求系统能识别"我无法判断"再由人裁决；潜在吸筹市值中性后为负、不升格（1.8c）。 - 预期：关注数减少；PMS 人工队列只剩关注态与不可用；人批与人拒理由全部落账本。 - 复核日期：随一致性检查表第三行。 + +## 014 · 2026-09-03 · 因果论断与研判结论进候选卡（数据基座两张只读视图建成） + +- 改动：数据基座库新增 v_factor_logic（因果论断，一行一条，客体是环节时展开到成员并标 via_segment）与 v_factor_judgement（研判结论，读评析表 logic_reviews）。选股系统候选卡把因果论断作证据线展示，不进判决。 +- 依据：方案 2.1"研究深度"与 3.3 数据基座表。建成日读数：论断 7,083 条；视图链接 1,996 行、724 只票；公司类论断只链上三成三（解析精确层），是后续提升点。 +- 预期：候选卡每票理由能看到"谁利好谁、机制、时效、出处"；一致性检查表第一行由"否"转"是"。 +- 复核日期：随一致性检查表第一、七行。 diff --git a/freeze.py b/freeze.py index b18b54a..077394a 100644 --- a/freeze.py +++ b/freeze.py @@ -29,13 +29,13 @@ from __future__ import annotations import datetime as dt import json -import subprocess from pathlib import Path import pandas as pd import config import db +import version try: # parquet 更省更快，但不强制装 pyarrow import pyarrow # noqa: F401 @@ -57,13 +57,11 @@ def _write(df: pd.DataFrame, path: Path) -> int: def _git_rev() -> str: - """记录冻结时的桥代码版本——快照可复算的前提是知道当时的口径。""" - try: - return subprocess.run(["git", "rev-parse", "--short", "HEAD"], - cwd="/app", capture_output=True, text=True, - timeout=5).stdout.strip() or "unknown" - except Exception: # noqa: BLE001 —— 无 git / 无 .git 都不该让冻结失败 - return "unknown" + """记录冻结时的代码版本——快照可复算的前提是知道当时的口径。 + 2026-09-03 改用 version.git_short_rev()：容器镜像没装 git，原来调 git 命令在容器里恒为 + unknown（155 的 manifest.json 实测）；version.py 直接解析挂载进来的 .git 文件，与计划快照的 + plan_version 同一来源，任何失败返回 "unknown"，不让冻结失败。""" + return version.git_short_rev() # ---------------------------------------------------------------- 各源抓取 diff --git a/plan.py b/plan.py index a8acf75..224e3e1 100644 --- a/plan.py +++ b/plan.py @@ -10,7 +10,14 @@ 基座 industry_pools —— 股票名称 升降档一节对比前一交易日的档位表——数据到达本身是信号（首次覆盖 / 新进传导链即升档）。 + +2026-09-03 起（《主观量化系统方案_2026-09-03》第 3.3 节）：候选卡每票多带数据基座的因果论断 +证据线（只展示不进判决，sources.logic_claims）；generate 出计划时把市场四项（两市成交额、广度、 +融资、恐贪，sources.market_context）写进快照的 market 段，与 08:45 追加的 regime 段并列， +接口 /plan 只从快照读这两段。 """ +from __future__ import annotations # 注解不在定义时求值：开发机的 Python 3.9 也能导入本模块跑离线单测 + import datetime as dt import json import os @@ -117,6 +124,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series, moved = sources.moved_members(ds) daily = sources.stock_daily(ds) night = sources.night_conclusions(codes, ds) + # 因果论断（2026-09-03）：数据基座抽取的论断挂在卡上作证据线，只展示不进判决；视图未建时为空。 + logic = sources.logic_claims(codes, ds) if risk is None: # collect 会传入读过一次的名单；单独调用时自己读 try: risk = factors._risk_set() or set() # noqa: SLF001 —— 同仓自用 @@ -136,7 +145,7 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series, "risk_name": k in risk, "accum_state": n.get("accum_state"), "accum_score": n.get("accum_score"), "accum_age": n.get("accum_age"), "y_signal": n.get("signal"), - "stale_snapshot": stale} + "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) @@ -144,6 +153,7 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series, **j, "theme": theme, "n_sources": n_sources, "chain_fit": evd["chain_fit"], "started_source": "moved_view" if mv else None, + "logic_claims": evd["logic"], "pct0": d.get("pct0"), "net_z": d.get("net_z"), "heat_chg": d.get("heat_chg"), "accum": ({"state": n.get("accum_state"), "score": n.get("accum_score"), "age": n.get("accum_age"), "pos_tag": n.get("accum_pos_tag"), @@ -258,12 +268,17 @@ def collect(date: str | None = None, top: int = 20, obs_top: int = 10, if with_tier: r["tier"] = _tier_label(s) if c: + # 2026-09-03 新增只联入：basis（判决依据一句话）、logic（因果论断带出处的文字行）、 + # card.chain_fit（链符，入池上下文用）、card.logic_claims（论断原值）、card.failed_gates。 r.update(verdict=c["verdict"], reasons=c["reasons"], missing=c["missing"], risk=c["risk"], card_rank=c["card_rank"], + basis=c.get("basis"), logic=c.get("logic") or [], card={"pct0": c.get("pct0"), "net_z": c.get("net_z"), "heat_chg": c.get("heat_chg"), "accum": c.get("accum"), "night": c.get("night"), "gates": c.get("gates"), - "confirm": c.get("confirm")}) + "confirm": c.get("confirm"), "failed_gates": c.get("failed_gates"), + "chain_fit": c.get("chain_fit"), + "logic_claims": c.get("logic_claims") or []}) return r def _full_rows(ranked: pd.Series, with_tier: bool) -> list: @@ -379,8 +394,11 @@ def collect(date: str | None = None, top: int = 20, obs_top: int = 10, "card_params": {"start_pct": config.CARD_START_PCT, "accum_max_age": config.CARD_ACCUM_MAX_AGE, "neg_tol": config.UPSIDE_NEG_TOLERANCE, + "logic_per_stock": config.LOGIC_CLAIMS_PER_STOCK, "rules": "候选=环节被指向∧当日涨幅达标∧券商覆盖且非ST∧明确吸筹∧无硬风险；" - "关注=无硬风险且(只差覆盖 或 门槛全过无确认)；其余仅展示"}, + "关注=三门槛全过∧无硬风险∧确认线缺失或陈旧（系统无法判断，交人裁决）；" + "其余仅展示（只差覆盖、潜在吸筹等非明确状态都不升格，台账 013）；" + "因果论断只展示不进判决"}, "card_counts": card_counts, "candidates": _by_verdict(card.VERDICT_CANDIDATE), "watch": _by_verdict(card.VERDICT_WATCH), @@ -417,6 +435,56 @@ def _fmt_accum(ac: dict) -> str: return f"{st}（{age} 日前）" if isinstance(age, int) else st +def _fmt_logic(lines, width: int = 60) -> str: + """候选单与关注单表格里的因果论断列：第一条截到 width 字，多于一条时带条数；表格里不能有竖线。""" + lines = [str(x) for x in (lines or []) if x] + if not lines: + return "—" + first = lines[0].replace("|", "／") + if len(first) > width: + first = first[:width] + "…" + return f"{first}（共 {len(lines)} 条）" if len(lines) > 1 else first + + +def _fmt_amount(v) -> str: + """成交额按元换算成亿显示；原表单位若不是元，读数需要按 market.turnover.unit 说明校正。""" + if v is None: + return "—" + return f"{v / 1e8:,.0f} 亿" if abs(v) >= 1e8 else f"{v:,.0f}" + + +def _fmt_market(m: dict) -> str: + """环境段市场四项的一行文字；缺的项写"—"，读数原因在 JSON 的 market.errors 里。""" + t, b, mg, fg = m.get("turnover") or {}, m.get("breadth") or {}, m.get("margin") or {}, m.get("fear_greed") or {} + parts = [] + if t: + ratio = t.get("ratio_vs_prev5") + parts.append(f"两市成交额 {_fmt_amount(t.get('amount'))}" + + (f"（前五日均值的 {ratio:.2f} 倍）" if ratio else "") + + (f"，数据日 {t['data_date']}" if t.get("data_date") and t.get("data_date") != m.get("date") else "")) + else: + parts.append("两市成交额 —") + if b: + med = b.get("pct_median") + parts.append(f"广度 上涨 {b.get('up')} / 下跌 {b.get('down')} 家，涨停近似 {b.get('limit_up_approx')} 家，" + f"涨幅中位数 {med:+.2f}%" if med is not None else + f"广度 上涨 {b.get('up')} / 下跌 {b.get('down')} 家") + else: + parts.append("广度 —") + if mg: + bal, chg = mg.get("financing_balance"), mg.get("change_percent_5d") + parts.append(f"融资余额 {_fmt_amount(bal)}" + + (f"（五日变化 {chg:+.2f}%）" if chg is not None else "") + + (f"，{mg['date']}" if mg.get("date") else "")) + else: + parts.append("融资余额 —") + if fg and fg.get("index_value") is not None: + parts.append(f"恐贪指数 {fg['index_value']:.0f}" + (f"（{fg['date']}）" if fg.get("date") else "")) + else: + parts.append("恐贪指数 —") + return "市场环境：" + "；".join(parts) + "（只展示与复盘分组，不作交易前置）。" + + def render_md(d: dict) -> str: L = [f"# 每日选股计划 · {d['date']}", ""] c = d["counts"] @@ -437,8 +505,8 @@ def render_md(d: dict) -> str: if not cands: L.append("（今日无候选——候选为空不是故障：环节没被指向、成员没启动或没有明确吸筹，都会为空。）") else: - L.append("| # | 代码 | 名称 | 环节 | 源数 | 当日涨幅 | 吸筹 | 预期空间 | 理由 |") - L.append("|---|------|------|------|------|----------|------|----------|------|") + L.append("| # | 代码 | 名称 | 环节 | 源数 | 当日涨幅 | 吸筹 | 预期空间 | 理由 | 因果论断（出处） |") + L.append("|---|------|------|------|------|----------|------|----------|------|------------------|") for r in cands: c = r.get("card") or {} ac = c.get("accum") or {} @@ -446,7 +514,7 @@ def render_md(d: dict) -> str: L.append(f"| {r['rank']} | {r['code']} | {r['name'] or '—'} | {ev_.get('theme') or '—'} " f"| {ev_.get('n_sources') or '—'} | {_fmt_pct0(c.get('pct0'))} " f"| {_fmt_accum(ac)} | {_fmt_pct(r.get('upside'))} " - f"| {'；'.join(r.get('reasons') or [])} |") + f"| {'；'.join(r.get('reasons') or [])} | {_fmt_logic(r.get('logic'))} |") L.append("") segs = d.get("segments_pointed") or [] L.append(f"## 关注环节（今日被传导指向的 {len(segs)} 个环节：定位对不对看这里，挑票看候选单）") @@ -463,23 +531,28 @@ def render_md(d: dict) -> str: f"| {s.get('candidates', 0)} |") L.append("") watch = d.get("watch") or [] - L.append(f"## 关注单（无硬风险，只差券商覆盖或缺明确吸筹；共 {cc.get('关注', len(watch))} 只，列前 20）") + L.append(f"## 关注单（三门槛全过、无硬风险，但吸筹确认线缺失或陈旧——系统无法判断，交人裁决；" + f"共 {cc.get('关注', len(watch))} 只，列前 20）") L.append("") if watch: - L.append("| # | 代码 | 名称 | 环节 | 当日涨幅 | 吸筹 | 缺什么 |") - L.append("|---|------|------|------|----------|------|--------|") + L.append("| # | 代码 | 名称 | 环节 | 当日涨幅 | 吸筹 | 缺什么 | 因果论断（出处） |") + L.append("|---|------|------|------|----------|------|--------|------------------|") for r in watch[:20]: c = r.get("card") or {} ev_ = r.get("evidence") or {} L.append(f"| {r['rank']} | {r['code']} | {r['name'] or '—'} | {ev_.get('theme') or '—'} " f"| {_fmt_pct0(c.get('pct0'))} | {_fmt_accum(c.get('accum') or {})} " - f"| {'；'.join(r.get('missing') or [])} |") + f"| {'；'.join(r.get('missing') or [])} | {_fmt_logic(r.get('logic'))} |") L.append("") reg = d.get("regime") if reg: L.append(f"环境标签：{reg.get('status')}，弱势指数 {reg.get('weak_count')}/8" f"{'，弱势日' if reg.get('weak_day') else ''}（只展示与复盘分组，不作交易前置）。") L.append("") + mk = d.get("market") + if mk: + L.append(_fmt_market(mk)) + L.append("") cap_txt = f"，每主题限额 {d['theme_cap']}" if d["theme_cap"] else "" gate_txt = "、在十五五赛道内" if d.get("gate_on") else "" @@ -542,6 +615,9 @@ def generate(date: str | None = None, top: int = 20, obs_top: int = 10, data = collect(date, top, obs_top, theme_cap) except RuntimeError as e: raise SystemExit(str(e)) + # 环境段的市场四项（2026-09-03）：出计划时读一次落进快照，接口 /plan 只从快照读，盘中不再取数； + # 每项读失败为空并把原因记在 market.errors，不阻断。区制段仍由 08:45 的追加步骤写入。 + data["market"] = sources.market_context(data["date"]) text = render_md(data) os.makedirs(config.PLAN_SNAPSHOT_DIR, exist_ok=True) out = os.path.join(config.PLAN_SNAPSHOT_DIR, f"plan_{data['date']}.md") diff --git a/plan_review.py b/plan_review.py index 3524fd7..d650a0c 100644 --- a/plan_review.py +++ b/plan_review.py @@ -23,6 +23,16 @@ score_lab（两套读数工具两种口径的坑），也承接了 09-02 手工 主榜等权 当日全部主榜 全池等权 基座行情快照当日全部个股（基准） +## 另四份名单（2026-09-03 方案第 3.3 节"复盘四份名单与对照节"） + + 机器通过名单 PMS 动作账本 pms_action_ledger 当日 action='OPEN'、arbiter='judge'、verdict='PASS' 的票 + 人批名单 同表当日 arbiter='user'、verdict='PASS' + 人拒名单 同表当日 arbiter='user'、verdict='REJECT' + 择时看多名单 择时决策系统结论表 strategy_daily_results 当日 signal_type='BUY' 的票 + 账本按 decided_at 的日历日筛，代码统一转前缀式；任一路读失败该名单为空并在逐日注记里说明。 + 周报第七节"三套对照"把选股系统候选单、择时决策系统自评、PMS 账本四份名单摆在一张表里； + 第八节"台账对表"列出 docs/复盘决定台账.md 的条目，留"一致 / 不一致"两列给人填。 + ## 口径 起点价 next_close（默认，实盘买得到）：T 日出计划，T+1 收盘买，收益 = Σ pct[T+2 .. T+1+h] @@ -45,6 +55,7 @@ import argparse import datetime as dt import json import os +import re import pandas as pd @@ -55,6 +66,9 @@ import plan HORIZONS_DEFAULT = (5, 10, 20) MAIN_MIN = 150.0 +LEDGER_LISTS = ("机器通过名单", "人批名单", "人拒名单") +TIMING_LIST = "择时看多名单" +DECISION_LEDGER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "docs", "复盘决定台账.md") # ============================================================================ @@ -122,6 +136,84 @@ def pms_roster(day: str) -> tuple[list[str], str]: f"PMS 快照名册（N={n} 为现值，历史 N 不可还原）" +def _uniq_codes(values) -> list[str]: + """代码列 -> 去重、去占位符（账本里宏观闸等行的 ts_code 是 "-"）、转前缀式，保持出现顺序。""" + out, seen = [], set() + for v in values: + s = str(v or "").strip() + if not s or s == "-" or s.lower() == "nan": + continue + k = common.to_prefix(s.upper()) + if k not in seen: + seen.add(k) + out.append(k) + return out + + +def ledger_lists(day: str) -> tuple[dict[str, list[str]], str]: + """PMS 动作账本当日新建仓评审的三份名单：机器通过（研判闸 judge 放行）、人批、人拒。 + decided_at 按日历日筛（day 零点到次日零点）；读失败三份都为空并返回原因。""" + empty = {name: [] for name in LEDGER_LISTS} + nxt = (dt.date.fromisoformat(day) + dt.timedelta(days=1)).isoformat() + try: + df = db.read_mysql( + "pms", "SELECT ts_code, arbiter, verdict FROM pms_action_ledger " + "WHERE action = 'OPEN' AND decided_at >= %s AND decided_at < %s", (day, nxt)) + except Exception as e: # noqa: BLE001 + return empty, f"PMS 账本读取失败（{e!r}），机器通过、人批、人拒三份名单为空" + if df.empty: + return empty, "PMS 账本当日无新建仓评审行" + arb = df["arbiter"].astype(str).str.strip().str.lower() + vd = df["verdict"].astype(str).str.strip().str.upper() + return { + "机器通过名单": _uniq_codes(df.loc[(arb == "judge") & (vd == "PASS"), "ts_code"]), + "人批名单": _uniq_codes(df.loc[(arb == "user") & (vd == "PASS"), "ts_code"]), + "人拒名单": _uniq_codes(df.loc[(arb == "user") & (vd == "REJECT"), "ts_code"]), + }, f"PMS 账本当日评审行 {len(df)} 条" + + +def timing_bullish(day: str) -> tuple[list[str], str]: + """择时决策系统结论表当日 signal_type='BUY' 的票（trade_date 是整数 YYYYMMDD）。读失败为空。""" + try: + df = db.read_mysql( + "pms", "SELECT stock_code FROM strategy_daily_results " + "WHERE trade_date = %s AND signal_type = 'BUY'", (int(day.replace("-", "")),)) + except Exception as e: # noqa: BLE001 + return [], f"择时决策系统结论表读取失败（{e!r}），择时看多名单为空" + return _uniq_codes(df["stock_code"]) if not df.empty else [], "" + + +def timing_self_eval(since: str, until: str | None) -> str: + """择时决策系统自评口径与本期读数：判分表 decision_outcome 里日终策略（ref_type='strategy'） + 五日方向命中率——命中按原始收益方向判，不是超额（方案第 1.3 节列为已知缺陷）。读不到写"未接入"。""" + lo = int(since.replace("-", "")) + hi = int((until or dt.date.today().isoformat()).replace("-", "")) + try: + df = db.read_mysql( + "pms", "SELECT COUNT(*) AS n, SUM(dir_hit) AS hits FROM decision_outcome " + "WHERE ref_type = 'strategy' AND horizon = 5 AND dir_hit IS NOT NULL " + "AND base_date >= %s AND base_date <= %s", (lo, hi)) + n = int(df.iloc[0]["n"] or 0) if not df.empty else 0 + hits = int(df.iloc[0]["hits"] or 0) if not df.empty else 0 + except Exception as e: # noqa: BLE001 + return f"未接入（decision_outcome 读取失败：{type(e).__name__}）" + if n == 0: + return "未接入（区间内无已判分的日终策略行）" + return f"五日方向命中 {hits}/{n} = {hits / n * 100:.0f}%（基准日 {since} 至 {until or '今日'}）" + + +def decision_ledger_entries(path: str = DECISION_LEDGER_PATH) -> list[dict]: + """解析 docs/复盘决定台账.md 的条目标题行 "## 0NN · 日期 · 标题" -> [{no, date, title}]。 + 文件不存在或没有条目返回空列表，周报对表节据此写"台账文件缺失"。""" + try: + with open(path, "r", encoding="utf-8") as f: + text = f.read() + except OSError: + return [] + pat = re.compile(r"^##\s+(\d{3})\s*·\s*(\d{4}-\d{2}-\d{2})\s*·\s*(.+?)\s*$", re.M) + return [{"no": m.group(1), "date": m.group(2), "title": m.group(3)} for m in pat.finditer(text)] + + # ============================================================================ # 收益 # ============================================================================ @@ -203,6 +295,8 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str prod_note = f"{prod_note}；无快照，交付名单当日剔除" cands = [r["code"] for r in data.get("candidates") or []] watch = [r["code"] for r in data.get("watch") or []] + ledger, ledger_note = ledger_lists(day) + bullish, bullish_note = timing_bullish(day) caps = cap_bucket(day) acc = accum_by_day.get(day, {}) reg = None @@ -232,6 +326,8 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str "强传导交付名单": prod, "候选单": cands, "关注单": watch, "环节名单": seg_codes, "主榜等权": main_codes, "观察档等权": obs_codes, "全池等权": list(ret.dropna().index), + # 2026-09-03：PMS 账本三份与择时看多一份，与其余名单同口径算收益 + **ledger, TIMING_LIST: bullish, } for name, codes in lists.items(): s = summarize(ret, codes, base_all, base_main, caps, cap_base) @@ -267,7 +363,10 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str rows.append({"date": day, "h": h, "list": "候选单", "group": f"市值={cap}", "regime_post": regime_post, "regime_pre": regime_pre, **s}) notes.append(f"{day}: 主榜 {len(main_codes)} 观察 {len(obs_rows)} 候选 {len(cands)} " - f"关注 {len(watch)} 生产 {len(prod)}（{prod_note}）") + f"关注 {len(watch)} 生产 {len(prod)}（{prod_note}）；" + f"账本 机器通过 {len(ledger['机器通过名单'])} 人批 {len(ledger['人批名单'])} " + f"人拒 {len(ledger['人拒名单'])}（{ledger_note}）；择时看多 {len(bullish)}" + + (f"（{bullish_note}）" if bullish_note else "")) df = pd.DataFrame(rows) if df.empty: @@ -327,7 +426,11 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str else: md += ["## 四、按事前线上标签分组", "", "（区间内没有带环境标签的快照，本节待环境标签上线后出现。）", ""] md += ["## 五、逐日注记", ""] + [f"- {n}" for n in notes] + ["", - "## 六、拍板建议", "", "（只列读数与选项，不改任何东西——由每周五人工填写。）", ""] + "## 六、待决定事项", "", "（只列读数与选项，不改任何东西——由每周五人工填写。）", ""] + md += ["## 七、三套对照（选股系统、择时决策系统、PMS 账本各自的读数摆在一张表里，只对照不合并）", "", + _md(compare_table(lists_tbl, since, until)), ""] + md += ["## 八、台账对表（一致 / 不一致两列由人填；对表依据是方案第 4.1 节一致性检查表）", "", + _md(ledger_table()), ""] md_path = os.path.join(out_dir, f"复盘_{stamp}_{start}.md") with open(md_path, "w", encoding="utf-8") as f: f.write("\n".join(md)) @@ -336,6 +439,48 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str return {"md": md_path, "csv": csv_path, "days": len(days_plan)} +def _list_reading(lists_tbl: pd.DataFrame, name: str) -> str: + """第一节汇总表里一份名单各期限的按日等权读数，拼成一句；没有样本写明。""" + if lists_tbl is None or lists_tbl.empty or "list" not in lists_tbl.columns: + return "无样本（区间内该名单没有可算的期限：名单为空、读失败或数据尾部不足）" + sub = lists_tbl[lists_tbl["list"] == name].sort_values("h") + if sub.empty: + return "无样本（区间内该名单没有可算的期限：名单为空、读失败或数据尾部不足）" + parts = [] + for _, r in sub.iterrows(): + parts.append(f"{int(r['h'])} 日：计划日 {int(r['days'])}，样本 {int(r['n'])}，" + f"超额（全池）{r['excess_all']:+.2f}，跑赢 {r['beat']:.1f}%，纪律 {r['grade']}") + return "；".join(parts) + + +def compare_table(lists_tbl: pd.DataFrame, since: str, until: str | None) -> pd.DataFrame: + """第七节"三套对照"：三行——选股系统候选单按日等权读数；择时决策系统自评口径与本期方向命中率； + PMS 账本四份名单读数。三套口径不同，只并列不合并。""" + rows = [ + {"系统": "选股系统", "口径": "候选单按计划日等权，相对全池等权超额与跑赢比例（第一节主口径）", + "读数": _list_reading(lists_tbl, "候选单")}, + {"系统": "择时决策系统", + "口径": "自评：判分表 decision_outcome 日终策略五日方向命中率，按原始收益方向判、非超额（其判分脚本口径）", + "读数": timing_self_eval(since, until)}, + {"系统": "PMS 账本", + "口径": "动作账本当日 OPEN 行：机器通过（judge PASS）、人批（user PASS）、人拒（user REJECT）；" + "另列择时看多（结论表 BUY）；四份名单与候选单同口径算收益", + "读数": "；".join(f"{name}—{_list_reading(lists_tbl, name)}" + for name in (*LEDGER_LISTS, TIMING_LIST))}, + ] + return pd.DataFrame(rows) + + +def ledger_table() -> pd.DataFrame: + """第八节"台账对表"：台账条目编号、日期、标题，加"一致""不一致"两列空着给人填。""" + entries = decision_ledger_entries() + if not entries: + return pd.DataFrame([{"编号": "—", "日期": "—", "标题": f"台账文件缺失或无条目（{DECISION_LEDGER_PATH}）", + "一致": "", "不一致": ""}]) + return pd.DataFrame([{"编号": e["no"], "日期": e["date"], "标题": e["title"], "一致": "", "不一致": ""} + for e in entries]) + + def main() -> int: ap = argparse.ArgumentParser(description="候选单复盘（只读，名单级观察收益，不是回测）") ap.add_argument("--since", default="2026-07-29") diff --git a/pool.py b/pool.py index 9b983bd..8efa02d 100644 --- a/pool.py +++ b/pool.py @@ -26,6 +26,8 @@ python run.py push-pool # 真写（写完顺手触发决策系统的增量补扫） python run.py push-pool --no-kick # 写库但不触发补扫（比如夜间已近 22:30 全量扫） """ +from __future__ import annotations # 注解不在定义时求值：开发机的 Python 3.9 也能导入本模块跑离线单测 + import datetime as dt import urllib.parse import urllib.request @@ -126,6 +128,49 @@ def build_remark(d: dict, plan_date: str, now_str: str, degraded: str = "") -> d } +def segment_started_ratios(segments: list) -> dict: + """关注环节表 -> {环节名: 已启动比例}。比例优先用数据基座台账的 moved / members_total + （与传导视图的 moved_ratio 同口径），台账没给时退回已动成员视图数出来的 started_count； + 成员数缺失则为 None。plan.collect 返回的 segments_pointed 直接喂进来。""" + out = {} + for s in segments or []: + total = s.get("members_total") + if not total: + out[s.get("segment")] = None + continue + moved = s.get("moved") + if moved is None: + moved = s.get("started_count") + out[s.get("segment")] = round(float(moved) / float(total), 3) if moved is not None else None + return out + + +def strategy_context_entry(r: dict, plan_date: str, seg_ratio=None) -> dict: + """一行计划 -> 分组文档 strategy_context 里该票的条目（纯函数，test_pool_logic.py 直接测）。 + + 2026-08 起的既有键：factor_code / score / tier / upside / theme / plan_date / verdict / + reasons / card_rank；2026-09-03 方案第 3.3 节"入池上下文补证据字段"再加：chain_fit（链符）、 + n_sources（源数）、pct0（数据日涨幅）、accum_state 与 accum_age（吸筹三态与评分日龄）、 + segment_started_ratio（所在环节已启动比例）。都是只加字段，择时决策系统读池只取代码列，不受影响。 + source 与 plan_version 两个键由调用方补，它们不来自计划行本身。""" + card_ = r.get("card") or {} + ev = r.get("evidence") or {} + accum = card_.get("accum") or {} + theme = ev.get("theme") + return {"factor_code": "akg_score", "score": r.get("score"), + "tier": r.get("tier"), "upside": r.get("upside"), + "theme": theme, "plan_date": plan_date, + "verdict": r.get("verdict"), + "reasons": (r.get("reasons") or [])[:4], + "card_rank": r.get("card_rank"), + "chain_fit": card_.get("chain_fit"), + "n_sources": ev.get("n_sources"), + "pct0": card_.get("pct0"), + "accum_state": accum.get("state"), + "accum_age": accum.get("age"), + "segment_started_ratio": (seg_ratio or {}).get(theme) if theme else None} + + def build_recycle_docs(d: dict, group_id: str, group_name: str, org_id: str, now: dt.datetime) -> list: """回收站文档，字段照抄现有格式（group_id/group_name/org_id/removal_batch/ @@ -330,16 +375,10 @@ def push(date: str | None = None, top: int | None = None, remark = build_remark(d, ds, now.strftime("%Y-%m-%d %H:%M:%S"), degraded) ctx = dict((old_doc or {}).get("strategy_context") or {}) ctx = {k: v for k, v in ctx.items() if k in set(d["pool"])} + seg_ratio = segment_started_ratios(data.get("segments_pointed") or []) for r in plan_rows: - ctx[r["code"]] = {"factor_code": "akg_score", "score": r.get("score"), - "tier": r.get("tier"), "upside": r.get("upside"), - "theme": (r.get("evidence") or {}).get("theme"), - "plan_date": ds, - # 候选卡摘要与来源标记（09-02 方案第五项第一步：只加字段） + ctx[r["code"]] = {**strategy_context_entry(r, ds, seg_ratio), "source": r.get("_pool_source"), - "verdict": r.get("verdict"), - "reasons": (r.get("reasons") or [])[:4], - "card_rank": r.get("card_rank"), "plan_version": data.get("plan_version")} for c in d["retained_holdings"]: ctx.setdefault(c, {"factor_code": "akg_score", "note": "持仓保留"}) @@ -347,7 +386,7 @@ def push(date: str | None = None, top: int | None = None, col = client[config.mongo().db][col_name] col.update_one( {"group_code": config.POOL_GROUP_CODE, "org_id": config.POOL_ORG_ID}, - {"$set": {"group_name": config.POOL_GROUP_NAME, "pool_type": "core", + {"$set": {"group_name": config.POOL_GROUP_NAME, "pool_type": config.POOL_TYPE, "is_public": False, "description": "akg-factor-bridge 每日选股计划池：当日计划(强传导主榜) + " "持仓保留 + 留池观察。决策系统每晚认知扫描按本组产出结论，" diff --git a/regime.py b/regime.py index a70b45f..b512721 100644 --- a/regime.py +++ b/regime.py @@ -115,11 +115,17 @@ def append_to_snapshot(day: str) -> dict: return reg -def read_from_snapshot(day: str) -> dict | None: - """/plan 用：只读当日快照里的 regime 段，没有就 None（应答里给 UNKNOWN）。""" +def read_section(day: str, key: str): + """只读当日快照里的某一段（regime、market 等顶层键），快照不存在或坏 JSON 返回 None。 + /plan 应答里的环境类字段一律从落盘快照取，盘中不向任何来源发请求。""" path = snapshot_path(day) try: with open(path, "r", encoding="utf-8") as f: - return json.load(f).get("regime") + return json.load(f).get(key) except (OSError, ValueError): return None + + +def read_from_snapshot(day: str) -> dict | None: + """/plan 用：只读当日快照里的 regime 段，没有就 None（应答里给 UNKNOWN）。""" + return read_section(day, "regime") diff --git a/sources.py b/sources.py index 6984883..a81da4b 100644 --- a/sources.py +++ b/sources.py @@ -10,21 +10,40 @@ 153 代理 strategy_daily_results 决策系统昨夜结论：信号、支撑压力位、吸筹块 （吸筹的评分、状态、评分日三项一次读齐；基座落库的吸筹版本没有评分日，所以不从基座取） 平台 MySQL gp_day_data 交易日历（算评分日龄用；取一只长期存在的票的日期序列） + 基座 PG v_factor_logic 数据基座抽取的因果论断（方向、机制、时效、出处），每票最多几条， + 只展示不作门槛（2026-09-03 方案第 3.3 节） + 平台 MySQL zs_day_data / eastmoney_rzrq_data / fear_greed_index + 计划环境段的市场四项：两市成交额、融资余额、恐贪指数； + 市场广度从基座 v_factor_stock_daily 当日行自算（同一节） 代码格式：基座是点后缀式 600000.SH，决策系统与桥是前缀式 SH600000，进出都过 common.to_prefix。 读失败的语义：每一路读不到都返回空字典并打印一行原因，候选卡按"缺失"处理（进关注或仅展示）， 不让计划断产——与 pool.py 的安全边界一致。 + +离线单测：logic_claims 与 market_context 都接受注入的读函数（read_pg / read_mysql）， +test_market_context.py 用假数据函数替换真实连接，不连库；两者内部只对"记录列表"做计算， +读函数返回 DataFrame 或普通的字典列表都可以（见 _records）。 """ from __future__ import annotations import datetime as dt import json +import statistics import pandas as pd import common +import config import db +# 上证指数与深证成指在指数日线表 zs_day_data 里的代码，两市成交额取二者当日 amount 之和。 +MARKET_INDEX_CODES = ("000001.SH", "399001.SZ") +# 市场广度里"涨停家数"的近似口径：涨幅达到 9.8%（不读涨跌停价表，与环节日行情视图注释一致）。 +LIMIT_UP_PCT = 9.8 +# 融资表与恐贪指数表的日期列名不在本仓库内核实过，按候选名逐个匹配（第一个命中的用）。 +_DATE_COL_CANDIDATES = ("trade_date", "date", "stat_date", "data_date", "report_date", + "dt", "timestamp", "update_date", "record_date", "created_at") + def moved_members(ds: str) -> dict[str, dict]: """数据日 ds 被传导指向的环节里，已启动（不在未动名单）的成员。 @@ -137,12 +156,20 @@ def night_conclusions(codes, ds: str) -> dict[str, dict]: def _ymd(v) -> str | None: - """strategy_daily_results.trade_date 是整数 YYYYMMDD（也可能是日期），统一成 ISO 串。""" - if v is None or (isinstance(v, float) and pd.isna(v)): + """各表的日期列形态不一（整数 YYYYMMDD、date、datetime、ISO 串），统一成 ISO 日期串。 + 前几种形态不经 pandas 就能认出来，这样离线单测里的假数据不依赖 pandas；认不出的最后才交给 + pandas 解析，仍失败返回 None。""" + if v is None or (isinstance(v, float) and v != v): return None + if isinstance(v, dt.datetime): + return v.date().isoformat() + if isinstance(v, dt.date): + return v.isoformat() s = str(v).strip() if len(s) == 8 and s.isdigit(): return f"{s[:4]}-{s[4:6]}-{s[6:]}" + if len(s) >= 10 and s[4] == "-" and s[7] == "-" and s[:4].isdigit(): + return s[:10] try: return pd.Timestamp(s).date().isoformat() except Exception: # noqa: BLE001 @@ -167,3 +194,219 @@ def _f(v): except (TypeError, ValueError): return None return None if x != x else x + + +def _records(df) -> list[dict]: + """把读函数的返回统一成字典列表：DataFrame 走 to_dict，普通列表原样返回，None 与空表返回空列表。 + 这样取数函数的计算部分只面对普通 Python 对象，离线单测的假读函数直接返回字典列表即可。""" + if df is None: + return [] + if isinstance(df, list): + return [dict(r) for r in df] + if hasattr(df, "to_dict"): + if getattr(df, "empty", False): + return [] + return list(df.to_dict("records")) + return list(df) + + +def _to_dot(code: str) -> str: + """前缀式 SH600000 转成数据基座的点后缀式 600000.SH；已是点后缀式或纯数字则原样返回。""" + s = str(code or "").strip().upper() + if "." in s or len(s) < 3: + return s + if s[:2] in ("SH", "SZ", "BJ") and s[2:].isdigit(): + return f"{s[2:]}.{s[:2]}" + return s + + +# ============================================================================ +# 因果论断（数据基座 v_factor_logic，2026-09-03 方案第 3.3 节"候选卡读因果论断"） +# ============================================================================ +_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") + + +def logic_claims(codes, ds: str, per_stock: int | None = None, read_pg=None) -> dict[str, list[dict]]: + """这批票在数据基座因果论断视图里、披露日不晚于数据日 ds 的论断，每票取最近披露日的最多 + per_stock 条（默认 config.LOGIC_CLAIMS_PER_STOCK），按前缀码索引。 + + 每条论断带：方向、机制、条件、时效、强度、层级、置信度、披露日、出处文档标题与编号、 + 论断编号、经由环节、主体与客体名。只作展示与出处，不进判决；读失败返回空字典并打印一行原因。 + read_pg 可注入（离线单测），默认走 db.read_pg。""" + n_per = config.LOGIC_CLAIMS_PER_STOCK if per_stock is None else int(per_stock) + if n_per <= 0: + return {} + wanted = sorted({common.to_prefix(str(c).strip()) for c in codes if c}) + if not wanted: + return {} + dots = [_to_dot(c) for c in wanted] + reader = read_pg or db.read_pg + try: + marks = ",".join(["%s"] * len(dots)) + rows = _records(reader( + f"SELECT {', '.join(_LOGIC_COLS)} FROM v_factor_logic " + f"WHERE ts_code IN ({marks}) AND disclosure_date <= %s", + tuple(dots) + (ds,))) + except Exception as e: # noqa: BLE001 + print(f" （因果论断视图 v_factor_logic 读取失败，候选卡的论断证据线整体缺席: {e!r}）") + return {} + if not rows: + return {} + want = set(wanted) + by_code: dict[str, list[dict]] = {} + seen: set[tuple[str, str]] = set() + for r in rows: + k = common.to_prefix(str(r.get("ts_code") or "").strip()) + if k not in want: # 只留请求的票，视图返回的多余行不带进卡 + continue + cid = _s(r.get("claim_id")) + if cid and (k, cid) in seen: # 视图里客体为环节的论断按成员展开，同票同论断只留一条 + continue + if cid: + seen.add((k, cid)) + by_code.setdefault(k, []).append({ + "direction": _s(r.get("direction")), "mechanism": _s(r.get("mechanism")), + "condition": _s(r.get("condition")), "horizon": _s(r.get("horizon")), + "strength": _s(r.get("strength")), "tier": _s(r.get("tier")), + "confidence": _f(r.get("confidence")), + "disclosure_date": _ymd(r.get("disclosure_date")), + "doc_id": _s(r.get("doc_id")), "doc_title": _s(r.get("doc_title")), + "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")), + }) + 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] + return out + + +def _s(v) -> str | None: + if v is None or (isinstance(v, float) and v != v): + return None + s = str(v).strip() + return s or None + + +# ============================================================================ +# 计划环境段的市场四项（2026-09-03 方案第 1.4 节清单里"有"与"可自算"的项） +# ============================================================================ +def market_context(ds: str, read_pg=None, read_mysql=None) -> dict: + """数据日 ds 的市场环境四项，全部只展示与复盘分组，不拦任何票。 + + turnover 两市成交额：指数日线表 zs_day_data 里上证与深成当日 amount 之和，以及相对前五个 + 交易日均值的比值（原表单位，未换算） + breadth 市场广度：基座个股日行情视图当日行自算——上涨、下跌、平盘家数，涨幅达 9.8% 的家数 + （涨停近似），涨幅中位数 + margin 融资：eastmoney_rzrq_data 最新一日的 financing_balance 与 change_percent_5d + fear_greed 恐贪指数：fear_greed_index 最新一日的 index_value 与日期 + + 每一项读失败为 None 并把原因记进 errors，不阻断出计划。read_pg / read_mysql 可注入（离线单测）。 + 融资与恐贪两张表按"最新一日"取，不按 ds 过滤：它们是 T+1 更新的情绪读数，计划日早晨看到的 + 就是最新一行；行里带日期，读者自己判断新鲜度。""" + rpg = read_pg or db.read_pg + rmy = read_mysql or db.read_mysql + src = config.MARKET_MYSQL_SOURCE + out = {"date": ds, "turnover": None, "breadth": None, "margin": None, "fear_greed": None, + "errors": {}, "fetched_at": dt.datetime.now().isoformat(timespec="seconds")} + + # 一、两市成交额。不按日期列过滤（列的类型未在本仓库内核实：DATE 与整数 YYYYMMDD 的比较 + # 语义不同），改为取每个指数最近的几十行，在 Python 里按归一化日期筛不晚于 ds 的行。 + try: + marks = ",".join(["%s"] * len(MARKET_INDEX_CODES)) + rows = _records(rmy( + src, f"SELECT symbol, `timestamp` AS d, amount FROM zs_day_data " + f"WHERE symbol IN ({marks}) ORDER BY `timestamp` DESC LIMIT 80", + tuple(MARKET_INDEX_CODES))) + out["turnover"] = _turnover(rows, ds) + if out["turnover"] is None: + out["errors"]["turnover"] = "zs_day_data 最近 40 个交易日内没有不晚于计划日、且两市齐全的行" + except Exception as e: # noqa: BLE001 + out["errors"]["turnover"] = repr(e) + print(f" （两市成交额读取失败，环境段该项为空: {e!r}）") + + # 二、市场广度：基座个股日行情视图当日全部行自算。 + try: + rows = _records(rpg( + "SELECT pct_change FROM v_factor_stock_daily WHERE trade_date = %s", (ds,))) + out["breadth"] = _breadth(rows) + if out["breadth"] is None: + out["errors"]["breadth"] = "v_factor_stock_daily 当日无行" + except Exception as e: # noqa: BLE001 + out["errors"]["breadth"] = repr(e) + print(f" （市场广度自算失败，环境段该项为空: {e!r}）") + + # 三、融资余额与五日变化；四、恐贪指数。两张表都取最新一日一行。 + for key, table, cols, label in ( + ("margin", "eastmoney_rzrq_data", ("financing_balance", "change_percent_5d"), "融资余额"), + ("fear_greed", "fear_greed_index", ("index_value",), "恐贪指数")): + try: + row, date_col = _latest_row(rmy, src, table) + if row is None: + out["errors"][key] = f"{table} 为空表" + continue + item = {"date": _ymd(row.get(date_col)) if date_col else None, + "date_col": date_col, "source": table} + for c in cols: + item[c] = _f(row.get(c)) + if all(item[c] is None for c in cols): + out["errors"][key] = f"{table} 最新行缺列 {cols}（实际列: {sorted(row)[:12]}）" + continue + out[key] = item + except Exception as e: # noqa: BLE001 + out["errors"][key] = repr(e) + print(f" （{label}读取失败，环境段该项为空: {e!r}）") + return out + + +def _turnover(rows: list[dict], ds: str) -> dict | None: + """两市成交额：按日期把两个指数的 amount 相加，只认两市齐全的日子；当日取不晚于 ds 的最近一日， + 前五日均值取它之前的五个交易日（不足五个按实际个数）。""" + by_day: dict[str, dict] = {} + for r in rows: + d = _ymd(r.get("d")) + a = _f(r.get("amount")) + if not d or a is None or d > ds: + continue + by_day.setdefault(d, {})[str(r.get("symbol") or "").strip()] = a + full = sorted((d for d, m in by_day.items() if all(c in m for c in MARKET_INDEX_CODES)), + reverse=True) + if not full: + return None + day0 = full[0] + amt0 = sum(by_day[day0].values()) + prev = [sum(by_day[d].values()) for d in full[1:6]] + avg5 = (sum(prev) / len(prev)) if prev else None + return {"data_date": day0, "amount": amt0, "prev5_avg": avg5, + "ratio_vs_prev5": (amt0 / avg5) if avg5 else None, "prev5_days": len(prev), + "unit": "zs_day_data 原表单位，未换算", + "source": "zs_day_data 上证 000001.SH 与深成 399001.SZ 当日 amount 之和"} + + +def _breadth(rows: list[dict]) -> dict | None: + pcts = [p for p in (_f(r.get("pct_change")) for r in rows) if p is not None] + if not pcts: + return None + return {"n": len(pcts), + "up": sum(1 for p in pcts if p > 0), "down": sum(1 for p in pcts if p < 0), + "flat": sum(1 for p in pcts if p == 0), + "limit_up_approx": sum(1 for p in pcts if p >= LIMIT_UP_PCT), + "pct_median": round(statistics.median(pcts), 3), + "limit_up_rule": f"涨幅达 {LIMIT_UP_PCT}% 记为涨停近似", + "source": "v_factor_stock_daily 当日行自算"} + + +def _latest_row(rmy, src: str, table: str) -> tuple[dict | None, str | None]: + """取一张表按日期列排序的最新一行。日期列名先用一行样本探出（候选名见 _DATE_COL_CANDIDATES）， + 探不到就按第一列倒序（通常是自增主键）并把 date_col 记为 None。""" + sample = _records(rmy(src, f"SELECT * FROM {table} LIMIT 1")) + if not sample: + return None, None + cols = list(sample[0].keys()) + date_col = next((c for c in _DATE_COL_CANDIDATES if c in cols), None) + order = f"`{date_col}`" if date_col else "1" + rows = _records(rmy(src, f"SELECT * FROM {table} ORDER BY {order} DESC LIMIT 1")) + return (rows[0] if rows else None), date_col diff --git a/test_card.py b/test_card.py index 7958615..32d7d7c 100644 --- a/test_card.py +++ b/test_card.py @@ -1,7 +1,8 @@ """card.judge() 纯逻辑单测（无需 DB、无需 pandas）。 -覆盖：候选 / 关注（缺覆盖）/ 关注（缺确认）/ 仅展示（未启动）/ 仅展示（硬风险否决） -/ 缺失文案 / 评分陈旧 / 旋钮 / 卡内序 / 坏信号集合与 pool 同源。 +覆盖（2026-09-03 台账 013 的关注定义细化）：候选 / 关注只剩"确认线缺失或陈旧"两种 / +只差覆盖与潜在吸筹归仅展示 / 仅展示（未启动、硬风险否决）/ 缺失文案与判决依据 / +因果论断证据线只展示不进判决 / 旋钮 / 卡内序 / 坏信号集合与 pool 同源。 跑法：python3 test_card.py 或 pytest test_card.py """ @@ -12,6 +13,10 @@ BASE = dict(pointed=True, theme="散热器件", n_sources=2, pct0=4.1, accum_state="明确吸筹·量在价先", accum_score=78, accum_age=3, y_signal="WATCH", stale_snapshot=False) +CLAIM = {"direction": "利好", "mechanism": "液冷散热渗透率提升带动订单", "horizon": "半年内", + "condition": None, "doc_title": "散热行业深度", "disclosure_date": "2026-08-20", + "claim_id": "clm_001"} + def t(name, cond): assert cond, name @@ -22,46 +27,75 @@ def main(): r = card.judge(dict(BASE)) t("三门槛全过且明确吸筹 -> 候选", r["verdict"] == "候选" and r["confirm"] and not r["risk"]) t("候选理由四条齐", len(r["reasons"]) == 4 and "已启动" in r["reasons"][1]) + t("候选的判决依据一句话", "明确吸筹" in r["basis"] and r["failed_gates"] == []) + t("无论断时 logic 为空列表", r["logic"] == []) + # ---- 关注只剩两种：确认线缺失（无评分）、确认线陈旧（明确吸筹但日龄超限或未知）---- + r = card.judge({**BASE, "accum_state": ""}) + t("缺评分 -> 仍关注且缺失文案", r["verdict"] == "关注" and "无吸筹评分（未入池）" in r["missing"]) + t("关注的判决依据写明无法判断、交人裁决", "无法判断" in r["basis"] and "交人裁决" in r["basis"]) + + r = card.judge({**BASE, "accum_age": 45}) + t("评分陈旧 -> 仍关注、不确认、缺失文案", r["verdict"] == "关注" and not r["confirm"] + and any("陈旧 45" in m for m in r["missing"])) + + r = card.judge({**BASE, "accum_age": None}) + t("明确吸筹但日龄未知 -> 关注（陈旧一类）", r["verdict"] == "关注" + and any("日龄未知" in m for m in r["missing"])) + + # ---- 只差覆盖与潜在吸筹不再升格关注（台账 013）---- r = card.judge({**BASE, "covered": False, "upside": None}) - t("只差覆盖 -> 关注", r["verdict"] == "关注" and "无券商覆盖" in r["missing"]) + t("只差覆盖 -> 仅展示，缺失文案标明不升格", r["verdict"] == "仅展示" and "covered" in r["failed_gates"] + and any(m.startswith("无券商覆盖") and "仅展示" in m for m in r["missing"])) r = card.judge({**BASE, "accum_state": "潜在吸筹·低位企稳"}) - t("门槛全过但潜在吸筹不算确认 -> 关注", r["verdict"] == "关注" and not r["confirm"]) + t("潜在吸筹 -> 仅展示，failed_gates 附加 confirm", r["verdict"] == "仅展示" and not r["confirm"] + and "confirm" in r["failed_gates"] and any("潜在吸筹" in m and "仅展示" in m for m in r["missing"])) + t("潜在吸筹的判决依据写明未达确认线", "未达确认线" in r["basis"]) - r = card.judge({**BASE, "accum_state": ""}) - t("无评分 -> 关注且缺失文案", r["verdict"] == "关注" and "无吸筹评分（未入池）" in r["missing"]) + r = card.judge({**BASE, "accum_state": "信号不明"}) + t("信号不明等非明确状态 -> 仅展示", r["verdict"] == "仅展示" and "confirm" in r["failed_gates"]) + r = card.judge({**BASE, "upside": -0.05}) + t("负容忍默认 0：轻微为负不过覆盖门槛 -> 仅展示", r["verdict"] == "仅展示" and "covered" in r["failed_gates"] + and any("负容忍线" in m for m in r["missing"])) + + # ---- 其余仅展示与硬风险 ---- r = card.judge({**BASE, "pct0": 1.2}) - t("未启动 -> 仅展示", r["verdict"] == "仅展示" and "started" in r["failed_gates"]) + t("未启动 -> 仅展示", r["verdict"] == "仅展示" and "started" in r["failed_gates"] and "门槛未过" in r["basis"]) r = card.judge({**BASE, "pointed": False}) t("未被指向 -> 仅展示且标无传导", r["verdict"] == "仅展示" and "无传导" in r["missing"]) r = card.judge({**BASE, "y_signal": "sell"}) - t("决策系统 SELL -> 硬风险否决为仅展示", r["verdict"] == "仅展示" and r["risk"]) + t("决策系统 SELL -> 硬风险否决为仅展示", r["verdict"] == "仅展示" and r["risk"] and "硬风险" in r["basis"]) r = card.judge({**BASE, "stale_snapshot": True}) t("快照日不符 -> 硬风险", r["verdict"] == "仅展示" and "传导快照日与计划日不符" in r["risk"]) r = card.judge({**BASE, "accum_state": "⚠️ 高位派发"}) - t("高位派发 -> 硬风险", r["verdict"] == "仅展示" and "高位派发" in r["risk"]) + t("高位派发 -> 硬风险，且不重复写进缺失", r["verdict"] == "仅展示" and "高位派发" in r["risk"] + and not any("派发" in m for m in r["missing"])) - r = card.judge({**BASE, "accum_age": 45}) - t("评分陈旧 -> 不确认、关注、缺失文案", r["verdict"] == "关注" and not r["confirm"] - and any("陈旧 45" in m for m in r["missing"])) + r = card.judge({**BASE, "risk_name": True}) + t("ST 族 -> 仅展示", r["verdict"] == "仅展示" and "clean_name" in r["failed_gates"]) + # ---- 因果论断：只展示，不改判决 ---- + r = card.judge({**BASE, "logic": [CLAIM, {**CLAIM, "claim_id": "clm_002", "doc_title": None}]}) + t("带论断仍是候选，logic 两行带出处", r["verdict"] == "候选" and len(r["logic"]) == 2 + and "《散热行业深度》" in r["logic"][0] and "2026-08-20" in r["logic"][0] and "clm_001" in r["logic"][0]) + t("论断不进理由", len(r["reasons"]) == 4) + r = card.judge({**BASE, "pct0": 1.2, "logic": [CLAIM]}) + t("论断不能把未启动拉成候选", r["verdict"] == "仅展示" and r["logic"]) + t("坏论断条目被跳过", card.logic_lines([None, "x", {}]) == ["因果论断"]) + + # ---- 旋钮 ---- r = card.judge({**BASE, "pct0": 2.5}, start_pct=2.0) t("启动阈值旋钮生效", r["verdict"] == "候选") r = card.judge({**BASE, "accum_age": 45}, accum_max_age=60) t("评分日龄旋钮生效", r["verdict"] == "候选") r = card.judge({**BASE, "upside": -0.05}, neg_tol=0.10) t("负容忍线传入生效", r["verdict"] == "候选") - r = card.judge({**BASE, "upside": -0.05}) - t("负容忍默认 0：轻微为负不过覆盖门槛 -> 关注", r["verdict"] == "关注") - - r = card.judge({**BASE, "risk_name": True}) - t("ST 族 -> 仅展示", r["verdict"] == "仅展示" and "clean_name" in r["failed_gates"]) rows = [{"verdict": "关注", "pct0": 9.0}, {"verdict": "候选", "pct0": 3.5}, {"verdict": "候选", "pct0": 7.2}, {"verdict": "仅展示", "pct0": None}] @@ -70,7 +104,8 @@ def main(): [x["pct0"] for x in rows] == [7.2, 3.5, 9.0, None]) t("坏信号集合口径", card.BAD_SIGNALS == {"SELL", "AVOID", "DROPPED"}) - print("ALL OK — 候选卡判决 / 关注两种 / 硬风险三种 / 缺失文案 / 旋钮 / 卡内序 全部通过") + print("ALL OK — 候选卡判决 / 关注两种（缺失、陈旧）/ 只差覆盖与潜在吸筹归仅展示 / 硬风险三种 / " + "因果论断只展示 / 缺失文案与依据 / 旋钮 / 卡内序 全部通过") if __name__ == "__main__": diff --git a/test_market_context.py b/test_market_context.py new file mode 100644 index 0000000..3a59dde --- /dev/null +++ b/test_market_context.py @@ -0,0 +1,189 @@ +"""sources.market_context 与 sources.logic_claims 的离线单测（不连库），另带复盘脚本的台账标题解析。 + +取数函数都接受注入的读函数：这里用返回字典列表的假函数替换 db.read_pg / db.read_mysql， +覆盖：两市成交额与前五日比值、日期筛选、两市不齐全的日子被跳过；广度四项；融资与恐贪按最新 +一行取且日期列自动探测；每一项读失败为空不阻断；因果论断按前缀码索引、只取披露日不晚于 +数据日的、每票最多三条按披露日倒序；读失败返回空字典。 + +开发机没有 pandas 与数据库驱动时，只给缺席的模块装最小桩（与 test_plan_verdict.py 同一约定： +仅在模块缺席时装桩，不覆盖真实模块），取数函数的计算部分不碰 pandas。 + +跑法：python3 test_market_context.py 或 pytest test_market_context.py +""" +import datetime as dt +import os +import sys +import types + +_STUBS = ("pandas", "psycopg", "pymysql", "dotenv") +for _n in _STUBS: + if _n not in sys.modules: + try: + __import__(_n) + except ImportError: + _m = types.ModuleType(_n) + if _n == "pandas": # db.py / plan.py 的函数签名在定义时引用这两个名字 + _m.DataFrame = type("DataFrame", (), {}) + _m.Series = type("Series", (), {}) + sys.modules[_n] = _m + +import config # noqa: E402 +import sources # noqa: E402 + + +def t(name, cond): + assert cond, name + print(" ok", name) + + +# ---------------------------------------------------------------- 假数据 +DS = "2026-09-02" + + +def _zs_rows(): + """指数日线：六个交易日两市齐全，另有一天只有上证（该日应被跳过），还有一天晚于数据日。""" + days = ["2026-08-25", "2026-08-26", "2026-08-27", "2026-08-28", "2026-08-31", "2026-09-01", "2026-09-02"] + rows = [] + for i, d in enumerate(days): + rows.append({"symbol": "000001.SH", "d": dt.date.fromisoformat(d), "amount": 6000.0 + i * 100}) + if d != "2026-08-27": # 这一天深成缺行 + rows.append({"symbol": "399001.SZ", "d": dt.date.fromisoformat(d), "amount": 8000.0 + i * 100}) + rows.append({"symbol": "000001.SH", "d": dt.date(2026, 9, 3), "amount": 99999.0}) # 晚于数据日 + rows.append({"symbol": "399001.SZ", "d": dt.date(2026, 9, 3), "amount": 99999.0}) + return rows + + +def _mysql_ok(src, sql, params=None): + s = " ".join(sql.split()) + if "zs_day_data" in s: + return _zs_rows() + if "eastmoney_rzrq_data" in s: + if "LIMIT 1" in s and "ORDER BY" not in s: + return [{"id": 1, "stat_date": 20260901, "financing_balance": 1.9e12, "change_percent_5d": 1.23}] + assert "ORDER BY `stat_date` DESC" in s, s + return [{"id": 9, "stat_date": 20260901, "financing_balance": 1.9e12, "change_percent_5d": 1.23}] + if "fear_greed_index" in s: + if "LIMIT 1" in s and "ORDER BY" not in s: + return [{"id": 1, "date": "2026-09-01", "index_value": 62.5}] + assert "ORDER BY `date` DESC" in s, s + return [{"id": 7, "date": "2026-09-01", "index_value": 62.5}] + raise AssertionError(f"意外的查询: {s}") + + +def _pg_ok(sql, params=None): + s = " ".join(sql.split()) + if "v_factor_stock_daily" in s: + assert params == (DS,) + return [{"pct_change": 9.95}, {"pct_change": 3.0}, {"pct_change": 0.0}, {"pct_change": -1.5}, + {"pct_change": None}, {"pct_change": 10.02}, {"pct_change": -4.0}] + if "v_factor_logic" in s: + assert params[-1] == DS and "600000.SH" in params and "SZ000001" not in params + return [ + {"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": "乙"}, + {"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, + "subject_name": "甲", "object_name": "丙"}, + {"ts_code": "600000.SH", "direction": "利空", "mechanism": "机制丙", "condition": None, "horizon": None, + "strength": None, "tier": None, "confidence": 0.9, "disclosure_date": "2026-08-30", + "doc_id": "d3", "doc_title": "文档三", "source_span": "", "claim_id": "c3", "via_segment": None, + "subject_name": None, "object_name": 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}, + {"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, + "subject_name": None, "object_name": None}, + ] + raise AssertionError(f"意外的查询: {s}") + + +def _boom(*a, **k): + raise OSError("connection refused") + + +# ---------------------------------------------------------------- 用例 +def test_market_context(): + config.MARKET_MYSQL_SOURCE = "price" + m = sources.market_context(DS, read_pg=_pg_ok, read_mysql=_mysql_ok) + tv = m["turnover"] + t("两市成交额取数据日、两市齐全的行：6600+8600", tv and tv["data_date"] == DS and tv["amount"] == 15200.0) + # 前五日：09-01(15000)、08-31(14800)、08-28(14600)、08-26(14200)；08-27 深成缺行被跳过 → 再补 08-25(14000) + t("前五日均值跳过两市不齐全的日子", tv["prev5_days"] == 5 and abs(tv["prev5_avg"] - 14520.0) < 1e-6) + t("比值 = 当日 / 前五日均值", abs(tv["ratio_vs_prev5"] - 15200.0 / 14520.0) < 1e-9) + t("晚于数据日的行不参与", tv["amount"] < 99999) + b = m["breadth"] + # 六个有效值排序：-4.0、-1.5、0.0、3.0、9.95、10.02，中位数 = (0.0 + 3.0) / 2 = 1.5 + t("广度：上涨 3 / 下跌 2 / 平盘 1，涨停近似 2，中位数 1.5（空值剔除）", + b["n"] == 6 and b["up"] == 3 and b["down"] == 2 and b["flat"] == 1 and b["limit_up_approx"] == 2 + and b["pct_median"] == 1.5) + mg = m["margin"] + t("融资：最新一行、日期列自动探到 stat_date、整数日期归一", + mg and mg["date"] == "2026-09-01" and mg["date_col"] == "stat_date" + and mg["financing_balance"] == 1.9e12 and mg["change_percent_5d"] == 1.23) + fg = m["fear_greed"] + t("恐贪：最新一行、日期列 date", fg and fg["index_value"] == 62.5 and fg["date"] == "2026-09-01") + t("四项齐全时 errors 为空", m["errors"] == {} and m["date"] == DS) + + m = sources.market_context(DS, read_pg=_boom, read_mysql=_boom) + t("四项读失败：全为空、原因入 errors、不抛错", + m["turnover"] is None and m["breadth"] is None and m["margin"] is None and m["fear_greed"] is None + and set(m["errors"]) == {"turnover", "breadth", "margin", "fear_greed"}) + + def _mysql_partial(src, sql, params=None): + if "zs_day_data" in sql: + raise OSError("proxy down") + return _mysql_ok(src, sql, params) + m = sources.market_context(DS, read_pg=_pg_ok, read_mysql=_mysql_partial) + t("单项失败不影响其余三项", m["turnover"] is None and "turnover" in m["errors"] + and m["breadth"] and m["margin"] and m["fear_greed"]) + + m = sources.market_context("2026-01-01", read_pg=lambda *a, **k: [], read_mysql=_mysql_ok) + t("数据日早于所有行、广度无行：两项为空并注明", m["turnover"] is None and m["breadth"] is None + and "turnover" in m["errors"] and "breadth" in m["errors"]) + + +def test_logic_claims(): + config.LOGIC_CLAIMS_PER_STOCK = 3 + got = sources.logic_claims(["SH600000", "600000.SH", "SZ300750"], DS, read_pg=_pg_ok) + t("按前缀码索引、去重后只查一次", set(got) == {"SH600000"}) + items = got["SH600000"] + t("每票最多三条、按披露日倒序（同日按置信度）", + [c["claim_id"] for c in items] == ["c3", "c2", "c1"]) + t("字段齐全：日期归一、出处、经由环节、原文截断到 200 字", + items[2]["disclosure_date"] == "2026-08-20" and items[2]["doc_title"] == "文档一" + and items[2]["via_segment"] == "环节甲" and len(items[2]["source_span"]) == 200 + and items[1]["source_span"] is None and items[0]["condition"] is None) + got = sources.logic_claims(["SH600000"], DS, per_stock=1, read_pg=_pg_ok) + t("条数上限参数生效", len(got["SH600000"]) == 1 and got["SH600000"][0]["claim_id"] == "c3") + t("上限 0 = 不读视图", sources.logic_claims(["SH600000"], DS, per_stock=0, read_pg=_boom) == {}) + 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") + t("前缀式转点后缀式", sources._to_dot("SH600000") == "600000.SH" and sources._to_dot("600000.SH") == "600000.SH") + + +def test_decision_ledger_titles(): + import plan_review + here = os.path.dirname(os.path.abspath(__file__)) + entries = plan_review.decision_ledger_entries(os.path.join(here, "docs", "复盘决定台账.md")) + t("台账标题行解析出编号、日期、标题且含 013", entries and entries[0]["no"] == "001" + and any(e["no"] == "013" and "关注" in e["title"] for e in entries) + and all(len(e["date"]) == 10 for e in entries)) + t("台账文件缺失返回空列表", plan_review.decision_ledger_entries("/nonexistent/台账.md") == []) + + +def main(): + test_market_context() + test_logic_claims() + test_decision_ledger_titles() + print("ALL OK — 市场四项 / 单项失败不阻断 / 因果论断索引与上限 / 台账标题解析 全部通过") + + +if __name__ == "__main__": + main() diff --git a/test_pool_logic.py b/test_pool_logic.py index aacab1c..696698a 100644 --- a/test_pool_logic.py +++ b/test_pool_logic.py @@ -3,13 +3,29 @@ 运行: docker compose exec -T akg-factor-bridge python test_pool_logic.py 全过输出 "ALL PASS (n cases)"，任一失败退出码 1。 -被测函数: pool.decide / pool.build_remark / pool.build_recycle_docs。 +被测函数: pool.decide / pool.build_remark / pool.build_recycle_docs / +pool.segment_started_ratios / pool.strategy_context_entry（后两个 2026-09-03 加）。 + +开发机没有 pandas 与数据库驱动时也能跑：只给缺席的模块装最小桩（与 test_market_context.py +同一约定，模块存在时不覆盖），被测函数本身不碰这些依赖。 """ import datetime as dt import sys import traceback +import types -import pool +for _n in ("pandas", "psycopg", "pymysql", "dotenv"): + if _n not in sys.modules: + try: + __import__(_n) + except ImportError: + _m = types.ModuleType(_n) + if _n == "pandas": + _m.DataFrame = type("DataFrame", (), {}) + _m.Series = type("Series", (), {}) + sys.modules[_n] = _m + +import pool # noqa: E402 RESULTS = [] @@ -135,6 +151,31 @@ def _(): assert "形态恶化" in doc["reason"] +@case("入池上下文: 2026-09-03 补的证据字段只加不改, 环节已启动比例优先用数据基座台账的 moved") +def _(): + segs = [{"segment": "散热器件", "members_total": 20, "moved": 5, "started_count": 4}, + {"segment": "液冷", "members_total": 10, "moved": None, "started_count": 3}, + {"segment": "无成员数", "members_total": None, "moved": 2, "started_count": 2}] + ratio = pool.segment_started_ratios(segs) + assert ratio == {"散热器件": 0.25, "液冷": 0.3, "无成员数": None}, ratio + row = {"code": "SH600000", "score": 231.5, "tier": "强传导", "upside": 0.25, + "evidence": {"theme": "散热器件", "n_sources": 2, "moved_ratio": 0.25}, + "verdict": "候选", "reasons": ["a", "b", "c", "d", "e"], "card_rank": 1, + "card": {"pct0": 4.1, "chain_fit": 0.8, "accum": {"state": "明确吸筹·量在价先", "age": 3}}} + e = pool.strategy_context_entry(row, "2026-09-03", ratio) + for k, v in {"factor_code": "akg_score", "score": 231.5, "tier": "强传导", "upside": 0.25, + "theme": "散热器件", "plan_date": "2026-09-03", "verdict": "候选", "card_rank": 1, + "chain_fit": 0.8, "n_sources": 2, "pct0": 4.1, "accum_state": "明确吸筹·量在价先", + "accum_age": 3, "segment_started_ratio": 0.25}.items(): + assert e[k] == v, (k, e[k]) + assert e["reasons"] == ["a", "b", "c", "d"] # 理由仍只留四条 + assert "source" not in e and "plan_version" not in e # 这两个键由调用方补 + # 证据线缺席的行 (无 card / 无 evidence): 新字段为 None, 不抛错 + e = pool.strategy_context_entry({"code": "SZ000001", "score": 100.0}, "2026-09-03", ratio) + assert e["theme"] is None and e["chain_fit"] is None and e["segment_started_ratio"] is None + assert e["accum_state"] is None and e["pct0"] is None and e["reasons"] == [] + + # ---------------------------------------------------------------- runner def main(): passed, failed = 0, 0