"""把真实行情快照（净值 + 费率）回填到 `fin_product`（默认 dry-run，只有 `--apply` 才写库）。 ## 为什么需要它 `fin_product` 里两个费率字段在本机 **20 只里 19 只为空**，客户问"管理费多少"只能转人工； `current_nav` 则有几只被人为改成演示值。这些都能从东方财富取到 （`EastmoneyFundAdapter.fetch_history` / `fetch_fees`），本脚本负责落到库里。 ## 口径 - **净值取 `DWJZ`（单位净值）**，不是场内交易价。这条口径由既有数据确定： 库里 19 只的 `current_nav` 与 `fetch_history` 的 `DWJZ` 逐只一致。 货币 ETF 是唯一容易搞混的 —— 它 `QUOTE_API` 的 `f2` 约 100 元（场内价）， 而 `DWJZ` 是 0.2332 这种量级，**两者不是一回事，不要混用**。 - 费率取 f10 基金概况页（管理费/托管费），单位是「%/年」。 ## 安全设计 1. **默认 dry-run**：只打印将执行的 UPDATE，不写库。真写要显式 `--apply`。 2. **默认只补空值**：`NULL` 才写，**已有值一律不动** —— 已有值可能是人为设定。 净值字段**从来不为空**，所以刷新净值必须显式给 `--overwrite`。 3. **`--codes` 限定范围**：只回填指定产品，避免"顺手"把全部快照刷成最新 （全量刷新是 `ProductHistorySyncService` / `MarketQuoteSyncService` 的职责）。 4. 写入走单个事务，任一失败整体回滚。 ## 用法 python tools/backfill_product_snapshot.py # 看将改什么 python tools/backfill_product_snapshot.py --apply # 只补空值（费率） python tools/backfill_product_snapshot.py --apply --overwrite --codes 510300 # 连净值和已有费率一起改 """ from __future__ import annotations import argparse import asyncio import sys from datetime import UTC, date, datetime, time from pathlib import Path from sqlalchemy import text PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from app.infrastructure.db import SessionFactory # noqa: E402 from app.infrastructure.fund_market_adapter import EastmoneyFundAdapter # noqa: E402 if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] #: 与 `MarketQuoteSyncService.sync` 同一批产品。 PRODUCT_QUERY = """ SELECT id, product_code, product_name, current_nav, current_nav_at, management_fee_rate, custodian_fee_rate FROM fin_product WHERE fund_manager = '南方基金' AND exchange_code IN ('SSE', 'SZSE') AND status = '上市' ORDER BY product_code """ #: 库里已有值 → 数据源字段。**净值排在最前**，便于阅读。 FEE_FIELDS = ( ("management_fee_rate", "management_fee_rate"), ("custodian_fee_rate", "custodian_fee_rate"), ) #: 净值快照写回时的时间部分：与库里既有记录的 `15:00:00` 保持一致（收盘时刻）。 NAV_AT_TIME = time(15, 0, 0) def _same(left: object, right: object) -> bool: """比较两个数值文本，容忍 `0.5` 与 `0.5000`、`4.5` 与 `4.500000` 这种表示差异。""" if left is None or right is None: return False try: return abs(float(str(left)) - float(str(right))) < 1e-9 except (TypeError, ValueError): return str(left) == str(right) async def main() -> int: parser = argparse.ArgumentParser(description="把真实净值与费率回填到 fin_product") parser.add_argument("--apply", action="store_true", help="真正写库（默认只看不改）") parser.add_argument( "--overwrite", action="store_true", help="连已有值也改（默认只补 NULL，不动人为设定过的值）", ) parser.add_argument( "--codes", default="", help="只处理这些产品代码（逗号分隔）；留空表示全部", ) args = parser.parse_args() wanted = {code.strip() for code in args.codes.split(",") if code.strip()} async with SessionFactory() as session: rows = (await session.execute(text(PRODUCT_QUERY))).mappings().all() if wanted: rows = [row for row in rows if str(row["product_code"]) in wanted] if not rows: print("没有匹配的产品。") return 1 adapter = EastmoneyFundAdapter() codes = [str(row["product_code"]) for row in rows] histories = await asyncio.gather( *(adapter.fetch_history(code, date.today()) for code in codes), return_exceptions=True, ) fees = await adapter.fetch_fees(codes) print(f"产品 {len(rows)} 只；取到净值 {sum(1 for h in histories if isinstance(h, dict) and not h.get('degraded'))} 只" f"、费率 {len(fees)} 只\n") updates: list[tuple[int, str, str, object]] = [] conflicts: list[str] = [] for row, history in zip(rows, histories, strict=True): code = str(row["product_code"]) live_nav: object = None nav_at: datetime | None = None if isinstance(history, dict) and not history.get("degraded"): live_nav = history.get("nav") nav_date = history.get("nav_date") if isinstance(nav_date, date): nav_at = datetime.combine(nav_date, NAV_AT_TIME) if live_nav is not None: stored_nav = row["current_nav"] if _same(stored_nav, live_nav): pass elif stored_nav is not None and not args.overwrite: conflicts.append( f"{code} {row['product_name']} 的 current_nav：" f"库里 {stored_nav} → 真实 {live_nav}" ) else: updates.append((int(row["id"]), code, "current_nav", live_nav)) if nav_at is not None: updates.append((int(row["id"]), code, "current_nav_at", nav_at)) live_fee = fees.get(code) or {} for column, key in FEE_FIELDS: live_value = live_fee.get(key) if live_value is None: continue stored = row[column] if _same(stored, live_value): continue if stored is not None and not args.overwrite: conflicts.append( f"{code} {row['product_name']} 的 {column}：库里 {stored} → 真实 {live_value}" ) continue updates.append((int(row["id"]), code, column, live_value)) if conflicts: print("与库里已有值冲突（默认不动，需要 --overwrite 才覆盖）：") for line in conflicts: print(f" * {line}") print() if not updates: print("没有需要回填的字段。") return 0 print(f"将执行 {len(updates)} 条 UPDATE：") for _, code, column, value in updates: print(f" · {code} {column} = {value}") if not args.apply: print("\n[dry-run] 未写库。加 --apply 执行。") return 0 now = datetime.now(UTC).replace(tzinfo=None) async with SessionFactory() as session, session.begin(): for product_id, _, column, value in updates: # 列名来自本模块常量，不是外部输入；值一律走参数绑定。 await session.execute( text( f"UPDATE fin_product SET {column} = :value, updated_at = :now " "WHERE id = :product_id" ), {"value": value, "now": now, "product_id": product_id}, ) print(f"\n已写入 {len(updates)} 个字段。") return 0 if __name__ == "__main__": sys.exit(asyncio.run(main()))