278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""每日选股计划（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
|