mirror of
https://github.com/shy3130/tick-stock-panel.git
synced 2026-11-02 10:26:13 +00:00
shy3130
a40c0780d7
用户反馈资金流扩展表点「手动拉取」报错、板块轮动的资金流下拉为空。 排查出三个独立缺陷: 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>
103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
"""扩展表 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"]
|