From 930dd59d6713eaf55580d9e08f24c67fceb594a8 Mon Sep 17 00:00:00 2001 From: Windows Date: Fri, 11 Sep 2026 18:39:57 +0800 Subject: [PATCH] feat: complete advisory market data refresh pipeline --- app/service/product_history_sync_service.py | 15 +- docs/21-投顾Agent迁移TODO.md | 8 + hq.py | 80 +++++++- .../test_fund_market_adapter.py | 29 +++ .../test_product_history_sync_service.py | 28 ++- tools/sync_advisor_market_data.py | 172 ++++++++++++++++++ 6 files changed, 327 insertions(+), 5 deletions(-) create mode 100644 tools/sync_advisor_market_data.py diff --git a/app/service/product_history_sync_service.py b/app/service/product_history_sync_service.py index e8a64b4..aae2dc4 100644 --- a/app/service/product_history_sync_service.py +++ b/app/service/product_history_sync_service.py @@ -1,4 +1,4 @@ -"""Synchronize public fund NAV history into the additive advisory history store.""" +"""Synchronize public fund history into the additive advisory history store.""" import asyncio from collections.abc import Callable @@ -24,7 +24,7 @@ class ProductHistorySyncResult: class ProductHistorySyncService: - """Writes fund NAV observations only; it never writes immutable baseline tables.""" + """Write NAV observations and verified exchange turnover only.""" SOURCE = "eastmoney_hq_nav" @@ -110,12 +110,21 @@ class ProductHistorySyncService: return None if close_price <= 0: return None + turnover = raw.get("turnover_amount") + parsed_turnover: Decimal | None = None + if turnover not in (None, ""): + try: + parsed_turnover = Decimal(str(turnover)) + except InvalidOperation: + parsed_turnover = None + if parsed_turnover is not None and parsed_turnover < 0: + parsed_turnover = None return { "product_id": product_id, "trade_date": trade_date, "price_kind": "fund_nav", "close_price": close_price, - "turnover_amount": None, + "turnover_amount": parsed_turnover, "source": cls.SOURCE, "source_updated_at": now, "created_at": now, diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index 173b74c..c3c9b4a 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -145,6 +145,14 @@ Redis 不可用时实测按设计降级放行；生产装配模式因本机 Milv 原 `jr_agent` 库的集成测试仍为 `25 passed, 3 failed, 1 skipped`，失败是旧约束、测试账号外键和 UTC 状态差异；其结构审计多出 8 张历史场外表，约束审计有 12 条历史唯一键差异，均未修改原库。 +阶段十四完成行情增量管道修复：`hq.py` 新增场内日线 K 线成交额解析（Eastmoney f56），净值历史同步 +按交易日合并真实成交额；新增 `tools/sync_advisor_market_data.py`，可一次完成历史增量、指标重算和 +数据质量快照 upsert。历史接口失败时保留净值、成交额为空，质量状态继续 `rejected`，不绕过推荐和动态 +配置的失败关闭门槛。专项测试 `7 passed`，全量单元测试 `494 passed, 3 warnings`，Ruff 和 MyPy +通过。独立迁移库真实刷新结果：历史 `5240` 条、`19` 个产品，成交额非空 `0` 条；东方财富历史 K 线 +端点批量请求出现 `RemoteProtocolError`，因此质量 `rejected=19`，产品推荐和动态配置真实验收仍待 +行情源恢复后复验。实现提交：待提交。 + ## 一、迁移准备 - [ ] 确认远程仓库可访问。（当前失败：连接 `47.106.207.27:3000` 被拒绝） diff --git a/hq.py b/hq.py index 728b3b1..c8b82d8 100644 --- a/hq.py +++ b/hq.py @@ -18,10 +18,12 @@ logger = logging.getLogger(__name__) NAV_API = "https://api.fund.eastmoney.com/f10/lsjz" RETURN_API = "https://api.fund.eastmoney.com/pinzhong/LJSYLZS" QUOTE_API = "https://push2.eastmoney.com/api/qt/ulist.np/get" +EXCHANGE_HISTORY_API = "https://push2his.eastmoney.com/api/qt/stock/kline/get" TENCENT_QUOTE_API = "https://qt.gtimg.cn/q=" DETAIL_API = "https://fund.eastmoney.com/pingzhongdata/{code}.js" SOUTHERN_COMPANY_API = "https://fund.eastmoney.com/company/80000220.html" REQUEST_TIMEOUT = 12.0 +EXCHANGE_HISTORY_RETRIES = 2 MAX_FUNDS_PER_CALL = 1000 FUND_TYPE_GROUPS = { "货币型": ("202308", "020480", "511810"), @@ -119,6 +121,15 @@ def get_southern_fund_nav_history( records = _fetch_nav_records( fund_code, start_date=start.isoformat(), end_date=end.isoformat(), all_pages=True ) + turnover_by_date: dict[str, str] = {} + try: + turnover_by_date = { + row["trade_date"]: row["turnover_amount"] + for row in get_southern_fund_exchange_history(fund_code, start_date, end_date) + if row.get("turnover_amount") + } + except (httpx.HTTPError, ValueError, TypeError) as exc: + logger.warning("历史成交额接口失败 code=%s error=%s", fund_code, type(exc).__name__) observations: list[dict[str, str]] = [] for record in records: value_date = str(record.get("FSRQ") or "") @@ -129,10 +140,77 @@ def get_southern_fund_nav_history( continue except ValueError: continue - observations.append({"fund_code": fund_code, "trade_date": value_date, "nav": nav}) + row = {"fund_code": fund_code, "trade_date": value_date, "nav": nav} + if value_date in turnover_by_date: + row["turnover_amount"] = turnover_by_date[value_date] + observations.append(row) return sorted(observations, key=lambda item: item["trade_date"]) +def get_southern_fund_exchange_history( + fund_code: str, start_date: str, end_date: str +) -> list[dict[str, str]]: + """Return exchange daily close and turnover for a listed fund. + + The NAV endpoint has no trading amount. Eastmoney's exchange K-line + endpoint exposes amount as field f56, which is the only source accepted + for historical liquidity calculations here. + """ + if fund_code not in SOUTHERN_FUND_CODES: + raise ValueError("基金代码不在南方基金白名单内") + start = date.fromisoformat(start_date) + end = date.fromisoformat(end_date) + if start > end: + raise ValueError("开始日期不能晚于结束日期") + secid = ("1." if fund_code.startswith(("5", "6", "9")) else "0.") + fund_code + params: dict[str, str | int] = { + "secid": secid, + "klt": 101, + "fqt": 1, + "beg": start.strftime("%Y%m%d"), + "end": end.strftime("%Y%m%d"), + "fields1": "f1,f2,f3,f4,f5,f6", + "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61", + } + for attempt in range(EXCHANGE_HISTORY_RETRIES): + try: + response = httpx.get( + EXCHANGE_HISTORY_API, + params=params, + headers=HEADERS, + timeout=REQUEST_TIMEOUT, + ) + response.raise_for_status() + break + except httpx.HTTPError: + if attempt == EXCHANGE_HISTORY_RETRIES - 1: + raise + time.sleep(0.2 * (attempt + 1)) + payload = response.json() + records = ((payload.get("data") or {}).get("klines") or []) + result: list[dict[str, str]] = [] + for raw in records: + fields = str(raw).split(",") + if len(fields) < 7: + continue + trade_date, close_price, turnover_amount = fields[0], fields[2], fields[6] + try: + parsed_date = date.fromisoformat(trade_date) + if parsed_date < start or parsed_date > end: + continue + if float(close_price) <= 0 or float(turnover_amount) < 0: + continue + except ValueError: + continue + result.append({ + "fund_code": fund_code, + "trade_date": trade_date, + "close_price": close_price, + "turnover_amount": turnover_amount, + }) + return result + + def get_southern_fund_catalog( fund_codes: list[str] | tuple[str, ...], ) -> list[dict[str, Any]]: diff --git a/tests/unit/infrastructure/test_fund_market_adapter.py b/tests/unit/infrastructure/test_fund_market_adapter.py index 55445e6..f576219 100644 --- a/tests/unit/infrastructure/test_fund_market_adapter.py +++ b/tests/unit/infrastructure/test_fund_market_adapter.py @@ -7,6 +7,35 @@ import pytest from app.infrastructure.fund_market_adapter import EastmoneyFundAdapter +def test_hq_exchange_history_parser_contract(monkeypatch: pytest.MonkeyPatch) -> None: + import hq + + class Response: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, object]: + return {"data": {"klines": [ + "2026-09-09,1.20,1.23,1.24,1.19,100000,1234567.89,4.1,2.5,0.03,1.2", + "2026-09-08,1.20,1.21,1.22,1.19,90000,1000000,2.5,1.0,0.01,1.0", + "malformed", + ]}} + + def get(*args: object, **kwargs: object) -> Response: + del args, kwargs + return Response() + + monkeypatch.setattr(hq.httpx, "get", get) + rows = hq.get_southern_fund_exchange_history("159511", "2026-09-09", "2026-09-09") + + assert rows == [{ + "fund_code": "159511", + "trade_date": "2026-09-09", + "close_price": "1.23", + "turnover_amount": "1234567.89", + }] + + @pytest.mark.asyncio async def test_adapter_parses_names_quotes_and_history() -> None: def handler(request: httpx.Request) -> httpx.Response: diff --git a/tests/unit/service/test_product_history_sync_service.py b/tests/unit/service/test_product_history_sync_service.py index 9c761ba..c87c46b 100644 --- a/tests/unit/service/test_product_history_sync_service.py +++ b/tests/unit/service/test_product_history_sync_service.py @@ -18,6 +18,33 @@ def test_history_sync_row_accepts_a_positive_public_nav() -> None: assert result["source"] == "eastmoney_hq_nav" +def test_history_sync_row_keeps_verified_exchange_turnover() -> None: + now = datetime.now(UTC).replace(tzinfo=None) + + result = ProductHistorySyncService._row( + 7, + {"trade_date": "2026-09-09", "nav": "1.234500", "turnover_amount": "123456.78"}, + now, + ) + + assert result is not None + assert result["turnover_amount"] == Decimal("123456.78") + + +def test_history_sync_row_rejects_negative_or_invalid_turnover() -> None: + now = datetime.now(UTC).replace(tzinfo=None) + + negative = ProductHistorySyncService._row( + 7, {"trade_date": "2026-09-09", "nav": "1", "turnover_amount": "-1"}, now + ) + invalid = ProductHistorySyncService._row( + 7, {"trade_date": "2026-09-09", "nav": "1", "turnover_amount": "bad"}, now + ) + + assert negative is not None and negative["turnover_amount"] is None + assert invalid is not None and invalid["turnover_amount"] is None + + def test_history_sync_row_rejects_invalid_or_non_positive_nav() -> None: now = datetime.now(UTC).replace(tzinfo=None) @@ -25,4 +52,3 @@ def test_history_sync_row_rejects_invalid_or_non_positive_nav() -> None: assert ProductHistorySyncService._row( 7, {"trade_date": "2026-09-09", "nav": "0"}, now ) is None - diff --git a/tools/sync_advisor_market_data.py b/tools/sync_advisor_market_data.py new file mode 100644 index 0000000..a24a8c5 --- /dev/null +++ b/tools/sync_advisor_market_data.py @@ -0,0 +1,172 @@ +"""Sync listed Southern Fund history and rebuild advisory data snapshots. + +The command writes only additive advisory tables. Baseline product and trading +tables remain read-only for this pipeline. +""" +# Imports intentionally follow the project-root path bootstrap below so the +# command works when invoked as ``python tools/sync_advisor_market_data.py``. +# ruff: noqa: E402 + +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 + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from sqlalchemy import select +from sqlalchemy.dialects.mysql import insert + +from app.infrastructure.db import SessionFactory +from app.model.advisor_product import ( + AdvisorProductDataQualitySnapshot, + AdvisorProductMetricSnapshot, + AdvisorProductPriceHistory, +) +from app.model.fund import FundProduct +from app.service.product_history_sync_service import ProductHistorySyncService +from app.service.product_metric_service import ProductMetricService + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Sync advisory listed-fund market data") + parser.add_argument("--days", type=int, default=400, help="Calendar days to refresh") + parser.add_argument("--limit", type=int, default=100, help="Maximum products") + parser.add_argument("--as-of-date", type=date.fromisoformat, default=date.today()) + return parser.parse_args() + + +def expected_trading_days(start: date, end: date) -> int: + return sum( + (start + timedelta(days=offset)).weekday() < 5 + for offset in range((end - start).days + 1) + ) + + +async def rebuild_snapshots(*, start: date, end: date, limit: int) -> tuple[int, int]: + now = datetime.now(UTC).replace(tzinfo=None) + expected = expected_trading_days(start, end) + async with SessionFactory() as session: + products = list(await session.scalars( + select(FundProduct).where( + FundProduct.fund_manager == "南方基金", + FundProduct.status == "上市", + ).order_by(FundProduct.id).limit(limit) + )) + + metric_rows: list[dict[str, object]] = [] + quality_rows: list[dict[str, object]] = [] + async with SessionFactory() as session: + for product in products: + history = list(await session.scalars( + select(AdvisorProductPriceHistory).where( + AdvisorProductPriceHistory.product_id == product.id, + AdvisorProductPriceHistory.trade_date >= start, + AdvisorProductPriceHistory.trade_date <= end, + AdvisorProductPriceHistory.price_kind == "fund_nav", + ).order_by(AdvisorProductPriceHistory.trade_date) + )) + snapshot = ProductMetricService.snapshot(product.id, history) + if snapshot is None: + continue + metric_rows.append({ + "product_id": snapshot.product_id, + "as_of_date": snapshot.as_of_date, + "trailing_20d_return_pct": snapshot.trailing_20d_return_pct, + "trailing_120d_return_pct": snapshot.trailing_120d_return_pct, + "annualized_volatility_pct": snapshot.annualized_volatility_pct, + "max_drawdown_pct": snapshot.max_drawdown_pct, + "average_daily_turnover_amount": snapshot.average_daily_turnover_amount, + "observation_count": snapshot.observation_count, + "source": snapshot.source, + "calculation_version": snapshot.calculation_version, + "created_at": snapshot.created_at, + }) + turnover_count = sum(item.turnover_amount is not None for item in history) + observation_count = len(history) + price_coverage = min( + Decimal("100"), Decimal(observation_count) * 100 / max(1, expected) + ).quantize(Decimal("0.0001")) + turnover_coverage = min( + Decimal("100"), Decimal(turnover_count) * 100 / max(1, expected) + ).quantize(Decimal("0.0001")) + reasons: list[str] = [] + if observation_count < 20: + reasons.append("INSUFFICIENT_OBSERVATIONS") + if price_coverage < 80: + reasons.append("LOW_PRICE_COVERAGE") + if turnover_coverage < 80: + reasons.append("LOW_TURNOVER_COVERAGE") + daily_returns = [ + abs((current.close_price / previous.close_price - 1) * 100) + for previous, current in zip(history, history[1:], strict=False) + if previous.close_price > 0 + ] + quality_rows.append({ + "product_id": product.id, + "as_of_date": end, + "observation_count": observation_count, + "expected_trading_days": expected, + "price_coverage_pct": price_coverage, + "turnover_coverage_pct": turnover_coverage, + "max_abs_daily_return_pct": ( + max(daily_returns).quantize(Decimal("0.0001")) if daily_returns else None + ), + "status": "accepted" if not reasons else "rejected", + "reason_codes": reasons, + "rule_version": "v1", + "created_at": now, + }) + + async with SessionFactory() as session, session.begin(): + if metric_rows: + statement = insert(AdvisorProductMetricSnapshot).values(metric_rows) + await session.execute(statement.on_duplicate_key_update( + trailing_20d_return_pct=statement.inserted.trailing_20d_return_pct, + trailing_120d_return_pct=statement.inserted.trailing_120d_return_pct, + annualized_volatility_pct=statement.inserted.annualized_volatility_pct, + max_drawdown_pct=statement.inserted.max_drawdown_pct, + average_daily_turnover_amount=statement.inserted.average_daily_turnover_amount, + observation_count=statement.inserted.observation_count, + source=statement.inserted.source, + calculation_version=statement.inserted.calculation_version, + created_at=statement.inserted.created_at, + )) + if quality_rows: + statement = insert(AdvisorProductDataQualitySnapshot).values(quality_rows) + await session.execute(statement.on_duplicate_key_update( + observation_count=statement.inserted.observation_count, + expected_trading_days=statement.inserted.expected_trading_days, + price_coverage_pct=statement.inserted.price_coverage_pct, + turnover_coverage_pct=statement.inserted.turnover_coverage_pct, + max_abs_daily_return_pct=statement.inserted.max_abs_daily_return_pct, + status=statement.inserted.status, + reason_codes=statement.inserted.reason_codes, + rule_version=statement.inserted.rule_version, + created_at=statement.inserted.created_at, + )) + return len(metric_rows), sum(item["status"] == "accepted" for item in quality_rows) + + +async def run(args: argparse.Namespace) -> None: + end = args.as_of_date + start = end - timedelta(days=max(1, args.days)) + result = await ProductHistorySyncService().sync( + days=args.days, limit=args.limit, as_of_date=end + ) + metric_count, accepted_count = await rebuild_snapshots( + start=start, end=end, limit=args.limit + ) + print( + f"history products={result.product_count} observations={result.observation_count}; " + f"metrics={metric_count} quality_accepted={accepted_count}" + ) + + +if __name__ == "__main__": + asyncio.run(run(parse_args()))