diff --git a/docs/G1实测与待办_2026-07-26.md b/docs/G1实测与待办_2026-07-26.md index 4105996..837c765 100644 --- a/docs/G1实测与待办_2026-07-26.md +++ b/docs/G1实测与待办_2026-07-26.md @@ -254,10 +254,19 @@ probe 三张读数提醒已钉 **07-30 周四 19:00**。 ① 基座 transmission.py 传导主题黑名单（高新技术企业/小型微利企业/联营企业/无/ 报告分部——联营企业 298 名成员不触 600 闸，必须显式拦；源与目标两侧都挡）； ② 桥 config/frontier_tracks.yml v0.1（6 confirmed + 4 candidate、61 主题映射、 - 排除清单、TODO 决策点待圈）；③ 桥 tracks.py + `run.py tracks`（赛道覆盖体检 + + 排除清单、TODO 决策点待圈；当日实测：61 主题全部在池中有名，体检表"命中 x/y" + 的缺口是报表把被同赛道更早主题完全覆盖的主题误计漏，当日已修）； + ③ 桥 tracks.py + `run.py tracks`（赛道覆盖体检 + 成员表版本化快照落 data/，即本清单"赛道覆盖体检"的正式工具）；④ 桥 akg_gate / akg_score 两个新因子（三档+两段式，进 build all 日更；C 闸默认关 =ENABLE_TRACK_GATE，yml 转正后开）；⑤ requirements.txt 加 PyYAML（**桥机需 rebuild 镜像**；不 rebuild 只影响 tracks 命令，其余照跑）。 + 当日实机判收：tracks 六赛道全非空（confirmed 合计 777 只 = 池的 32.5%）、 + register 六因子成、gate 2386 行、score 1068 行（主榜 961 + 观察档 107）✅。 + **score 编码当日修正**：档位步长 10 → 20——组内分 clip±9.9 全幅 19.8， + 步长 10 会档间重叠、先档后分的词典序破功；07-29 的 score 表需重 build 覆盖。 + ⑥ plan.py + `run.py plan`（R4 首版）：每日选股计划 Markdown——主榜/观察档 + 榜单（名称、传导证据、热度、预期空间）+ 今日升降档 + T−1 口径声明， + 落 data/plan/，终端同步打印。 **工作哲学（07-30 用户令）**：数据渐进式暴露是常态——不等全量抽完、不等干净 截面、不等覆盖完备；缺锚降档不弃用，滞后交采信规则消化。 \ No newline at end of file diff --git a/factors.py b/factors.py index 851e432..d4684cc 100644 --- a/factors.py +++ b/factors.py @@ -408,8 +408,9 @@ def build_score(start, end): 主榜 = 200 + 传导档位×10 + 组内分：传导票按 log 传导分的组内中位数分成 强(2)/弱(1)两档，无传导 0 档；组内分 = 0.6·z(−热度) + 0.4·z(upside)， 即"还没热、还便宜"。观察档 = 100 + 0.6·z(传导) + 0.4·z(−热度)——无估值锚， - 低置信。组内分夹在 ±9.9 保证档位永不重叠；z 都在各自档内算。 - 缺失处置同设计 §3.4：传导缺=0，热度缺=档内中位数。""" + 低置信。**档位步长 20、组内分夹 ±9.9**：组内分全幅 19.8 小于步长， + 先档后分的词典序才真正成立（步长 10 会档间重叠，07-30 交付当天修正）。 + z 都在各自档内算；缺失处置同设计 §3.4：传导缺=0，热度缺=档内中位数。""" ts = _track_set() out = [] for ds in _days_of(start, end): @@ -429,7 +430,7 @@ def build_score(start, end): med = lt[hit].median() tier[hit] = np.where(lt[hit] >= med, 2.0, 1.0) inner = 0.6 * (-_robust_z(P["heat"])) + 0.4 * _robust_z(P["upside"]) - score = 200.0 + tier * 10.0 + inner.clip(-9.9, 9.9) + score = 200.0 + tier * 20.0 + inner.clip(-9.9, 9.9) out.append(pd.DataFrame({"trade_date": ds, "stock_code": P.index, "factor_value": score.values})) diff --git a/plan.py b/plan.py new file mode 100644 index 0000000..c2c89f4 --- /dev/null +++ b/plan.py @@ -0,0 +1,190 @@ +"""每日选股计划（R4 首版）：把 akg_gate / akg_score 变成一份人能读的榜单。 + +数据全部来自已落库的表，不重算： + 平台因子表 t_factor_akg_score / _gate / _upside / _heat —— 当日截面 + 基座只读视图 v_factor_transmission —— 传导证据（主题、源数、已动比例） + 基座 industry_pools —— 股票名称 +产出：终端打印 + data/plan/plan_<日期>.md。升降档一节对比前一交易日的档位表， +体现"数据到达本身是信号"（首次覆盖 / 新进传导链即升档）。 +""" +import json +import os + +import pandas as pd + +import common +import db + + +# 分数编码（与 factors.build_score 一致）：主榜 = 200 + 传导档位×20 + 组内分， +# 观察档 = 100 + 组内分，组内分 clip ±9.9。150 落在两带中间的空档上，用作分界。 +_MAIN_MIN = 150.0 + + +def _factor(table: str, ds: str) -> pd.Series: + df = db.read_mysql( + "factor", + f"SELECT stock_code, factor_value FROM {table} WHERE trade_date = %s", (ds,)) + if df.empty: + return pd.Series(dtype=float) + return df.set_index("stock_code")["factor_value"].astype(float) + + +def _latest_date(table: str, upto: str | None = None): + if upto: + df = db.read_mysql( + "factor", f"SELECT MAX(trade_date) d FROM {table} " + f"WHERE trade_date <= %s", (upto,)) + else: + df = db.read_mysql("factor", f"SELECT MAX(trade_date) d FROM {table}") + v = None if df.empty else df.iloc[0, 0] + return None if v is None or pd.isna(v) else pd.Timestamp(v).date().isoformat() + + +def _prev_date(table: str, before: str): + df = db.read_mysql( + "factor", f"SELECT MAX(trade_date) d FROM {table} " + f"WHERE trade_date < %s", (before,)) + v = None if df.empty else df.iloc[0, 0] + return None if v is None or pd.isna(v) else pd.Timestamp(v).date().isoformat() + + +def _names() -> dict: + out = {} + pools = db.read_pg("SELECT members FROM industry_pools") + for _, r in pools.iterrows(): + ms = r["members"] + if isinstance(ms, str): + ms = json.loads(ms) + for m in ms or []: + ts, name = (m or {}).get("ts_code"), (m or {}).get("name") + if ts and name: + out.setdefault(common.to_prefix(ts), str(name)) + return out + + +def _evidence(ds: str): + """每股最强一条传导证据：主题、源数、已动比例；另返回涉及的行情快照日。""" + tr = db.read_pg( + "SELECT ts_code, target, n_sources, moved_ratio, mkt_trade_date " + "FROM v_factor_transmission WHERE scan_date = %s", (ds,)) + if tr.empty: + return {}, set() + tr["k"] = tr["ts_code"].map(common.to_prefix) + tr["n_sources"] = pd.to_numeric(tr["n_sources"], errors="coerce").fillna(0) + tr["moved_ratio"] = pd.to_numeric(tr["moved_ratio"], errors="coerce").fillna(0) + tr["strength"] = tr["n_sources"] * (1.0 - tr["moved_ratio"]) + tr = tr.sort_values("strength", ascending=False).drop_duplicates("k") + ev = {r.k: (str(r.target), int(r.n_sources), float(r.moved_ratio)) + for r in tr.itertuples()} + days = {str(x) for x in tr["mkt_trade_date"].dropna().unique()} + return ev, days + + +def _tier_label(score: float) -> str: + return {0: "无传导", 1: "弱传导", 2: "强传导"}.get( + int((score - 190.0) // 20), "?") + + +def _pct(v) -> str: + return "—" if v is None or pd.isna(v) else f"{v:+.0%}" + + +def _num(v) -> str: + return "—" if v is None or pd.isna(v) else f"{v:.2f}" + + +def generate(date: str | None = None, top: int = 20, obs_top: int = 10) -> str: + ds = date or _latest_date("t_factor_akg_score") + if not ds: + raise SystemExit("t_factor_akg_score 还没有数据——先 build akg_score。") + score = _factor("t_factor_akg_score", ds) + gate = _factor("t_factor_akg_gate", ds) + if score.empty or gate.empty: + raise SystemExit(f"{ds} 缺 akg_score / akg_gate——先 build 该日再出计划。") + upside = _factor("t_factor_akg_upside", ds) + hd = _latest_date("t_factor_akg_heat", ds) + heat = _factor("t_factor_akg_heat", hd) if hd else pd.Series(dtype=float) + names = _names() + ev, mkt_days = _evidence(ds) + + main = score[score >= _MAIN_MIN].sort_values(ascending=False) + obs = score[score < _MAIN_MIN].sort_values(ascending=False) + + L = [f"# 每日选股计划 · {ds}", ""] + L.append(f"主榜 {len(main)} 只 / 观察档 {len(obs)} 只 / 全池档位覆盖 {len(gate)} 只。") + stale = sorted(d for d in mkt_days if d != ds) + if stale: + L.append(f"⚠️ 本日传导用的行情快照 = {'、'.join(stale)}（T−1 口径：" + f"\"谁已经动了\"看的是上个交易日收盘；拍点方案定版前均如此）。") + L.append("") + + L.append(f"## 主榜 Top {min(top, len(main))}（有券商预期、目标价不低于现价）") + L.append("") + L.append("| # | 代码 | 名称 | 总分 | 档位 | 传导证据 | 热度 | 预期空间 |") + L.append("|---|------|------|------|------|----------|------|----------|") + for i, (k, s) in enumerate(main.head(top).items(), 1): + e = ev.get(k) + etxt = f"{e[0]}（{e[1]} 源，已动 {e[2]:.0%}）" if e else "—" + L.append(f"| {i} | {k} | {names.get(k, '—')} | {s:.1f} | {_tier_label(s)} " + f"| {etxt} | {_num(heat.get(k))} | {_pct(upside.get(k))} |") + L.append("") + + L.append(f"## 观察档 Top {min(obs_top, len(obs))}" + f"（无券商预期、但在传导链上——没有估值锚，置信度低）") + L.append("") + L.append("| # | 代码 | 名称 | 分 | 传导证据 | 热度 |") + L.append("|---|------|------|----|----------|------|") + for i, (k, s) in enumerate(obs.head(obs_top).items(), 1): + e = ev.get(k) + etxt = f"{e[0]}（{e[1]} 源，已动 {e[2]:.0%}）" if e else "—" + L.append(f"| {i} | {k} | {names.get(k, '—')} | {s:.1f} " + f"| {etxt} | {_num(heat.get(k))} |") + L.append("") + + L.append("## 今日升降档") + L.append("") + prev_ds = _prev_date("t_factor_akg_gate", ds) + if not prev_ds: + L.append("（没有更早的档位表可比，升降档从下一个交易日开始。）") + else: + prev = _factor("t_factor_akg_gate", prev_ds) + both = pd.concat([prev.rename("prev"), gate.rename("cur")], axis=1) + both = both.fillna(-1.0) # -1 = 当日不在面板 + up = both[both["cur"] > both["prev"]] + down = both[both["cur"] < both["prev"]] + lab = {-1.0: "池外", 0.0: "不采纳", 1.0: "观察档", 2.0: "主榜"} + L.append(f"对比 {prev_ds}：升档 {len(up)} 只，降档 {len(down)} 只。" + f"升档=拿到新锚（首次覆盖 / 新进传导链），本身就是值得看的信号。") + + def _rows(d: pd.DataFrame, cap: int = 15): + lines = [] + for k, r in d.iterrows(): + lines.append(f"- {k} {names.get(k, '')}：" + f"{lab.get(r['prev'], '?')} → {lab.get(r['cur'], '?')}") + if len(lines) >= cap: + lines.append(f"- ……共 {len(d)} 只，其余见档位表") + break + return lines + + if not up.empty: + L.append("") + L.append("**升档**：") + L += _rows(up.sort_values("cur", ascending=False)) + if not down.empty: + L.append("") + L.append("**降档**：") + L += _rows(down.sort_values("prev", ascending=False)) + L.append("") + L.append("---") + L.append("口径：主榜分 = 200 + 传导档位×20 + 组内分（还没热、还便宜）；" + "观察档分 = 100 + 0.6z(传导) + 0.4z(−热度)。研究用途，不构成投资建议。") + + text = "\n".join(L) + os.makedirs("data/plan", exist_ok=True) + out = f"data/plan/plan_{ds}.md" + with open(out, "w", encoding="utf-8") as f: + f.write(text + "\n") + print(text) + print(f"\n已写入 {out}") + return out diff --git a/run.py b/run.py index b522eb1..3769d60 100644 --- a/run.py +++ b/run.py @@ -5,6 +5,7 @@ python run.py probe # G1 体检（只读，详见 probe.py） python run.py freeze [--date D] # 输入冻结（G0.5，详见 freeze.py） python run.py tracks # 赛道覆盖体检 + 成员表快照（G2） + python run.py plan [--date D] [--top N] # 每日选股计划（R4，读已落库因子表） python run.py register # 注册全部因子到 factor_metadata python run.py build all --mode history --start 2024-01-01 --end 2025-12-31 python run.py build akg_heat --mode daily --date 2026-07-24 @@ -94,9 +95,9 @@ _META = { "转正后再交赛道成员), 1=观察档(无券商覆盖但在图谱传导链上, 无估值锚, " "低置信), 0=不采纳(两锚皆无, 或upside<0)。全池出行, 让平台看得见门槛"), "akg_score": ("astock-kg 景气度漏斗", - "先档后分: 主榜=200+传导档位x10+组内分(0.6z(-热度)+0.4z(upside), " - "两段式07-30拍板); 观察档=100+0.6z(传导)+0.4z(-热度)。" - "仅gate>0出行, 数值直接可排序"), + "先档后分: 主榜=200+传导档位x20+组内分(0.6z(-热度)+0.4z(upside), " + "两段式07-30拍板, 组内分clip±9.9故档间不重叠); " + "观察档=100+0.6z(传导)+0.4z(-热度)。仅gate>0出行, 数值直接可排序"), } @@ -140,6 +141,10 @@ def main(): sub.add_parser("views") sub.add_parser("register") sub.add_parser("tracks") # 赛道覆盖体检 + confirmed 成员表快照（G2） + pl = sub.add_parser("plan") # 每日选股计划（R4） + pl.add_argument("--date", help="默认取 score 表最新日") + pl.add_argument("--top", type=int, default=20, help="主榜条数") + pl.add_argument("--obs-top", type=int, default=10, help="观察档条数") p = sub.add_parser("probe") p.add_argument("--section", choices=["all", "pools", "price", "upside", "corr"], default="all", @@ -171,6 +176,9 @@ def main(): freeze.snapshot(a.date) elif a.cmd == "register": cmd_register() + elif a.cmd == "plan": + import plan + plan.generate(a.date, a.top, a.obs_top) elif a.cmd == "tracks": import tracks tracks.coverage_report() diff --git a/tracks.py b/tracks.py index 3ec07cd..03b6521 100644 --- a/tracks.py +++ b/tracks.py @@ -37,7 +37,8 @@ def load_yml(path: str | None = None) -> dict: return yaml.safe_load(f) -def resolve_members(only_confirmed: bool = True, path: str | None = None): +def resolve_members(only_confirmed: bool = True, path: str | None = None, + dedup: bool = True): """yml + industry_pools 当前态 → (成员表, 未命中主题清单)。 成员表列：ts_code, name, track, status, theme。同股同赛道多主题只留一行， @@ -69,9 +70,10 @@ def resolve_members(only_confirmed: bool = True, path: str | None = None): if ts: rows.append((ts, m.get("name"), tr["name"], tr.get("status"), t)) - df = (pd.DataFrame(rows, columns=["ts_code", "name", "track", - "status", "theme"]) - .drop_duplicates(["ts_code", "track"])) + df = pd.DataFrame(rows, columns=["ts_code", "name", "track", + "status", "theme"]) + if dedup: + df = df.drop_duplicates(["ts_code", "track"]) return df, missing @@ -88,9 +90,11 @@ def snapshot(only_confirmed: bool = True, path: str | None = None): def coverage_report(path: str | None = None) -> pd.DataFrame: """赛道覆盖体检（G1 清单的首件事）：逐赛道成员数与主题命中率，含 candidate。""" - df, missing = resolve_members(only_confirmed=False, path=path) + # dedup=False：主题命中率要在去重前数——成员完全被同赛道更早主题收进来的 + # 主题（如卫星导航之于卫星），去重后一行不剩，会被误计成"没命中"（07-30 实测）。 + df, missing = resolve_members(only_confirmed=False, path=path, dedup=False) d = load_yml(path) - print("赛道覆盖体检（industry_pools 当前态；成员数已去重）：") + print("赛道覆盖体检（industry_pools 当前态；成员数已去重，主题命中按去重前算）：") for tr in d.get("tracks") or []: sub = df[df["track"] == tr["name"]] n_theme = len(tr.get("kg_themes") or [])