diff --git a/backend/app/api/ext_data.py b/backend/app/api/ext_data.py index c0288c32..e31037dc 100644 --- a/backend/app/api/ext_data.py +++ b/backend/app/api/ext_data.py @@ -115,6 +115,9 @@ class PullConfigReq(BaseModel): # 日内序列表时间列名 (字段映射后的列名, 如 "ts"): 配置后同 symbol 允许多行, # 按 [symbol, time_field] 去重; 留空 = 每日快照表 (按 symbol 去重) time_field: str | None = Field(None, max_length=32) + # 增量续传参数名 (如 "since"): 每次拉取自动带上本地目标日分区 time_field 的 + # 最大值, 源只返回其后数据 (盘中多盘表免全量重拉); 需先配置 time_field + since_param: str | None = Field(None, min_length=1, max_length=16, pattern=r"^[A-Za-z_][A-Za-z0-9_]*$") # 鉴权方式; 请求中缺省 (None) = 保留现有配置, {"type":"none"} = 关闭鉴权 auth: PullAuthReq | None = None # 单次拉取请求超时 (秒), 默认 30 与历史行为一致; 大响应接口可调高 @@ -414,6 +417,53 @@ def _date_range(config: ExtConfig, data_dir: Path) -> list[str] | None: return [dates[0], dates[-1]] +def _plate_stats(config: ExtConfig, data_dir: Path) -> dict | None: + """日内多盘表最新分区的盘次概览 (源卡片展示用)。 + + 只读最新分区的时间列: 盘数 = 时刻值去重计数, from/to = 时刻范围 (HH:MM:SS, + 与 collapse_ext_rows_at 的时刻切片同口径, 兼容 ISO 前缀/纯时刻两种列)。 + 非多盘表 (非 timeseries 或未配 time_field)、无分区、列缺失或文件损坏 → None。 + """ + if config.mode != "timeseries": + return None + tf = config.pull.time_field if config.pull else None + if not tf: + return None + base = _ext_data_dir(config, data_dir) + if not base.exists(): + return None + parts = sorted( + d.name[5:] for d in base.iterdir() + if d.is_dir() and d.name.startswith("date=") and (d / "part.parquet").exists() + ) + if not parts: + return None + try: + col = pl.read_parquet( + base / f"date={parts[-1]}" / "part.parquet", columns=[tf] + )[tf].cast(pl.Utf8) + except Exception: # noqa: BLE001 — 列缺失/文件损坏 → 无多盘信息, 不阻断列表 + return None + if col.is_empty(): + return None + tod = ( + pl.DataFrame({tf: col}) + .select( + pl.when(pl.col(tf).str.len_chars() >= 11) + .then(pl.col(tf).str.slice(11, 8)) + .otherwise(pl.col(tf)) + .alias("_tod") + ) + .select( + plates=pl.col("_tod").n_unique(), + first=pl.col("_tod").min(), + last=pl.col("_tod").max(), + ) + .row(0, named=True) + ) + return {"date": parts[-1], "plates": tod["plates"], "from": tod["first"], "to": tod["last"]} + + @router.get("") def list_configs(request: Request): """列出所有扩展数据配置。""" @@ -424,6 +474,7 @@ def list_configs(request: Request): d = c.to_dict() d["latest_sync_date"] = _latest_sync_date(c, data_dir) d["date_range"] = _date_range(c, data_dir) + d["plate"] = _plate_stats(c, data_dir) items.append(d) return {"items": items} @@ -1057,6 +1108,7 @@ def configure_pull(request: Request, config_id: str, body: PullConfigReq): date_param=body.date_param, date_format=body.date_format, time_field=body.time_field, + since_param=body.since_param, auth=new_auth, timeout_seconds=body.timeout_seconds, page_param=body.page_param, diff --git a/backend/app/services/ext_data.py b/backend/app/services/ext_data.py index 51fbd2d6..b3569507 100644 --- a/backend/app/services/ext_data.py +++ b/backend/app/services/ext_data.py @@ -12,7 +12,7 @@ from typing import Literal import polars as pl -from app.market_time import cn_today +from app.market_time import cn_now, cn_today from app.services.fs_utils import atomic_write_parquet, atomic_write_text logger = logging.getLogger(__name__) @@ -49,6 +49,16 @@ SCHEDULE_TIME_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$") # 多盘扩展列选盘时刻 (ext_columns 的 "@HH:MM[:SS]" 后缀) 的合法格式 EXT_SNAPSHOT_AT_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d(?::[0-5]\d)?$") +# time_field 保留名: 源数据没有时间列时 (人气榜/资金流等每日快照表), 配置它 +# 让写入路径注入拉取时刻, 快照表同样获得日内多盘能力 (去重键 [symbol, 拉取时刻])。 +# 注入值是北京时间 ISO 串, 与 collapse_ext_rows_at 的时刻切片口径兼容。 +PLATE_TS_FIELD = "__pull_ts__" + + +def _pull_ts_value() -> str: + """注入用拉取时刻 (北京时间 ISO 串); 测试 monkeypatch 本函数固定时刻。""" + return cn_now().isoformat() + def norm_ext_snapshot_at(raw: str) -> str: """选盘时刻归一为 "HH:MM:SS" ("09:20" → "09:20:00"); 非法值抛 ValueError。""" @@ -103,7 +113,7 @@ class PullConfig: "date_format", "time_field", "auth", "timeout_seconds", "page_param", "page_size_param", "page_size", "page_start", "max_pages", "fanout", "schedule_time", "schedule_period", "schedule_day", - "schedule_trading_only", + "schedule_trading_only", "since_param", ) def __init__( @@ -138,6 +148,7 @@ class PullConfig: schedule_period: str | None = None, schedule_day: int | None = None, schedule_trading_only: bool = False, + since_param: str | None = None, ) -> None: self.fanout = fanout or None # 两级拉取配方 (见 fanout.py), None=单请求 self.url = url @@ -164,6 +175,14 @@ class PullConfig: # symbol 允许多行 (按 [symbol, time_field] 去重), 用于集合竞价等日内多盘 # 数据; None = 每日快照表 (按 symbol 去重, 默认) self.time_field = (time_field or "").strip() or None + # 增量续传参数名 (如 "since"): 配置后每次拉取自动带上本地目标日分区 + # time_field 列的最大值, 源只返回该值之后的数据 —— 盘中多盘表免全量重拉 + # (缺口感知: 本地缺多少源补多少; 无本地数据时不带参数=全量, 见 ext_pull)。 + # 依赖 time_field (游标值取自该列), 未配置 time_field 时归一 None (失效); + # PLATE_TS_FIELD 是本地注入的拉取时刻, 源无法按它过滤, 同样失效。 + self.since_param = ( + (since_param or "").strip() or None + ) if self.time_field and self.time_field != PLATE_TS_FIELD else None # 拉取接口鉴权方式 {"type": "none|bearer|header|query", "header": ..., "param": ...}, # 与自定义行情源 AuthConfig 同口径; Key 本体存 secrets_store, 不落 config.json self.auth = auth @@ -237,6 +256,7 @@ class PullConfig: "date_param": self.date_param, "date_format": self.date_format, "time_field": self.time_field, + "since_param": self.since_param, "auth": self.auth, "timeout_seconds": self.timeout_seconds, "page_param": self.page_param, @@ -274,6 +294,7 @@ class PullConfig: date_param=d.get("date_param"), date_format=d.get("date_format", "iso"), time_field=d.get("time_field"), + since_param=d.get("since_param"), auth=d.get("auth"), timeout_seconds=d.get("timeout_seconds", 30), page_param=d.get("page_param"), @@ -798,13 +819,18 @@ def write_ext_parquet( out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / "part.parquet" + # time_field 配置为保留名: 注入拉取时刻, 盘中每次拉取各存一盘 + # (同批一次注入 → 批内仍按 symbol 收敛; 不同批次不同盘)。 + tf = config.pull.time_field if config.pull else None + if tf == PLATE_TS_FIELD and PLATE_TS_FIELD not in df.columns: + df = df.with_columns(pl.lit(_pull_ts_value()).alias(PLATE_TS_FIELD)) + # 如果已有文件，合并去重 if out_path.exists(): try: existing = pl.read_parquet(out_path) key: str | list[str] = "symbol" if "symbol" in df.columns else df.columns[0] # 日内序列表 (配置 time_field): 同 symbol 同时刻去重, 不同盘并存 - tf = config.pull.time_field if config.pull else None if tf and tf in df.columns: key = [key, tf] df = pl.concat([existing, df]).unique( diff --git a/backend/app/services/ext_pull.py b/backend/app/services/ext_pull.py index 82ad8979..9c70d6b8 100644 --- a/backend/app/services/ext_pull.py +++ b/backend/app/services/ext_pull.py @@ -158,6 +158,35 @@ def _with_date_param(url: str, date_param: str | None, day: date, date_format: s return f"{url}{sep}{date_param}={_format_date_value(day, date_format)}" +def since_cursor(data_dir, config: ExtConfig, day: date) -> str | None: + """增量续传游标: 目标日本地分区 time_field 列的最大值 (字符串化)。 + + 盘中多盘表 (配置 pull.since_param + time_field) 用它让源只返回新于本地的 + 盘次。缺口感知天然成立 —— 本地缺多少源补多少; 下列情况返回 None (不带 + 参数 = 整日全量), 全部是安全的回退方向: + - 未配置 since_param/time_field (PullConfig 构造时已归一联动) + - 本地无该日分区 (首拉/新交易日/分区被删修复) + - 分区缺 time_field 列 (schema 漂移) 或文件损坏、列为空 + snapshot 模式无日期分区, 恒返回 None —— 增量仅对 timeseries 有意义。 + """ + pull = config.pull + if not pull or not pull.since_param or not pull.time_field or pull.fanout: + return None # fanout 两级拉取 (目录→明细) 无整表游标语义 + part = _day_partition(Path(data_dir), config.id, day) + if not part.exists(): + return None + import polars as pl + + try: + col = pl.read_parquet(part, columns=[pull.time_field])[pull.time_field] + value = col.max() + except Exception: # noqa: BLE001 — 列缺失/文件损坏 → 全量, 不阻断拉取 + return None + if value is None: + return None + return str(value).strip() or None + + # Hub 管理后台的会话 Cookie 名。它是有有效期的浏览器会话(默认 24h), # 换机器/换部署即失效, 不能当作定时拉取的长期凭据。 _HUB_SESSION_COOKIES = frozenset({"tickflow_hub_access"}) @@ -351,23 +380,28 @@ def _parse_rows_payload(config: ExtConfig, pull: PullConfig, data: Any) -> list[ return _apply_field_map(rows, pull.field_map) -async def fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict]: +async def fetch_rows_for_date(config: ExtConfig, target_date: date, data_dir=None) -> list[dict]: async with _pull_client_scope(): - return await _fetch_rows_for_date(config, target_date) + return await _fetch_rows_for_date(config, target_date, data_dir) -async def _fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict]: +async def _fetch_rows_for_date(config: ExtConfig, target_date: date, data_dir=None) -> list[dict]: """按日期请求外部 API 并解析为行 (不写盘)。空数据返回 []。 与 fetch_and_ingest 共用同一解析链 (response_path/字段映射/关联字段校验), 历史回补与当日拉取不产生第二套口径。 配置了 pull.page_param 时按页循环拉取 (仅 GET), 空页/短页/达 max_pages 停。 配置了 pull.fanout 时走两级拉取 (目录 → 逐条目明细), 见 ext_fanout。 + 传入 data_dir 且配置了 pull.since_param 时带增量游标 (?since=本地最大盘次); + 历史回补不传 data_dir → 恒全量 (回补只拉缺失分区, 游标无意义)。 """ pull = config.pull if not pull or not pull.url: raise ValueError("拉取未配置或 URL 为空") + cursor = since_cursor(data_dir, config, target_date) if data_dir is not None else None + since_extra: dict[str, str] | None = {pull.since_param: cursor} if cursor else None + if pull.fanout: from app.services.ext_fanout import fetch_fanout_rows @@ -376,7 +410,8 @@ async def _fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dic collected: list[dict] = [] page = pull.page_start for _ in range(pull.max_pages): - extra: dict[str, str] = {pull.page_param: str(page)} + # since 是结果集过滤器, 与页码参数一起每页携带 + extra: dict[str, str] = {**(since_extra or {}), pull.page_param: str(page)} if pull.page_size_param and pull.page_size > 0: extra[pull.page_size_param] = str(pull.page_size) data = await _request_json(pull, config.id, day=target_date, extra_params=extra) @@ -395,7 +430,7 @@ async def _fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dic ) rows = collected else: - data = await _request_json(pull, config.id, day=target_date) + data = await _request_json(pull, config.id, day=target_date, extra_params=since_extra) rows = _parse_rows_payload(config, pull, data) # 校验可关联标的的字段：直接 symbol/code，或配置里声明的映射源列。 @@ -432,7 +467,7 @@ async def fetch_and_ingest( """ # 同上: 落盘分区按北京日期, 否则 UTC 容器在北京时间 08:00 之前写的是前一天。 day = target_date or cn_today() - rows = await fetch_rows_for_date(config, day) + rows = await fetch_rows_for_date(config, day, data_dir) if not rows: raise ValueError("提取到的行数为 0") n = rows_to_parquet( diff --git a/backend/tests/test_ext_pull_since.py b/backend/tests/test_ext_pull_since.py new file mode 100644 index 00000000..bab4b6eb --- /dev/null +++ b/backend/tests/test_ext_pull_since.py @@ -0,0 +1,190 @@ +"""增量续传参数 (since_param): 盘中多盘表免每次全量重拉。 + +机制: 配置 pull.since_param + time_field 后, 每次拉取前读本地目标日分区 +time_field 列最大值, 以 ?{since_param}={值} 附到请求 — 源只返回其后数据, +缺口感知天然成立 (本地缺多少源补多少)。本地无数据/列缺失/文件损坏时不带 +参数 = 整日全量, 回退方向永远安全。 + +契约: + - PullConfig 归一: 未配置 time_field 时 since_param 失效 (None) + - 首拉 (无分区) 不带参数; 有分区带 max(time_field); 分区删除后又回全量 + - 分页拉取: since 作为过滤器与页码参数一起每页携带 + - 历史回补 (fetch_rows_for_date 不传 data_dir) 恒不带参数 + - 两级拉取 (fanout) 无整表游标语义, 不注入 + - API PullConfigReq.since_param 校验参数名合法性 +mock httpx.MockTransport, 不发真实请求。 +""" +from __future__ import annotations + +from datetime import date + +import httpx +import polars as pl +import pytest + +from app.services import ext_pull +from app.services.ext_data import ExtConfig, ExtField, PullConfig + +TARGET = date(2026, 10, 26) + + +def _config(**pull_over) -> ExtConfig: + kwargs = dict( + url="https://example.test/flow", + response_path="data", + time_field="ts", + since_param="since", + ) + kwargs.update(pull_over) + return ExtConfig( + id="flow", + label="资金流", + mode="timeseries", + fields=[ExtField("symbol", "string"), ExtField("val", "float"), ExtField("ts", "string")], + pull=PullConfig(**kwargs), + ) + + +def _install_transport(monkeypatch: pytest.MonkeyPatch, handler) -> list[str]: + """MockTransport 记录每次请求 URL; 返回 URL 清单。""" + urls: list[str] = [] + + def _record(request: httpx.Request) -> httpx.Response: + urls.append(str(request.url)) + return handler(request) + + real = httpx.AsyncClient + + def _factory(*args, **kwargs): + return real(*args, transport=httpx.MockTransport(_record), **kwargs) + + monkeypatch.setattr(ext_pull.httpx, "AsyncClient", _factory) + return urls + + +# ---------------------------------------------------------------- PullConfig 归一 +def test_since_param_requires_time_field() -> None: + p = PullConfig(time_field="ts", since_param="since") + assert p.since_param == "since" + assert PullConfig(since_param="since").since_param is None # 无 time_field → 失效 + assert PullConfig(time_field="ts").since_param is None + + d = p.to_dict() + assert d["since_param"] == "since" + assert PullConfig.from_dict(d).since_param == "since" # 往返不丢 + + +def test_since_cursor_fallbacks(tmp_path) -> None: + cfg = _config() + # 无分区 → None (首拉/新交易日/删分区修复 → 全量) + assert ext_pull.since_cursor(tmp_path, cfg, TARGET) is None + + part = ext_pull._day_partition(tmp_path, "flow", TARGET) + part.parent.mkdir(parents=True, exist_ok=True) + # 分区缺 ts 列 (schema 漂移) → None + pl.DataFrame({"symbol": ["000001.SZ"]}).write_parquet(part) + assert ext_pull.since_cursor(tmp_path, cfg, TARGET) is None + + # fanout 两级拉取无整表游标语义 → None + fanout_cfg = _config(fanout={"url": "https://example.test/members"}) + assert ext_pull.since_cursor(tmp_path, fanout_cfg, TARGET) is None + + +# ---------------------------------------------------------------- E2E: 首拉全量 → 增量 +def _rows(*slots: str) -> dict: + return {"data": [ + {"symbol": "000001.SZ", "val": 10.0, "ts": s} for s in slots + ]} + + +async def test_first_pull_full_then_incremental(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + """首拉无分区不带 since (全量); 落库后二次拉取带 since=本地最大 ts; + 增量行并入同日分区, 按 [symbol, ts] 去重。""" + urls = _install_transport(monkeypatch, lambda req: httpx.Response( + 200, json=_rows("1030") if "since=" in str(req.url) else _rows("0930", "1000"), + )) + + n1, d1 = await ext_pull.fetch_and_ingest(_config(), tmp_path, TARGET) + assert n1 == 2 and d1 == TARGET.isoformat() + assert urls[0].endswith("/flow") and "since" not in urls[0] # 首拉全量 + + n2, _ = await ext_pull.fetch_and_ingest(_config(), tmp_path, TARGET) + assert "since=1000" in urls[1] # 游标 = 本地最大 ts + assert n2 == 3 # 返回合并后分区总行数: 2 存量 + 1 增量 + + part = ext_pull._day_partition(tmp_path, "flow", TARGET) + df = pl.read_parquet(part) + assert df.sort("ts")["ts"].to_list() == ["0930", "1000", "1030"] + assert df.height == 3 # 不同盘并存, 同盘去重 + + +async def test_partition_deleted_resumes_full(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + """删分区修复: 游标随分区消失, 自动退回整日全量 (自愈路径)。""" + mode = {"full": False} + urls = _install_transport(monkeypatch, lambda req: httpx.Response( + 200, json=_rows("1000", "1030") if mode["full"] else _rows("0930"), + )) + cfg = _config() + await ext_pull.fetch_and_ingest(cfg, tmp_path, TARGET) + + part = ext_pull._day_partition(tmp_path, "flow", TARGET) + part.unlink() + mode["full"] = True # 换挡: 删分区后的下一次拉取走全量响应 + n, _ = await ext_pull.fetch_and_ingest(cfg, tmp_path, TARGET) + assert n == 2 + assert "since" not in urls[1] # 分区没了 → 不带游标 → 整日全量 + assert pl.read_parquet(part).height == 2 # 全量重建, 不残留旧盘 + + +# ---------------------------------------------------------------- 与其他机制共存 +async def test_since_with_date_param_and_pagination(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + """since 与 date_param / 分页参数共存: 每页都带 since + 页码。""" + cfg = _config( + url="https://example.test/day", + date_param="day", + page_param="page", + page_size_param="pageSize", + page_size=1, + ) + seen: list[dict | None] = [] + + async def fake(pull, config_id, day=None, extra_params=None): + seen.append(dict(extra_params) if extra_params else None) + page = int((extra_params or {}).get("page", 1)) + return _rows("0930") if page == 1 else {"data": []} + + monkeypatch.setattr(ext_pull, "_request_json", fake) + + part = ext_pull._day_partition(tmp_path, "flow", TARGET) + part.parent.mkdir(parents=True, exist_ok=True) + pl.DataFrame({"symbol": ["000001.SZ", "000002.SZ"], "ts": ["0900", "0900"]}).write_parquet(part) + + n, _ = await ext_pull.fetch_and_ingest(cfg, tmp_path, TARGET) + assert n == 1 + assert seen[0] == {"since": "0900", "page": "1", "pageSize": "1"} # 每页携带 + + +async def test_backfill_path_never_sends_cursor(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + """历史回补不传 data_dir → 恒全量 (回补只拉缺失分区, 游标无意义)。""" + part = ext_pull._day_partition(tmp_path, "flow", TARGET) + part.parent.mkdir(parents=True, exist_ok=True) + pl.DataFrame({"symbol": ["000001.SZ"], "ts": ["0900"]}).write_parquet(part) + + seen: list[dict | None] = [] + + async def fake(pull, config_id, day=None, extra_params=None): + seen.append(dict(extra_params) if extra_params else None) + return _rows("0930") + + monkeypatch.setattr(ext_pull, "_request_json", fake) + rows = await ext_pull.fetch_rows_for_date(_config(), TARGET) # 公开入口 (回补同款) + assert len(rows) == 1 and seen == [None] # 本地有数据也不带游标 + + +# ---------------------------------------------------------------- API 参数校验 +def test_api_request_model_validates_since_param() -> None: + from app.api.ext_data import PullConfigReq + + assert PullConfigReq(url="https://x", since_param="since").since_param == "since" + with pytest.raises(ValueError): + PullConfigReq(url="https://x", since_param="bad-name") # 参数名只允许字母数字下划线 diff --git a/backend/tests/test_ext_pull_time_window.py b/backend/tests/test_ext_pull_time_window.py index 0a1f908b..73ddea72 100644 --- a/backend/tests/test_ext_pull_time_window.py +++ b/backend/tests/test_ext_pull_time_window.py @@ -66,7 +66,7 @@ async def test_default_landing_date_is_the_beijing_date(at_beijing, monkeypatch) at_beijing(datetime(2026, 3, 2, 0, 30, tzinfo=CN_TZ)) seen: dict[str, date] = {} - async def fake_fetch(config, target_date): + async def fake_fetch(config, target_date, data_dir=None): seen["fetched"] = target_date return [{"symbol": "000001.SZ", "v": 1}] @@ -90,7 +90,7 @@ async def test_an_explicit_target_date_still_wins(at_beijing, monkeypatch) -> No at_beijing(datetime(2026, 3, 2, 10, 0, tzinfo=CN_TZ)) seen: dict[str, date] = {} - async def fake_fetch(config, target_date): + async def fake_fetch(config, target_date, data_dir=None): seen["fetched"] = target_date return [{"symbol": "000001.SZ", "v": 1}] diff --git a/backend/tests/test_ext_time_field.py b/backend/tests/test_ext_time_field.py index aa2c93c2..e131ec33 100644 --- a/backend/tests/test_ext_time_field.py +++ b/backend/tests/test_ext_time_field.py @@ -4,9 +4,11 @@ from __future__ import annotations from datetime import date from app.services.ext_data import ( + PLATE_TS_FIELD, ExtConfig, ExtField, PullConfig, + collapse_ext_rows_at, rows_to_parquet, ) @@ -82,3 +84,109 @@ def test_rows_to_parquet_without_time_field_unchanged(tmp_path): ) assert out.height == 1 assert out.row(0, named=True)["v"] == 2.0 + + +# ---------------------------------------------------------------- __pull_ts__ 注入 +def test_plate_ts_field_disables_since_param(): + """__pull_ts__ 是本地注入的拉取时刻, 源无法按它过滤 → since_param 归一失效。""" + p = PullConfig(url="https://x", time_field=PLATE_TS_FIELD, since_param="since") + assert p.time_field == PLATE_TS_FIELD + assert p.since_param is None + # 往返不丢 time_field 本身 + assert PullConfig.from_dict(p.to_dict()).time_field == PLATE_TS_FIELD + + +def test_rows_to_parquet_pull_ts_injection(tmp_path, monkeypatch): + """无时间列的每日快照表配 __pull_ts__: 盘中每次拉取各存一盘; 同秒重拉覆盖。""" + import app.services.ext_data as ext_data_module + + cfg = _config(time_field=PLATE_TS_FIELD) + day = date(2026, 9, 15) + clock = {"now": "2026-09-15T09:30:00+08:00"} + monkeypatch.setattr(ext_data_module, "_pull_ts_value", lambda: clock["now"]) + + first = [{"symbol": "000001.SZ", "ts": "x", "v": 1.0}] + assert rows_to_parquet(first, cfg, tmp_path, snapshot_date=day) == 1 + + # 同一秒内重拉: [symbol, __pull_ts__] 同键 → keep=last 覆盖, 不追加盘 + assert rows_to_parquet( + [{"symbol": "000001.SZ", "ts": "x", "v": 1.5}], cfg, tmp_path, snapshot_date=day + ) == 1 + + # 不同时刻再拉: 新的一盘追加 + clock["now"] = "2026-09-15T14:00:00+08:00" + assert rows_to_parquet( + [{"symbol": "000001.SZ", "ts": "x", "v": 2.0}], cfg, tmp_path, snapshot_date=day + ) == 2 + + import polars as pl + + out = pl.read_parquet( + tmp_path / "ext_data" / "auction" / "timeseries" + / f"date={day.isoformat()}" / "part.parquet" + ).sort(PLATE_TS_FIELD) + assert out.height == 2 + assert [r["v"] for r in out.to_dicts()] == [1.5, 2.0] # 第一盘被同秒重拉修正为 1.5 + + # collapse 语义与真实时间列一致: @时刻选盘 / 缺省末盘 + at_13 = collapse_ext_rows_at(out, "13:00", PLATE_TS_FIELD) + assert at_13.height == 1 and at_13.row(0, named=True)["v"] == 1.5 + last = collapse_ext_rows_at(out, None, PLATE_TS_FIELD) + assert last.height == 1 and last.row(0, named=True)["v"] == 2.0 + + +def test_plate_stats(tmp_path): + """盘次概览: 多盘表返回 {date, plates, from, to}; 非多盘表/无分区为 None。""" + from app.api.ext_data import _plate_stats + + cfg = _config(time_field="ts") + day = date(2026, 9, 15) + rows_to_parquet( + [ + {"symbol": "000001.SZ", "ts": "2026-09-15T09:15:03+08:00", "v": 1.0}, + {"symbol": "000001.SZ", "ts": "2026-09-15T09:25:05+08:00", "v": 2.0}, + {"symbol": "600519.SH", "ts": "2026-09-15T09:15:03+08:00", "v": 3.0}, + ], + cfg, tmp_path, snapshot_date=day, + ) + assert _plate_stats(cfg, tmp_path) == { + "date": "2026-09-15", "plates": 2, "from": "09:15:03", "to": "09:25:05", + } + # 未配 time_field 的每日快照表 → None + assert _plate_stats(_config(time_field=None), tmp_path) is None + # 多盘表但无任何分区 → None + assert _plate_stats(_config(time_field="ts"), tmp_path / "empty") is None + + +def test_list_configs_plate_field(tmp_path): + """列表端点为多盘表附带 plate 概览, 非多盘表为 None (源卡片展示契约)。""" + from types import SimpleNamespace + + from app.api import ext_data as ext_api + from app.services.ext_data import ExtConfigStore + + day = date(2026, 9, 15) + multi = _config(time_field="ts") + single = _config(time_field=None) + single.id = "pop" + rows_to_parquet( + [ + {"symbol": "000001.SZ", "ts": "09:15:03", "v": 1.0}, + {"symbol": "000001.SZ", "ts": "09:25:05", "v": 2.0}, + ], + multi, tmp_path, snapshot_date=day, + ) + rows_to_parquet( + [{"symbol": "000001.SZ", "ts": "09:25:05", "v": 2.0}], + single, tmp_path, snapshot_date=day, + ) + # rows_to_parquet 按传入 config 入库; store 注册两份配置供 load_all + ExtConfigStore(tmp_path).upsert(multi) + ExtConfigStore(tmp_path).upsert(single) + + state = SimpleNamespace(repo=SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path))) + items = {c["id"]: c for c in ext_api.list_configs(SimpleNamespace(app=SimpleNamespace(state=state)))["items"]} + assert items["auction"]["plate"] == { + "date": "2026-09-15", "plates": 2, "from": "09:15:03", "to": "09:25:05", + } + assert items["pop"]["plate"] is None diff --git a/docs/features.md b/docs/features.md index 1da7673e..37df2eb7 100644 --- a/docs/features.md +++ b/docs/features.md @@ -214,6 +214,16 @@ timeseries 模式的表 (如人气排行) 默认只能从开启拉取之日起 防污染契约: 若接口忽略日期参数返回当日数据 (以响应行 `date` 字段为准校验), 该日**拒绝写入**并计入失败清单 —— 否则当日值会静默污染整个历史时序。接口侧需提供: 传日期参数返回该日数据、无数据返回空数组或 404 (均视为该日无数据, 计入 empty 跳过)、每行带 `date` 字段;不传参数保持当日 (向后兼容)。 +##### 日内多盘表增量续传 + +盘中多盘表 (配置**时间字段**`time_field` 后同标的允许多行, 按标的+时刻去重) 若每次全量重拉, 越到收盘越重。拉取配置可声明**增量参数名**(`since_param`, 如 `since`): 每次拉取自动带上本地目标日分区时间列的最大值 `?since=<值>`, 源只返回其后数据 —— 缺口感知天然成立, 本地缺多少源补多少。本地无该日数据 (首拉/新交易日/分区被删修复) 时不带参数即整日全量, 回退方向永远安全; 分页拉取下 `since` 作为过滤器与页码参数一起每页携带; 两级拉取 (fanout) 不适用。接口侧契约: 传参数返回该值之后的行、无新数据返回空数组。 + +##### 快照表多盘化 (拉取时刻注入) + +源数据没有时间列的每日快照表 (人气榜、资金流等) 同样可以多盘: **时间字段**填保留名 `__pull_ts__`, 写入时自动注入拉取时刻 (北京时间 ISO), 去重键变为 [标的, 拉取时刻] —— 盘中每次拉取各存一盘, 读取侧 `@时刻` 选盘与末盘收敛语义和真实时间列完全一致。注入列是本地时刻, `since_param` 对它自动失效 (源无法按我们的拉取时刻过滤); 历史回补一天一拉, 注入回补时刻后每标的恰一行, 与单盘行为等价; 同一秒内重复写入按末次覆盖不追加盘。 + +扩展数据页源卡片为多盘表显示「多盘」徽章与最新分区盘次概览 (盘数 · 时刻范围); 列表扩展列可按 折线/柱状/文字@时刻 三种形态展示盘次序列 (见选股列配置)。 + ##### 分页拉取协议 数据量大的接口 (全市场排行、批量快照) 单次响应装不下时, 拉取配置可声明分页 (仅 GET): **页码参数名**(`page_param`, 如 `page`)、**每页条数参数名**(`page_size_param`, 如 `pageSize`)与**每页条数**(`page_size`)。配置后单次拉取按页循环请求并合并结果, 空页/不足一页自动判停; **最多页数**(`max_pages`, 默认 20, 上限 200)防止接口永远返回数据拖死循环。POST body 分页不支持 (请求体是固定模板)。 diff --git a/frontend/src/components/ext-data/ExtDataPullPanel.tsx b/frontend/src/components/ext-data/ExtDataPullPanel.tsx index 46c74fd7..34116b0e 100644 --- a/frontend/src/components/ext-data/ExtDataPullPanel.tsx +++ b/frontend/src/components/ext-data/ExtDataPullPanel.tsx @@ -82,6 +82,7 @@ export function ExtDataPullPanel({ config, onSaved }: { date_param: pull.date_param ?? null, date_format: pull.date_format, time_field: pull.time_field ?? null, + since_param: pull.since_param ?? null, timeout_seconds: pull.timeout_seconds, page_param: pull.page_param ?? null, page_size_param: pull.page_size_param ?? null, @@ -267,6 +268,7 @@ function ExtPullConfigDialog({ config, onSaved, onClose }: { const [dateParam, setDateParam] = useState(pull?.date_param ?? '') const [dateFormat, setDateFormat] = useState(pull?.date_format ?? 'iso') const [timeField, setTimeField] = useState(pull?.time_field ?? '') + const [sinceParam, setSinceParam] = useState(pull?.since_param ?? '') const [timeoutSec, setTimeoutSec] = useState(pull?.timeout_seconds ?? 30) // 分页协议 (仅 GET): 页码参数名配置后按页循环拉取 const [pageParam, setPageParam] = useState(pull?.page_param ?? '') @@ -342,6 +344,7 @@ function ExtPullConfigDialog({ config, onSaved, onClose }: { date_param: dateParam.trim() || null, date_format: dateFormat, time_field: timeField.trim() || null, + since_param: sinceParam.trim() || null, timeout_seconds: effTimeoutSec, page_param: pageParam.trim() || null, page_size_param: pageSizeParam.trim() || null, @@ -798,12 +801,12 @@ function ExtPullConfigDialog({ config, onSaved, onClose }: {
时间字段 - 日内多行数据填，如 ts + 日内多行数据填，如 ts；无时间列填 __pull_ts__
setTimeField(e.target.value)} - placeholder="ts · 留空=每日快照表" - title="配置后同一代码允许一天多行, 按代码+时间列去重" + placeholder="ts / __pull_ts__ · 留空=每日快照表" + title="配置后同一代码允许一天多行, 按代码+时间列去重; 填 __pull_ts__ 时自动注入每次拉取的时刻, 盘中多次拉取各存一盘" className={`${INPUT_CLS} font-mono`} />
@@ -844,6 +847,22 @@ function ExtPullConfigDialog({ config, onSaved, onClose }: { +
+
+ 增量参数名 + 盘中多盘表填，如 since · 免每次全量重拉 +
+ setSinceParam(e.target.value)} + placeholder="since · 留空=每次全量" + disabled={!timeField.trim()} + title="每次拉取自动带上本地已有最大时间值 (?参数=值), 源只返回其后数据; 缺多少补多少, 本地无数据时自动整日全量。需先配置时间字段; 两级拉取不适用" + className="h-9 w-full rounded-lg border border-border bg-base px-3 text-xs font-mono text-foreground placeholder:text-muted/40 focus:outline-none focus:ring-1 focus:ring-accent/25 focus:border-accent/50 transition-shadow disabled:opacity-40" + /> + {!timeField.trim() && ( +
需先填「时间字段」才能增量续传
+ )} +
分页（仅 GET） diff --git a/frontend/src/components/ext-data/ExtDataStatCard.tsx b/frontend/src/components/ext-data/ExtDataStatCard.tsx index 820734d4..f32d9058 100644 --- a/frontend/src/components/ext-data/ExtDataStatCard.tsx +++ b/frontend/src/components/ext-data/ExtDataStatCard.tsx @@ -26,17 +26,50 @@ export function ExtDataStatCard({ config, onDelete, deleting, onEdit }: { const [fieldsExpanded, setFieldsExpanded] = useState(false) const fieldsRef = useRef(null) const [fieldsOverflow, setFieldsOverflow] = useState(false) + const [collapsedH, setCollapsedH] = useState(48) useEffect(() => { const el = fieldsRef.current if (!el) return - const prev = el.style.maxHeight - el.style.maxHeight = 'none' - const full = el.scrollHeight - el.style.maxHeight = prev - setFieldsOverflow(full > 68) + const chip = el.firstElementChild as HTMLElement | null + // 折叠高度按整行 chip 对齐 (行高 = chip 高 + gap-1), 只露完整行, 不拦腰截断 + const rowH = (chip ? chip.offsetHeight : 20) + 4 + const h = 2 * rowH - 4 + setCollapsedH(h) + setFieldsOverflow(el.scrollHeight > h + 1) }, [config.fields]) + // 描述固定两行: 内嵌引擎有最小字号 (text-[10px] 实际行距 ≈17px 而非 16.25), + // -webkit-line-clamp 与 px 硬编码都不可靠 — 实测前两行文本的包络高度定容器高 + const descRef = useRef(null) + const [descMaxH, setDescMaxH] = useState(undefined) + useEffect(() => { + const el = descRef.current + if (!el) return + const range = document.createRange() + range.selectNodeContents(el) + // 文本矩形按 top 聚类成行 (同行内中英文片段 top 差 < 8px) + const lines: Array<{ top: number; bottom: number }> = [] + for (const r of [...range.getClientRects()].sort((a, b) => a.top - b.top)) { + const last = lines[lines.length - 1] + if (last && r.top - last.top < 8) { + last.top = Math.min(last.top, r.top) + last.bottom = Math.max(last.bottom, r.bottom) + } else { + lines.push({ top: r.top, bottom: r.bottom }) + } + } + if (lines.length < 3) { + setDescMaxH(undefined) // 本就不足三行, 不设限 + return + } + // 裁剪线定在第 3 行顶之上: 无论行距/行盒是否重叠, 第三行笔画一个像素都进不了盒内。 + // getBoundingClientRect 受页面缩放 (实测 ~1.1×) 影响, 须除回 CSS 像素再设置高度 + const r = el.getBoundingClientRect() + const zoom = r.height / el.clientHeight || 1 + setDescMaxH(Math.max(24, Math.ceil((lines[2].top - r.top - 0.5) / zoom))) + }, [config.description]) + const upload = useMutation({ mutationFn: ({ id, file }: { id: string; file: File }) => api.extDataUpload(id, file), onMutate: () => { setUploading(true); setUploadResult(null) }, @@ -64,37 +97,66 @@ export function ExtDataStatCard({ config, onDelete, deleting, onEdit }: { if (file) doUpload(file) } + // 模式主色: 快照=蓝 / 时序=琥珀 (延续历史语义)。展示为三处层次而非整卡平涂: + // 染色边框 + 顶部发光边线 + 自主色到卡底的纵向渐变 — 有色而不闹 + const accent = config.mode === 'snapshot' ? { + border: 'border-blue-400/25 hover:border-blue-400/45', + bg: 'bg-gradient-to-b from-blue-400/[0.07] via-surface to-surface', + line: 'via-blue-400/60', + } : { + border: 'border-amber-400/25 hover:border-amber-400/45', + bg: 'bg-gradient-to-b from-amber-400/[0.07] via-surface to-surface', + line: 'via-amber-400/60', + } + return ( -
-
-

{config.label}

-
- - {config.mode === 'snapshot' ? '快照' : '时序'} - - +
+
+
+
+

+ {config.label} +

+
+ + {config.mode === 'snapshot' ? '快照' : '时序'} + + {config.mode === 'timeseries' && config.pull?.time_field && ( + + 多盘 + + )} +
+
-
+ {/* 描述固定两行: 行距由运行时实测 (见上方 descMaxH), 52px 为测量前的兜底 */} +
{config.description || `扩展数据 · ${config.mode === 'snapshot' ? '与标的 JOIN' : '与日K JOIN'}`}
-
-
+
+
f.name !== 'symbol' && f.name !== 'code') .map((f) => ( - - - {f.label || f.name} + + {f.label || f.name} )) }
- {fieldsOverflow && !fieldsExpanded && ( -
setFieldsExpanded(true)} - > - -
- )} {fieldsExpanded && fieldsOverflow && ( )}
+ {fieldsOverflow && !fieldsExpanded && ( + + )}
-
-
- 标识 - {config.id} +
+
+ 标识 + {config.id}
-
- 最新 - {config.latest_sync_date ?? '—'} +
+ 最新 + + {config.latest_sync_date ?? '—'} +
+ {config.plate && ( +
+ 盘次 + + {config.plate.plates} 盘 · {config.plate.from.slice(0, 5)}–{config.plate.to.slice(0, 5)} + +
+ )}
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index c83c1d00..b557edfc 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -3938,6 +3938,7 @@ export const api = { time_window_start?: string | null; time_window_end?: string | null; date_param?: string | null; time_field?: string | null; + since_param?: string | null; auth?: ExtPullAuth; timeout_seconds?: number; page_param?: string | null; @@ -4874,6 +4875,8 @@ export interface PullConfig { date_format?: string /** 日内序列表时间列名 (如 "ts"): 配置后同 symbol 允许多行 (按 symbol+时间列去重), 用于集合竞价等多盘数据 */ time_field?: string | null + /** 增量续传参数名 (如 "since"): 每次拉取自动带上本地目标日分区时间列最大值, 源只返回其后数据; 需先配置 time_field */ + since_param?: string | null auth?: ExtPullAuth | null /** 单次拉取请求超时 (秒), 默认 30 */ timeout_seconds?: number @@ -4966,6 +4969,8 @@ export interface ExtDataConfig { updated_at: string latest_sync_date?: string | null date_range?: string[] | null + /** 日内多盘表最新分区的盘次概览; 非多盘表为 null */ + plate?: { date: string; plates: number; from: string; to: string } | null pull?: PullConfig | null }