"""每日选股计划（R4）：数据装配 / Markdown 渲染 / 产出，三段分离。 collect() -> dict 结构化计划——api.py 直接当 JSON 返回 render_md() -> str 从 dict 渲染 Markdown generate() CLI 与 cron 的入口：collect + render + 落盘 + 打印 数据全部来自已落库的表，不重算： 平台因子表 t_factor_akg_score / _gate / _upside / _heat —— 当日截面 基座只读视图 v_factor_transmission —— 传导证据 基座 industry_pools —— 股票名称 升降档一节对比前一交易日的档位表——数据到达本身是信号（首次覆盖 / 新进传导链即升档）。 """ import datetime as dt import json import os import pandas as pd import card import common import config import db import sources import version # 分数编码（与 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 _val(series: pd.Series, k: str): v = series.get(k) return None if v is None or pd.isna(v) else float(v) def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series, mkt_days: set) -> tuple[dict, list]: """候选卡装配（2026-09-02 方案第 2.2 节第三项）：对档位表里的全部票（主榜与观察档， 裁剪之前）读三路证据、逐票判决。规则在 card.py，取数在 sources.py，这里只做对齐。 返回 (cards, segments_pointed)：cards 按前缀码索引，含判决、理由、缺失、风险、卡内序 与证据线原值；segments_pointed 是"关注环节"聚合——今日被传导指向的每个环节的源数、 链符、成员数、已启动成员、领涨者、候选数。这是拍板记录第一项"强传导档降为关注环节" 在计划文本层的落地。""" import factors moved = sources.moved_members(ds) daily = sources.stock_daily(ds) night = sources.night_conclusions(codes, ds) try: risk = factors._risk_set() or set() # noqa: SLF001 —— 同仓自用 except Exception: # noqa: BLE001 —— 风险名单拿不到时不当 ST 处理，与 gate 的宽容一致 risk = set() stale = bool(mkt_days) and any(x != ds for x in mkt_days) cards: dict = {} for k in codes: mv, e, d, n = moved.get(k), ev.get(k), daily.get(k, {}), night.get(k, {}) theme = (mv or {}).get("theme") or (e[0] if e else None) n_sources = (mv or {}).get("n_sources") or (e[1] if e else None) up = _val(upside, k) evd = {"pointed": bool(mv or e), "theme": theme, "n_sources": n_sources, "chain_fit": (mv or {}).get("chain_fit"), "pct0": d.get("pct0"), "covered": up is not None, "upside": up, "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} j = card.judge(evd, start_pct=config.CARD_START_PCT, accum_max_age=config.CARD_ACCUM_MAX_AGE, neg_tol=config.UPSIDE_NEG_TOLERANCE) cards[k] = { **j, "theme": theme, "n_sources": n_sources, "chain_fit": evd["chain_fit"], "started_source": "moved_view" if mv else None, "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"), "date": n.get("conclusion_date")} if n else None), "night": ({"signal": n.get("signal"), "support": n.get("support"), "pressure": n.get("pressure"), "date": n.get("conclusion_date")} if n else None), } for i, (k, c) in enumerate(sorted(cards.items(), key=lambda kv: card.sort_key(kv[1])), 1): c["card_rank"] = i segs: dict = {} for k, mv in moved.items(): s = segs.setdefault(mv["theme"], { "segment": mv["theme"], "n_sources": mv["n_sources"], "chain_fit": mv["chain_fit"], "members_total": mv.get("members_total"), "moved": mv.get("moved"), "started": [], "candidates": 0, "leader": None}) s["started"].append(k) if cards.get(k, {}).get("verdict") == card.VERDICT_CANDIDATE: s["candidates"] += 1 for e in ev.values(): # 只有未动名单的环节也是"被指向"，列入但无已启动 segs.setdefault(e[0], {"segment": e[0], "n_sources": e[1], "chain_fit": None, "members_total": None, "moved": None, "started": [], "candidates": 0, "leader": None}) for s in segs.values(): best, best_pct = None, None for k in s["started"]: p = cards.get(k, {}).get("pct0") if p is not None and (best_pct is None or p > best_pct): best, best_pct = k, p s["leader"] = {"code": best, "pct0": best_pct} if best else None s["started_count"] = len(s["started"]) ordered = sorted(segs.values(), key=lambda s: (-s["candidates"], -(s["n_sources"] or 0), -(s["chain_fit"] or 0), s["segment"])) return cards, ordered def collect(date: str | None = None, top: int = 20, obs_top: int = 10, theme_cap: int = 5) -> dict: """装配一天的计划为结构化字典。数据缺失抛 RuntimeError（api 侧转 404）。 2026-09-02 起附带候选卡：main / observe 的装配、排序、裁剪一字不动（下游 PMS 只读 这两段的既有字段），每行只是多联入判决类字段；顶层新增 generated_at、plan_version、 card_counts、candidates（候选单全量，不受裁剪）、watch、segments_pointed。 `_full` 是全量主榜与观察档行，只给 generate 落快照用，api 返回前会去掉。""" ds = date or _latest_date("t_factor_akg_score") if not ds: raise RuntimeError("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 RuntimeError(f"{ds} 缺 akg_score / akg_gate——先 build 该日再出计划。") upside = _factor("t_factor_akg_upside", ds) if upside.empty: # 当日 upside 表为空时现算兜底（as-of 口径不变：consensus<=当日、当日收盘价） import factors df_up = factors.build_upside(ds, ds) if df_up is not None and not df_up.empty: x = df_up.copy() x["k"] = x["stock_code"].map(common.to_prefix) upside = x.groupby("k")["factor_value"].max().astype(float) 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) generated_at = dt.datetime.now().isoformat(timespec="seconds") cards, segments_pointed = _assemble_cards( ds, list(main.index) + list(obs.index), ev, upside, mkt_days) def _pick(ranked: pd.Series, n: int): """分数从高到低取 n 条；每个传导主题最多 theme_cap 条（0=不设限）—— 传导目标是环节级、同环节成员共享同一条证据，不限额会被少数环节刷屏。""" out, cnt = [], {} for k, s in ranked.items(): e = ev.get(k) theme = e[0] if e else "（无传导）" if theme_cap and cnt.get(theme, 0) >= theme_cap: continue cnt[theme] = cnt.get(theme, 0) + 1 out.append((k, s)) if len(out) >= n: break return out def _row(rank: int, k: str, s: float, with_tier: bool) -> dict: e = ev.get(k) c = cards.get(k) or {} evidence = ({"theme": e[0], "n_sources": e[1], "moved_ratio": round(e[2], 4)} if e else None) # 已启动成员在传导视图里没有证据行（视图只摊平未动名单）：只在原本为空时用 # 已动成员视图的目标环节与源数补上，不覆盖已有值——否则到 PMS 会全落进"无主题"桶。 if evidence is None and c.get("started_source") == "moved_view" and c.get("theme"): evidence = {"theme": c["theme"], "n_sources": c.get("n_sources"), "moved_ratio": None, "source": "moved_view"} r = {"rank": rank, "code": k, "name": names.get(k), "score": round(float(s), 2), "evidence": evidence, "heat": _val(heat, k), "upside": _val(upside, k)} if with_tier: r["tier"] = _tier_label(s) if c: r.update(verdict=c["verdict"], reasons=c["reasons"], missing=c["missing"], risk=c["risk"], card_rank=c["card_rank"], 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")}) return r def _full_rows(ranked: pd.Series, with_tier: bool) -> list: return [_row(i, k, s, with_tier) for i, (k, s) in enumerate(ranked.items(), 1)] def _by_verdict(v: str) -> list: rows = [_row(0, k, score[k], k in main.index) for k, c in cards.items() if c.get("verdict") == v] rows.sort(key=lambda r: r["card_rank"]) for i, r in enumerate(rows, 1): r["rank"] = i return rows changes = None prev_ds = _prev_date("t_factor_akg_gate", ds) if prev_ds: prev = _factor("t_factor_akg_gate", prev_ds) both = pd.concat([prev.rename("prev"), gate.rename("cur")], axis=1).fillna(-1.0) # -1 = 当日不在面板 lab = {-1.0: "池外", 0.0: "不采纳", 1.0: "观察档", 2.0: "主榜"} up_df = both[both["cur"] > both["prev"]].sort_values("cur", ascending=False) down_df = both[both["cur"] < both["prev"]].sort_values("prev", ascending=False) # ---- 升降原因（07-31 加）：区分首次覆盖 / 估值转正 / 新进传导链等。 # 依据前一日的 upside / 传导因子表；表空时现算兜底。原因是启发式归类 # （取最主要的一条），精确审计以档位表与因子表为准。 prev_up = _factor("t_factor_akg_upside", prev_ds) if prev_up.empty: try: import factors dfu = factors.build_upside(prev_ds, prev_ds) if dfu is not None and not dfu.empty: x = dfu.copy() x["k"] = x["stock_code"].map(common.to_prefix) prev_up = x.groupby("k")["factor_value"].max().astype(float) except Exception: # noqa: BLE001 —— 兜底失败则原因退化为通用文案 pass prev_tr = _factor("t_factor_akg_transmission", prev_ds) # 赛道闸与风险闸的成员集合（07-31 修：赛道闸开启后，"不在赛道"曾被 # 误标成"风险闸/档位调整"、"赛道锚生效"曾被误标成"新进传导链"）。 # 两个集合都尊重各自开关：闸没开时对应原因自然不会出现。 tset, risk = None, set() try: import factors tset = factors._track_set() # noqa: SLF001 —— 同仓自用 risk = factors._risk_set() # noqa: SLF001 except Exception: # noqa: BLE001 —— 拿不到就退化为通用文案 pass def _why(k: str, pg: float, cg: float) -> str: if cg == 2.0: # 升入主榜 pu = _val(prev_up, k) if pu is None: return "首次覆盖" return "估值转正" if pu < 0 else "重获资格" if cg == 1.0: # 升入观察档 if k in ev: # 今天真在传导链上 pt = _val(prev_tr, k) return "新进传导链" if pt is None or pt <= 0 else "档位调整" if tset is not None and k in tset: return "赛道锚生效" # 赛道成员身份给的锚（闸切换/成员变动） return "档位调整" if pg == 2.0: # 从主榜降出 if k in risk: return "风险闸" cu = _val(upside, k) if cu is None: return "覆盖脱落" if cu < 0: return "估值转负" if tset is not None and k not in tset: return "赛道闸外" # 有覆盖也不贵，但不在十赛道内 return "档位调整" if pg == 1.0: # 从观察档降出 if k in risk: return "风险闸" if k not in ev and (tset is None or k not in tset): return "离开传导链" return "档位调整" return "出入面板" def _mv(d: pd.DataFrame): return [{"code": k, "name": names.get(k), "from": lab.get(r["prev"], "?"), "to": lab.get(r["cur"], "?"), "reason": _why(k, r["prev"], r["cur"])} for k, r in d.iterrows()] changes = {"base_date": prev_ds, "upgrades_total": int(len(up_df)), "downgrades_total": int(len(down_df)), "upgrades": _mv(up_df.head(15)), "downgrades": _mv(down_df.head(15))} card_counts = {v: sum(1 for c in cards.values() if c.get("verdict") == v) for v in (card.VERDICT_CANDIDATE, card.VERDICT_WATCH, card.VERDICT_SHOW)} return { "date": ds, "generated_at": generated_at, "plan_version": version.git_short_rev(), "counts": {"main": int(len(main)), "observe": int(len(obs)), "gate_covered": int(len(gate))}, "market_snapshot_days": sorted(mkt_days), "heat_date": hd, "theme_cap": theme_cap, "main": [_row(i, k, s, True) for i, (k, s) in enumerate(_pick(main, top), 1)], "observe": [_row(i, k, s, False) for i, (k, s) in enumerate(_pick(obs, obs_top), 1)], "changes": changes, "gate_on": bool(config.ENABLE_TRACK_GATE), "encoding": "主榜分=200+传导档位×20+组内分（还没热、还便宜）；" "观察档分=100+0.6z(传导)+0.4z(−热度)", # ---- 候选卡（2026-09-02）：分数与档位之外的另一份产物，不受 top / theme_cap 裁剪 ---- "card_params": {"start_pct": config.CARD_START_PCT, "accum_max_age": config.CARD_ACCUM_MAX_AGE, "neg_tol": config.UPSIDE_NEG_TOLERANCE, "rules": "候选=环节被指向∧当日涨幅达标∧券商覆盖且非ST∧明确吸筹∧无硬风险；" "关注=无硬风险且(只差覆盖 或 门槛全过无确认)；其余仅展示"}, "card_counts": card_counts, "candidates": _by_verdict(card.VERDICT_CANDIDATE), "watch": _by_verdict(card.VERDICT_WATCH), "segments_pointed": segments_pointed, "_full": {"main": _full_rows(main, True), "observe": _full_rows(obs, False)}, } def _fmt_pct(v) -> str: return "—" if v is None else f"{v:+.0%}" def _fmt_num(v) -> str: return "—" if v is None else f"{v:.2f}" def _fmt_ev(e) -> str: if not e: return "—" if e.get("moved_ratio") is None: # 已动成员视图补的证据行，没有已动比例 return f"{e['theme']}（{e.get('n_sources') or '—'} 源，本票已启动）" return f"{e['theme']}（{e['n_sources']} 源，已动 {e['moved_ratio']:.0%}）" def _fmt_pct0(v) -> str: return "—" if v is None else f"{v:+.1f}%" def _fmt_accum(ac: dict) -> str: if not ac or not ac.get("state"): return "无评分" st = str(ac["state"]).split("·")[0] age = ac.get("age") return f"{st}（{age} 日前）" if isinstance(age, int) else st def render_md(d: dict) -> str: L = [f"# 每日选股计划 · {d['date']}", ""] c = d["counts"] L.append(f"主榜 {c['main']} 只 / 观察档 {c['observe']} 只 / " f"全池档位覆盖 {c['gate_covered']} 只。") stale = [x for x in d["market_snapshot_days"] if x != d["date"]] if stale: L.append(f"注：本日传导用的行情快照 = {'、'.join(stale)}" f"（与计划日不同——历史降级日口径）。") L.append("") # ---- 候选单与关注环节（2026-09-02）：放在主榜之前，这是新的主产物 ---- cc = d.get("card_counts") or {} cands = d.get("candidates") or [] L.append(f"## 候选单（环节被指向、当日已启动、券商覆盖且非 ST、明确吸筹、无硬风险；" f"共 {cc.get('候选', len(cands))} 只，全量列出不受裁剪）") L.append("") if not cands: L.append("（今日无候选——候选为空不是故障：环节没被指向、成员没启动或没有明确吸筹，都会为空。）") else: L.append("| # | 代码 | 名称 | 环节 | 源数 | 当日涨幅 | 吸筹 | 预期空间 | 理由 |") L.append("|---|------|------|------|------|----------|------|----------|------|") for r in cands: c = r.get("card") or {} ac = c.get("accum") or {} ev_ = r.get("evidence") or {} 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 [])} |") L.append("") segs = d.get("segments_pointed") or [] L.append(f"## 关注环节（今日被传导指向的 {len(segs)} 个环节：定位对不对看这里，挑票看候选单）") L.append("") if segs: L.append("| 环节 | 源数 | 链符 | 成员 | 已启动 | 领涨 | 候选 |") L.append("|------|------|------|------|--------|------|------|") for s in segs: ld = s.get("leader") or {} L.append(f"| {s['segment']} | {s.get('n_sources') or '—'} | {_fmt_num(s.get('chain_fit'))} " f"| {s.get('members_total') if s.get('members_total') is not None else '—'} " f"| {s.get('started_count', 0)} " f"| {(ld.get('code') or '—') + (' ' + _fmt_pct0(ld.get('pct0')) if ld.get('code') else '')} " f"| {s.get('candidates', 0)} |") L.append("") watch = d.get("watch") or [] L.append(f"## 关注单（无硬风险，只差券商覆盖或缺明确吸筹；共 {cc.get('关注', len(watch))} 只，列前 20）") L.append("") if watch: 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 [])} |") 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("") cap_txt = f"，每主题限额 {d['theme_cap']}" if d["theme_cap"] else "" gate_txt = "、在十五五赛道内" if d.get("gate_on") else "" L.append(f"## 主榜 Top {len(d['main'])}" f"（有券商预期、目标价不低于现价{gate_txt}{cap_txt}）") L.append("") L.append("| # | 代码 | 名称 | 总分 | 档位 | 传导证据 | 热度 | 预期空间 |") L.append("|---|------|------|------|------|----------|------|----------|") for r in d["main"]: L.append(f"| {r['rank']} | {r['code']} | {r['name'] or '—'} | {r['score']:.1f} " f"| {r['tier']} | {_fmt_ev(r['evidence'])} " f"| {_fmt_num(r['heat'])} | {_fmt_pct(r['upside'])} |") L.append("") L.append(f"## 观察档 Top {len(d['observe'])}" f"（无券商预期、但{'在赛道或传导链上' if d.get('gate_on') else '在传导链上'}" f"——没有估值锚，置信度低{cap_txt}）") L.append("") L.append("| # | 代码 | 名称 | 分 | 传导证据 | 热度 |") L.append("|---|------|------|----|----------|------|") for r in d["observe"]: L.append(f"| {r['rank']} | {r['code']} | {r['name'] or '—'} | {r['score']:.1f} " f"| {_fmt_ev(r['evidence'])} | {_fmt_num(r['heat'])} |") L.append("") L.append("## 今日升降档") L.append("") ch = d["changes"] if not ch: L.append("（没有更早的档位表可比，升降档从下一个交易日开始。）") else: L.append(f"对比 {ch['base_date']}：升档 {ch['upgrades_total']} 只，" f"降档 {ch['downgrades_total']} 只。" f"升档=拿到新锚（首次覆盖 / 新进传导链），本身就是值得看的信号。") if ch["upgrades"]: L.append("") L.append("**升档**：") L += [f"- {m['code']} {m['name'] or ''}：{m['from']} → {m['to']}" f"（{m.get('reason', '—')}）" for m in ch["upgrades"]] if ch["upgrades_total"] > len(ch["upgrades"]): L.append(f"- ……共 {ch['upgrades_total']} 只，其余见档位表") if ch["downgrades"]: L.append("") L.append("**降档**：") L += [f"- {m['code']} {m['name'] or ''}：{m['from']} → {m['to']}" f"（{m.get('reason', '—')}）" for m in ch["downgrades"]] if ch["downgrades_total"] > len(ch["downgrades"]): L.append(f"- ……共 {ch['downgrades_total']} 只，其余见档位表") L.append("") L.append("---") L.append(f"口径：{d['encoding']}。") return "\n".join(L) def generate(date: str | None = None, top: int = 20, obs_top: int = 10, theme_cap: int = 5) -> str: try: data = collect(date, top, obs_top, theme_cap) except RuntimeError as e: raise SystemExit(str(e)) 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") with open(out, "w", encoding="utf-8") as f: f.write(text + "\n") # ---- 当日 JSON 快照（2026-09-02 方案第 2.2 节第一项）：主榜与观察档全部行、全部证据线， # 不裁剪、不设主题限额；带生成时刻与代码版本。它是复盘与对账的唯一底本； # 08:45 的 regime-append 步骤会往里追加 regime 段，/plan 的 regime 段只读这份。---- full = data.pop("_full", None) or {} snap = {**data, "main_shown": data["main"], "observe_shown": data["observe"], "main": full.get("main", []), "observe": full.get("observe", []), "shown_params": {"top": top, "obs_top": obs_top, "theme_cap": theme_cap}} jpath = os.path.join(config.PLAN_SNAPSHOT_DIR, f"plan_{data['date']}.json") tmp = jpath + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(snap, f, ensure_ascii=False, indent=1, default=str) os.replace(tmp, jpath) print(text) print(f"\n已写入 {out} 与快照 {jpath}" f"（主榜 {len(snap['main'])} 行、观察档 {len(snap['observe'])} 行、" f"候选 {len(data.get('candidates') or [])} 只，版本 {data.get('plan_version')}）") return out