diff --git a/app/service/market_price_sync_service.py b/app/service/market_price_sync_service.py index 03285fd..b6d49c4 100644 --- a/app/service/market_price_sync_service.py +++ b/app/service/market_price_sync_service.py @@ -48,9 +48,11 @@ from app.infrastructure.db import SessionFactory from app.infrastructure.fund_market_adapter import EastmoneyFundAdapter from app.model.fund import FundMarketPrice, FundProduct -#: 来源口径。带 `+shares_estimated` 表示 `total_fund_shares` 是推算的。 +#: 来源口径。带 `+shares_estimated` 表示 `total_fund_shares` 是推算的; +#: `eastmoney_nav_fallback` 表示腾讯行情没有这只是,改用东财净值构造(见 `_nav_fallback`)。 SOURCE_QUOTE = "tencent_quote" SOURCE_QUOTE_ESTIMATED = "tencent_quote+shares_estimated" +SOURCE_NAV_FALLBACK = "eastmoney_nav_fallback" PRICE_QUANT = Decimal("0.000001") AMOUNT_QUANT = Decimal("0.01") @@ -100,6 +102,14 @@ class MarketPriceSyncService: codes = [str(row["product_code"]) for row in products] code_to_id = {str(row["product_code"]): int(row["id"]) for row in products} quotes = await self.adapter.fetch_tencent_quotes(codes) + # 行情源没有的产品,**在事务外**用净值源补上(在事务里做网络请求会拉长事务)。 + fallbacks: dict[str, dict[str, Any]] = {} + for code in codes: + if quotes.get(code): + continue + fallback = await self._nav_fallback(code) + if fallback is not None: + fallbacks[code] = fallback # 只为**既没有已登记份额、行情里也没有总市值**的产品去抓季度规模兜底。 need_scale = [ code @@ -116,9 +126,10 @@ class MarketPriceSyncService: for row in products: code = str(row["product_code"]) product_id = int(row["id"]) - quote = quotes.get(code) + quote = quotes.get(code) or fallbacks.get(code) + used_fallback = code in fallbacks if not quote: - skipped.append({"product_code": code, "reason": "行情源未返回该产品"}) + skipped.append({"product_code": code, "reason": "行情源与净值源都没有它"}) continue trade_date = self._trade_date(quote, now) if trade_date is None: @@ -134,7 +145,8 @@ class MarketPriceSyncService: }) continue if await self._upsert( - session, product_id, trade_date, quote, shares, estimated, now + session, product_id, trade_date, quote, shares, + self._source_of(used_fallback, estimated), now, ): written += 1 @@ -162,15 +174,54 @@ class MarketPriceSyncService: @staticmethod def _trade_date(quote: Mapping[str, Any], fallback: datetime) -> date | None: - """从腾讯的行情时间戳(`YYYYMMDDHHMMSS`)取交易日;解析不出就用当天。""" + """取交易日:优先腾讯行情时间戳(`YYYYMMDDHHMMSS`),其次净值日期,最后当天。""" raw = str(quote.get("quoted_at") or "") if len(raw) >= 8 and raw[:8].isdigit(): try: return date(int(raw[:4]), int(raw[4:6]), int(raw[6:8])) except ValueError: return fallback.date() + nav_date = quote.get("nav_date") + if isinstance(nav_date, date): + return nav_date return fallback.date() + @staticmethod + def _source_of(used_fallback: bool, estimated: bool) -> str: + """这一行的来源口径:净值降级 > 含估算份额 > 纯行情。""" + if used_fallback: + return SOURCE_NAV_FALLBACK + return SOURCE_QUOTE_ESTIMATED if estimated else SOURCE_QUOTE + + async def _nav_fallback(self, code: str) -> dict[str, Any] | None: + """行情源没有这只是时,用东财历史净值构造一条**降级**行情。 + + 为什么可以接受:`source` 会写成 `eastmoney_nav_fallback` 明确标注来源, + `volume`/`turnover` 留空(净值接口不提供量额),不编造任何数字; + 开高低用净值同值(该接口只给一个价格)。 + + ⚠️ **净值不等于市价**:若该产品确实在交易所交易,用它当收盘价会有折溢价偏差。 + 所以这条路径只应落在"行情源没有覆盖"的产品上,且值得业务侧复核 —— + 实测触发它的 `160129` 是 C 类份额(其 A 类 `160128` 在腾讯源有行情), + 它很可能未上市交易,那它本就不该出现在场内产品列表里。 + """ + history = await self.adapter.fetch_history(code, date.today()) + nav = history.get("nav") + if nav is None or history.get("degraded"): + return None + return { + "code": code, + "open": nav, + "close": nav, + "high": nav, + "low": nav, + "volume": None, + "turnover": None, + "total_market_value_yi": None, + "quoted_at": None, + "nav_date": history.get("nav_date"), + } + @staticmethod def _resolve_shares( existing: Decimal | None, @@ -203,7 +254,7 @@ class MarketPriceSyncService: trade_date: date, quote: Mapping[str, Any], shares: Decimal, - estimated: bool, + source: str, now: datetime, ) -> bool: """按 `(product_id, trade_date)` 覆盖写一行;返回是否真的写了。""" @@ -218,7 +269,7 @@ class MarketPriceSyncService: "volume": MarketPriceSyncService._q_optional(quote.get("volume")), "turnover_amount": MarketPriceSyncService._q_optional(quote.get("turnover")), "total_fund_shares": shares, - "source": SOURCE_QUOTE_ESTIMATED if estimated else SOURCE_QUOTE, + "source": source, "source_updated_at": now, } existing = await session.scalar(