From 0c4129738db8a150463ec0c103c2a3439ded7630 Mon Sep 17 00:00:00 2001 From: zlt Date: Wed, 9 Sep 2026 10:29:56 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=AA=E8=82=A1=E6=B7=B1=E5=BA=A6=E8=AF=84?= =?UTF-8?q?=E6=9E=90=E7=9B=AE=E6=A0=87=E5=90=8D=E5=8D=95=EF=BC=9A=E5=87=BA?= =?UTF-8?q?=E8=AE=A1=E5=88=92=E6=97=B6=E5=86=99=20t=5Fakg=5Freview=5Ftarge?= =?UTF-8?q?ts=EF=BC=8C=E6=89=8B=E5=B7=A5=E8=A1=A5=E5=86=99=E5=AD=90?= =?UTF-8?q?=E5=91=BD=E4=BB=A4=EF=BC=8C=E5=8D=95=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- plan.py | 7 ++++ review_targets.py | 77 ++++++++++++++++++++++++++++++++++++++++++ run.py | 5 +++ test_review_targets.py | 24 +++++++++++++ 4 files changed, 113 insertions(+) create mode 100644 review_targets.py create mode 100644 test_review_targets.py diff --git a/plan.py b/plan.py index 5c647f8..e9e740b 100644 --- a/plan.py +++ b/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'])} 行、" diff --git a/review_targets.py b/review_targets.py new file mode 100644 index 0000000..3f1a64d --- /dev/null +++ b/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) diff --git a/run.py b/run.py index f63d5fa..2a74bc8 100644 --- a/run.py +++ b/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_.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 diff --git a/test_review_targets.py b/test_review_targets.py new file mode 100644 index 0000000..8654bb7 --- /dev/null +++ b/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()