group_fqcd_jr/tools/backfill_product_snapshot.py

"""把真实行情快照（净值 + 费率）回填到 `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()))