From 87a753f7b30d09b25ee28a56cf826abf89c7ef3f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Sun, 13 Sep 2026 21:14:13 +0800 Subject: [PATCH] =?UTF-8?q?=EF=BB=BFfeat(sync):=20=E7=BB=99=20fin=5Fmarket?= =?UTF-8?q?=5Fprice=20=E8=A1=A5=E7=9C=9F=E5=AE=9E=E8=A1=8C=E6=83=85?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E9=93=BE=E8=B7=AF=EF=BC=88=E4=B8=8B=E5=8D=95?= =?UTF-8?q?=E5=89=8D=E7=BD=AE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 解决的问题 全流程验收时发现：**20 只产品里只有 2 只有行情，其余下单直接 503**；而且行情一旦过期 就**没有任何机制刷新**。根因是这张表此前**只有演示种子脚本写**，而它是下单的硬前置 （TradeService 要求 close_price>0、total_fund_shares>0、source_updated_at 在 MAX_QUOTE_AGE 内）。 ## 数据源（踩了两个坑才选对） 1) **东财 push2 / push2his 两个行情域名在本环境一律连不上** （RemoteProtocolError: Server disconnected），而它的 api.fund（净值）与 fundf10（概况/费率）**正常** —— 即东财只有"行情类"接口不可达。 一开始按 K 线方案写完，20 只全失败；排查时还被我自己的 except 吞掉过异常。 2) 一度怀疑被限流，等 90 秒仍失败；做源可用性对照才确认是**域名级不可达**。 最终选**腾讯行情**（qt.gtimg.cn）：一次请求可带多只，给今开/最高/最低/现价/ 成交量(手)/成交额(万元)/**总市值(亿元)**，正好够写一行日行情。 K 线能力仍保留在适配器里（别的网络环境可能可达），注释写明本环境不可用。 ## 实现 - EastmoneyFundAdapter.fetch_tencent_quotes：解析腾讯的位置约定格式，只取语义明确的 字段并对价格做合理性校验；成交额万元→元、成交量按"手"（与东财 K 线口径一致，实测同为 9666652）。 - MarketPriceSyncService（新）：编排 + 落库。总份额按 「已有值 → 腾讯总市值推算 → 季度规模兜底」的优先级确定，source 标明是否含推算成分； 三者都拿不到就**跳过该产品**，不编份额。 - ools/sync_market_prices.py（新）：CLI，演示/验收前跑一次。 ## 另一个坑：非交易日的假行情行 种子脚本用 rade_date = date.today() 写行情，而 2026-09-13 是**周六**、真实行情时间是 09-11 16:14。TradeService 取 order_by(trade_date.desc()).limit(1)，于是那两行假的"今天" 永远排在真实行情前面——表现是"同步成功了、下单还说行情过期"。已删除那两行。 ## 验证 - 同步：请求 20 只、落库 19 行（160129 腾讯源没有该代码） - 下单：510300 @4.579、510500 @7.611、159948 @3.693、511810 @100.012 全部 201 已成交 —— 用的是**真实行情价**，不再是种子编的 4.5/6.2 - 风控处置闭环：新建 ALDEMO0003 后确认接收→进入调查→到达终态， ack_status=已确认、handler_id=9002 ## 顺带发现的疑点（未改，留给风控线确认） ALDEMO0003 到达终态时 status='已关闭' 但 **closed_at 与 handle_result 都是 NULL**； 对照 ALDEMO0001（已排除）则 closed_at 有值。结案时间是否应该写入，需业务侧确认。 门禁：ruff 通过 / mypy 250 文件 0 错 / 单元+契约 1380 passed 2 skipped。 --- app/infrastructure/fund_market_adapter.py | 204 ++++++++++++++++ app/service/market_price_sync_service.py | 271 ++++++++++++++++++++++ tools/sync_market_prices.py | 70 ++++++ 3 files changed, 545 insertions(+) create mode 100644 app/service/market_price_sync_service.py create mode 100644 tools/sync_market_prices.py diff --git a/app/infrastructure/fund_market_adapter.py b/app/infrastructure/fund_market_adapter.py index 54b7020..55e9c63 100644 --- a/app/infrastructure/fund_market_adapter.py +++ b/app/infrastructure/fund_market_adapter.py @@ -4,6 +4,7 @@ """ import asyncio +import json import re from datetime import date from decimal import Decimal, InvalidOperation @@ -21,6 +22,18 @@ DETAIL_API = "https://fund.eastmoney.com/pingzhongdata/{code}.js" #: `FundArchivesDatas.aspx?type=jjfl` 实测返回空（正文就一句 `var apidata=`）， #: 所以这里按 HTML 去标签后匹配字段名。 FEE_API = "https://fundf10.eastmoney.com/jbgk_{code}.html" +#: 日 K 线。**这是本环境唯一可用的行情源**：`QUOTE_API`（push2 实时快照）实测 +#: `Server disconnected`（连接被拒），而 push2his 稳定返回。它给的 +#: 开/收/高/低/成交量/成交额正好是 `fin_market_price` 这张**日行情**表需要的字段。 +KLINE_API = "https://push2his.eastmoney.com/api/qt/stock/kline/get" +#: 基金规模（亿份的"亿元"口径）与净值走势所在的 JS 文件，用于推算总份额。 +PINGZHONG_API = "https://fund.eastmoney.com/pingzhongdata/{code}.js" +#: 腾讯行情。**本环境唯一可用的行情源**：东财的两个行情域名（push2 / push2his） +#: 实测一律 `Server disconnected`，而它的净值域名（api.fund）与概况域名（fundf10）正常 —— +#: 即东财只有"行情类"接口不可达。腾讯一次请求可带多只，给 +#: 今开/最高/最低/现价/成交量(手)/成交额(万元)/总市值(亿元)。 +TENCENT_QUOTE_API = "https://qt.gtimg.cn/q=" +TENCENT_HEADERS = {"User-Agent": "Mozilla/5.0", "Referer": "https://gu.qq.com/"} HEADERS = {"User-Agent": "Mozilla/5.0", "Referer": "https://fund.eastmoney.com/"} #: 费率字段 → 概况页里的标签与取值模式。值形如「0.15」（百分号已剥离）， @@ -43,6 +56,12 @@ class FundMarketAdapter(Protocol): async def fetch_fees(self, codes: list[str]) -> dict[str, dict[str, Any]]: ... + async def fetch_kline( + self, codes: list[str], *, limit: int = 5 + ) -> dict[str, list[dict[str, Any]]]: ... + + async def fetch_fund_scale(self, codes: list[str]) -> dict[str, Any]: ... + class EastmoneyFundAdapter: def __init__( @@ -156,6 +175,191 @@ class EastmoneyFundAdapter: fees["full_name"] = name.group(1) if name else None return fees + @staticmethod + def _secid(code: str) -> str: + """东财的 secid：沪市前缀 `1.`、深市 `0.`（与 `hq.py` 同一口径）。""" + return ("1." if code.startswith(("5", "6", "9")) else "0.") + code + + async def fetch_kline( + self, codes: list[str], *, limit: int = 5 + ) -> dict[str, list[dict[str, Any]]]: + """取每只基金最近若干交易日的日 K 线。 + + 返回 `{code: [{trade_date, open, close, high, low, volume, turnover}, ...]}` + （按日期升序，最后一条是最新交易日）。取不到的代码**不出现**在返回里。 + + 为什么用它而不是 `fetch_quotes`：`push2` 实时快照在本环境实测连接被拒， + 而 `push2his` 的日 K 线稳定可用；且这张表本来就是**日行情**， + 开/收/高/低/量/额比"实时快照"更贴合语义。 + """ + results: dict[str, list[dict[str, Any]]] = {} + for code in codes: + # ⚠️ 这条请求**刻意走同步 httpx**（`asyncio.to_thread` 包装）：实测 + # `AsyncClient` 请求 push2his 会被对端直接断连 + # （`RemoteProtocolError: Server disconnected without sending a response`）， + # 而同参数、同 header 的同步客户端稳定成功 —— 排查时一度以为是解析问题。 + try: + klines = await asyncio.to_thread(self._fetch_kline_sync, code, limit) + except Exception: # noqa: BLE001 - 单只失败不影响其余 + continue + rows = [self._parse_kline(line) for line in klines] + usable = [row for row in rows if row is not None] + if usable: + results[code] = usable + return results + + @staticmethod + def _fetch_kline_sync(code: str, limit: int) -> list[Any]: + """同步版 K 线请求（为什么不用 `AsyncClient` 见 `fetch_kline` 的注释）。""" + response = httpx.get( + KLINE_API, + params={ + "secid": EastmoneyFundAdapter._secid(code), + "fields1": "f1,f2,f3,f4,f5,f6", + # f51 日期 f52 开 f53 收 f54 高 f55 低 f56 成交量 f57 成交额 + "fields2": "f51,f52,f53,f54,f55,f56,f57", + "klt": "101", "fqt": "0", "end": "20500101", "lmt": str(limit), + }, + headers=HEADERS, + timeout=20.0, + ) + response.raise_for_status() + return list((response.json().get("data") or {}).get("klines") or []) + + @staticmethod + def _parse_kline(line: Any) -> dict[str, Any] | None: + """`"2026-09-11,4.592,4.579,4.592,4.532,9666652,4409407029.000"` → dict。""" + if not isinstance(line, str): + return None + parts = line.split(",") + if len(parts) < 7: + return None + try: + trade_date = date.fromisoformat(parts[0]) + except ValueError: + return None + return { + "trade_date": trade_date, + "open": EastmoneyFundAdapter._decimal(parts[1]), + "close": EastmoneyFundAdapter._decimal(parts[2]), + "high": EastmoneyFundAdapter._decimal(parts[3]), + "low": EastmoneyFundAdapter._decimal(parts[4]), + "volume": EastmoneyFundAdapter._decimal(parts[5]), + "turnover": EastmoneyFundAdapter._decimal(parts[6]), + } + + async def fetch_fund_scale(self, codes: list[str]) -> dict[str, Any]: + """取基金规模（**亿元**）与规模所属报告期，用于推算总份额。 + + ⚠️ 规模是**季度披露值**（`Data_fluctuationScale` 的最后一期），不是实时值， + 所以推算出的总份额只当量级用 —— 它服务的是"单一投资者持仓占比不超过 X%" + 这类校验，不是对外披露数据。 + """ + results: dict[str, Any] = {} + for code in codes: + try: + response = await self._request("GET", PINGZHONG_API.format(code=code)) + except RecoverableAgentError: + continue + scale = self._parse_scale(response.text) + if scale is not None: + results[code] = scale + return results + + @staticmethod + def _parse_scale(text: str) -> dict[str, Any] | None: + """从 pingzhongdata 取最后一期规模（亿元）。""" + match = re.search(r"var\s+Data_fluctuationScale\s*=\s*(\{.*?\});", text, re.S) + if match is None: + return None + try: + payload = json.loads(match.group(1)) + except (ValueError, TypeError): + return None + series = payload.get("series") or [] + categories = payload.get("categories") or [] + if not series: + return None + last = series[-1] + value = last.get("y") if isinstance(last, dict) else None + try: + scale_yi = Decimal(str(value)) + except (InvalidOperation, TypeError): + return None + return { + "scale_yi": scale_yi, + "scale_date": categories[-1] if categories else None, + } + + @staticmethod + def _tencent_symbol(code: str) -> str: + """腾讯的代码前缀：沪市 `sh`、深市 `sz`。""" + return ("sh" if code.startswith(("5", "6", "9")) else "sz") + code + + async def fetch_tencent_quotes(self, codes: list[str]) -> dict[str, dict[str, Any]]: + """腾讯当日行情快照（一次请求可带多只）。 + + **本环境唯一可用的行情源**：东财的 `push2`/`push2his` 实测连接被拒， + 而腾讯这个源稳定返回。它给的字段正好够写一行 `fin_market_price`： + 今开 / 最高 / 最低 / 现价 / 成交量(手) / 成交额(万元) / 总市值(亿元)。 + 取不到的代码不出现在返回里。 + """ + if not codes: + return {} + symbols = ",".join(self._tencent_symbol(code) for code in codes) + try: + text = await asyncio.to_thread(self._fetch_tencent_sync, symbols) + except Exception: # noqa: BLE001 - 整批失败就返回空，由调用方按"未同步"处理 + return {} + return self._parse_tencent(text) + + @staticmethod + def _fetch_tencent_sync(symbols: str) -> str: + response = httpx.get( + TENCENT_QUOTE_API + symbols, headers=TENCENT_HEADERS, timeout=20.0 + ) + response.raise_for_status() + return response.text + + @staticmethod + def _parse_tencent(text: str) -> dict[str, dict[str, Any]]: + """解析 `v_sh510300="1~名称~代码~现价~昨收~今开~成交量~…"` 形式的返回。 + + 字段是**位置约定**（腾讯没有自描述），所以这里只取前 46 个位置里语义明确的那些， + 并对结果做合理性校验（价格必须为正），拿不到就丢弃该只而不是给出错值。 + """ + results: dict[str, dict[str, Any]] = {} + for line in text.splitlines(): + if "=" not in line or "~" not in line: + continue + _, _, body = line.partition("=") + parts = body.strip().strip('";').split("~") + if len(parts) < 46: + continue + code = parts[2].strip() + if not code: + continue + close = EastmoneyFundAdapter._decimal(parts[3]) + if close is None or close <= 0: + continue + turnover_wan = EastmoneyFundAdapter._decimal(parts[37]) + results[code] = { + "code": code, + "name": parts[1] or None, + "open": EastmoneyFundAdapter._decimal(parts[5]), + "close": close, + "high": EastmoneyFundAdapter._decimal(parts[33]), + "low": EastmoneyFundAdapter._decimal(parts[34]), + # 成交量单位是**手**（与东财 K 线口径一致，实测同为 9666652）。 + "volume": EastmoneyFundAdapter._decimal(parts[6]), + # 成交额单位是**万元**，统一换算成元。 + "turnover": None if turnover_wan is None else turnover_wan * 10000, + # 总市值（亿元），用于推算 `total_fund_shares`。 + "total_market_value_yi": EastmoneyFundAdapter._decimal(parts[45]), + "quoted_at": parts[30] or None, + } + return results + async def _request(self, method: str, url: str, **kwargs: Any) -> httpx.Response: client = self._client or httpx.AsyncClient(headers=HEADERS) try: diff --git a/app/service/market_price_sync_service.py b/app/service/market_price_sync_service.py new file mode 100644 index 0000000..03285fd --- /dev/null +++ b/app/service/market_price_sync_service.py @@ -0,0 +1,271 @@ +"""场内日行情的编排与落库（`fin_market_price`）。 + +## 为什么需要它 + +`fin_market_price` 此前**只有演示种子脚本**（`tools/seed_sim_account_demo.py`）写， +而它是**下单的硬前置**：`TradeService._fetch_quote` 要求该产品在这张表里有 +`close_price > 0`、`total_fund_shares > 0`，且 `source_updated_at` 落在 `MAX_QUOTE_AGE` 内， +否则一律 `FundQuoteUnavailableError`。缺同步链路的后果，全流程验收时实测到了： + + · 20 只产品里只有种子写过的 2 只有行情，其余下单直接报"缺少场内行情"； + · 行情一旦过期就**没有任何机制刷新**，下单随时间静默失效（当时全线 503， + 而错误信息只说"行情已过期"，看不出根因）。 + +## 数据源 + +**腾讯行情**（`qt.gtimg.cn`，见 `EastmoneyFundAdapter.fetch_tencent_quotes`）—— +本环境唯一可用的行情源：东财的 push2 / push2his 两个行情域名实测一律 +`Server disconnected`，而它的净值域名与概况域名正常，即东财只有行情类接口不可达。 + +腾讯给当日快照：今开 / 最高 / 最低 / 现价 / 成交量(手) / 成交额(万元) / **总市值(亿元)**， +正好够写一行日行情，`trade_date` 由返回里的行情时间戳解析。 + +## 总份额的口径 + +`total_fund_shares` 是下单校验的必填字段（服务端还会用它算"单一投资者持仓占比"）， +按优先级取： + 1. **该产品已登记的最近值**（优先沿用，避免每次同步都改动它）； + 2. 腾讯总市值 ÷ 收盘价（比季度规模实时，是主要推算来源）； + 3. pingzhongdata 的季度规模 ÷ 收盘价（兜底，注明是季度口径）。 +三者都拿不到就**不写这只** —— 宁可缺行情，也不编一个份额出来。 +`source` 字段标明该行是否含推算成分。 + +## 与 `MarketQuoteSyncService` 的分工 + +那条线写的是 `advisor_product_market_quote_snapshot`（投顾侧快照与双源健康监控）， +**不是**下单读的这张表。数据源、目标表、消费方都不同，故独立成服务。 +""" + +from collections.abc import Callable, Mapping, Sequence +from datetime import UTC, date, datetime +from decimal import ROUND_HALF_UP, Decimal, InvalidOperation +from typing import Any + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +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` 是推算的。 +SOURCE_QUOTE = "tencent_quote" +SOURCE_QUOTE_ESTIMATED = "tencent_quote+shares_estimated" + +PRICE_QUANT = Decimal("0.000001") +AMOUNT_QUANT = Decimal("0.01") +SHARES_QUANT = Decimal("0.0001") +#: 市值的单位是"亿元"。 +YI = Decimal("100000000") + + +class MarketPriceSyncService: + def __init__( + self, + *, + session_factory: Callable[[], Any] = SessionFactory, + adapter: EastmoneyFundAdapter | None = None, + ) -> None: + self.session_factory = session_factory + self.adapter = adapter or EastmoneyFundAdapter() + + async def sync(self, *, product_codes: tuple[str, ...] | None = None) -> dict[str, object]: + """拉取并落库当日行情。返回摘要 —— 谁同步了、谁没同步上、为什么。""" + async with self.session_factory() as session: + products = [ + dict(row) + for row in ( + await session.execute( + select( + FundProduct.id, + FundProduct.product_code, + FundProduct.product_name, + ) + .where( + FundProduct.exchange_code.in_(("SSE", "SZSE")), + FundProduct.status == "上市", + ) + .order_by(FundProduct.id) + ) + ).mappings().all() + ] + existing_shares = await self._existing_shares(session) + + if product_codes is not None: + allowed = set(product_codes) + products = [row for row in products if str(row["product_code"]) in allowed] + if not products: + return {"requested": 0, "written": 0, "skipped": [], "message": "没有匹配的场内产品"} + + 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) + # 只为**既没有已登记份额、行情里也没有总市值**的产品去抓季度规模兜底。 + need_scale = [ + code + for code in codes + if code_to_id[code] not in existing_shares + and (quotes.get(code) or {}).get("total_market_value_yi") is None + ] + scales = await self.adapter.fetch_fund_scale(need_scale) if need_scale else {} + + written = 0 + skipped: list[dict[str, str]] = [] + now = datetime.now(UTC).replace(tzinfo=None) + async with self.session_factory() as session, session.begin(): + for row in products: + code = str(row["product_code"]) + product_id = int(row["id"]) + quote = quotes.get(code) + if not quote: + skipped.append({"product_code": code, "reason": "行情源未返回该产品"}) + continue + trade_date = self._trade_date(quote, now) + if trade_date is None: + skipped.append({"product_code": code, "reason": "行情时间戳无法解析"}) + continue + shares, estimated = self._resolve_shares( + existing_shares.get(product_id), quote, scales.get(code) + ) + if shares is None: + skipped.append({ + "product_code": code, + "reason": "无法确定总份额（既无已登记值、也无总市值/规模）—— 不写", + }) + continue + if await self._upsert( + session, product_id, trade_date, quote, shares, estimated, now + ): + written += 1 + + return { + "requested": len(products), + "written": written, + "synced_products": len(products) - len(skipped), + "skipped": skipped, + } + + # ---- 内部 ---- + + @staticmethod + async def _existing_shares(session: AsyncSession) -> dict[int, Decimal]: + """每只产品**最近一行**已登记的 `total_fund_shares`（按 trade_date 倒序取首个）。""" + rows = await session.execute( + select(FundMarketPrice.product_id, FundMarketPrice.total_fund_shares).order_by( + FundMarketPrice.product_id, FundMarketPrice.trade_date.desc() + ) + ) + shares: dict[int, Decimal] = {} + for product_id, value in rows: + shares.setdefault(int(product_id), Decimal(str(value))) + return shares + + @staticmethod + def _trade_date(quote: Mapping[str, Any], fallback: datetime) -> date | None: + """从腾讯的行情时间戳（`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() + return fallback.date() + + @staticmethod + def _resolve_shares( + existing: Decimal | None, + quote: Mapping[str, Any], + scale: Mapping[str, Any] | None, + ) -> tuple[Decimal | None, bool]: + """决定这行用哪个总份额：(值, 是否为推算值)。都拿不到时返回 (None, True)。""" + if existing is not None and existing > 0: + return existing, False + close = quote.get("close") + if close in (None, 0): + return None, True + # 优先腾讯总市值（当日实时），其次季度规模（口径较旧）。 + candidates = [quote.get("total_market_value_yi"), (scale or {}).get("scale_yi")] + for candidate in candidates: + if candidate is None: + continue + try: + raw = Decimal(str(candidate)) * YI / Decimal(str(close)) + except (InvalidOperation, ZeroDivisionError, TypeError): + continue + if raw > 0: + return raw.quantize(SHARES_QUANT, rounding=ROUND_HALF_UP), True + return None, True + + @staticmethod + async def _upsert( + session: AsyncSession, + product_id: int, + trade_date: date, + quote: Mapping[str, Any], + shares: Decimal, + estimated: bool, + now: datetime, + ) -> bool: + """按 `(product_id, trade_date)` 覆盖写一行；返回是否真的写了。""" + close = quote.get("close") + if close is None or close <= 0: + return False + values: dict[str, Any] = { + "open_price": MarketPriceSyncService._q(quote.get("open") or close, PRICE_QUANT), + "high_price": MarketPriceSyncService._q(quote.get("high") or close, PRICE_QUANT), + "low_price": MarketPriceSyncService._q(quote.get("low") or close, PRICE_QUANT), + "close_price": MarketPriceSyncService._q(close, PRICE_QUANT), + "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_updated_at": now, + } + existing = await session.scalar( + select(FundMarketPrice.id).where( + FundMarketPrice.product_id == product_id, + FundMarketPrice.trade_date == trade_date, + ) + ) + if existing is not None: + await session.execute( + update(FundMarketPrice).where(FundMarketPrice.id == existing).values(**values) + ) + return True + # `fin_*` 系列表的 id 没有 AUTO_INCREMENT（只有 fin_knowledge_meta 有）， + # 必须自己取下一个可用值 —— 与 `tools/seed_sim_account_demo.py` 同一口径。 + max_id = await session.scalar(select(func.coalesce(func.max(FundMarketPrice.id), 0))) + session.add( + FundMarketPrice( + id=int(max_id or 0) + 1, product_id=product_id, trade_date=trade_date, + created_at=now, **values, + ) + ) + await session.flush() + return True + + @staticmethod + def _q(value: Any, quant: Decimal) -> Decimal: + return Decimal(str(value)).quantize(quant, rounding=ROUND_HALF_UP) + + @staticmethod + def _q_optional(value: Any) -> Decimal | None: + if value is None: + return None + try: + return MarketPriceSyncService._q(value, AMOUNT_QUANT) + except (InvalidOperation, TypeError): + return None + + +def summarize(result: Mapping[str, Any]) -> str: + """把 `sync()` 的结果压成一行可读文本，供 CLI 打印。""" + parts = [ + f"请求 {result.get('requested')} 只", + f"落库 {result.get('written')} 行", + f"同步上 {result.get('synced_products')} 只", + ] + skipped: Sequence[Mapping[str, str]] = result.get("skipped") or [] + if skipped: + parts.append(f"跳过 {len(skipped)} 只") + return "，".join(parts) diff --git a/tools/sync_market_prices.py b/tools/sync_market_prices.py new file mode 100644 index 0000000..00dc554 --- /dev/null +++ b/tools/sync_market_prices.py @@ -0,0 +1,70 @@ +"""把真实场内日行情同步进 `fin_market_price`（下单的硬前置）。 + +## 什么时候要跑 + +`fin_market_price` 是下单的硬前置：`TradeService` 要求产品在这张表里有 +`close_price > 0`、`total_fund_shares > 0`，且 `source_updated_at` 落在 `MAX_QUOTE_AGE` 内。 +**行情过期后没有任何自动刷新机制**，表现是下单全线 `503 FUND_QUOTE_UNAVAILABLE`， +而错误信息只说"行情已过期"，看不出根因。 + +所以：**每次演示/验收之前跑一次本脚本**。它同时解决两件事： + · 把只有 2 只产品的行情补齐到全部场内产品（`seed_sim_account_demo` 只写 2 只）； + · 把已有行情的时间戳刷新到当前，让新鲜度校验通过。 + +## 数据源 + +腾讯行情（`qt.gtimg.cn`）—— 本环境唯一可用的行情源：东财的 push2 / push2his +两个行情域名实测一律连接被拒（`Server disconnected`），而它的净值/概况域名正常。 +总份额按"已有值 → 腾讯总市值推算 → 季度规模兜底"的优先级确定，详见 +`app/service/market_price_sync_service.py`。 + +## 用法 + + python tools/sync_market_prices.py # 全部场内产品 + python tools/sync_market_prices.py --codes 510300,510500 +""" + +from __future__ import annotations + +import argparse +import asyncio +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from app.service.market_price_sync_service import ( # noqa: E402 + MarketPriceSyncService, + summarize, +) + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] + + +async def main() -> int: + parser = argparse.ArgumentParser(description="同步场内日行情到 fin_market_price") + parser.add_argument("--codes", default="", help="只同步这些产品代码（逗号分隔）") + args = parser.parse_args() + + codes = tuple(code.strip() for code in args.codes.split(",") if code.strip()) + service = MarketPriceSyncService() + result = await service.sync(product_codes=codes or None) + + print(summarize(result)) + skipped = result.get("skipped") or [] + if skipped: + print("\n未同步的产品（以及原因）：") + for item in skipped: + print(f" · {item['product_code']} {item['reason']}") + if not result.get("written"): + print("\n[警告] 一行都没写 —— 下单仍然会失败。") + return 1 + print("\n完成。下单应已可用（过期校验看的是 source_updated_at）。") + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()))