From 25724003d1875230ce8edba676466d5eb7803009 Mon Sep 17 00:00:00 2001 From: zlt Date: Fri, 4 Sep 2026 11:40:56 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=BB=E6=95=B0=E8=84=9A=E6=9C=AC=E5=88=86?= =?UTF-8?q?=E5=BC=80=E6=95=B0=E6=8D=AE=E6=97=A5=E4=B8=8E=E8=AE=A1=E5=88=92?= =?UTF-8?q?=E6=97=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 行业观点快照按计划日落库，行情与论断按数据日取，两者在生产里差一天。 原先用同一个日期取三样东西，结果是要么读不到当天快照、要么读不到行情。 Co-Authored-By: Claude Opus 5 --- logic_state_coverage.py | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/logic_state_coverage.py b/logic_state_coverage.py index 34d2ae3..edae7bb 100644 --- a/logic_state_coverage.py +++ b/logic_state_coverage.py @@ -18,7 +18,8 @@ 三，同一家机构在窗口里可能发多篇，先按机构取最近一篇再算中位数，否则发得勤的 机构会被重复计入。 -跑法（155 上）：docker exec akg_factor_bridge python3 logic_state_coverage.py [数据日] +跑法（155 上）：docker exec akg_factor_bridge python3 logic_state_coverage.py [数据日] [计划日] + 计划日不给就取数据日的次日。 """ from __future__ import annotations @@ -114,9 +115,20 @@ def snapshot_of(day: str) -> dict: for _, r in df.iterrows()} -def main(ds: str | None = None) -> None: +def main(ds: str | None = None, plan_day: str | None = None) -> None: + """ds 是数据日（行情与论断按它取），plan_day 是计划日（行业观点快照按它取）。 + + 这两个日期在生产里本来就差一天：计划日凌晨构建，用的是上一个交易日的数据， + 而当天的行业观点快照按计划日落库。早先这个脚本用同一个日期取三样东西， + 结果是要么读不到当天的快照，要么读不到行情。plan_day 不给就取数据日的次日。 + """ ds = ds or (dt.date.today() - dt.timedelta(days=1)).isoformat() - print(f"=== 逻辑状态四态覆盖率读数 · 数据日 {ds} ===\n") + if not plan_day: + try: + plan_day = (dt.date.fromisoformat(ds) + dt.timedelta(days=1)).isoformat() + except ValueError: + plan_day = ds + print(f"=== 逻辑状态四态覆盖率读数 · 数据日 {ds} · 计划日 {plan_day} ===\n") codes = db.read_pg("SELECT DISTINCT ts_code FROM v_factor_stock_daily " "WHERE trade_date = %s", (ds,))["ts_code"].tolist() @@ -127,7 +139,7 @@ def main(ds: str | None = None) -> None: brokers = broker_paths(codes, ds) # 要的是这个计划日当天那一版快照（带迁移与陈旧两列），不是它之前的那一版—— # load_previous 读的是严格早于计划日的，用它会在快照首日读到空。当天没有就退回上一版。 - snaps = snapshot_of(ds) or judgement.load_previous(ds) + snaps = snapshot_of(plan_day) or judgement.load_previous(plan_day) print(f"甲路取到 {len(claims)} 只票的论断；丙路算得出 {len(brokers)} 只票；" f"乙路快照 {len(snaps)} 个簇\n") @@ -194,4 +206,5 @@ def main(ds: str | None = None) -> None: if __name__ == "__main__": - main(sys.argv[1] if len(sys.argv) > 1 else None) + main(sys.argv[1] if len(sys.argv) > 1 else None, + sys.argv[2] if len(sys.argv) > 2 else None)