AI260626/group_fqcd_jr
13
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: complete advisory market data refresh pipeline

Browse Source
This commit is contained in:
Windows
2026-09-11 18:39:57 +08:00
parent 95633188f1
commit 930dd59d67
6 changed files with 327 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/service/product_history_sync_service.py
+12 -3
View File
@@ -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,
docs/21-投顾Agent迁移TODO.md
+8
View File
@@ -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` 被拒绝）
hq.py
+79 -1
View File
@@ -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]]:
tests/unit/infrastructure/test_fund_market_adapter.py
+29
View File
@@ -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:
tests/unit/service/test_product_history_sync_service.py
+27 -1
View File
@@ -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
tools/sync_advisor_market_data.py
+172
View File
@@ -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()))