feat(ext): 盘中多盘表增量续传 — pull.since_param 游标免全量重拉

拉取前读本地目标日分区 time_field 列最大值, 以 ?{since_param}={值} 附到
请求, 源只返回新于本地的盘次; 缺口感知天然成立 (本地缺多少源补多少)。
未配置/首拉/分区缺失/schema 漂移/文件损坏一律不带参数回退整日全量,
回退方向永远安全; 历史回补与 fanout 两级拉取不注入游标。

- PullConfig 归一: 未配置 time_field 时 since_param 失效
- 分页拉取每页携带 since 过滤器
- API since_param 参数名合法性校验
- 前端拉取面板/统计卡片与 features.md 同步
This commit is contained in:
shy3130
2026-10-26 22:10:39 +08:00
parent c932098ae7
commit 5909155fb1
10 changed files with 590 additions and 62 deletions
+52
View File
@@ -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,
+29 -3
View File
@@ -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(
+41 -6
View File
@@ -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(
+190
View File
@@ -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") # 参数名只允许字母数字下划线
+2 -2
View File
@@ -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}]
+108
View File
@@ -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
+10
View File
@@ -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 分页不支持 (请求体是固定模板)。
@@ -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 }: {
<div>
<div className="text-[11px] font-medium text-secondary mb-1.5">
时间字段
<span className="ml-1.5 text-[10px] font-normal text-muted">日内多行数据填,如 ts</span>
<span className="ml-1.5 text-[10px] font-normal text-muted">日内多行数据填,如 ts;无时间列填 __pull_ts__</span>
</div>
<input
value={timeField} onChange={e => setTimeField(e.target.value)}
placeholder="ts · 留空=每日快照表"
title="配置后同一代码允许一天多行, 按代码+时间列去重"
placeholder="ts / __pull_ts__ · 留空=每日快照表"
title="配置后同一代码允许一天多行, 按代码+时间列去重; 填 __pull_ts__ 时自动注入每次拉取的时刻, 盘中多次拉取各存一盘"
className={`${INPUT_CLS} font-mono`}
/>
</div>
@@ -844,6 +847,22 @@ function ExtPullConfigDialog({ config, onSaved, onClose }: {
</select>
</div>
</div>
<div>
<div className="text-[11px] font-medium text-secondary mb-1.5">
增量参数名
<span className="ml-1.5 text-[10px] font-normal text-muted">盘中多盘表填,如 since · 免每次全量重拉</span>
</div>
<input
value={sinceParam} onChange={e => 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() && (
<div className="mt-1 text-[10px] text-muted/70">需先填「时间字段」才能增量续传</div>
)}
</div>
<div>
<div className="text-[11px] font-medium text-secondary mb-1.5">
分页(仅 GET)
@@ -26,17 +26,50 @@ export function ExtDataStatCard({ config, onDelete, deleting, onEdit }: {
const [fieldsExpanded, setFieldsExpanded] = useState(false)
const fieldsRef = useRef<HTMLDivElement>(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<HTMLDivElement>(null)
const [descMaxH, setDescMaxH] = useState<number | undefined>(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 (
<div className={`rounded-card border flex flex-col transition-all duration-300 ${
config.mode === 'snapshot'
? 'border-blue-500/30 bg-blue-500/[0.03]'
: 'border-amber-500/30 bg-amber-500/[0.03]'
}`}>
<div className="flex items-center justify-between px-4 pt-4 pb-2">
<h3 className="text-sm font-medium text-foreground">{config.label}</h3>
<div className="flex items-center gap-1.5">
<span className={`text-[10px] px-1.5 py-px rounded font-medium ${
config.mode === 'snapshot'
? 'bg-blue-500/10 text-blue-400'
: 'bg-amber-500/10 text-amber-400'
}`}>
{config.mode === 'snapshot' ? '快照' : '时序'}
</span>
<button
onClick={() => setSettingsOpen(v => !v)}
className="p-0.5 rounded hover:bg-elevated transition-colors text-secondary"
>
<Settings className="h-3.5 w-3.5" />
</button>
<div className={`rounded-card border ${accent.border} ${accent.bg} relative flex flex-col transition-colors duration-300`}>
<div className={`pointer-events-none absolute inset-x-3 top-0 h-px bg-gradient-to-r from-transparent ${accent.line} to-transparent`} />
<div className="flex items-start justify-between gap-2 px-4 pt-4">
<div className="min-w-0 flex-1">
<h3 className="text-sm font-medium text-foreground truncate" title={config.label}>
{config.label}
</h3>
<div className="flex items-center gap-1.5 mt-1.5">
<span className={`whitespace-nowrap text-[10px] leading-none font-medium px-1.5 py-1 rounded-md ${
config.mode === 'snapshot'
? 'bg-blue-500/10 text-blue-400'
: 'bg-amber-500/10 text-amber-400'
}`}>
{config.mode === 'snapshot' ? '快照' : '时序'}
</span>
{config.mode === 'timeseries' && config.pull?.time_field && (
<span
className="whitespace-nowrap text-[10px] leading-none font-medium px-1.5 py-1 rounded-md bg-emerald-500/10 text-emerald-500 dark:text-emerald-400"
title={`日内多盘 · 时间列 ${config.pull.time_field}`}
>
多盘
</span>
)}
</div>
</div>
<button
onClick={() => setSettingsOpen(v => !v)}
className="shrink-0 p-1.5 rounded-md hover:bg-elevated/70 transition-colors text-muted hover:text-secondary"
>
<Settings className="h-3.5 w-3.5" />
</button>
</div>
<div className="px-4 pb-1.5 text-[10px] text-muted/70 leading-relaxed line-clamp-2">
{/* 描述固定两行: 行距由运行时实测 (见上方 descMaxH), 52px 为测量前的兜底 */}
<div
ref={descRef}
style={descMaxH !== undefined ? { maxHeight: descMaxH } : undefined}
className="px-4 pt-2 pb-1.5 text-[10px] text-muted leading-relaxed max-h-[52px] shrink-0 overflow-hidden break-words"
>
{config.description || `扩展数据 · ${config.mode === 'snapshot' ? '与标的 JOIN' : '与日K JOIN'}`}
</div>
<div className="flex-1 min-h-[81px] px-4 pb-2">
<div className={`relative overflow-hidden transition-all duration-300 ${fieldsExpanded ? '' : 'h-[75px]'}`}>
<div className="flex-1 min-h-0 px-4 pb-2">
<div
className="relative overflow-hidden transition-[height] duration-300"
style={fieldsExpanded || !fieldsOverflow ? undefined : { height: collapsedH }}
>
<div
ref={fieldsRef}
className="flex flex-wrap gap-1"
@@ -102,41 +164,62 @@ export function ExtDataStatCard({ config, onDelete, deleting, onEdit }: {
{config.fields
.filter(f => f.name !== 'symbol' && f.name !== 'code')
.map((f) => (
<span key={f.name} className="inline-flex items-center gap-1 text-[10px] bg-elevated rounded px-1.5 py-0.5">
<Tag className="h-2.5 w-2.5 text-muted" />
<span className="text-secondary">{f.label || f.name}</span>
<span
key={f.name}
className="text-[10px] leading-none py-1 px-1.5 rounded-md bg-elevated/70 border border-border/40 text-secondary"
title={f.name}
>
{f.label || f.name}
</span>
))
}
</div>
{fieldsOverflow && !fieldsExpanded && (
<div
className="absolute bottom-0 inset-x-0 h-6 bg-gradient-to-t from-surface to-transparent cursor-pointer flex items-end justify-center"
onClick={() => setFieldsExpanded(true)}
>
<ChevronDown className="h-3 w-3 text-muted" />
</div>
)}
{fieldsExpanded && fieldsOverflow && (
<button
onClick={() => setFieldsExpanded(false)}
className="w-full flex items-center justify-center pt-1 text-[10px] text-muted hover:text-secondary transition-colors"
className="w-full flex items-center justify-center pt-1.5 text-muted hover:text-secondary transition-colors"
title="收起字段"
>
<ChevronUp className="h-3 w-3" />
</button>
)}
</div>
{fieldsOverflow && !fieldsExpanded && (
<button
onClick={() => setFieldsExpanded(true)}
className="mx-auto mt-1 flex items-center rounded-full border border-border/60 bg-elevated/60 px-2.5 py-0.5 text-muted hover:text-secondary hover:border-border transition-colors"
title="展开全部字段"
>
<ChevronDown className="h-3 w-3" />
</button>
)}
</div>
<div className="mt-auto px-4 pb-4 pt-2 border-t border-border/50">
<div className="flex justify-between text-[11px]">
<span className="text-muted">标识</span>
<span className="font-mono text-secondary">{config.id}</span>
<div className="mt-auto px-4 pb-4 pt-2.5 border-t border-border/60 space-y-1 text-[11px]">
<div className="flex items-baseline justify-between gap-3">
<span className="text-muted shrink-0">标识</span>
<span className="font-mono text-secondary/80 truncate" title={config.id}>{config.id}</span>
</div>
<div className="flex justify-between text-[11px] mt-1">
<span className="text-muted">最新</span>
<span className="text-secondary">{config.latest_sync_date ?? '—'}</span>
<div className="flex items-baseline justify-between gap-3">
<span className="text-muted shrink-0">最新</span>
<span
className="text-secondary/80 tabular-nums truncate"
title={config.latest_sync_date ?? undefined}
>
{config.latest_sync_date ?? '—'}
</span>
</div>
{config.plate && (
<div className="flex items-baseline justify-between gap-3">
<span className="text-muted shrink-0">盘次</span>
<span
className="font-mono text-secondary/80 tabular-nums whitespace-nowrap"
title={`${config.plate.date} 多盘保留, @时刻可选盘`}
>
{config.plate.plates} 盘 · {config.plate.from.slice(0, 5)}–{config.plate.to.slice(0, 5)}
</span>
</div>
)}
</div>
<AnimatePresence>
+5
View File
@@ -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
}