个股深度评析目标名单：出计划时写 t_akg_review_targets，手工补写子命令，单测

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

plan.py

@@ -870,6 +870,13 @@ def generate(date: str | None = None, top: int = 20, obs_top: int = 10,
                                   data.get("plan_version"))
     except Exception as e:  # noqa: BLE001 —— 表写失败不能拖垮出计划
         print(f"  （逐票逻辑状态表写入失败，计划照出，抗抖动明天少一天历史: {e!r}）")
+    # ---- 个股深度评析的目标名单（2026-09-09 方案）：主榜与观察档写进 153 库小表，数据基座 00:45 读它。
+    # 写失败只打印，不断产。----
+    try:
+        import review_targets
+        review_targets.persist(data["date"], snap)
+    except Exception as e:  # noqa: BLE001
+        print(f"  （评析目标名单写入失败，计划照出: {e!r}）")
     print(text)
     print(f"\n已写入 {out} 与快照 {jpath}"
           f"（主榜 {len(snap['main'])} 行、观察档 {len(snap['observe'])} 行、"

review_targets.py

@@ -0,0 +1,77 @@
+"""个股深度评析的目标名单（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:
+    rows = build_rows(ds, snap.get("main") or [], snap.get("observe") or [])
+    (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)

run.py

@@ -194,6 +194,8 @@ def main():
     pr.add_argument("--horizons", default="5,10,20")
     pr.add_argument("--start-price", choices=["next_close", "signal_close"], default="next_close")
     pr.add_argument("--out", default="data/review")
+    rv = sub.add_parser("review-targets")  # 个股深度评析的目标名单：从已生成的计划文件补写（2026-09-09）
+    rv.add_argument("--date", required=True, help="计划日 YYYY-MM-DD，读 data/plan/plan_<date>.json")
     f = sub.add_parser("freeze")
     f.add_argument("--date", help="默认今天")
     b = sub.add_parser("build")
@@ -234,6 +236,9 @@ def main():
     elif a.cmd == "judgement-snapshot":
         import judgement
         judgement.snapshot(a.date)
+    elif a.cmd == "review-targets":
+        import review_targets
+        print(review_targets.from_plan_file(a.date))
     elif a.cmd == "plan-review":
         import datetime as _dt
         import plan_review

test_review_targets.py

@@ -0,0 +1,24 @@
+"""评析目标名单：主榜与观察档转点后缀式、去重、名次；写入可注入。"""
+import review_targets as rt
+
+
+def t(name, cond):
+    print(("  ok  " if cond else "  FAIL") + name)
+    assert cond, name
+
+
+def main():
+    snap = {"main": [{"code": "SH600519", "name": "贵州茅台"}, {"code": "SZ300750", "name": "宁德时代"}, {"code": "SH600519"}],
+            "observe": [{"code": "SZ000001", "name": "平安银行"}, {"code": "bad"}]}
+    rows = rt.build_rows("2026-09-08", snap["main"], snap["observe"])
+    t("点后缀式且去重", [r["ts_code"] for r in rows] == ["600519.SH", "300750.SZ", "000001.SZ"])
+    t("来源与名次", rows[0]["source"] == "main" and rows[0]["rank"] == 1 and rows[2]["source"] == "observe" and rows[2]["rank"] == 1)
+    got = {}
+    out = rt.persist("2026-09-08", snap, write=lambda ds, rs: got.setdefault(ds, rs))
+    t("写入可注入且读数正确", out["rows"] == 3 and out["main"] == 2 and len(got["2026-09-08"]) == 3)
+    t("认不出的代码丢弃", all(len(r["ts_code"]) == 9 for r in rows))
+    print("ALL OK — 评析目标名单")
+
+
+if __name__ == "__main__":
+    main()