"""日频行业观点快照:每个计划日把数据基座最新一版的产业研判与环节评析原样抄一行,只写不判。 ## 为什么要有它 产业研判的结论存在数据基座的评析表里,那张表按主题唯一:每次重新评估就整行覆盖,并把生成 时间改成当前。所以库里永远只有最新一版,查不到"这个观点上一次是什么样、什么时候变的"。 而逻辑状态四态里"采信倾向从偏多变成偏空"这条判据依赖的正是这种变化,变化必须由读的一方自己 建立版本史。不先攒,这条判据永远看不到方向变过。 本模块就是攒版本史的那一步(2026-09-03 方案第四之五之三节,落地顺序第一步):每个计划日为 每个主题或环节写一行,只抄不判,不产生任何判决、不改任何排序、不拦任何票。 ## 落点与表名 写在平台因子库(写 t_factor_akg_* 的同一个 MySQL,config.factor_mysql)。理由两条:桥对数据 基座 PostgreSQL 只有只读账号(见 config 模块说明),写不进去;这张表又是选股系统自己的派生 记录,不是数据基座的事实,本来也不该回写基座。表名默认 t_akg_judgement_snapshot,不带 t_factor_ 前缀,免得平台的因子清单把它当成一张因子表收进去。 ## 幂等 同一计划日重跑先删当日行再整批插,不追加重复行。删与插在同一个事务里(一天几十行,量小, 不必像因子表那样分块提交)。上一版的比对只看"早于本计划日"的行,所以重跑得到的迁移与陈旧 天数与首次跑完全一样。 ## 迁移与陈旧的判定口径(写进列里,但本模块不判任何状态) 只有材料指纹(input_version)变化才算一次迁移:migrated 记 1,陈旧天数归零。 指纹没变而日期变老只累加陈旧天数:migrated 记 0,stale_days 加上距上一个计划日的自然日数。 本表里第一次见到这个主题,无从比对:migrated 留空,陈旧天数从零起算。 指纹本身缺失(视图给了空值)时不判迁移:migrated 留空,陈旧天数照常累加——陈旧只依赖时间。 采信倾向的上一版值单独存一列 leaning_prev,"从偏多变成偏空"这类问题不必回表自己拼。 往后要算逻辑状态四态,入口是本模块的 history 与 migrations 两个函数:它们只回答"什么时候 变过、从什么变成什么",不回答"该判成哪个状态"——四态合成是后续步骤(落地顺序第四步)的事, 现在不判。 """ from __future__ import annotations import datetime as dt import config import db import sources # 落库列的顺序(插入语句按此顺序拼参数)。plan_date 是计划日,snapshot_at 是写入时刻。 COLUMNS = ("plan_date", "cluster_key", "scope", "subject_name", "segment_name", "leaning", "leaning_prev", "n_bull", "n_bear", "n_flags", "verified", "n_verify_problems", "n_materials", "input_version", "migrated", "stale_days", "review_date", "reviewed_at", "model", "rounds", "review_id", "snapshot_at") # 建表语句与服务器上手工执行的那一份逐字相同(docs 里给的就是这一段)。 # 主键取(计划日,簇键):一个计划日一个簇最多一行,重跑覆盖;另建(簇键,计划日)索引, # 按主题回看版本史时走它。 _CREATE_TABLE = """ CREATE TABLE IF NOT EXISTS {t} ( plan_date DATE NOT NULL, cluster_key VARCHAR(191) NOT NULL, scope VARCHAR(16), subject_name VARCHAR(128), segment_name VARCHAR(128), leaning VARCHAR(32), leaning_prev VARCHAR(32), n_bull INT, n_bear INT, n_flags INT, verified TINYINT, n_verify_problems INT, n_materials INT, input_version VARCHAR(80), migrated TINYINT, stale_days INT, review_date DATE, reviewed_at DATETIME, model VARCHAR(64), rounds INT, review_id VARCHAR(64), snapshot_at DATETIME, PRIMARY KEY (plan_date, cluster_key), KEY idx_cluster (cluster_key, plan_date) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ # 回看上一版时读的列(不必把全部列拉回来)。 _PREV_COLS = ("plan_date", "cluster_key", "leaning", "input_version", "stale_days") def _day_gap(earlier: str | None, later: str | None) -> int: """两个计划日之间的自然日数;认不出日期时按一天算(陈旧只会少算不会多算)。""" try: return max(0, (dt.date.fromisoformat(later) - dt.date.fromisoformat(earlier)).days) except (TypeError, ValueError): return 1 def build_rows(day: str, fetched: list, prev: dict, now: str | None = None) -> list: """纯函数:把取回来的最新一版结论,配上一版的比对结果,拼成可落库的行。 day 是计划日(ISO 日期串),fetched 是 sources.judgement_rows 的返回,prev 是 簇键到上一版行的字典(load_previous 的返回),now 是写入时刻(不传取当前时刻)。 迁移与陈旧两列的口径见模块说明。同一簇键在 fetched 里出现多次时只留最后一条。""" stamp = now or dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S") by_key = {} for r in fetched or []: key = (r.get("cluster_key") or "").strip() if not key: continue by_key[key] = r out = [] for key in sorted(by_key): r = by_key[key] p = prev.get(key) if prev else None fp = r.get("input_version") if not p: migrated, stale, leaning_prev = None, 0, None else: inc = _day_gap(sources._ymd(p.get("plan_date")), day) # noqa: SLF001 —— 同仓自用 base = int(p.get("stale_days") or 0) prev_fp = (p.get("input_version") or None) leaning_prev = p.get("leaning") or None if fp is None or prev_fp is None: migrated, stale = None, base + inc # 指纹缺失无从比对,但日子照样变老 elif fp == prev_fp: migrated, stale = 0, base + inc else: migrated, stale = 1, 0 out.append({ "plan_date": day, "cluster_key": key[:191], "scope": r.get("scope"), "subject_name": (r.get("subject_name") or "")[:128] or None, "segment_name": (r.get("segment_name") or "")[:128] or None, "leaning": r.get("leaning"), "leaning_prev": leaning_prev, "n_bull": r.get("n_bull"), "n_bear": r.get("n_bear"), "n_flags": r.get("n_flags"), "verified": None if r.get("verified") is None else int(bool(r.get("verified"))), "n_verify_problems": r.get("n_verify_problems"), "n_materials": r.get("n_materials"), "input_version": fp, "migrated": migrated, "stale_days": stale, "review_date": r.get("review_date"), "reviewed_at": r.get("reviewed_at"), "model": r.get("model"), "rounds": r.get("rounds"), "review_id": r.get("review_id"), "snapshot_at": stamp, }) return out def load_previous(day: str, read_mysql=None) -> dict: """本计划日之前、回看窗口之内,每个簇最近一次快照的行,按簇键索引。 表还没建、或读不到,返回空字典并打印一行原因:那样全部主题都按"第一次见到"处理, 快照照写,只是这一天算不出迁移。read_mysql 可注入(离线单测)。""" table = config.JUDGEMENT_SNAPSHOT_TABLE reader = read_mysql or db.read_mysql try: start = (dt.date.fromisoformat(day) - dt.timedelta(days=config.JUDGEMENT_PREV_LOOKBACK_DAYS)).isoformat() except ValueError: print(f" (计划日 {day!r} 不是合法日期,跳过上一版比对)") return {} try: rows = sources._records(reader( # noqa: SLF001 —— 同仓自用 "factor", f"SELECT {', '.join(_PREV_COLS)} FROM {table} " f"WHERE plan_date >= %s AND plan_date < %s ORDER BY plan_date", (start, day))) except Exception as e: # noqa: BLE001 —— 头一次跑时表还不存在,属正常 print(f" ({table} 读不到历史行,本次全部按第一次见到处理: {e!r})") return {} out = {} for r in rows: # 按计划日升序读,同一簇键后来的覆盖先来的 key = str(r.get("cluster_key") or "").strip() if key: out[key] = dict(r) return out def save(day: str, rows: list, conn_factory=None) -> None: """幂等落库:建表(已存在就跳过)、删当日行、整批插当日行,删与插同一事务。 rows 为空也照样删当日行:那表示这一天视图里没有可抄的结论,当日不该留下上一次重跑的残行。 conn_factory 可注入(离线单测),默认走 db.factor_conn。""" table = config.JUDGEMENT_SNAPSHOT_TABLE factory = conn_factory or db.factor_conn cols = ",".join(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_TABLE.format(t=table)) conn.commit() with conn.cursor() as cur: cur.execute(f"DELETE FROM {table} WHERE plan_date = %s", (day,)) if payload: cur.executemany(f"INSERT INTO {table} ({cols}) VALUES ({marks})", payload) conn.commit() def snapshot(day: str | None = None, fetch=None, load_prev=None, write=None) -> dict: """一个计划日的完整一步:取最新一版结论、配上一版比对、幂等写当日行、打印读数。 day 不传取今天。补写一个早于今天的计划日要当心:视图给的永远是此刻的最新一版, 把它记在过去某一天名下,等于把今天的观点写成那天的观察结果,版本史会失真—— 所以这种情况下会多打印一行提醒,但仍照写(重跑与补跑都幂等)。 三个函数都可注入(离线单测):取数、读上一版、落库。""" day = day or dt.date.today().isoformat() if day != dt.date.today().isoformat(): print(f" ⚠️ 计划日 {day} 不是今天:抄下来的是此刻的最新一版结论," f"记在过去的日期名下会让版本史失真,确认是有意为之再用。") rows_in = (fetch or sources.judgement_rows)() prev = (load_prev or load_previous)(day) rows = build_rows(day, rows_in, prev) (write or save)(day, rows) moved = [r for r in rows if r["migrated"] == 1] first = [r for r in rows if r["migrated"] is None] stale_max = max([r["stale_days"] for r in rows], default=0) by_scope = {} for r in rows: by_scope[r["scope"] or "未标"] = by_scope.get(r["scope"] or "未标", 0) + 1 print(f"行业观点快照 {day}:写入 {len(rows)} 行" f"({'、'.join(f'{k} {v}' for k, v in sorted(by_scope.items())) or '无'})" f",指纹变化 {len(moved)} 个,第一次见到 {len(first)} 个," f"最长陈旧 {stale_max} 天 -> {config.JUDGEMENT_SNAPSHOT_TABLE}") for r in moved: print(f" 指纹变化:{r['subject_name']}(采信倾向 {r['leaning_prev']} -> {r['leaning']}," f"生成日 {r['review_date']})") return {"date": day, "rows": len(rows), "migrated": len(moved), "first_seen": len(first), "stale_max": stale_max, "by_scope": by_scope, "table": config.JUDGEMENT_SNAPSHOT_TABLE} # ============================================================================ # 版本史的读取入口。给后面的逻辑状态合成用,现在只回答"变过没有、什么时候变的"。 # ============================================================================ def history(cluster_key: str, upto: str | None = None, read_mysql=None) -> list: """一个簇在本表里的全部版本,按计划日升序。upto 给了就只要不晚于它的行。""" table = config.JUDGEMENT_SNAPSHOT_TABLE reader = read_mysql or db.read_mysql sql = (f"SELECT plan_date, leaning, leaning_prev, input_version, migrated, stale_days, " f"review_date, verified, n_bull, n_bear FROM {table} WHERE cluster_key = %s") params = (cluster_key,) if upto: sql += " AND plan_date <= %s" params = (cluster_key, upto) try: rows = sources._records(reader("factor", sql + " ORDER BY plan_date", params)) # noqa: SLF001 except Exception as e: # noqa: BLE001 print(f" ({table} 版本史读取失败: {e!r})") return [] return [dict(r) for r in rows] def migrations(rows: list) -> list: """纯函数:从一个簇的版本史里摘出被判为迁移的那些计划日。 判定口径就是落库时那一条,这里只是把已经算好的列摘出来,不重新判、也不据此判任何状态。 每条给出:发生在哪个计划日、采信倾向从什么变成什么、当时的材料指纹与生成日期。 采信倾向没变的迁移也照样列出来(材料换了但结论一样,本身就是一条信息)。""" out = [] for r in rows or []: if r.get("migrated") != 1: continue out.append({"plan_date": sources._ymd(r.get("plan_date")), # noqa: SLF001 "leaning_from": r.get("leaning_prev"), "leaning_to": r.get("leaning"), "input_version": r.get("input_version"), "review_date": sources._ymd(r.get("review_date"))}) # noqa: SLF001 return out