"""每日选股计划(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 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 _val(series: pd.Series, k: str): v = series.get(k) return None if v is None or pd.isna(v) else float(v) def collect(date: str | None = None, top: int = 20, obs_top: int = 10, theme_cap: int = 5) -> dict: """装配一天的计划为结构化字典。数据缺失抛 RuntimeError(api 侧转 404)。""" 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) 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) r = {"rank": rank, "code": k, "name": names.get(k), "score": round(float(s), 2), "evidence": ({"theme": e[0], "n_sources": e[1], "moved_ratio": round(e[2], 4)} if e else None), "heat": _val(heat, k), "upside": _val(upside, k)} if with_tier: r["tier"] = _tier_label(s) return r 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) def _mv(d: pd.DataFrame): return [{"code": k, "name": names.get(k), "from": lab.get(r["prev"], "?"), "to": lab.get(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))} return { "date": ds, "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, "encoding": "主榜分=200+传导档位×20+组内分(还没热、还便宜);" "观察档分=100+0.6z(传导)+0.4z(−热度)", } 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 "—" return f"{e['theme']}({e['n_sources']} 源,已动 {e['moved_ratio']:.0%})" 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("") cap_txt = f",每主题限额 {d['theme_cap']}" if d["theme_cap"] else "" L.append(f"## 主榜 Top {len(d['main'])}(有券商预期、目标价不低于现价{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"(无券商预期、但在传导链上——没有估值锚,置信度低{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']}" 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']}" 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("data/plan", exist_ok=True) out = f"data/plan/plan_{data['date']}.md" with open(out, "w", encoding="utf-8") as f: f.write(text + "\n") print(text) print(f"\n已写入 {out}") return out