"""拉取场内基金的真实行情并核对库里的快照。 ## 数据源 走项目自带的 `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()))