82 lines
3.4 KiB
Python
82 lines
3.4 KiB
Python
"""个股深度评析的目标名单（2026-09-09 方案附录甲）：每早出计划时把主榜与观察档写进 153 库一张小表，
|
||
数据基座的 00:45 任务读它选目标（178 的容器打不到 155 的接口，只能走库）。
|
||
|
||
表 t_akg_review_targets：plan_date、ts_code（点后缀式 600000.SH）、source（main / observe）、rank、name。
|
||
幂等：删该日行再整批插。写失败不拦计划（try 包住）。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
|
||
import config
|
||
import db
|
||
|
||
TABLE = os.environ.get("REVIEW_TARGETS_TABLE", "t_akg_review_targets")
|
||
TOP_MAIN = int(os.environ.get("REVIEW_TARGETS_MAIN", "100")) # 主榜前一百只
|
||
TOP_OBS = int(os.environ.get("REVIEW_TARGETS_OBS", "20")) # 观察档前二十只
|
||
COLUMNS = ("plan_date", "ts_code", "source", "rank", "name")
|
||
_CREATE = """
|
||
CREATE TABLE IF NOT EXISTS {t} (
|
||
plan_date DATE NOT NULL,
|
||
ts_code VARCHAR(16) NOT NULL,
|
||
source VARCHAR(16) NOT NULL,
|
||
`rank` INT NOT NULL,
|
||
name VARCHAR(64),
|
||
PRIMARY KEY (plan_date, ts_code)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
"""
|
||
|
||
|
||
def to_dot(code: str) -> str:
|
||
s = (code or "").strip().upper()
|
||
if len(s) == 8 and s[:2] in ("SH", "SZ", "BJ") and s[2:].isdigit():
|
||
return f"{s[2:]}.{s[:2]}"
|
||
return s
|
||
|
||
|
||
def build_rows(ds: str, main_rows: list, obs_rows: list) -> list[dict]:
|
||
seen: set = set()
|
||
out: list[dict] = []
|
||
for src, rows in (("main", main_rows), ("observe", obs_rows)):
|
||
for i, r in enumerate(rows or [], 1):
|
||
code = to_dot(str(r.get("code") or ""))
|
||
if len(code) != 9 or code in seen:
|
||
continue
|
||
seen.add(code)
|
||
out.append({"plan_date": ds, "ts_code": code, "source": src, "rank": i, "name": (r.get("name") or "")[:64]})
|
||
return out
|
||
|
||
|
||
def save(ds: str, rows: list, conn_factory=None) -> None:
|
||
factory = conn_factory or db.factor_conn
|
||
cols = ",".join(f"`{c}`" for c in COLUMNS)
|
||
marks = ",".join(["%s"] * len(COLUMNS))
|
||
payload = [tuple(r.get(c) for c in COLUMNS) for r in rows]
|
||
with factory() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(_CREATE.format(t=TABLE))
|
||
conn.commit()
|
||
with conn.cursor() as cur:
|
||
cur.execute(f"DELETE FROM {TABLE} WHERE plan_date = %s", (ds,))
|
||
if payload:
|
||
cur.executemany(f"INSERT INTO {TABLE} ({cols}) VALUES ({marks})", payload)
|
||
conn.commit()
|
||
|
||
|
||
def persist(ds: str, snap: dict, write=None) -> dict:
|
||
"""目标是主榜前 TOP_MAIN 只加观察档前 TOP_OBS 只（快照里的 main / observe 已按名次排），约一百二十家，
|
||
与方案说的"一两百家"一致；07:10 计划默认只展示 20 加 10，那份太窄。"""
|
||
rows = build_rows(ds, (snap.get("main") or [])[:TOP_MAIN], (snap.get("observe") or [])[:TOP_OBS])
|
||
(write or save)(ds, rows)
|
||
n_main = sum(1 for r in rows if r["source"] == "main")
|
||
print(f" 评析目标名单 {ds}：主榜 {n_main}，观察 {len(rows) - n_main}，写入 {TABLE}")
|
||
return {"date": ds, "rows": len(rows), "main": n_main}
|
||
|
||
|
||
def from_plan_file(ds: str, plan_dir: str | None = None, write=None) -> dict:
|
||
"""从已生成的计划文件补写（手工入口 run.py review-targets --date）。"""
|
||
path = os.path.join(plan_dir or config.PLAN_SNAPSHOT_DIR, f"plan_{ds}.json")
|
||
with open(path, encoding="utf-8") as f:
|
||
snap = json.load(f)
|
||
return persist(ds, snap, write=write)
|