group_fqcd_jr/tools/fetch_live_quotes.py

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

## 数据源

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

  · 实时/收盘价  `https://push2.eastmoney.com/api/qt/ulist.np/get`
  · 历史净值     `https://api.fund.eastmoney.com/f10/lsjz`

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

## 用法

    python tools/fetch_live_quotes.py            # 与库里快照逐只对比
    python tools/fetch_live_quotes.py --json     # 只输出 JSON（供生成知识条目等）

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

`fin_product.current_nav` 是**快照缓存**，正常情况下与数据源一致（本机 20 只里 18 只
完全一致，只是小数位表示不同）。**不一致的那两只才是重点**：它们要么被人为改过
（演示需要），要么字段口径本身有问题 —— 直接覆盖会把线索一起抹掉。
所以本脚本只**报告**差异，不写库；要不要刷新由 `ProductHistorySyncService` 决定。
"""

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
    FROM fin_product
    WHERE fund_manager = '南方基金'
      AND exchange_code IN ('SSE', 'SZSE')
      AND status = '上市'
    ORDER BY product_code
"""


async def collect() -> 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,
    )

    records: list[dict[str, object]] = []
    for row, history in zip(rows, histories, strict=True):
        if isinstance(history, BaseException):
            history = {"degraded": True, "error": type(history).__name__}
        live_nav = history.get("nav")
        stored_nav = row["current_nav"]
        records.append({
            "product_code": str(row["product_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"]),
            "management_fee_rate": (
                None if row["management_fee_rate"] is None else str(row["management_fee_rate"])
            ),
            "custodian_fee_rate": (
                None if row["custodian_fee_rate"] is None else str(row["custodian_fee_rate"])
            ),
            "live_nav": None if live_nav is None else str(live_nav),
            "live_nav_date": (
                None if history.get("nav_date") is None else str(history["nav_date"])
            ),
            "live_change_pct": (
                None if history.get("daily_change") is None else str(history["daily_change"])
            ),
            "stored_nav": str(stored_nav),
            "stored_nav_at": str(row["current_nav_at"]),
            "matches_stored": (
                live_nav is not None and float(live_nav) == float(stored_nav)
            ),
            "degraded": bool(history.get("degraded")),
        })
    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(f"场内产品 {len(records)} 只；取到真实行情 {len(ok)} 只"
          f"（与库里一致 {len(same)}、不一致 {len(diff)}）\n")
    header = f"{'代码':<8}{'名称':<26}{'真实净值':>10}{'涨跌%':>8}{'净值日期':>12}   {'库里净值':>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}"
        )

    if diff:
        print("\n不一致的条目（需要人工判断，不要直接覆盖）：")
        for item in diff:
            print(f"  · {item['product_code']} {item['product_name']}")
            print(f"      真实 {item['live_nav']}（{item['live_nav_date']}）"
                  f" vs 库里 {item['stored_nav']}（快照于 {item['stored_nav_at']}）")

    missing = [item["product_code"] for item in records if item["degraded"]]
    if missing:
        print(f"\n取数失败：{missing}")


async def main() -> int:
    parser = argparse.ArgumentParser(description="拉取场内基金真实行情并核对库里快照")
    parser.add_argument("--json", action="store_true", help="只输出 JSON")
    args = parser.parse_args()

    records = await collect()
    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()))