group_fqcd_jr/tools/fetch_live_quotes.py

"""拉取场内基金的真实行情与费率，并核对库里的快照。

## 数据源

走项目自带的 `EastmoneyFundAdapter`（`app/infrastructure/fund_market_adapter.py`），
不自己拼 HTTP、不引第三方行情库：

  · 历史净值  `api.fund.eastmoney.com/f10/lsjz`
  · 费率      `fundf10.eastmoney.com/jbgk_{code}.html`（管理费/托管费/销售服务费/申赎费）
  · 实时行情  `push2.eastmoney.com/api/qt/ulist.np/get`（见适配器 `fetch_quotes`）

同一套适配器也是下单链路（`TradeService` 取 `quote_price`/`quote_source`）
与 `MarketQuoteSyncService`（东财 + 腾讯双源）在用的，所以这里拿到的值与成交价同源。

## 用法

    python tools/fetch_live_quotes.py              # 净值 + 费率，与库里逐只对比
    python tools/fetch_live_quotes.py --no-fees    # 跳过费率（费率要串行抓 45KB/只，慢）
    python tools/fetch_live_quotes.py --json       # 输出 JSON（供生成知识条目等）

## 为什么要做对比而不是直接覆盖

`fin_product.current_nav` / `management_fee_rate` 都是**快照缓存**。多数情况下与数据源
一致，**不一致的那几只才是重点**：它们要么被人为改过（演示需要），要么字段本身填错。
直接覆盖会把线索一起抹掉。所以本脚本只**报告**差异、不写库
（刷新快照属于 `ProductHistorySyncService` / `MarketQuoteSyncService` 的职责）。

本机实测的两处不一致（2026-09-13），都是这么发现的：
  · `510300` 库里净值 4.500000、真实 4.5794（且库里快照时间是当天，其余 19 只停在 09-11）
  · `510300` 库里管理费 0.5000 / 托管费 0.1000、真实 0.15 / 0.05
    —— 而 `0.50/0.10` 恰好是另外几只 ETF 的真实值，像是照抄过来的
"""

from __future__ import annotations

import argparse
import asyncio
import json
import sys
from datetime import date
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 product_code, product_name, exchange_code, product_category, risk_level,
           lot_size, min_amount, current_nav, current_nav_at,
           management_fee_rate, custodian_fee_rate, transaction_fee_rate
    FROM fin_product
    WHERE fund_manager = '南方基金'
      AND exchange_code IN ('SSE', 'SZSE')
      AND status = '上市'
    ORDER BY product_code
"""

#: 库里已有值的费率字段 ↔ 数据源字段（用于逐项核对）。
FEE_STORED_TO_LIVE = {
    "management_fee_rate": "management_fee_rate",
    "custodian_fee_rate": "custodian_fee_rate",
}


def _text(value: object) -> str | None:
    return None if value is None else str(value)


def _same(left: object, right: object) -> bool:
    """比较两个费率/净值文本，容忍 `0.5` 与 `0.5000` 这种表示差异。"""
    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 collect(*, with_fees: bool = True) -> list[dict[str, object]]:
    async with SessionFactory() as session:
        rows = (await session.execute(text(PRODUCT_QUERY))).mappings().all()

    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,
    )
    # 费率走 f10 概况页：约 45KB/只、且适配器内部是串行抓取（不对公开接口并发施压），
    # 因此只在需要时拉 —— `--no-fees` 可跳过。
    fees: dict[str, dict[str, object]] = {}
    if with_fees:
        fees = await adapter.fetch_fees(codes)

    records: list[dict[str, object]] = []
    for row, history in zip(rows, histories, strict=True):
        code = str(row["product_code"])
        if isinstance(history, BaseException):
            history = {"degraded": True, "error": type(history).__name__}
        live_nav = history.get("nav")
        stored_nav = row["current_nav"]
        live_fee = fees.get(code) or {}
        record: dict[str, object] = {
            "product_code": code,
            "product_name": str(row["product_name"]),
            "exchange_code": str(row["exchange_code"]),
            "product_category": str(row["product_category"]),
            "risk_level": str(row["risk_level"]),
            "lot_size": str(row["lot_size"]),
            "min_amount": str(row["min_amount"]),
            "live_nav": _text(live_nav),
            "live_nav_date": _text(history.get("nav_date")),
            "live_change_pct": _text(history.get("daily_change")),
            "stored_nav": str(stored_nav),
            "stored_nav_at": str(row["current_nav_at"]),
            "matches_stored": _same(live_nav, stored_nav),
            "degraded": bool(history.get("degraded")),
        }
        for stored_field, live_field in FEE_STORED_TO_LIVE.items():
            record[f"stored_{stored_field}"] = _text(row[stored_field])
            record[f"live_{live_field}"] = _text(live_fee.get(live_field))
        record["live_service_fee_rate"] = _text(live_fee.get("service_fee_rate"))
        record["live_subscribe_fee_rate"] = _text(live_fee.get("subscribe_fee_rate"))
        record["live_redeem_fee_rate"] = _text(live_fee.get("redeem_fee_rate"))
        record["live_full_name"] = _text(live_fee.get("full_name"))
        records.append(record)
    return records


def render(records: list[dict[str, object]]) -> None:
    ok = [item for item in records if not item["degraded"]]
    same = [item for item in ok if item["matches_stored"]]
    diff = [item for item in ok if not item["matches_stored"]]
    print("=" * 100)
    print("净值核对（数据源 vs 库里快照）")
    print("=" * 100)
    print(f"场内产品 {len(records)} 只；取到真实行情 {len(ok)} 只"
          f"（与库里一致 {len(same)}、不一致 {len(diff)}）\n")
    header = (f"{'代码':<8}{'名称':<26}{'真实净值':>10}{'涨跌%':>8}{'净值日期':>12}"
              f"   {'库里净值':>11}  核对")
    print(header)
    print("-" * len(header))
    for item in records:
        if item["degraded"]:
            verdict = "!! 取数失败"
        elif item["matches_stored"]:
            verdict = "一致"
        else:
            verdict = "**不一致**"
        print(
            f"{str(item['product_code']):<8}{str(item['product_name'])[:24]:<26}"
            f"{str(item['live_nav']):>10}{str(item['live_change_pct']):>8}"
            f"{str(item['live_nav_date']):>12}   {str(item['stored_nav']):>11}  {verdict}"
        )

    fee_rows = [item for item in records if item.get("live_management_fee_rate") is not None]
    if fee_rows:
        print()
        print("=" * 100)
        print("费率核对（数据源 vs 库里字段）")
        print("=" * 100)
        fee_header = (f"{'代码':<8}{'名称':<24}{'真实管理费':>11}{'库里':>8}"
                      f"{'真实托管费':>11}{'库里':>8}  核对")
        print(fee_header)
        print("-" * len(fee_header))
        fee_diff = []
        for item in fee_rows:
            live_manage = item["live_management_fee_rate"]
            stored_manage = item["stored_management_fee_rate"]
            if stored_manage is None:
                verdict = "库里为空"
            elif _same(live_manage, stored_manage) and _same(
                item["live_custodian_fee_rate"], item["stored_custodian_fee_rate"]
            ):
                verdict = "一致"
            else:
                verdict = "**不一致**"
                fee_diff.append(item)
            print(
                f"{str(item['product_code']):<8}{str(item['product_name'])[:22]:<24}"
                f"{str(live_manage):>11}{str(stored_manage):>8}"
                f"{str(item['live_custodian_fee_rate']):>11}"
                f"{str(item['stored_custodian_fee_rate']):>8}  {verdict}"
            )
        empty = [item for item in fee_rows if item["stored_management_fee_rate"] is None]
        print(f"\n真实费率全部取到 {len(fee_rows)} 只；库里为空 {len(empty)} 只、"
              f"与真实值冲突 {len(fee_diff)} 只")

    if diff or fee_diff:
        print()
        print("需要人工判断的条目（不要直接覆盖）：")
        for item in diff:
            print(f"  · 净值 {item['product_code']} {item['product_name']}："
                  f"真实 {item['live_nav']}（{item['live_nav_date']}） vs "
                  f"库里 {item['stored_nav']}（{item['stored_nav_at']}）")
        for item in fee_diff:
            print(f"  · 费率 {item['product_code']} {item['product_name']}："
                  f"真实 管理 {item['live_management_fee_rate']} / 托管 "
                  f"{item['live_custodian_fee_rate']} vs 库里 管理 "
                  f"{item['stored_management_fee_rate']} / 托管 "
                  f"{item['stored_custodian_fee_rate']}")

    full_names = [
        (str(item["product_code"]), item["live_full_name"])
        for item in records if item.get("live_full_name")
    ]
    if full_names:
        print()
        print("数据源返回的基金全称（可用于核对 fund_manager 标注是否正确）：")
        for code, name in full_names:
            print(f"  · {code}  {name}")


async def main() -> int:
    parser = argparse.ArgumentParser(description="拉取场内基金真实行情与费率，核对库里快照")
    parser.add_argument("--json", action="store_true", help="只输出 JSON")
    parser.add_argument("--no-fees", action="store_true", help="跳过费率抓取（较快）")
    args = parser.parse_args()

    records = await collect(with_fees=not args.no_fees)
    if args.json:
        print(json.dumps(records, ensure_ascii=False, indent=1))
    else:
        render(records)
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))