From cd9e20610f9691bb05afe751611eb54a0a563e74 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Sun, 6 Sep 2026 22:35:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(ext-data):=20=E6=97=B6=E5=BA=8F=E8=A1=A8?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E5=9B=9E=E8=A1=A5=20=E2=80=94=20date=5Fparam?= =?UTF-8?q?=20=E6=8C=89=E6=97=A5=E6=8B=89=E5=8F=96=20+=20tickflow-hub=2040?= =?UTF-8?q?4=20=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 人气排行等 timeseries 扩展表此前只能从开启拉取之日起逐日累积。拉取配置 新增日期参数名 (date_param), 接口支持按日查询时即可历史回补: 按本地交易日 逐日请求 ?{date_param}=YYYY-MM-DD 写入对应分区, 幂等 (已有分区跳过), 单次上限 120 天, 429 限流退避重试 + 连续限流中止可续补。 适配 tickflow-hub PR#19 (/exports、/fuyao-rank 按日历史截面): - 该日无快照返回 404 → 计入 empty 跳过, 不进失败清单 (主循环与 429 退避重试两条路径均覆盖) - 防污染契约: 响应行 date 与请求日期不一致 (接口忽略日期参数) 时 fail-closed 拒写该分区, 避免当日值污染整个历史时序 - 修复调度器 next_run 计算的 UTC 未导入 NameError 前端拉取面板新增日期参数名配置与回补入口; 回补落盘的历史数据立即进入 信号/因子/回测按日对齐通道 (PIT)。 验证: pytest tests/test_ext_backfill.py 16 通过 (含 404/429/幂等/防污染), ext 全套 144 通过; pnpm build 通过。 --- backend/app/api/ext_data.py | 35 ++ backend/app/services/ext_data.py | 8 +- backend/app/services/ext_pull.py | 186 ++++++++++- backend/tests/test_ext_backfill.py | 313 ++++++++++++++++++ docs/features.md | 6 + .../components/ext-data/ExtDataPullPanel.tsx | 90 ++++- frontend/src/lib/api.ts | 20 ++ 7 files changed, 638 insertions(+), 20 deletions(-) create mode 100644 backend/tests/test_ext_backfill.py diff --git a/backend/app/api/ext_data.py b/backend/app/api/ext_data.py index 174389d..e0a2d56 100644 --- a/backend/app/api/ext_data.py +++ b/backend/app/api/ext_data.py @@ -82,6 +82,8 @@ class PullConfigReq(BaseModel): enabled: bool = False time_window_start: str | None = None # "HH:MM", None=不限 time_window_end: str | None = None # "HH:MM", None=不限 + # 接口按日查询的参数名 (如 "date"): 配置后支持历史回补, 且当日拉取也带日期参数 + date_param: str | None = Field(None, min_length=1, max_length=16, pattern=r"^[A-Za-z_][A-Za-z0-9_]*$") class DetectUrlReq(BaseModel): @@ -830,6 +832,7 @@ def configure_pull(request: Request, config_id: str, body: PullConfigReq): enabled=body.enabled, time_window_start=body.time_window_start, time_window_end=body.time_window_end, + date_param=body.date_param, last_run=old_pull.last_run if old_pull else None, last_status=old_pull.last_status if old_pull else None, last_message=old_pull.last_message if old_pull else None, @@ -924,6 +927,38 @@ async def run_pull(request: Request, config_id: str): raise HTTPException(400, f"拉取失败: {e}") from e +@router.post("/{config_id}/backfill") +async def backfill_history_ep( + request: Request, + config_id: str, + start: str = Query(..., description="开始日期 YYYY-MM-DD"), + end: str = Query(..., description="结束日期 YYYY-MM-DD (含)"), +): + """历史回补: 按本地交易日逐日拉取并写入 timeseries 分区。 + + 前提: 配置为 timeseries 模式且拉取配置了 date_param (接口支持按日期 + 查询)。幂等 —— 已存在的分区跳过, 失败单日不中断, 结果逐项返回。 + """ + store = _store(request) + config = store.get(config_id) + if not config: + raise HTTPException(404, f"配置 '{config_id}' 不存在") + try: + start_d = date.fromisoformat(start) + end_d = date.fromisoformat(end) + except ValueError as e: + raise HTTPException(422, f"日期格式错误 (应为 YYYY-MM-DD): {e}") from e + + from app.services.ext_pull import backfill_history + + try: + result = await backfill_history(config, _data_dir(request), start_d, end_d) + except ValueError as e: + raise HTTPException(400, str(e)) from e + _refresh_views(request) + return {"status": "ok", **result} + + # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Symbol 格式修复 diff --git a/backend/app/services/ext_data.py b/backend/app/services/ext_data.py index 34836af..0037d3f 100644 --- a/backend/app/services/ext_data.py +++ b/backend/app/services/ext_data.py @@ -40,7 +40,7 @@ class PullConfig: "url", "method", "headers", "body", "response_path", "field_map", "schedule_minutes", "enabled", "last_run", "last_status", "last_message", "last_rows", - "next_run", "time_window_start", "time_window_end", + "next_run", "time_window_start", "time_window_end", "date_param", ) def __init__( @@ -60,6 +60,7 @@ class PullConfig: next_run: str | None = None, time_window_start: str | None = None, time_window_end: str | None = None, + date_param: str | None = None, ) -> None: self.url = url self.method = method # GET | POST @@ -76,6 +77,9 @@ class PullConfig: self.next_run = next_run # 下次预计运行 (ISO, 调度器写入) self.time_window_start = time_window_start # 每日拉取窗口起始 "HH:MM", None=不限 self.time_window_end = time_window_end # 每日拉取窗口结束 "HH:MM", None=不限 + # 接口按日期查询的参数名 (如 "date"): 非 None 时请求 + # 带 ?{date_param}=YYYY-MM-DD, 支持历史回补; None = 接口只有当日快照 + self.date_param = date_param def to_dict(self) -> dict: return { @@ -94,6 +98,7 @@ class PullConfig: "next_run": self.next_run, "time_window_start": self.time_window_start, "time_window_end": self.time_window_end, + "date_param": self.date_param, } @classmethod @@ -116,6 +121,7 @@ class PullConfig: next_run=d.get("next_run"), time_window_start=d.get("time_window_start"), time_window_end=d.get("time_window_end"), + date_param=d.get("date_param"), ) diff --git a/backend/app/services/ext_pull.py b/backend/app/services/ext_pull.py index a57054d..16a7a8a 100644 --- a/backend/app/services/ext_pull.py +++ b/backend/app/services/ext_pull.py @@ -5,8 +5,9 @@ import asyncio import json import logging import threading -from datetime import date, datetime, timezone +from datetime import UTC, date, datetime, timezone from functools import reduce +from pathlib import Path from typing import Any import httpx @@ -102,19 +103,46 @@ def _apply_preset_flatten(config_id: str, rows: list[dict]) -> list[dict]: return flatten(rows) -async def fetch_and_ingest( - config: ExtConfig, - data_dir, -) -> tuple[int, str]: - """执行一次拉取: 请求外部 API → 解析响应 → 写入 Parquet。 +def _with_date_param(url: str, date_param: str | None, day: date) -> str: + """接口按日查询参数: ?{date_param}=YYYY-MM-DD (已有 query 用 &)。""" + if not date_param: + return url + sep = "&" if "?" in url else "?" + return f"{url}{sep}{date_param}={day.isoformat()}" - Returns: - (rows_written, date_str) + +def _assert_rows_date(rows: list[dict], day: date) -> None: + """金融契约: 响应行的 date 字段 (若提供) 必须与请求日期一致。 + + 服务端忽略日期参数返回当日数据时会静默把当日值写进历史分区, + 造成整个时序口径错乱 —— 此处 fail-closed 拒绝 (实测确实有忽略 + ?date= 的接口)。date 字段缺省的接口不做校验。 + """ + want = day.isoformat() + for r in rows[:20]: + if not isinstance(r, dict): + continue + raw = r.get("date") + if raw is None: + continue + if str(raw)[:10] != want: + raise ValueError( + f"接口返回的日期 {str(raw)[:10]!r} 与请求日期 {want} 不一致 " + "(接口可能不支持日期参数), 已拒绝写入该分区" + ) + + +async def fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict]: + """按日期请求外部 API 并解析为行 (不写盘)。空数据返回 []。 + + 与 fetch_and_ingest 共用同一解析链 (response_path/预设转换/字段映射/ + 关联字段校验), 历史回补与当日拉取不产生第二套口径。 """ pull = config.pull if not pull or not pull.url: raise ValueError("拉取未配置或 URL 为空") + url = _with_date_param(pull.url, pull.date_param, target_date) async with httpx.AsyncClient(timeout=30) as client: headers = pull.headers or {} kwargs: dict[str, Any] = {"headers": headers} @@ -124,7 +152,7 @@ async def fetch_and_ingest( if "content-type" not in {k.lower() for k in headers}: kwargs["headers"]["Content-Type"] = "application/json" - resp = await client.request(pull.method.upper(), pull.url, **kwargs) + resp = await client.request(pull.method.upper(), url, **kwargs) resp.raise_for_status() # 解析 JSON @@ -135,8 +163,6 @@ async def fetch_and_ingest( # 提取行 rows = _extract_rows(data, pull.response_path) - if not rows: - raise ValueError("提取到的行数为 0") # 内置预设 (概念/行业): 应用结构转换, 让产出 schema 与分析页一致。 # 否则 raw 接口列 (concepts/industries 数组、name) 会直接覆盖正确的 part.parquet, @@ -157,10 +183,136 @@ async def fetch_and_ingest( if rows and not ({"symbol", "code"} & row_keys or mapped_cols & row_keys): raise ValueError("数据行中缺少 symbol/code 字段，请配置字段映射或标的映射") - # 写入 - snap = date.today() - n = rows_to_parquet(rows, config, data_dir, snapshot_date=snap) - return n, snap.isoformat() + _assert_rows_date(rows, target_date) + return rows + + +async def fetch_and_ingest( + config: ExtConfig, + data_dir, + target_date: date | None = None, +) -> tuple[int, str]: + """执行一次拉取: 请求外部 API → 解析响应 → 写入 Parquet。 + + target_date 默认当日; 历史回补传入目标日期 (写入对应分区)。 + Returns: + (rows_written, date_str) + """ + day = target_date or date.today() + rows = await fetch_rows_for_date(config, day) + if not rows: + raise ValueError("提取到的行数为 0") + n = rows_to_parquet(rows, config, data_dir, snapshot_date=day) + return n, day.isoformat() + + +MAX_BACKFILL_DAYS = 120 # 单次回补上限: 同步端点, 控制请求时长 +_BACKFILL_DAY_INTERVAL_S = 0.3 # 相邻请求间隔 (对数据源限速) +_BACKFILL_429_WAIT_S = 30.0 # 429 限流退避时长 (服务端按分钟配额) +_BACKFILL_MAX_CONSECUTIVE_429 = 3 # 连续 429 天数达到阈值 → 中止本次回补 +_RATE_LIMIT_ABORT_REASON = "限流中止 (429), 稍后重跑回补可自动续补剩余日期" + + +def _status_code(e: BaseException) -> int | None: + """从 httpx.HTTPStatusError 提取状态码; 非该类异常返回 None。""" + resp = getattr(e, "response", None) + return getattr(resp, "status_code", None) + + +def _day_partition(data_dir, config_id: str, day: date) -> Path: + return Path(data_dir) / "ext_data" / config_id / "timeseries" / f"date={day.isoformat()}" / "part.parquet" + + +async def backfill_history( + config: ExtConfig, + data_dir, + start: date, + end: date, +) -> dict: + """按本地交易日逐日回补 timeseries 历史分区 (幂等, 已有分区跳过)。 + + 前提: 接口支持按日期查询 (pull.date_param 已配置)。交易日取本地日K + 分区日期 —— 非交易日无人气数据, 也避免无谓请求。单日失败不中断, + 汇总进 failed 清单返回; 该日无数据 (空响应或 404) 计入 empty 跳过。 + """ + if config.mode != "timeseries": + raise ValueError("仅 timeseries 模式支持历史回补 (snapshot 无历史概念)") + pull = config.pull + if not pull or not pull.url: + raise ValueError("拉取未配置或 URL 为空") + if not pull.date_param: + raise ValueError("接口未配置日期参数 (date_param) —— 需接口支持 ?日期参数= 历史查询") + if start > end: + raise ValueError("开始日期不能晚于结束日期") + if (end - start).days + 1 > MAX_BACKFILL_DAYS: + raise ValueError(f"单次回补上限 {MAX_BACKFILL_DAYS} 天, 请分段执行") + + from app.services.dragon_tiger import _local_trading_days + + days = [d for d in _local_trading_days(data_dir) if start <= d <= end] + if not days: + raise ValueError("范围内无本地交易日 (需先同步日K以确定交易日历)") + + fetched = skipped = empty = 0 + rows_written = 0 + failed: list[dict] = [] + consecutive_429 = 0 # 连续限流天数 (重试成功即清零); 达到阈值中止本次回补 + for i, d in enumerate(days): + part = _day_partition(data_dir, config.id, d) + if part.exists(): + skipped += 1 + continue + try: + rows = await fetch_rows_for_date(config, d) + consecutive_429 = 0 + if not rows: + empty += 1 # 该日无数据 (服务端未归档), 不是错误 + else: + rows_written += rows_to_parquet(rows, config, data_dir, snapshot_date=d) + fetched += 1 + except httpx.HTTPStatusError as e: + if _status_code(e) == 404: + # 接口契约 (tickflow-hub /exports、/fuyao-rank): 该日无快照 + # 返回 404 —— 视为该日无数据跳过, 不计入失败 + empty += 1 + elif _status_code(e) != 429: + failed.append({"date": d.isoformat(), "reason": str(e)[:200]}) + else: + # 服务端按分钟配额限流: 退避后原地重试一次; 连续多日 429 + # 说明配额窗口已耗尽, 中止剩余天数 (幂等, 重跑即可续补)。 + consecutive_429 += 1 + if consecutive_429 >= _BACKFILL_MAX_CONSECUTIVE_429: + remaining = [dd for dd in days[i:] if not _day_partition(data_dir, config.id, dd).exists()] + failed.extend({"date": dd.isoformat(), "reason": _RATE_LIMIT_ABORT_REASON} + for dd in remaining) + break + await asyncio.sleep(_BACKFILL_429_WAIT_S) + try: + rows = await fetch_rows_for_date(config, d) + consecutive_429 = 0 + if not rows: + empty += 1 + else: + rows_written += rows_to_parquet(rows, config, data_dir, snapshot_date=d) + fetched += 1 + except Exception as e2: + if _status_code(e2) == 404: # 退避重试后无该日快照 → 同样视为无数据 + empty += 1 + else: + failed.append({"date": d.isoformat(), "reason": str(e2)[:200]}) + except Exception as e: + failed.append({"date": d.isoformat(), "reason": str(e)[:200]}) + if i + 1 < len(days): + await asyncio.sleep(_BACKFILL_DAY_INTERVAL_S) # 限速, 对数据源礼貌 + return { + "total_days": len(days), + "fetched": fetched, + "skipped_existing": skipped, + "empty": empty, + "failed": failed, + "rows_written": rows_written, + } + # --------------------------------------------------------------------------- @@ -298,11 +450,11 @@ class PullScheduler: # 间隔取自最新配置 (每次重新读取, 修复改间隔不生效) interval = max(pull.schedule_minutes * 60, 60) # 至少 60s # 预告下次运行时间, 供前端展示 - next_dt = datetime.now(timezone.utc).timestamp() + interval + next_dt = datetime.now(UTC).timestamp() + interval latest = store.get(config.id) if latest and latest.pull: latest.pull.next_run = datetime.fromtimestamp( - next_dt, tz=timezone.utc + next_dt, tz=UTC ).isoformat() store.upsert(latest) diff --git a/backend/tests/test_ext_backfill.py b/backend/tests/test_ext_backfill.py new file mode 100644 index 0000000..8313e3c --- /dev/null +++ b/backend/tests/test_ext_backfill.py @@ -0,0 +1,313 @@ +"""扩展数据历史回补测试 — date_param 按日拉取 / 日期一致性防污染 / 幂等回补。 + +背景: 人气排行等时序扩展表历史只能从开启拉取之日起累积; backfill_history +按本地交易日逐日回补。金融契约重点: 接口忽略日期参数返回当日数据时 +必须拒写历史分区 (实测确有此类接口), 否则整个时序口径错乱。 +""" +from __future__ import annotations + +from datetime import date +from pathlib import Path +from typing import ClassVar + +import httpx +import polars as pl +import pytest + +from app.services import ext_pull +from app.services.ext_data import ExtConfig, ExtConfigStore, ExtField, PullConfig +from app.services.ext_pull import ( + _assert_rows_date, + _with_date_param, + backfill_history, + fetch_rows_for_date, +) + + +def _cfg(mode: str = "timeseries", date_param: str | None = "date") -> ExtConfig: + return ExtConfig( + id="hot", label="人气", mode=mode, + fields=[ + ExtField("symbol", "string"), ExtField("rank", "int"), + ExtField("date", "string"), ExtField("heat", "float"), + ], + pull=PullConfig(url="https://example.test/rank", date_param=date_param), + ) + + +def _row(sym: str, day: str) -> dict: + return {"symbol": sym, "rank": 1, "date": day, "heat": 9.9} + + +class _FakeResp: + def __init__(self, payload): + self._payload = payload + + def raise_for_status(self) -> None: + pass + + def json(self): + return self._payload + + +class _FakeClient: + """按请求 URL 中的日期参数返回预置数据; 记录全部请求 URL。 + + responses/calls 用函数属性 (class-annotation) 而非实例属性: + ext_pull 以 ``httpx.AsyncClient(**kw)`` 工厂方式构造, fixture 借类属性 + 注入预置数据, RUF012 mutable-default 面由此声明为 ClassVar 语义。 + """ + + responses: ClassVar[dict[str, list]] = {} + calls: ClassVar[list[str]] = [] + errors: ClassVar[dict[str, Exception]] = {} + fail_times: ClassVar[dict[str, int]] = {} # url -> 还需失败的次数 + error_sequence: ClassVar[dict[str, list[Exception]]] = {} # url -> 按序抛出后耗尽 + + def __init__(self, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def request(self, method: str, url: str, **kwargs): + _FakeClient.calls.append(url) + seq = _FakeClient.error_sequence.get(url) + if seq: + raise seq.pop(0) + if url in _FakeClient.errors and _FakeClient.fail_times.get(url, 1) > 0: + _FakeClient.fail_times[url] = _FakeClient.fail_times.get(url, 1) - 1 + raise _FakeClient.errors[url] + return _FakeResp(_FakeClient.responses.get(url, [])) + + +@pytest.fixture() +def fake_http(monkeypatch): + _FakeClient.responses = {} + _FakeClient.calls = [] + _FakeClient.errors = {} + _FakeClient.fail_times = {} + _FakeClient.error_sequence = {} + monkeypatch.setattr(ext_pull.httpx, "AsyncClient", _FakeClient) + return _FakeClient + + +# ── 纯函数 ──────────────────────────────────────────────── + +def test_with_date_param_url_building(): + d = date(2026, 1, 5) + assert _with_date_param("https://x/api", "date", d) == "https://x/api?date=2026-01-05" + assert _with_date_param("https://x/api?a=1", "date", d) == "https://x/api?a=1&date=2026-01-05" + assert _with_date_param("https://x/api", None, d) == "https://x/api" + + +def test_assert_rows_date_contract(): + d = date(2026, 1, 5) + _assert_rows_date([_row("A", "2026-01-05")], d) # 一致 + _assert_rows_date([_row("A", "2026-01-05 00:00:00")], d) # 带时间前缀 + _assert_rows_date([{"symbol": "A"}], d) # 无 date 字段: 不校验 + with pytest.raises(ValueError, match="不一致"): + _assert_rows_date([_row("A", "2026-09-06")], d) # 接口忽略参数返回当日 + + +def test_pull_config_date_param_roundtrip(): + p = PullConfig(url="u", date_param="date") + assert p.to_dict()["date_param"] == "date" + legacy = PullConfig.from_dict({"url": "u"}) # 旧 JSON 无该键 + assert legacy.date_param is None + + +# ── fetch_rows_for_date ─────────────────────────────────── + +async def test_fetch_rows_builds_dated_url(fake_http): + fake_http.responses["https://example.test/rank?date=2026-01-05"] = [_row("A", "2026-01-05")] + rows = await fetch_rows_for_date(_cfg(), date(2026, 1, 5)) + assert rows and rows[0]["symbol"] == "A" + assert fake_http.calls == ["https://example.test/rank?date=2026-01-05"] + + +async def test_fetch_rows_rejects_mismatched_date(fake_http): + # 接口忽略 ?date= 返回当日数据 → 必须拒收, 不给 backfill 写历史分区的机会 + fake_http.responses["https://example.test/rank?date=2026-01-05"] = [_row("A", "2026-09-06")] + with pytest.raises(ValueError, match="不一致"): + await fetch_rows_for_date(_cfg(), date(2026, 1, 5)) + + +async def test_fetch_rows_empty_returns_empty_list(fake_http): + assert await fetch_rows_for_date(_cfg(), date(2026, 1, 5)) == [] + + +async def test_fetch_rows_requires_symbol_or_code(fake_http): + fake_http.responses["https://example.test/rank?date=2026-01-05"] = [{"rank": 1}] + with pytest.raises(ValueError, match="symbol/code"): + await fetch_rows_for_date(_cfg(), date(2026, 1, 5)) + + +# ── backfill_history ────────────────────────────────────── + +def _seed_trading_days(data_dir: Path, *days: str) -> None: + root = data_dir / "kline_daily" + for d in days: + (root / f"date={d}").mkdir(parents=True, exist_ok=True) + + +async def test_backfill_rejects_snapshot_and_missing_date_param(tmp_path): + with pytest.raises(ValueError, match="timeseries"): + await backfill_history(_cfg(mode="snapshot"), tmp_path, date(2026, 1, 5), date(2026, 1, 9)) + with pytest.raises(ValueError, match="date_param"): + await backfill_history(_cfg(date_param=None), tmp_path, date(2026, 1, 5), date(2026, 1, 9)) + + +async def test_backfill_rejects_bad_range_and_no_trading_days(tmp_path): + with pytest.raises(ValueError, match="晚于"): + await backfill_history(_cfg(), tmp_path, date(2026, 1, 9), date(2026, 1, 5)) + with pytest.raises(ValueError, match="上限"): + await backfill_history(_cfg(), tmp_path, date(2025, 1, 1), date(2026, 9, 1)) + with pytest.raises(ValueError, match="交易日"): + await backfill_history(_cfg(), tmp_path, date(2026, 1, 5), date(2026, 1, 9)) + + +async def test_backfill_writes_partitions_and_is_idempotent(tmp_path, fake_http): + _seed_trading_days(tmp_path, "2026-01-02", "2026-01-05", "2026-01-06") + # 01-02: 已有分区 → 跳过; 01-05: 接口有数据 → 写入; 01-06: 接口空 → empty + part_102 = tmp_path / "ext_data" / "hot" / "timeseries" / "date=2026-01-02" / "part.parquet" + part_102.parent.mkdir(parents=True) + part_102.write_bytes(b"x") + fake_http.responses["https://example.test/rank?date=2026-01-05"] = [ + _row("600000.SH", "2026-01-05"), _row("000001.SZ", "2026-01-05"), + ] + + result = await backfill_history(_cfg(), tmp_path, date(2026, 1, 1), date(2026, 1, 9)) + assert result["total_days"] == 3 + assert result["fetched"] == 1 and result["rows_written"] == 2 + assert result["skipped_existing"] == 1 + assert result["empty"] == 1 and result["failed"] == [] + written = tmp_path / "ext_data" / "hot" / "timeseries" / "date=2026-01-05" / "part.parquet" + assert written.exists() + + # 幂等: 已写入的分区不再请求; empty 日 (无分区文件) 允许重试 + fake_http.calls.clear() + again = await backfill_history(_cfg(), tmp_path, date(2026, 1, 1), date(2026, 1, 9)) + assert again["skipped_existing"] == 2 and again["fetched"] == 0 and again["empty"] == 1 + assert fake_http.calls == ["https://example.test/rank?date=2026-01-06"] + + +async def test_backfill_collects_failures_without_abort(tmp_path, fake_http): + _seed_trading_days(tmp_path, "2026-01-05", "2026-01-06") + # 01-05: 接口忽略参数返回当日 → 拒写 (failed); 01-06: 正常写入 + fake_http.responses["https://example.test/rank?date=2026-01-05"] = [_row("A", "2026-09-06")] + fake_http.responses["https://example.test/rank?date=2026-01-06"] = [_row("A", "2026-01-06")] + + result = await backfill_history(_cfg(), tmp_path, date(2026, 1, 5), date(2026, 1, 6)) + assert result["fetched"] == 1 + assert [f["date"] for f in result["failed"]] == ["2026-01-05"] + assert "不一致" in result["failed"][0]["reason"] + assert not (tmp_path / "ext_data/hot/timeseries/date=2026-01-05").exists() + + +# ── 与扩展消费链路的衔接 ───────────────────────────────── + +async def test_backfilled_partition_feeds_signal_frame(tmp_path, fake_http): + """回补落盘的历史分区, ext_factors 按日对齐立即可用 (PIT)。""" + from app.factors import ext_factors + + ExtConfigStore(tmp_path).upsert(_cfg()) + _seed_trading_days(tmp_path, "2026-01-05", "2026-01-06") + fake_http.responses["https://example.test/rank?date=2026-01-05"] = [ + {"symbol": "600000.SH", "rank": 3, "date": "2026-01-05", "heat": 88.0}, + ] + fake_http.responses["https://example.test/rank?date=2026-01-06"] = [ + {"symbol": "600000.SH", "rank": 1, "date": "2026-01-06", "heat": 99.0}, + ] + await backfill_history(_cfg(), tmp_path, date(2026, 1, 5), date(2026, 1, 6)) + + frame = pl.DataFrame( + {"symbol": ["600000.SH", "600000.SH"], "date": ["2026-01-05", "2026-01-06"]}, + schema={"symbol": pl.Utf8, "date": pl.Utf8}, + ) + out = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=tmp_path) + assert out["ext_hot_rank"].to_list() == [3.0, 1.0] # rank int → Float64 + assert out["ext_hot_heat"].to_list() == [88.0, 99.0] # 每日各自的值, 无串日 + + +# ── 429 限流退避与中止 ───────────────────── + + +def _err_429(url: str) -> httpx.HTTPStatusError: + req = httpx.Request("GET", url) + return httpx.HTTPStatusError("429", request=req, response=httpx.Response(429, request=req)) + + +async def test_backfill_429_retries_once_and_succeeds(tmp_path, fake_http, monkeypatch): + monkeypatch.setattr(ext_pull, "_BACKFILL_429_WAIT_S", 0) + monkeypatch.setattr(ext_pull, "_BACKFILL_DAY_INTERVAL_S", 0) + _seed_trading_days(tmp_path, "2026-01-05") + url = "https://example.test/rank?date=2026-01-05" + fake_http.errors[url] = _err_429(url) + fake_http.fail_times[url] = 1 # 仅首请 429, 重试成功 + fake_http.responses[url] = [_row("A", "2026-01-05")] + + result = await backfill_history(_cfg(), tmp_path, date(2026, 1, 5), date(2026, 1, 5)) + assert result["fetched"] == 1 and result["failed"] == [] + assert fake_http.calls.count(url) == 2 # 首请 429 + 退避重试 + + +async def test_backfill_aborts_after_consecutive_429(tmp_path, fake_http, monkeypatch): + monkeypatch.setattr(ext_pull, "_BACKFILL_429_WAIT_S", 0) + monkeypatch.setattr(ext_pull, "_BACKFILL_DAY_INTERVAL_S", 0) + days = ["2026-01-05", "2026-01-06", "2026-01-07", "2026-01-08", "2026-01-09"] + _seed_trading_days(tmp_path, *days) + for d in days: + u = f"https://example.test/rank?date={d}" + fake_http.errors[u] = _err_429(u) + fake_http.fail_times[u] = 2 # 首请 + 重试均 429 + + result = await backfill_history(_cfg(), tmp_path, date(2026, 1, 5), date(2026, 1, 9)) + assert result["fetched"] == 0 + # 连续计数达到 3 即中止: 前两日各 2 次请求, 第 3 日首个 429 立即中止 + assert len(fake_http.calls) == 5 + aborts = [f for f in result["failed"] if f["reason"].startswith("限流中止")] + assert len(result["failed"]) == 5 and len(aborts) == 3 # 剩余未请求的日标记为可续补 + + +# ── 404 无快照日 (tickflow-hub 契约) ────────────────────── + + +def _err_404(url: str) -> httpx.HTTPStatusError: + req = httpx.Request("GET", url) + return httpx.HTTPStatusError("404", request=req, response=httpx.Response(404, request=req)) + + +async def test_backfill_404_counts_as_empty_not_failed(tmp_path, fake_http, monkeypatch): + """hub /exports、/fuyao-rank 契约: 该日无快照返回 404 → empty 跳过, 不进失败清单。""" + monkeypatch.setattr(ext_pull, "_BACKFILL_DAY_INTERVAL_S", 0) + _seed_trading_days(tmp_path, "2026-01-05", "2026-01-06", "2026-01-07") + u5, u6, u7 = (f"https://example.test/rank?date={d}" for d in ("2026-01-05", "2026-01-06", "2026-01-07")) + fake_http.errors[u5] = _err_404(u5) + fake_http.responses[u6] = [_row("600000.SH", "2026-01-06")] + fake_http.errors[u7] = _err_404(u7) + + result = await backfill_history(_cfg(), tmp_path, date(2026, 1, 5), date(2026, 1, 7)) + assert result["fetched"] == 1 and result["empty"] == 2 + assert result["failed"] == [] # 404 不是失败 + assert result["rows_written"] == 1 + assert (tmp_path / "ext_data/hot/timeseries/date=2026-01-06/part.parquet").exists() + + +async def test_backfill_404_after_429_retry_counts_as_empty(tmp_path, fake_http, monkeypatch): + """429 退避重试后返回 404: 同样视为该日无数据, 不进失败清单。""" + monkeypatch.setattr(ext_pull, "_BACKFILL_429_WAIT_S", 0) + monkeypatch.setattr(ext_pull, "_BACKFILL_DAY_INTERVAL_S", 0) + _seed_trading_days(tmp_path, "2026-01-05", "2026-01-06") + u5 = "https://example.test/rank?date=2026-01-05" + u6 = "https://example.test/rank?date=2026-01-06" + fake_http.error_sequence[u5] = [_err_429(u5), _err_404(u5)] # 首请 429 → 重试 404 + fake_http.responses[u6] = [_row("A", "2026-01-06")] + + result = await backfill_history(_cfg(), tmp_path, date(2026, 1, 5), date(2026, 1, 6)) + assert result["fetched"] == 1 and result["empty"] == 1 + assert result["failed"] == [] + assert fake_http.calls.count(u5) == 2 # 确认重试确实发生 diff --git a/docs/features.md b/docs/features.md index 51564b3..495b78a 100644 --- a/docs/features.md +++ b/docs/features.md @@ -183,6 +183,12 @@ 接入后自动 schema 发现 + 符号归一,页面可视化配置,最终并入 DuckDB 同台分析。 +##### 时序表历史回补 + +timeseries 模式的表 (如人气排行) 默认只能从开启拉取之日起逐日累积。拉取配置新增**日期参数名**(`date_param`): 接口支持按日查询时填参数名 (如 `date`), 即可在拉取面板使用**历史回补** —— 按本地交易日逐日请求 `?date=YYYY-MM-DD` 写入对应分区,已有分区自动跳过 (幂等,可重复执行; 单次上限 120 天,单日失败不中断并逐项列出原因)。回补落盘的历史数据立即进入信号/因子/回测的按日对齐通道。 + +防污染契约: 若接口忽略日期参数返回当日数据 (以响应行 `date` 字段为准校验), 该日**拒绝写入**并计入失败清单 —— 否则当日值会静默污染整个历史时序。接口侧需提供: 传日期参数返回该日数据、无数据返回空数组或 404 (均视为该日无数据, 计入 empty 跳过)、每行带 `date` 字段;不传参数保持当日 (向后兼容)。 + #### 扩展字段接入信号/因子 扩展表的数值字段(int/float)可直接用作**自定义信号条件**与**因子**——无需任何额外配置: diff --git a/frontend/src/components/ext-data/ExtDataPullPanel.tsx b/frontend/src/components/ext-data/ExtDataPullPanel.tsx index f15be3d..2434b0b 100644 --- a/frontend/src/components/ext-data/ExtDataPullPanel.tsx +++ b/frontend/src/components/ext-data/ExtDataPullPanel.tsx @@ -1,6 +1,6 @@ import { useState } from 'react' -import { Loader2, Search, Check, Clock, Zap, Settings2, AlertCircle, CheckCircle2, Calendar } from 'lucide-react' -import { api, type ExtDataConfig } from '@/lib/api' +import { Loader2, Search, Check, Clock, Zap, Settings2, AlertCircle, CheckCircle2, Calendar, History } from 'lucide-react' +import { api, type ExtDataBackfillResult, type ExtDataConfig } from '@/lib/api' import { toast } from '@/components/Toast' export function ExtDataPullPanel({ config, onSaved }: { @@ -21,6 +21,7 @@ export function ExtDataPullPanel({ config, onSaved }: { const [schedule, setSchedule] = useState(pull?.schedule_minutes ?? 1440) const [timeWindowStart, setTimeWindowStart] = useState(pull?.time_window_start ?? '') const [timeWindowEnd, setTimeWindowEnd] = useState(pull?.time_window_end ?? '') + const [dateParam, setDateParam] = useState(pull?.date_param ?? '') const [enabled, setEnabled] = useState(pull?.enabled ?? false) const [saving, setSaving] = useState(false) const [testing, setTesting] = useState(false) @@ -29,6 +30,14 @@ export function ExtDataPullPanel({ config, onSaved }: { const [testResult, setTestResult] = useState<{ total_rows: number; preview: Record[]; has_symbol: boolean } | null>(null) const [error, setError] = useState('') + // 历史回补 (仅 timeseries + 接口支持按日查询) + const today = new Date().toISOString().slice(0, 10) + const monthAgo = new Date(Date.now() - 30 * 86400_000).toISOString().slice(0, 10) + const [bfStart, setBfStart] = useState(monthAgo) + const [bfEnd, setBfEnd] = useState(today) + const [bfRunning, setBfRunning] = useState(false) + const [bfResult, setBfResult] = useState(null) + // 解析 JSON 输入, 失败时设置 error 并返回 null const parseJson = (str: string, label: string): Record | undefined | null => { if (!str.trim()) return undefined @@ -48,6 +57,7 @@ export function ExtDataPullPanel({ config, onSaved }: { schedule_minutes: schedule, enabled: enabledOverride ?? enabled, time_window_start: timeWindowStart || null, time_window_end: timeWindowEnd || null, + date_param: dateParam.trim() || null, } } @@ -87,6 +97,19 @@ export function ExtDataPullPanel({ config, onSaved }: { .finally(() => setRunning(false)) } + const handleBackfill = () => { + setBfRunning(true); setError(''); setBfResult(null) + api.extDataBackfill(config.id, bfStart, bfEnd) + .then(r => { + setBfResult(r) + onSaved() + if (r.failed.length === 0) toast(`回补完成 · 写入 ${r.fetched} 日 ${r.rows_written} 行`, 'success') + else toast(`回补完成 · ${r.failed.length} 日失败 (见详情)`, 'error') + }) + .catch(e => setError(e.message || '回补失败')) + .finally(() => setBfRunning(false)) + } + // 开关 toggle: 自动保存全量配置 (切换 enabled), 后端 refresh 后立即首次拉取 const [toggling, setToggling] = useState(false) const handleToggle = (next: boolean) => { @@ -203,6 +226,15 @@ export function ExtDataPullPanel({ config, onSaved }: { +
+
日期参数名 (接口支持按日查询时填, 如 date)
+ setDateParam(e.target.value)} + placeholder="date · 留空=接口只有当日快照" + className="w-full rounded-btn border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40" + /> +
+
字段映射 (外部名 → 内部名，JSON，可选)