diff --git a/backend/app/services/ext_fanout.py b/backend/app/services/ext_fanout.py index 7d86290b..baeda0bc 100644 --- a/backend/app/services/ext_fanout.py +++ b/backend/app/services/ext_fanout.py @@ -75,6 +75,13 @@ def _raise_on_envelope_error(data: Any) -> None: async def fetch_fanout_rows(config: ExtConfig, target_date=None) -> list[dict]: + from app.services.ext_pull import _pull_client_scope + + async with _pull_client_scope(): + return await _fetch_fanout_rows(config, target_date) + + +async def _fetch_fanout_rows(config: ExtConfig, target_date=None) -> list[dict]: """执行两级拉取并反转为「成员 → 类别清单」行。""" from app.services.ext_pull import ( _append_query_params, diff --git a/backend/app/services/ext_pull.py b/backend/app/services/ext_pull.py index a3f78373..c55f523b 100644 --- a/backend/app/services/ext_pull.py +++ b/backend/app/services/ext_pull.py @@ -5,6 +5,8 @@ import asyncio import json import logging import threading +from contextlib import asynccontextmanager +from contextvars import ContextVar from datetime import UTC, date, datetime, timezone from functools import reduce from pathlib import Path @@ -22,6 +24,22 @@ from app.services.ext_data import ( logger = logging.getLogger(__name__) +_pull_client: ContextVar[httpx.AsyncClient | None] = ContextVar("ext_pull_client", default=None) + + +@asynccontextmanager +async def _pull_client_scope(): + client = _pull_client.get() + if client is not None: + yield client + return + async with httpx.AsyncClient() as client: + token = _pull_client.set(client) + try: + yield client + finally: + _pull_client.reset(token) + def outbound_headers(user_headers: dict[str, str] | None = None) -> dict[str, str]: """扩展数据出站请求的默认标识头。 @@ -230,10 +248,10 @@ async def _request_json( # getattr 兜底测试用的简化 pull 对象 timeout = getattr(pull, "timeout_seconds", None) timeout = timeout if isinstance(timeout, (int, float)) and timeout > 0 else 30 - async with httpx.AsyncClient(timeout=timeout) as client: + async with _pull_client_scope() as client: headers = outbound_headers(pull.headers) url = _apply_auth(config_id, pull.auth, url, headers) - kwargs: dict[str, Any] = {"headers": headers} + kwargs: dict[str, Any] = {"headers": headers, "timeout": timeout} if pull.method.upper() == "POST" and pull.body: kwargs["content"] = pull.body @@ -270,6 +288,11 @@ def _parse_rows_payload(config: ExtConfig, pull: PullConfig, data: Any) -> list[ async def fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict]: + async with _pull_client_scope(): + return await _fetch_rows_for_date(config, target_date) + + +async def _fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict]: """按日期请求外部 API 并解析为行 (不写盘)。空数据返回 []。 与 fetch_and_ingest 共用同一解析链 (response_path/字段映射/关联字段校验), @@ -377,6 +400,16 @@ async def backfill_history( data_dir, start: date, end: date, +) -> dict: + async with _pull_client_scope(): + return await _backfill_history(config, data_dir, start, end) + + +async def _backfill_history( + config: ExtConfig, + data_dir, + start: date, + end: date, ) -> dict: """按本地交易日逐日回补 timeseries 历史分区 (幂等, 已有分区跳过)。 diff --git a/backend/scripts/benchmark_ext_pull_client.py b/backend/scripts/benchmark_ext_pull_client.py new file mode 100644 index 00000000..1924c3e7 --- /dev/null +++ b/backend/scripts/benchmark_ext_pull_client.py @@ -0,0 +1,74 @@ +"""Compare per-request HTTP clients with one client per pull using a mock server. + +Run: uv run --frozen python -m scripts.benchmark_ext_pull_client +No external network requests or user data are used. +""" +from __future__ import annotations + +import asyncio +import statistics +import time +from datetime import date + +import httpx + +from app.services import ext_pull +from app.services.ext_data import ExtConfig, PullConfig + +PAGES = 30 +RUNS = 5 + + +async def run(shared: bool) -> tuple[float, int, int]: + clients = 0 + requests = 0 + real_client = httpx.AsyncClient + + def respond(request: httpx.Request) -> httpx.Response: + nonlocal requests + requests += 1 + page = int(request.url.params["page"]) + return httpx.Response(200, json=[{"symbol": "000001.SZ"}] if page <= PAGES else []) + + def factory(*args, **kwargs): + nonlocal clients + clients += 1 + return real_client(*args, transport=httpx.MockTransport(respond), **kwargs) + + original = ext_pull.httpx.AsyncClient + ext_pull.httpx.AsyncClient = factory + config = ExtConfig( + id="benchmark", label="benchmark", mode="timeseries", fields=[], + pull=PullConfig(url="https://example.test/rank", page_param="page", max_pages=PAGES + 1), + ) + try: + started = time.perf_counter() + if shared: + rows = await ext_pull.fetch_rows_for_date(config, date(2026, 9, 1)) + else: + for page in range(1, PAGES + 2): + payload = await ext_pull._request_json(config.pull, config.id, extra_params={"page": str(page)}) + if not payload: + break + if page == 1: + rows = [] + rows.extend(payload) + elapsed = time.perf_counter() - started + assert len(rows) == PAGES and requests == PAGES + 1 + return elapsed, clients, requests + finally: + ext_pull.httpx.AsyncClient = original + + +async def main() -> None: + timings: dict[bool, list[float]] = {False: [], True: []} + for _ in range(RUNS): + for shared in (False, True): + elapsed, clients, requests = await run(shared) + timings[shared].append(elapsed) + print(f"{'shared' if shared else 'per-request'}: {elapsed:.4f}s, {requests} requests, {clients} clients") + print(f"median: per-request={statistics.median(timings[False]):.4f}s, shared={statistics.median(timings[True]):.4f}s") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/backend/tests/test_ext_pull_pagination.py b/backend/tests/test_ext_pull_pagination.py index 375549f0..091707c6 100644 --- a/backend/tests/test_ext_pull_pagination.py +++ b/backend/tests/test_ext_pull_pagination.py @@ -3,8 +3,10 @@ max_pages 安全上限、非分页配置行为不变。mock _request_json, 不 """ from __future__ import annotations +import asyncio from datetime import date +import httpx import pytest from app.services import ext_pull @@ -38,6 +40,107 @@ def _install_pages(monkeypatch: pytest.MonkeyPatch, pages: dict[int, list[dict]] TARGET = date(2026, 9, 20) +async def test_pagination_reuses_client_and_closes_on_failure(monkeypatch: pytest.MonkeyPatch) -> None: + created = [] + + def handler(request: httpx.Request) -> httpx.Response: + page = request.url.params["page"] + if page == "2": + return httpx.Response(503) + return httpx.Response(200, json={"data": [{"symbol": "000001.SZ"}]}) + + real_client = httpx.AsyncClient + + def client_factory(*args, **kwargs): + client = real_client(*args, transport=httpx.MockTransport(handler), **kwargs) + created.append(client) + return client + + monkeypatch.setattr(ext_pull.httpx, "AsyncClient", client_factory) + cfg = _cfg(PullConfig(url="https://example.test/rank", response_path="data", page_param="page")) + with pytest.raises(httpx.HTTPStatusError): + await ext_pull.fetch_rows_for_date(cfg, TARGET) + assert len(created) == 1 + assert created[0].is_closed + + +async def test_fanout_reuses_client_for_concurrent_requests(monkeypatch: pytest.MonkeyPatch) -> None: + from app.services.ext_fanout import fetch_fanout_rows + + created = [] + real_client = httpx.AsyncClient + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/catalog": + return httpx.Response(200, json=[{"id": "one"}, {"id": "two"}]) + return httpx.Response(200, json={"items": [{"symbol": request.url.params["id"]}]}) + + def client_factory(*args, **kwargs): + client = real_client(*args, transport=httpx.MockTransport(handler), **kwargs) + created.append(client) + return client + + monkeypatch.setattr(ext_pull.httpx, "AsyncClient", client_factory) + fanout = { + "url": "https://example.test/members", "param": "id", "id_field": "id", + "items_path": "items", "member_id_field": "symbol", "symbol_col": "symbol", + "name_col": "name", "value_col": "groups", + } + cfg = _cfg(PullConfig(url="https://example.test/catalog", fanout=fanout)) + rows = await fetch_fanout_rows(cfg) + assert {row["symbol"] for row in rows} == {"one", "two"} + assert len(created) == 1 + assert created[0].is_closed + + +async def test_cancelled_pull_closes_client(monkeypatch: pytest.MonkeyPatch) -> None: + created = [] + real_client = httpx.AsyncClient + + async def handler(request: httpx.Request) -> httpx.Response: + raise asyncio.CancelledError + + def client_factory(*args, **kwargs): + client = real_client(*args, transport=httpx.MockTransport(handler), **kwargs) + created.append(client) + return client + + monkeypatch.setattr(ext_pull.httpx, "AsyncClient", client_factory) + with pytest.raises(asyncio.CancelledError): + await ext_pull.fetch_rows_for_date(_cfg(PullConfig(url="https://example.test/rank")), TARGET) + assert len(created) == 1 + assert created[0].is_closed + + +async def test_backfill_reuses_client_across_days(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + created = [] + real_client = httpx.AsyncClient + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=[{"symbol": "000001.SZ", "date": request.url.params["date"]}]) + + def client_factory(*args, **kwargs): + client = real_client(*args, transport=httpx.MockTransport(handler), **kwargs) + created.append(client) + return client + + monkeypatch.setattr(ext_pull.httpx, "AsyncClient", client_factory) + monkeypatch.setattr( + "app.services.dragon_tiger._local_trading_days", + lambda _: [date(2026, 9, 17), date(2026, 9, 18)], + ) + + async def no_sleep(_): + return None + + monkeypatch.setattr(ext_pull.asyncio, "sleep", no_sleep) + cfg = _cfg(PullConfig(url="https://example.test/rank", date_param="date")) + result = await ext_pull.backfill_history(cfg, tmp_path, date(2026, 9, 17), date(2026, 9, 18)) + assert result["fetched"] == 2 + assert len(created) == 1 + assert created[0].is_closed + + async def test_pagination_collects_pages_and_stops_on_short_page(monkeypatch: pytest.MonkeyPatch) -> None: # page_size=2: 第 2 页只有 1 行 (短页) → 停, 不再请求第 3 页 pull = PullConfig( diff --git a/backend/tests/test_ext_pull_timeout_config.py b/backend/tests/test_ext_pull_timeout_config.py index cc64423b..3cf39606 100644 --- a/backend/tests/test_ext_pull_timeout_config.py +++ b/backend/tests/test_ext_pull_timeout_config.py @@ -10,9 +10,8 @@ from types import SimpleNamespace import pytest -from app.services.ext_data import PullConfig from app.services import ext_pull - +from app.services.ext_data import PullConfig # ── 模型: 默认值 / 序列化 / 非法值归一 ────────────────────────────── @@ -47,9 +46,10 @@ class _FakeResponse: class _FakeClient: - """捕获 timeout 构造参数的 httpx.AsyncClient 替身。""" + """捕获每次请求的 timeout。""" last_timeout: object = None + request_timeout: object = None def __init__(self, timeout=None, **kwargs): _FakeClient.last_timeout = timeout @@ -61,6 +61,7 @@ class _FakeClient: return False async def request(self, method, url, **kwargs): + _FakeClient.request_timeout = kwargs.get("timeout") return _FakeResponse() @@ -69,7 +70,7 @@ async def test_request_json_uses_configured_timeout(monkeypatch): pull = PullConfig(url="https://x", timeout_seconds=90) await ext_pull._request_json(pull, "cfg1") - assert _FakeClient.last_timeout == 90 + assert _FakeClient.request_timeout == 90 async def test_request_json_defaults_to_30_without_attr(monkeypatch): @@ -81,7 +82,7 @@ async def test_request_json_defaults_to_30_without_attr(monkeypatch): method="GET", body=None, auth=None, ) await ext_pull._request_json(pull, "cfg1") - assert _FakeClient.last_timeout == 30 + assert _FakeClient.request_timeout == 30 # ── 探测路径: detect-url 请求体带 timeout_seconds ─────────────────── diff --git a/docs/backend-data-performance-plan.md b/docs/backend-data-performance-plan.md new file mode 100644 index 00000000..6f28db3c --- /dev/null +++ b/docs/backend-data-performance-plan.md @@ -0,0 +1,33 @@ +# 后端数据获取与处理性能计划 + +本计划只处理已排查到的四条链路。按相同环境、相同数据量对比修改前后耗时和内存;没有实测数据前不宣称整套回测或页面加速。每项单独验证、单独提交,保留数据口径、缓存失效和原子写语义。 + +## 1. 扩展数据拉取复用连接(本次实施) + +- 场景:扩展表分页、历史回补和两级拉取会连续请求同一数据源。 +- 问题:每次请求都新建 HTTP 客户端,重复建立连接。 +- 改动:一次拉取任务内共用客户端,请求仍按各配置单独设置超时;任务结束、失败或取消时关闭。保留现有限流、重试、鉴权和两级拉取并发上限。 +- 验证:模拟接口分别覆盖分页、按天回补、并发明细与失败关闭;对比请求数、客户端创建次数和耗时。真实网络收益需另行测量。 + +## 2. 扩展表按范围读取时减少无用数据(待测) + +- 场景:按日期范围查询扩展数据明细或字段取值。 +- 问题:先读取所有分区和列并合并,然后才过滤、选列、分页。 +- 计划:先测分区数、读取耗时、峰值内存;若读取占主要时间,再将必要的过滤与选列提前,并只把返回页转为 Python 对象。 +- 验证:跨分区字段缺失、日期列补全、排序、总数与分页结果须与现有实现一致。 + +## 3. 日 K 流式同步减少临时文件读写(待测) + +- 场景:按批获取历史日 K,分日暂存后再写正式分区。 +- 问题:临时文件写入后按日读回,再与正式分区合并写入。 +- 计划:先分别记录下载、暂存、读回和合并耗时与文件数;只有落盘占比明显时才优化批量提交。 +- 验证:部分批次失败不能破坏旧数据;原子写和并发更新语义不变。 + +## 4. 扩展表更新与历史计算减少全量重复工作(待测) + +- 场景:扩展表频繁更新快照/当日分区,策略读取多日扩展因子。 +- 问题:每次写入都读取并合并已有文件;缓存失效后历史因子逐分区加载完整表。 +- 计划:先按更新频率、分区大小测写放大和读取峰值;再评估只读必要日期/字段或减少重复合并。 +- 验证:同日去重、历史时点对齐、缓存失效和失败恢复不变。 + +现有 `market-data-performance-checklist.md` 是另一份未提交的工作文件,本计划不修改它。 \ No newline at end of file