akg-factor-bridge/review_targets.py

"""个股深度评析的目标名单（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")
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:
    """目标是计划里展示给人的那一份候选单（主榜前几十只加观察档），不是全部主榜；快照里叫 main_shown / observe_shown，
    老快照没有这两个键就退回 main / observe。"""
    main = snap.get("main_shown") if snap.get("main_shown") is not None else (snap.get("main") or [])
    obs = snap.get("observe_shown") if snap.get("observe_shown") is not None else (snap.get("observe") or [])
    rows = build_rows(ds, main, 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)