group_fqcd_jr/tools/sync_nav_history.py

"""把真实历史净值同步进 `fin_nav_history`（产品详情页净值走势图的数据源）。

## 为什么需要它

`fin_nav_history` 一直是**空表**（0 行），所以产品详情页画不出净值走势。
此前那条曲线来自前端 `common/mock-data.js` 里编的 12 个点位 ——
2026-09-13 接入公开产品接口时把它去掉了（走势图最容易被当成真数据），
页面改为显式显示"历史净值数据尚未接入"。本脚本补上这条数据链路。

## 数据源

东方财富历史净值接口（`api.fund.eastmoney.com/f10/lsjz`）。
注意它与**行情**源不同：东财的 `push2` / `push2his` 两个行情域名在本环境实测
连接被拒，而这个净值域名可用 —— 所以历史走势走净值这条路。

不校验"南方基金白名单"：`fin_product` 里有非南方基金的产品（510300 是华泰柏瑞的），
走 `hq.py` 的 `get_southern_fund_nav_history` 会被白名单直接拒绝。

## 幂等

按 `(product_id, nav_date)` upsert（表上有复合唯一键
`uk_fin_nav_history_product_id_nav_date`），重复跑不会产生重复行。

## 用法

    python tools/sync_nav_history.py                    # 全部在售产品，近 180 天
    python tools/sync_nav_history.py --days 365         # 取更长的历史
    python tools/sync_nav_history.py --codes 510300,515450
"""

from __future__ import annotations

import argparse
import asyncio
import sys
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from pathlib import Path

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

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
from app.model.fund import FundNavHistory, FundProduct  # noqa: E402

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]

LISTED_STATUS = "上市"


async def _next_id(session: AsyncSession) -> int:
    """`fin_nav_history` 没有 AUTO_INCREMENT（`fin_*` 表一贯如此），自己算主键。"""
    result = await session.execute(select(func.coalesce(func.max(FundNavHistory.id), 0)))
    return int(result.scalar_one()) + 1


async def sync(codes: tuple[str, ...], days: int) -> dict[str, object]:
    end = date.today()
    start = end - timedelta(days=days)
    now = datetime.now(UTC).replace(tzinfo=None)
    adapter = EastmoneyFundAdapter()

    written = 0
    skipped: list[dict[str, str]] = []
    per_product: list[dict[str, object]] = []

    async with SessionFactory() as session:
        query = select(FundProduct).where(FundProduct.status == LISTED_STATUS)
        if codes:
            query = query.where(FundProduct.product_code.in_(codes))
        products = (
            (await session.execute(query.order_by(FundProduct.product_code))).scalars().all()
        )

        next_id = await _next_id(session)
        for product in products:
            rows = await adapter.fetch_nav_history(
                product.product_code, start_date=start, end_date=end
            )
            if not rows:
                skipped.append({
                    "product_code": product.product_code,
                    "reason": "净值源没有返回数据",
                })
                continue
            # 该产品已有的日期，避免重复插入（幂等）
            existing = set(
                (
                    await session.execute(
                        select(FundNavHistory.nav_date).where(
                            FundNavHistory.product_id == product.id
                        )
                    )
                )
                .scalars()
                .all()
            )
            added = 0
            for row in rows:
                nav_date = date.fromisoformat(str(row["nav_date"]))
                if nav_date in existing:
                    continue
                session.add(FundNavHistory(
                    id=next_id,
                    product_id=int(product.id),
                    nav_date=nav_date,
                    nav=Decimal(str(row["nav"])),
                    created_at=now,
                ))
                next_id += 1
                added += 1
            written += added
            per_product.append({
                "product_code": product.product_code,
                "fetched": len(rows),
                "inserted": added,
            })

        await session.commit()

    return {
        "requested": len(products),
        "written": written,
        "per_product": per_product,
        "skipped": skipped,
        "window": f"{start} ~ {end}",
    }


def summarize(result: dict[str, object]) -> str:
    lines = [
        f"区间：{result['window']}",
        f"产品 {result['requested']} 只；新写入 {result['written']} 行",
    ]
    per_product = result.get("per_product") or []
    if per_product:
        lines.append("")
        lines.append("每只产品（取到 / 新写入）：")
        for item in per_product:  # type: ignore[union-attr]
            lines.append(
                f"  · {item['product_code']}  {item['fetched']} 条 / 新增 {item['inserted']} 行"
            )
    skipped = result.get("skipped") or []
    if skipped:
        lines.append("")
        lines.append("没取到数据的产品：")
        for item in skipped:  # type: ignore[union-attr]
            lines.append(f"  · {item['product_code']}  {item['reason']}")
    return "\n".join(lines)


async def main() -> int:
    parser = argparse.ArgumentParser(description="同步历史净值到 fin_nav_history")
    parser.add_argument("--codes", default="", help="只同步这些产品代码（逗号分隔）")
    parser.add_argument("--days", type=int, default=180, help="回溯天数，默认 180")
    args = parser.parse_args()

    codes = tuple(code.strip() for code in args.codes.split(",") if code.strip())
    result = await sync(codes, args.days)
    print(summarize(result))

    if not result.get("written") and not result.get("per_product"):
        print("\n[警告] 一行都没写 —— 详情页走势图仍然没有数据。")
        return 1
    print("\n完成。产品详情页的净值走势图现在有数据了。")
    return 0


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