fix(ext-data): 快照接口误拒写 + schema 类型丢失 + 分页配置静默丢弃

用户反馈资金流扩展表点「手动拉取」报错、板块轮动的资金流下拉为空。
排查出三个独立缺陷:

1. 日期一致性校验用错判据 (ext_pull.py)
   报错: 「接口返回的日期与请求日期不一致, 已拒绝写入该分区」。
   _assert_rows_date 无条件校验响应 date 必须等于请求日期, 但该契约的
   目的只是防「接口忽略 ?date= 却把当日值写进历史分区」, 仅对按日查询
   接口有意义。资金流/人气等当日快照接口只有最新一个交易日的数据,
   周末/节假日/数据源 T+1 时响应日期本就落后于请求日期, 被误判成
   「接口忽略日期参数」而整表拒写。
   修复: 新增 _assert_rows_date_if_dated, 仅在 pull.date_param 已配置
   时校验; 历史回补的 fail-closed 防污染契约不变。

2. schema 端点一直抛异常走兜底 (api/ext_data.py)
   「SELECT column_name, data_type FROM (DESCRIBE ...)」在 DuckDB 1.5
   抛 BinderException —— 该版本输出列名是 column_type (1.4 及以前才是
   data_type)。异常被吞后回退到配置字段, 那些对象只有 dtype 没有 type,
   前端按 type 判断数值列的资金流下拉因此永远为空。
   修复: 按实际输出列名取值, 两个版本都可用; 兜底分支键名统一为 type。
   同时关掉 hive 分区推断 —— date=YYYY-MM-DD 目录会让 DuckDB 凭空多出
   一个 date 列, 污染列选择器。
   前端 NUMERIC_TYPES 由写死的 5 个简写改为类型族匹配, 覆盖 DOUBLE /
   BIGINT / DECIMAL 等 Parquet 真实类型名。

3. 分页配置保存后被静默丢弃 (api/ext_data.py)
   PullConfigReq 缺少 page_param / page_size_param / page_size /
   page_start / max_pages, 前端「拉取设置」提交这些值后 Pydantic 直接
   忽略, 保存后再打开退回默认的单次请求, 且不报错。
   修复: 补齐请求模型字段并透传到 PullConfig。

验证: 三处均先写复现用例 (未修复代码上 7 个用例失败), 修复后全绿;
扩展相关后端测试 202 passed; 前端 SectorRotationCard 7 passed;
pnpm build 通过; 真实接口验证 —— 修复前 400「日期不一致」, 修复后
200 写入 5580 行, schema-all 返回 DOUBLE/BIGINT/VARCHAR 真实类型。

Signed-off-by: shy3130 <shy3130@qq.com>
This commit is contained in:
shy3130
2026-10-24 22:52:40 +08:00
parent 7812c02ec4
commit a40c0780d7
7 changed files with 273 additions and 17 deletions
+49 -13
View File
@@ -119,6 +119,14 @@ class PullConfigReq(BaseModel):
auth: PullAuthReq | None = None
# 单次拉取请求超时 (秒), 默认 30 与历史行为一致; 大响应接口可调高
timeout_seconds: int = Field(30, ge=5, le=300)
# ── 分页协议 (仅 GET; 页码经 query 参数传递) ──
# 必须与本类其它字段一样显式声明: Pydantic 默认忽略未声明键, 前端保存时
# 提交的 page_param/page_size_* 会被静默丢弃, 回读退化成单次请求。
page_param: str | None = Field(None, max_length=64)
page_size_param: str | None = Field(None, max_length=64)
page_size: int = Field(0, ge=0)
page_start: int = Field(1, ge=0)
max_pages: int = Field(20, ge=1, le=200)
# 两级拉取配方 (目录 → 逐条目明细); 缺省 (None) = 保留现有配置
fanout: dict | None = None
@@ -1051,6 +1059,11 @@ def configure_pull(request: Request, config_id: str, body: PullConfigReq):
time_field=body.time_field,
auth=new_auth,
timeout_seconds=body.timeout_seconds,
page_param=body.page_param,
page_size_param=body.page_size_param,
page_size=body.page_size,
page_start=body.page_start,
max_pages=body.max_pages,
fanout=body.fanout if body.fanout is not None else (old_pull.fanout if old_pull else None),
last_run=old_pull.last_run if old_pull else None,
last_status=old_pull.last_status if old_pull else None,
@@ -1421,6 +1434,27 @@ async def detect_url(body: DetectUrlReq):
}
def _describe_parquet_columns(glob: str) -> list[tuple[str, str]]:
"""读取 Parquet 的 (列名, 类型), 类型用 DuckDB 类型名 (DOUBLE/BIGINT/VARCHAR…)。
DESCRIBE 的输出列名随 DuckDB 版本变化: 1.4 及以前是 data_type, 1.5 起是
column_type。写死任一名字都会在另一版本整体抛 BinderException, 使 schema
端点静默退回配置字段 (列对象只有 dtype 没有 type), 前端按 type 判断数值列
的下拉随之变空。这里按实际输出列名取值, 两个版本都可用。
"""
import duckdb
# hive_partitioning=false: 时序表目录名形如 date=YYYY-MM-DD, DuckDB 默认据此
# 推断分区并在真实列之外多出一个 date 列, 污染列选择器里的扩展字段。
rel = duckdb.query(
f"DESCRIBE SELECT * FROM read_parquet('{glob}', union_by_name=true, hive_partitioning=false)"
)
names = rel.columns
type_idx = names.index("column_type" if "column_type" in names else "data_type")
name_idx = names.index("column_name")
return [(row[name_idx], row[type_idx]) for row in rel.fetchall()]
@router.get("/schema/{config_id}")
def discover_schema(request: Request, config_id: str):
"""发现扩展数据的实际 Parquet schema(基于已有数据)。"""
@@ -1432,14 +1466,15 @@ def discover_schema(request: Request, config_id: str):
glob = _parquet_glob(config, data_dir)
try:
import duckdb
rows = duckdb.query(
f"SELECT column_name, data_type FROM (DESCRIBE SELECT * FROM read_parquet('{glob}', union_by_name=true))"
).fetchall()
return {"columns": [{"name": r[0], "type": r[1]} for r in rows]}
return {"columns": [
{"name": name, "type": dtype}
for name, dtype in _describe_parquet_columns(glob)
]}
except Exception:
# 无数据时返回配置中定义的字段
return {"columns": [f.to_dict() for f in config.fields]}
# 无数据时返回配置中定义的字段 — 键名与真实 schema 对齐 (type 而非 dtype)
return {"columns": [
{"name": f.name, "type": f.dtype, "label": f.label} for f in config.fields
]}
@router.get("/schema-all")
@@ -1452,14 +1487,15 @@ def discover_all_schemas(request: Request):
glob = _parquet_glob(config, data_dir)
try:
import duckdb
cols = duckdb.query(
f"SELECT column_name, data_type FROM (DESCRIBE SELECT * FROM read_parquet('{glob}', union_by_name=true))"
).fetchall()
field_labels = {f.name: f.label for f in config.fields}
columns = [{"name": r[0], "type": r[1], "label": field_labels.get(r[0], r[0])} for r in cols]
columns = [
{"name": name, "type": dtype, "label": field_labels.get(name, name)}
for name, dtype in _describe_parquet_columns(glob)
]
except Exception:
columns = [f.to_dict() for f in config.fields]
columns = [
{"name": f.name, "type": f.dtype, "label": f.label} for f in config.fields
]
result.append({
"id": config.id,
+15 -2
View File
@@ -252,6 +252,8 @@ def _assert_rows_date(rows: list[dict], day: date) -> None:
服务端忽略日期参数返回当日数据时会静默把当日值写进历史分区,
造成整个时序口径错乱 —— 此处 fail-closed 拒绝 (实测确实有忽略
?date= 的接口)。date 字段缺省的接口不做校验。
仅在 pull.date_param 已配置时调用 (见 _assert_rows_date_if_dated)。
"""
want = day.isoformat()
for r in rows[:20]:
@@ -267,6 +269,17 @@ def _assert_rows_date(rows: list[dict], day: date) -> None:
)
def _assert_rows_date_if_dated(pull: PullConfig, rows: list[dict], day: date) -> None:
"""仅对「按日查询」接口做响应日期一致性校验 (pull.date_param 已配置)。
未配置 date_param 的接口只有当日快照, 响应日期本就可能落后于请求日期
(周末/节假日/数据源 T+1), 校验会把正常快照误判成「接口忽略日期参数」
而整表拒写。该契约的目的是防历史分区被当日值污染, 对快照接口无意义。
"""
if pull.date_param:
_assert_rows_date(rows, day)
async def _request_json(
pull: PullConfig, config_id: str, day: date | None = None,
extra_params: dict[str, str] | None = None,
@@ -388,7 +401,7 @@ async def _fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dic
# 校验可关联标的的字段:直接 symbol/code,或配置里声明的映射源列。
# 市场级表 (market_level, 行=全市场一条, 无标的列) 跳过该校验。
if config.market_level:
_assert_rows_date(rows, target_date)
_assert_rows_date_if_dated(pull, rows, target_date)
return rows
row_keys = set(rows[0]) if rows else set()
mapped_cols = {
@@ -399,7 +412,7 @@ async def _fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dic
if rows and not ({"symbol", "code"} & row_keys or mapped_cols & row_keys):
raise ValueError("数据行中缺少 symbol/code 字段,请配置字段映射或标的映射")
_assert_rows_date(rows, target_date)
_assert_rows_date_if_dated(pull, rows, target_date)
return rows
+43
View File
@@ -177,6 +177,49 @@ def test_assert_rows_date_contract():
_assert_rows_date([_row("A", "2026-09-06")], d) # 接口忽略参数返回当日
async def test_snapshot_api_without_date_param_accepts_previous_trading_day(fake_http):
"""当日快照接口 (未配 date_param) 在周末返回上一交易日数据, 不得拒写。
回归: 资金流/人气类接口只有当日快照, 周末/节假日/数据源 T+1 时响应日期
落后于请求日期是正常现象。无 date_param 却做日期一致性校验会把整表拉取
全部拒掉 (报「接口可能不支持日期参数」)。
"""
cfg = ExtConfig(
id="flow", label="资金流", mode="timeseries",
fields=[ExtField("symbol", "string"), ExtField("date", "string"),
ExtField("net", "float")],
pull=PullConfig(url="https://example.test/flow"), # date_param 未配置
)
# 请求 2026-10-24 (周六), 接口返回 2026-10-23 (周五) 的快照
fake_http.responses["https://example.test/flow"] = [
{"symbol": "600000.SH", "date": "2026-10-23", "net": 1.0},
]
rows = await fetch_rows_for_date(cfg, date(2026, 10, 24))
assert [r["symbol"] for r in rows] == ["600000.SH"]
async def test_dated_api_still_rejects_mismatched_date(fake_http):
"""配了 date_param 的接口保持 fail-closed: 历史分区防污染契约不能松。"""
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_market_level_snapshot_without_date_param_accepts_stale_date(fake_http):
"""市场级表同样按 date_param 判定, 不因 market_level 分支绕开修复。"""
cfg = ExtConfig(
id="regime", label="情绪", mode="timeseries",
fields=[ExtField("date", "string"), ExtField("score", "float")],
pull=PullConfig(url="https://example.test/regime"),
market_level=True,
)
fake_http.responses["https://example.test/regime"] = [
{"date": "2026-10-23", "score": 55.0},
]
rows = await fetch_rows_for_date(cfg, date(2026, 10, 24))
assert rows == [{"date": "2026-10-23", "score": 55.0}]
def test_pull_config_date_param_roundtrip():
p = PullConfig(url="u", date_param="date")
assert p.to_dict()["date_param"] == "date"
+25
View File
@@ -275,6 +275,31 @@ def test_configure_pull_omitted_auth_preserves_existing(monkeypatch, tmp_path) -
assert store.get("demo").pull.auth == {"type": "none", "header": "Authorization", "param": "token"}
def test_configure_pull_persists_pagination_fields(monkeypatch, tmp_path) -> None:
"""PUT /pull 必须保存分页设置 (回归: PullConfigReq 缺 page_* 字段时静默丢弃)。
前端「拉取设置」会提交 page_param/page_size_param/page_size/page_start/
max_pages; 请求模型漏声明这些键时 Pydantic 直接忽略, 保存后回读退化成
单次请求且不报错。
"""
monkeypatch.setattr("app.api.ext_data.pull_scheduler.refresh", lambda *a, **k: None)
store = _make_config(tmp_path)
req = _request(tmp_path)
configure_pull(req, "demo", PullConfigReq(
url="https://x.test/d",
page_param="page",
page_size_param="size",
page_size=100,
page_start=0,
max_pages=7,
))
pull = store.get("demo").pull
assert (pull.page_param, pull.page_size_param, pull.page_size) == ("page", "size", 100)
assert (pull.page_start, pull.max_pages) == (0, 7)
def test_delete_config_clears_residual_key(monkeypatch, tmp_path) -> None:
from app.api.ext_data import delete_config
+102
View File
@@ -0,0 +1,102 @@
"""扩展表 schema 端点 — 必须返回 Parquet 里的真实列类型。
回归: schema 端点用 `SELECT column_name, data_type FROM (DESCRIBE ...)` 取类型,
而 DuckDB 1.5 起 DESCRIBE 的输出列名是 `column_type`。写死旧名会让整个查询抛
BinderException, except 分支退回配置字段 —— 那些对象只有 `dtype` 没有 `type`,
前端按 `type` 判断数值列的扩展列/资金流下拉因此全部为空。
"""
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
import polars as pl
import pytest
from app.api import ext_data as ext_api
from app.services.ext_data import ExtConfig, ExtConfigStore, ExtField, PullConfig
def _config(mode: str = "timeseries") -> ExtConfig:
return ExtConfig(
id="flow", label="资金流", mode=mode,
fields=[
ExtField("symbol", "string", "标的代码"),
ExtField("net", "float", "主力净流入"),
ExtField("rank", "int", "排名"),
],
pull=PullConfig(url="https://example.test/flow"),
)
def _seed(tmp_path: Path, config: ExtConfig) -> None:
ExtConfigStore(tmp_path).upsert(config)
part = tmp_path / "ext_data" / config.id / "timeseries" / "date=2026-10-23"
part.mkdir(parents=True)
pl.DataFrame({
"symbol": ["600000.SH"],
"net": [1.5],
"rank": [3],
}).write_parquet(part / "part.parquet")
def _request(tmp_path: Path) -> SimpleNamespace:
state = SimpleNamespace(repo=SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)))
return SimpleNamespace(app=SimpleNamespace(state=state))
@pytest.fixture()
def seeded(tmp_path, monkeypatch):
config = _config()
_seed(tmp_path, config)
monkeypatch.setattr(ext_api, "_store", lambda request: ExtConfigStore(tmp_path))
monkeypatch.setattr(ext_api, "_data_dir", lambda request: tmp_path)
return tmp_path
def test_schema_returns_real_parquet_types(seeded):
"""有数据时返回 DuckDB 真实类型, 且列对象带 type 键 (前端按 type 判数值列)。"""
res = ext_api.discover_schema(_request(seeded), "flow")
types = {c["name"]: c["type"] for c in res["columns"]}
assert types["symbol"] == "VARCHAR"
assert types["net"] == "DOUBLE"
assert types["rank"] == "BIGINT"
assert all("type" in c for c in res["columns"])
def test_schema_all_returns_real_parquet_types(seeded):
res = ext_api.discover_all_schemas(_request(seeded))
item = next(i for i in res["items"] if i["id"] == "flow")
types = {c["name"]: c["type"] for c in item["columns"]}
assert types["net"] == "DOUBLE"
assert types["rank"] == "BIGINT"
# label 取自配置字段定义
labels = {c["name"]: c["label"] for c in item["columns"]}
assert labels["net"] == "主力净流入"
def test_schema_all_fallback_uses_type_key_when_no_data(tmp_path, monkeypatch):
"""无数据 (读 Parquet 失败) 时退回配置字段, 键名必须仍是 type, 不能是 dtype。"""
ExtConfigStore(tmp_path).upsert(_config())
monkeypatch.setattr(ext_api, "_store", lambda request: ExtConfigStore(tmp_path))
monkeypatch.setattr(ext_api, "_data_dir", lambda request: tmp_path)
res = ext_api.discover_all_schemas(_request(tmp_path))
item = res["items"][0]
assert [c["type"] for c in item["columns"]] == ["string", "float", "int"]
assert all("type" in c and "dtype" not in c for c in item["columns"])
def test_schema_describe_helper_works_on_current_duckdb(seeded):
"""直接验证 DESCRIBE 取值: 不依赖具体版本的输出列名。"""
glob = str(seeded / "ext_data" / "flow" / "timeseries" / "**" / "*.parquet")
cols = ext_api._describe_parquet_columns(glob)
assert ("net", "DOUBLE") in cols
def test_schema_does_not_leak_hive_partition_column(seeded):
"""`date=YYYY-MM-DD` 目录不得变成真实列 (DuckDB 默认会推断 hive 分区)。"""
res = ext_api.discover_schema(_request(seeded), "flow")
names = [c["name"] for c in res["columns"]]
assert "date" not in names
assert names == ["symbol", "net", "rank"]
@@ -204,3 +204,36 @@ it('heatmap 模式: 领涨格子亮琥珀描边+辉光, 格间细缝, 中性中
// 图例条: 轨迹语义以真实色块表达
expect(host.textContent).toContain('领涨轨迹')
})
// ===== 资金流下拉: schema 类型名口径 =====
it('资金流下拉收录 Parquet 真实类型名 (DOUBLE/BIGINT/DECIMAL) 的数值列', async () => {
Object.assign(globalThis, { IS_REACT_ACT_ENVIRONMENT: true })
const { api } = await import('@/lib/api')
vi.mocked(api.extDataSchemaAll).mockResolvedValueOnce({
items: [{
id: 'ext_flow', label: '资金流', mode: 'timeseries', time_field: null,
columns: [
{ name: 'symbol', type: 'VARCHAR', label: '标的代码' },
// 后端 schema 端点返回的是 Parquet 真实类型, 不是配置里的 float/int 简写
{ name: 'net', type: 'DOUBLE', label: '主力净流入' },
{ name: 'rank', type: 'BIGINT', label: '排名' },
{ name: 'ratio', type: 'DECIMAL(18,4)', label: '净占比' },
{ name: 'note', type: 'VARCHAR', label: '备注' },
],
}],
} as any)
await renderCard()
// 资金流下拉在「更多」折叠区里, 先展开
const moreBtn = Array.from(host.querySelectorAll('button'))
.find(b => (b.textContent || '').includes('更多'))
await act(async () => { (moreBtn as HTMLButtonElement).click() })
const select = host.querySelector('select[aria-label="资金流扩展列"]')
expect(select).toBeTruthy()
const values = Array.from(select!.querySelectorAll('option')).map(o => (o as HTMLOptionElement).value)
expect(values).toContain('ext_flow.net')
expect(values).toContain('ext_flow.rank')
expect(values).toContain('ext_flow.ratio')
expect(values).not.toContain('ext_flow.symbol')
expect(values).not.toContain('ext_flow.note')
})
@@ -41,7 +41,11 @@ function parseSectorSource(raw: string | null): SectorSource {
if (raw === 'custom' || raw === 'score' || raw === 'pct' || raw === 'rank_change' || raw === 'momentum' || raw === 'flow') return raw
return 'score' // 未设置过 / 旧默认值 'auto' → 默认综合分 (显式选择过的值已持久化, 不受影响)
}
const NUMERIC_TYPES = new Set(['float', 'int', 'number', 'double', 'long'])
// 扩展表 schema 的类型名可能是配置里的简写 (float/int), 也可能是 Parquet 的
// 真实类型 (DOUBLE/BIGINT/DECIMAL(18,4))。两种口径都要认, 否则会漏掉整数列。
const NUMERIC_TYPE_PATTERN =
/^(float|int|number|double|long|real|decimal|numeric|hugeint|ubigint|bigint|u?smallint|u?tinyint|u?integer)/
const isNumericColumnType = (type: unknown) => NUMERIC_TYPE_PATTERN.test(String(type ?? '').toLowerCase())
// 强度图与展示图跨实例联动分组 (x 轴指针同步)
const CHART_CONNECT_GROUP = 'sector-rotation-card'
@@ -311,7 +315,7 @@ export function SectorRotationCard({ kind }: { kind: 'concept' | 'industry' }) {
const options: { value: string; label: string }[] = []
for (const item of schemaQuery.data?.items ?? []) {
for (const column of item.columns ?? []) {
if (!NUMERIC_TYPES.has(String(column.type).toLowerCase())) continue
if (!isNumericColumnType(column.type)) continue
options.push({
value: `${item.id}.${column.name}`,
label: `${item.label || item.id} · ${column.label || column.name}`,