feat(ext-data): 数据出口强化 + 核心开放设计方案 (0.3.2 起点)

- /rows 查询五件套: filter 等值过滤 / sort / offset 分页 /
  start_date+end_date 范围合并 (自动补行级 date 列, diagonal 容忍分区
  schema 漂移), 快照模式范围查询明确 400
- /values 字段取值枚举端点 (filter 下拉配套, 去重+计数降序)
- 市场级表 (market_level): 行=全市场每日一条, 无标的列; 创建/更新 API、
  ingest 与拉取链路放行、创建向导开关跳过标的关联 —— 市场环境类数据
  发布为扩展数据的底座
- 拉取分页协议 (仅 GET): page_param/page_size_param/page_size/max_pages,
  空页/短页判停; 拉取面板新增分页配置输入
- docs: features.md 分页与 /rows 契约章节; open-platform-plan.md
  核心能力开放与插件化改造设计稿 (Token/scope/限流/Tier A/分期路线)
- tests: 22 个新用例 (查询语义/分页判停/市场级全链路/values),
  ext 全量 261 通过
This commit is contained in:
shy3130 committed 2026-09-25 21:29:23 +08:00
1 parent 2acf025652
commit 5f39f7dd8d
11 files changed
+942 -29

No files matched your search

+136 -6
View File
@@ -57,6 +57,7 @@ class CreateExtReq(BaseModel):
description: str = ""
symbol_map: dict = {} # {"type": "mapped", "col": "..."} 或 {"type": "computed", "from": "code", "method": "append_exchange"}
code_map: dict = {} # {"type": "mapped", "col": "..."} 或 {"type": "computed", "from": "symbol", "method": "strip_exchange"}
market_level: bool = False # 市场级表: 行=全市场每日一条, 无 symbol 列 (如择时状态序列)
class UpdateExtReq(BaseModel):
@@ -65,6 +66,7 @@ class UpdateExtReq(BaseModel):
description: str | None = None
symbol_map: dict | None = None
code_map: dict | None = None
market_level: bool | None = None
class IngestReq(BaseModel):
@@ -232,10 +234,21 @@ def _read_ext_dataframe(
config: ExtConfig,
data_dir: Path,
snapshot_date: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
) -> tuple[pl.DataFrame, str | None]:
"""读取扩展表数据。
- snapshot: 当前快照 (不支持日期范围)。
- timeseries + snapshot_date: 单日分区 (历史行为)。
- timeseries + start_date/end_date (可只给一端): 范围内全部分区 concat,
active_date 返回 "首..末"。date 与范围同时给出时 date 优先。
"""
cfg_dir = data_dir / "ext_data" / config.id
if config.mode == "snapshot":
if start_date or end_date:
raise HTTPException(400, "快照模式无历史分区, 不支持日期范围查询")
path = cfg_dir / "part.parquet"
if not path.exists():
return pl.DataFrame(), None
@@ -259,6 +272,34 @@ def _read_ext_dataframe(
if not partitions:
return pl.DataFrame(), None
if start_date or end_date:
lo = _partition_date(start_date) if start_date else None
hi = _partition_date(end_date) if end_date else None
if lo and hi and lo > hi:
raise HTTPException(400, f"start_date {lo} 晚于 end_date {hi}")
# ISO 日期字符串字典序 == 时间序, 分区名直接比较
selected = [
d for d in partitions
if (not lo or d.name[5:] >= lo) and (not hi or d.name[5:] <= hi)
]
if not selected:
return pl.DataFrame(), None
# 每分区补行级日期列 (分区名): 表自带 date 列则用 data_date, 两者都有不再加。
# 没有它, 范围合并后消费方无法区分行来自哪一天。
frames = []
for d in selected:
f = pl.read_parquet(d / "part.parquet")
day = d.name[5:]
if "date" not in f.columns:
f = f.with_columns(pl.lit(day).alias("date"))
elif "data_date" not in f.columns:
f = f.with_columns(pl.lit(day).alias("data_date"))
frames.append(f)
# diagonal: 各分区 schema 可能不一致 (配置中途加字段 → 早期分区少列),
# 缺列自动补 null 并取超类型; 单分区时无需 concat
df = pl.concat(frames, how="diagonal") if len(frames) > 1 else frames[0]
return df, f"{selected[0].name[5:]}..{selected[-1].name[5:]}"
latest = partitions[-1]
latest_date = latest.name[5:]
return pl.read_parquet(latest / "part.parquet"), latest_date
@@ -401,6 +442,7 @@ def create_config(request: Request, body: CreateExtReq):
description=body.description,
symbol_map=body.symbol_map,
code_map=body.code_map,
market_level=body.market_level,
)
store.upsert(config)
_refresh_views(request)
@@ -424,6 +466,8 @@ def update_config(request: Request, config_id: str, body: UpdateExtReq):
config.symbol_map = body.symbol_map
if body.code_map is not None:
config.code_map = body.code_map
if body.market_level is not None:
config.market_level = body.market_level
store.upsert(config)
_refresh_views(request)
return config.to_dict()
@@ -443,6 +487,42 @@ def delete_config(request: Request, config_id: str):
return {"status": "deleted"}
def _apply_row_filters(df: pl.DataFrame, filters: list[str] | None) -> pl.DataFrame:
"""等值过滤: 每个条目 `field:v1|v2` (值间 OR); 多条目间 AND; 按字符串比较。
数值列的等值匹配走字符串口径 (1 → "1", 1.0 → "1.0"), 覆盖二开最常见的
"维度取值筛选"场景; 区间/模糊匹配暂不做, 留给真实需求出现。
"""
if not filters:
return df
for entry in filters:
field, sep, raw_values = entry.partition(":")
field = field.strip()
values = [v for v in (raw_values.split("|") if sep else []) if v != ""]
if not field or not values:
raise HTTPException(400, f"过滤条件格式错误 (应为 字段:值1|值2): {entry}")
if field not in df.columns:
raise HTTPException(400, f"过滤字段不存在: {field} (可用: {df.columns})")
df = df.filter(pl.col(field).cast(pl.Utf8).is_in(values))
return df
def _apply_row_sort(df: pl.DataFrame, sort: str | None) -> pl.DataFrame:
"""排序: `field` 升序 / `field:desc` 降序; 未知字段 400。"""
if not sort:
return df
field, _, direction = sort.partition(":")
field = field.strip()
descending = direction.strip().lower() == "desc"
if field not in df.columns:
raise HTTPException(400, f"排序字段不存在: {field} (可用: {df.columns})")
return df.sort(field, descending=descending, nulls_last=True)
# 模块级常量避免 B008 (参数默认值内的函数调用); 语义与行内 Query(...) 完全一致
_ROW_FILTER_QUERY = Query(None, description="过滤 条件`字段:值1|值2`, 可重复多条 (AND)")
@router.get("/{config_id}/rows")
def list_rows(
request: Request,
@@ -450,31 +530,39 @@ def list_rows(
snapshot_date: str | None = Query(None, alias="date"),
columns: str | None = Query(None, description="逗号分隔的字段列表"),
limit: int = Query(1000, ge=1, le=20000),
offset: int = Query(0, ge=0, le=1_000_000, description="分页偏移 (配合 limit/total)"),
filter: list[str] | None = _ROW_FILTER_QUERY,
sort: str | None = Query(None, description="排序字段, 降序加 :desc"),
start_date: str | None = Query(None, description="时序起始日期 YYYY-MM-DD (含)"),
end_date: str | None = Query(None, description="时序结束日期 YYYY-MM-DD (含)"),
):
"""读取扩展数据明细。
- snapshot: 返回当前快照。
- timeseries: 默认返回最新日期分区,也可通过 date=YYYY-MM-DD 指定。
- timeseries: 默认返回最新日期分区; date=YYYY-MM-DD 指定单日;
start_date/end_date 取日期范围 (可只给一端), 全部分区合并返回。
- filter/sort/offset: 服务端过滤/排序/分页, 二开方无需一次拉全量自行筛。
"""
config = _store(request).get(config_id)
if not config:
raise HTTPException(404, f"配置 '{config_id}' 不存在")
data_dir = _data_dir(request)
df, active_date = _read_ext_dataframe(config, data_dir, snapshot_date)
df, active_date = _read_ext_dataframe(config, data_dir, snapshot_date, start_date, end_date)
df = _with_instrument_name(df, data_dir)
# 日内序列表: 按时间列升序, 保证分页/截断取到的是最早的盘
tf = config.pull.time_field if config.pull else None
if tf and tf in df.columns:
df = df.sort(tf)
df = _apply_row_filters(df, filter)
df = _apply_row_sort(df, sort)
total = len(df)
df = df.slice(offset, limit)
requested = [c.strip() for c in (columns or "").split(",") if c.strip()]
if requested:
keep = [c for c in ["symbol", "code", "name", *requested] if c in df.columns]
if keep:
df = df.select(list(dict.fromkeys(keep)))
total = len(df)
if total > limit:
df = df.head(limit)
rows = []
for row in df.to_dicts():
@@ -486,12 +574,53 @@ def list_rows(
"mode": config.mode,
"date": active_date,
"total": total,
"offset": offset,
"limit": limit,
"fields": [f.to_dict() for f in config.fields],
"rows": rows,
}
@router.get("/{config_id}/values")
def field_values(
request: Request,
config_id: str,
field: str = Query(..., min_length=1, description="要枚举去重的字段名"),
snapshot_date: str | None = Query(None, alias="date"),
start_date: str | None = Query(None),
end_date: str | None = Query(None),
limit: int = Query(200, ge=1, le=1000),
):
"""字段取值枚举 (去重 + 计数, 按出现次数降序)。
`/rows?filter=` 的配套端点: 二开方先取某字段的候选值做筛选下拉,
再带 filter 查明细。日期语义与 /rows 相同 (date 单日 / 范围 / 默认最新)。
"""
config = _store(request).get(config_id)
if not config:
raise HTTPException(404, f"配置 '{config_id}' 不存在")
data_dir = _data_dir(request)
df, active_date = _read_ext_dataframe(config, data_dir, snapshot_date, start_date, end_date)
if field not in df.columns:
raise HTTPException(400, f"字段不存在: {field} (可用: {df.columns})")
counted = (
df.group_by(field).len().sort(["len", field], descending=[True, False]).head(limit)
)
values = [
{"value": _safe_json_value(row[field]), "count": row["len"]}
for row in counted.to_dicts()
]
return {
"id": config.id,
"field": field,
"date": active_date,
"total": df.height,
"distinct": df[field].n_unique(),
"values": values,
}
@router.get("/{config_id}/dimension-members")
def dimension_members(
request: Request,
@@ -856,7 +985,8 @@ def ingest_data(request: Request, config_id: str, body: IngestReq):
configured = {f.name for f in config.fields}
required = configured - {"symbol"}
for i, row in enumerate(body.rows):
if "symbol" not in row:
# 市场级表 (无标的列) 不强制 symbol; 标的表仍要求逐行携带
if not config.market_level and "symbol" not in row:
raise HTTPException(400, f"第 {i + 1} 行缺少 symbol 字段")
missing = required - set(row.keys())
if missing:
+43 -1
View File
@@ -52,6 +52,7 @@ class PullConfig:
"last_run", "last_status", "last_message", "last_rows",
"next_run", "time_window_start", "time_window_end", "date_param",
"date_format", "time_field", "auth", "timeout_seconds",
"page_param", "page_size_param", "page_size", "page_start", "max_pages",
)
def __init__(
@@ -76,6 +77,11 @@ class PullConfig:
time_field: str | None = None,
auth: dict | None = None,
timeout_seconds: int = 30,
page_param: str | None = None,
page_size_param: str | None = None,
page_size: int = 0,
page_start: int = 1,
max_pages: int = 20,
) -> None:
self.url = url
self.method = method # GET | POST
@@ -110,6 +116,25 @@ class PullConfig:
int(timeout_seconds) if isinstance(timeout_seconds, (int, float))
and 5 <= timeout_seconds <= 300 else 30
)
# ── 分页协议 (仅 GET; 页码经 query 参数传递) ──
# page_param: 页码参数名 (如 "page"), 配置后按页循环拉取直到空页/短页/上限;
# None = 单次请求 (默认, 历史行为)。POST body 分页不支持 (body 是固定模板)
self.page_param = (page_param or "").strip() or None
# page_size_param: 每页条数参数名 (如 "pageSize"/"per_page"); 配合 page_size 一起发送
self.page_size_param = (page_size_param or "").strip() or None
# page_size: 每页条数值 (>0 且 page_size_param 已配置才发送); 也用于短页判停
self.page_size = (
int(page_size) if isinstance(page_size, (int, float)) and page_size > 0 else 0
)
# page_start: 起始页码 (有的接口从 0 计数)
self.page_start = (
int(page_start) if isinstance(page_start, (int, float)) and page_start >= 0 else 1
)
# max_pages: 安全上限, 防止接口永远返回数据把拉取循环拖死; 手改非法值归一 20
self.max_pages = (
int(max_pages) if isinstance(max_pages, (int, float))
and 1 <= max_pages <= 200 else 20
)
def to_dict(self) -> dict:
return {
@@ -133,6 +158,11 @@ class PullConfig:
"time_field": self.time_field,
"auth": self.auth,
"timeout_seconds": self.timeout_seconds,
"page_param": self.page_param,
"page_size_param": self.page_size_param,
"page_size": self.page_size,
"page_start": self.page_start,
"max_pages": self.max_pages,
}
@classmethod
@@ -160,6 +190,11 @@ class PullConfig:
time_field=d.get("time_field"),
auth=d.get("auth"),
timeout_seconds=d.get("timeout_seconds", 30),
page_param=d.get("page_param"),
page_size_param=d.get("page_size_param"),
page_size=d.get("page_size", 0),
page_start=d.get("page_start", 1),
max_pages=d.get("max_pages", 20),
)
@@ -181,7 +216,7 @@ class ExtConfig:
"""一个扩展数据源的完整配置。"""
__slots__ = (
"id", "label", "mode", "fields", "description",
"symbol_map", "code_map",
"symbol_map", "code_map", "market_level",
"created_at", "updated_at", "pull",
)
@@ -197,6 +232,7 @@ class ExtConfig:
created_at: str | None = None,
updated_at: str | None = None,
pull: PullConfig | None = None,
market_level: bool = False,
) -> None:
self.id = id
self.label = label
@@ -206,6 +242,10 @@ class ExtConfig:
# 映射关系: {"type": "mapped", "col": "原始列名"} 或 {"type": "computed", "from": "symbol|code", "method": "strip_exchange|append_exchange"}
self.symbol_map = symbol_map or {}
self.code_map = code_map or {}
# 市场级表: 行 = 全市场每日一条 (无 symbol 列), 如市场环境/情绪指数/择时
# 状态序列。上传/写入/拉取跳过标的关联强制; 供 /rows API 与未来的
# "市场级过滤数据集" 契约消费 (策略 JOIN 面向 symbol, 不适用此类表)。
self.market_level = bool(market_level)
self.created_at = created_at or datetime.now().isoformat()
self.updated_at = updated_at or datetime.now().isoformat()
self.pull = pull
@@ -219,6 +259,7 @@ class ExtConfig:
"description": self.description,
"symbol_map": self.symbol_map,
"code_map": self.code_map,
"market_level": self.market_level,
"created_at": self.created_at,
"updated_at": self.updated_at,
}
@@ -239,6 +280,7 @@ class ExtConfig:
created_at=d.get("created_at"),
updated_at=d.get("updated_at"),
pull=PullConfig.from_dict(d["pull"]) if d.get("pull") else None,
market_level=bool(d.get("market_level", False)),
)
+59 -12
View File
@@ -205,13 +205,19 @@ def _assert_rows_date(rows: list[dict], day: date) -> None:
)
async def _request_json(pull: PullConfig, config_id: str, day: date | None = None) -> Any:
async def _request_json(
pull: PullConfig, config_id: str, day: date | None = None,
extra_params: dict[str, str] | None = None,
) -> Any:
"""发起一次拉取请求并返回解析后的 JSON。
正式拉取 (带日期参数) 与设置页"测试" (不带) 共用同一实现,
保证 UA 标识头与 API Key 鉴权注入只有一套口径。
extra_params: 追加的 query 参数 (分页页码/每页条数), 值经 URL 编码。
"""
url = _with_date_param(pull.url, pull.date_param, day, pull.date_format) if day else pull.url
if extra_params:
url = _append_query_params(url, extra_params)
# 每配置超时 (PullConfig.timeout_seconds, 默认 30 与历史行为一致);
# getattr 兜底测试用的简化 pull 对象
timeout = getattr(pull, "timeout_seconds", None)
@@ -234,19 +240,18 @@ async def _request_json(pull: PullConfig, config_id: str, day: date | None = Non
raise ValueError(f"响应不是有效 JSON: {e}") from e
async def fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict]:
"""按日期请求外部 API 并解析为行 (不写盘)。空数据返回 []。
def _append_query_params(url: str, params: dict[str, str]) -> str:
"""把追加参数并入 URL query (值经编码, 保留既有参数与 fragment)。"""
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
与 fetch_and_ingest 共用同一解析链 (response_path/预设转换/字段映射/
关联字段校验), 历史回补与当日拉取不产生第二套口径。
"""
pull = config.pull
if not pull or not pull.url:
raise ValueError("拉取未配置或 URL 为空")
parts = urlsplit(url)
merged = dict(parse_qsl(parts.query, keep_blank_values=True))
merged.update({str(k): str(v) for k, v in params.items()})
return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(merged), parts.fragment))
data = await _request_json(pull, config.id, day=target_date)
# 提取行
def _parse_rows_payload(config: ExtConfig, pull: PullConfig, data: Any) -> list[dict]:
"""单次响应 → 行列表 (提取/预设转换/字段映射)。与写入链共用, 不含日期校验。"""
rows = _extract_rows(data, pull.response_path)
# 内置预设 (概念/行业): 应用结构转换, 让产出 schema 与分析页一致。
@@ -256,9 +261,51 @@ async def fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict
rows = _apply_preset_flatten(config.id, rows)
# 字段映射
rows = _apply_field_map(rows, pull.field_map)
return _apply_field_map(rows, pull.field_map)
async def fetch_rows_for_date(config: ExtConfig, target_date: date) -> list[dict]:
"""按日期请求外部 API 并解析为行 (不写盘)。空数据返回 []。
与 fetch_and_ingest 共用同一解析链 (response_path/预设转换/字段映射/
关联字段校验), 历史回补与当日拉取不产生第二套口径。
配置了 pull.page_param 时按页循环拉取 (仅 GET), 空页/短页/达 max_pages 停。
"""
pull = config.pull
if not pull or not pull.url:
raise ValueError("拉取未配置或 URL 为空")
if pull.page_param and pull.method.upper() == "GET":
collected: list[dict] = []
page = pull.page_start
for _ in range(pull.max_pages):
extra: dict[str, str] = {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)
page_rows = _parse_rows_payload(config, pull, data)
if not page_rows:
break
collected.extend(page_rows)
# 短页 (不足一页) 说明已是最后一页, 免去再一次空请求
if pull.page_size > 0 and len(page_rows) < pull.page_size:
break
page += 1
else:
logger.warning(
"ext pull %s: 分页拉取达到 max_pages=%d 上限, 结果可能不完整",
config.id, pull.max_pages,
)
rows = collected
else:
data = await _request_json(pull, config.id, day=target_date)
rows = _parse_rows_payload(config, pull, data)
# 校验可关联标的的字段:直接 symbol/code,或配置里声明的映射源列。
# 市场级表 (market_level, 行=全市场一条, 无标的列) 跳过该校验。
if config.market_level:
_assert_rows_date(rows, target_date)
return rows
row_keys = set(rows[0]) if rows else set()
mapped_cols = {
m.get("col")
+117
View File
@@ -0,0 +1,117 @@
"""市场级扩展表 (market_level, 无 symbol 列) 全链路 + /values 枚举端点。
市场级表面向"行 = 全市场每日一条"的数据 (择时状态/情绪指数), 是把
市场环境类数据发布为扩展数据、供 /rows API 与未来"市场级过滤数据集"
契约消费的前提。标的表的既有强制 (逐行 symbol) 行为不变。
"""
from __future__ import annotations
from datetime import date
from pathlib import Path
from types import SimpleNamespace
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.ext_data import router
from app.services.ext_data import ExtConfig, ExtField, PullConfig
@pytest.fixture
def client(tmp_path: Path) -> TestClient:
app = FastAPI()
app.include_router(router)
# db=None: _refresh_views 的 db.execute 全在 try/except 内, 裸 app 下安全跳过
app.state.repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path, db=None))
return TestClient(app)
def _create(client: TestClient, market_level: bool = False, cid: str = "regime_ts") -> None:
r = client.post("/api/ext-data", json={
"id": cid,
"label": "市场状态",
"mode": "timeseries",
"fields": [{"name": "state", "dtype": "string"}, {"name": "score", "dtype": "float"}],
"market_level": market_level,
})
assert r.status_code == 200, r.text
def test_market_level_ingest_and_rows_roundtrip(client: TestClient):
_create(client, market_level=True)
r = client.post("/api/ext-data/regime_ts/ingest", json={
"date": "2026-09-24",
"rows": [{"state": "bull", "score": 80.0}],
})
assert r.status_code == 200, r.text
rows = client.get("/api/ext-data/regime_ts/rows").json()
assert rows["total"] == 1 and rows["rows"][0]["state"] == "bull"
def test_symbol_table_still_requires_symbol(client: TestClient):
_create(client, market_level=False, cid="per_stock")
r = client.post("/api/ext-data/per_stock/ingest", json={
"date": "2026-09-24",
"rows": [{"state": "bull", "score": 80.0}], # 无 symbol
})
assert r.status_code == 400 and "缺少 symbol" in r.json()["detail"]
def test_market_level_flag_persists(client: TestClient):
_create(client, market_level=True)
cfg = client.get("/api/ext-data").json()["items"][0]
assert cfg["market_level"] is True
# update 可切换
r = client.put("/api/ext-data/regime_ts", json={"market_level": False})
assert r.json()["market_level"] is False
async def test_market_level_pull_skips_symbol_check(monkeypatch: pytest.MonkeyPatch):
"""拉取链路: 市场级表响应行无 symbol/code 也不报错 (标的表会 400)。"""
from app.services import ext_pull
async def fake(pull, config_id, day=None, extra_params=None):
return [{"state": "bull", "score": 80.0}] # 直接根数组
monkeypatch.setattr(ext_pull, "_request_json", fake)
cfg = ExtConfig(
id="regime_ts", label="市场状态", mode="timeseries",
fields=[ExtField("state", "string"), ExtField("score", "float")],
pull=PullConfig(url="https://api.example.com/regime"),
market_level=True,
)
rows = await ext_pull.fetch_rows_for_date(cfg, date(2026, 9, 24))
assert rows == [{"state": "bull", "score": 80.0}]
# 对照: 标的表同样响应 → 报错
cfg2 = ExtConfig(
id="per_stock", label="个股", mode="timeseries",
fields=[ExtField("state", "string")],
pull=PullConfig(url="https://api.example.com/x"),
)
with pytest.raises(ValueError, match="symbol"):
await ext_pull.fetch_rows_for_date(cfg2, date(2026, 9, 24))
def test_values_endpoint_counts_and_400(client: TestClient):
r = client.post("/api/ext-data", json={
"id": "tags", "label": "标签", "mode": "timeseries",
"fields": [{"name": "state", "dtype": "string"}],
})
assert r.status_code == 200
client.post("/api/ext-data/tags/ingest", json={
"date": "2026-09-24",
"rows": [
{"symbol": "000001.SZ", "state": "bull"},
{"symbol": "600519.SH", "state": "bull"},
{"symbol": "300750.SZ", "state": "bear"},
],
})
r = client.get("/api/ext-data/tags/values", params={"field": "state"})
assert r.status_code == 200
body = r.json()
assert body["distinct"] == 2
assert body["values"] == [{"value": "bull", "count": 2}, {"value": "bear", "count": 1}]
# 未知字段 → 400
assert client.get("/api/ext-data/tags/values", params={"field": "nope"}).status_code == 400
+101
View File
@@ -0,0 +1,101 @@
"""分页拉取协议 (page_param/page_size_param): 按页循环、空页/短页判停、
max_pages 安全上限、非分页配置行为不变。mock _request_json, 不发真实请求。
"""
from __future__ import annotations
from datetime import date
import pytest
from app.services import ext_pull
from app.services.ext_data import ExtConfig, ExtField, PullConfig
def _cfg(pull: PullConfig) -> ExtConfig:
return ExtConfig(
id="hot",
label="人气",
mode="timeseries",
fields=[ExtField("symbol", "string"), ExtField("heat", "float")],
pull=pull,
)
def _install_pages(monkeypatch: pytest.MonkeyPatch, pages: dict[int, list[dict]]) -> list[dict]:
"""按 extra_params["page"] 返回预置页; 记录每次调用收到的 extra_params。"""
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)
key = (extra_params or {}).get("page")
page = int(key) if key is not None else 1
return {"data": pages.get(page, [])}
monkeypatch.setattr(ext_pull, "_request_json", fake)
return seen
TARGET = date(2026, 9, 20)
async def test_pagination_collects_pages_and_stops_on_short_page(monkeypatch: pytest.MonkeyPatch) -> None:
# page_size=2: 第 2 页只有 1 行 (短页) → 停, 不再请求第 3 页
pull = PullConfig(
url="https://api.example.com/hot", response_path="data",
page_param="page", page_size_param="pageSize", page_size=2, max_pages=10,
)
seen = _install_pages(monkeypatch, {
1: [{"symbol": "000001.SZ", "heat": 1}, {"symbol": "600519.SH", "heat": 2}],
2: [{"symbol": "300750.SZ", "heat": 3}],
3: [{"symbol": "不oshould到达", "heat": 9}],
})
rows = await ext_pull.fetch_rows_for_date(_cfg(pull), TARGET)
assert [r["symbol"] for r in rows] == ["000001.SZ", "600519.SH", "300750.SZ"]
assert len(seen) == 2
assert seen[0] == {"page": "1", "pageSize": "2"}
assert seen[1] == {"page": "2", "pageSize": "2"}
async def test_pagination_stops_on_empty_page(monkeypatch: pytest.MonkeyPatch) -> None:
# 未知 page_size: 只能靠空页判停
pull = PullConfig(url="https://api.example.com/hot", response_path="data", page_param="page")
seen = _install_pages(monkeypatch, {
1: [{"symbol": "000001.SZ", "heat": 1}],
2: [],
})
rows = await ext_pull.fetch_rows_for_date(_cfg(pull), TARGET)
assert len(rows) == 1 and len(seen) == 2
assert seen[0] == {"page": "1"} # 未配置 page_size_param → 不发送 size 参数
async def test_pagination_respects_max_pages_cap(monkeypatch: pytest.MonkeyPatch) -> None:
pull = PullConfig(url="https://api.example.com/hot", response_path="data", page_param="page", max_pages=3)
seen = _install_pages(monkeypatch, {
1: [{"symbol": "a", "heat": 1}], 2: [{"symbol": "b", "heat": 1}], 3: [{"symbol": "c", "heat": 1}],
})
# 每页都满 (无 page_size 信息) → 达上限 3 页停
rows = await ext_pull.fetch_rows_for_date(_cfg(pull), TARGET)
assert len(rows) == 3 and len(seen) == 3
async def test_pagination_page_start_zero(monkeypatch: pytest.MonkeyPatch) -> None:
pull = PullConfig(url="https://api.example.com/hot", response_path="data", page_param="page", page_start=0)
seen = _install_pages(monkeypatch, {0: [{"symbol": "000001.SZ", "heat": 1}]})
# page 0 有 1 行, page 1 空 → 停
rows = await ext_pull.fetch_rows_for_date(_cfg(pull), TARGET)
assert len(rows) == 1 and seen[0] == {"page": "0"}
async def test_no_page_param_keeps_single_request(monkeypatch: pytest.MonkeyPatch) -> None:
pull = PullConfig(url="https://api.example.com/hot", response_path="data")
seen = _install_pages(monkeypatch, {1: [{"symbol": "000001.SZ", "heat": 1}]})
rows = await ext_pull.fetch_rows_for_date(_cfg(pull), TARGET)
assert len(rows) == 1 and seen == [None] # 单次请求不带 extra_params
async def test_post_with_page_param_ignored(monkeypatch: pytest.MonkeyPatch) -> None:
# 分页仅支持 GET; POST 配了 page_param 也按单次请求 (不注页参数)
pull = PullConfig(url="https://api.example.com/hot", method="POST", body="{}", response_path="data", page_param="page")
seen = _install_pages(monkeypatch, {1: [{"symbol": "000001.SZ", "heat": 1}]})
rows = await ext_pull.fetch_rows_for_date(_cfg(pull), TARGET)
assert len(rows) == 1 and seen == [None]
+154
View File
@@ -0,0 +1,154 @@
"""`/rows` 查询增强 (V-开放能力): filter 等值过滤 / sort 排序 / offset 分页 /
timeseries 日期范围合并读取。
范围读取是路径敏感操作 (拼 `date=` 分区目录), 与 date_guard 同一套防穿越
校验 (_partition_date); 过滤/排序的 400 语义单独覆盖。
"""
from __future__ import annotations
from pathlib import Path
import polars as pl
import pytest
from fastapi import HTTPException
from app.api.ext_data import _apply_row_filters, _apply_row_sort, _read_ext_dataframe
from app.services.ext_data import ExtConfig, ExtField
def _cfg(mode: str = "timeseries") -> ExtConfig:
return ExtConfig(
id="hot",
label="人气",
mode=mode, # type: ignore[arg-type]
fields=[ExtField("symbol", "string"), ExtField("concept", "string"), ExtField("heat", "float")],
)
def _layout(tmp_path: Path) -> Path:
"""三个时序分区: 09-09 / 09-10 / 09-11。"""
data_dir = tmp_path / "data"
rows_by_day = {
"2026-09-09": [("000001.SZ", "AI", 1.0), ("600519.SH", "白酒", 2.0)],
"2026-09-10": [("000001.SZ", "AI", 3.0)],
"2026-09-11": [("300750.SZ", "AI", 5.0), ("600519.SH", "白酒", 4.0)],
}
for day, rows in rows_by_day.items():
part = data_dir / "ext_data" / "hot" / "timeseries" / f"date={day}"
part.mkdir(parents=True)
pl.DataFrame(
{"symbol": [r[0] for r in rows], "concept": [r[1] for r in rows], "heat": [r[2] for r in rows]}
).write_parquet(part / "part.parquet")
return data_dir
# ── 日期范围读取 ──────────────────────────────────────────
def test_read_range_merges_partitions(tmp_path: Path) -> None:
df, active = _read_ext_dataframe(_cfg(), _layout(tmp_path), start_date="2026-09-09", end_date="2026-09-10")
assert len(df) == 3
assert active == "2026-09-09..2026-09-10"
# 范围合并自动补行级分区日期 (表本身无 date 列 → 补 date 列)
assert "date" in df.columns
assert set(df.filter(pl.col("symbol") == "600519.SH")["date"].to_list()) == {"2026-09-09"}
def test_read_range_open_ended(tmp_path: Path) -> None:
data_dir = _layout(tmp_path)
df, active = _read_ext_dataframe(_cfg(), data_dir, start_date="2026-09-10")
assert len(df) == 3 and active == "2026-09-10..2026-09-11"
df, active = _read_ext_dataframe(_cfg(), data_dir, end_date="2026-09-09")
assert len(df) == 2 and active == "2026-09-09..2026-09-09"
def test_read_range_reversed_rejected(tmp_path: Path) -> None:
with pytest.raises(HTTPException) as e:
_read_ext_dataframe(_cfg(), _layout(tmp_path), start_date="2026-09-11", end_date="2026-09-09")
assert e.value.status_code == 400
def test_read_range_bad_date_rejected(tmp_path: Path) -> None:
with pytest.raises(HTTPException) as e:
_read_ext_dataframe(_cfg(), _layout(tmp_path), start_date="../../kline_daily")
assert e.value.status_code == 400
def test_snapshot_mode_rejects_range(tmp_path: Path) -> None:
data_dir = _layout(tmp_path)
snap = data_dir / "ext_data" / "hot" / "part.parquet"
snap.parent.mkdir(parents=True, exist_ok=True)
pl.DataFrame({"symbol": ["000001.SZ"], "concept": ["AI"], "heat": [1.0]}).write_parquet(snap)
with pytest.raises(HTTPException) as e:
_read_ext_dataframe(_cfg(mode="snapshot"), data_dir, start_date="2026-09-09")
assert e.value.status_code == 400
def test_exact_date_still_wins_over_range(tmp_path: Path) -> None:
df, active = _read_ext_dataframe(
_cfg(), _layout(tmp_path),
snapshot_date="2026-09-10", start_date="2026-09-09", end_date="2026-09-11",
)
assert len(df) == 1 and active == "2026-09-10"
def test_read_range_tolerates_schema_drift(tmp_path: Path) -> None:
"""分区 schema 不一致 (配置中途加字段 → 早期分区少列): diagonal 合并, 缺列补 null。
真实场景: ext_fuyao_hot 早期 4 列、后期 7 列, vertical concat 直接 ShapeError。
"""
data_dir = _layout(tmp_path)
# 给最后一天追加一列新字段 (模拟中途加列)
last = data_dir / "ext_data" / "hot" / "timeseries" / "date=2026-09-11" / "part.parquet"
df = pl.read_parquet(last).with_columns(pl.lit("x").alias("extra_tag"))
df.write_parquet(last)
out, _active = _read_ext_dataframe(_cfg(), data_dir, start_date="2026-09-09", end_date="2026-09-11")
assert len(out) == 5
assert "extra_tag" in out.columns
# 早期分区的 extra_tag 为 null, 后期有值
tags = out.sort("symbol").get_column("extra_tag").to_list()
assert None in tags and "x" in tags
# ── 过滤 / 排序 ────────────────────────────────────────────
def _df() -> pl.DataFrame:
return pl.DataFrame({
"symbol": ["000001.SZ", "600519.SH", "300750.SZ"],
"concept": ["AI", "白酒", "AI"],
"heat": [1.0, 2.5, 3.0],
})
def test_filter_single_value_and_in_list() -> None:
out = _apply_row_filters(_df(), ["concept:AI"])
assert out["symbol"].to_list() == ["000001.SZ", "300750.SZ"]
# 值间 OR
out = _apply_row_filters(_df(), ["concept:AI|白酒"])
assert len(out) == 3
# 数值列按字符串等值匹配
out = _apply_row_filters(_df(), ["heat:2.5"])
assert out["symbol"].to_list() == ["600519.SH"]
def test_filter_multiple_conditions_are_and() -> None:
out = _apply_row_filters(_df(), ["concept:AI", "heat:3.0"])
assert out["symbol"].to_list() == ["300750.SZ"]
def test_filter_unknown_field_and_malformed_rejected() -> None:
with pytest.raises(HTTPException) as e:
_apply_row_filters(_df(), ["nope:1"])
assert e.value.status_code == 400
with pytest.raises(HTTPException):
_apply_row_filters(_df(), ["concept"]) # 缺 :值
with pytest.raises(HTTPException):
_apply_row_filters(_df(), [":AI"]) # 缺字段
def test_sort_asc_desc_and_unknown_rejected() -> None:
out = _apply_row_sort(_df(), "heat")
assert out["heat"].to_list() == [1.0, 2.5, 3.0]
out = _apply_row_sort(_df(), "heat:desc")
assert out["heat"].to_list() == [3.0, 2.5, 1.0]
with pytest.raises(HTTPException) as e:
_apply_row_sort(_df(), "nope")
assert e.value.status_code == 400
+21
View File
@@ -189,6 +189,27 @@ timeseries 模式的表 (如人气排行) 默认只能从开启拉取之日起
防污染契约: 若接口忽略日期参数返回当日数据 (以响应行 `date` 字段为准校验), 该日**拒绝写入**并计入失败清单 —— 否则当日值会静默污染整个历史时序。接口侧需提供: 传日期参数返回该日数据、无数据返回空数组或 404 (均视为该日无数据, 计入 empty 跳过)、每行带 `date` 字段;不传参数保持当日 (向后兼容)。
##### 分页拉取协议
数据量大的接口 (全市场排行、批量快照) 单次响应装不下时, 拉取配置可声明分页 (仅 GET): **页码参数名**(`page_param`, 如 `page`)、**每页条数参数名**(`page_size_param`, 如 `pageSize`)与**每页条数**(`page_size`)。配置后单次拉取按页循环请求并合并结果, 空页/不足一页自动判停; **最多页数**(`max_pages`, 默认 20, 上限 200)防止接口永远返回数据拖死循环。POST body 分页不支持 (请求体是固定模板)。
##### `/rows` 查询接口 (面向二次开发)
每张扩展表天然是标准 HTTP 接口 `GET /api/ext-data/{id}/rows`, 供页面之外的任意消费方调用:
- `date=YYYY-MM-DD` 单日; `start_date`/`end_date` 日期范围 (可只给一端, 时序全分区合并, 快照模式不支持范围) —— 范围合并自动补行级 `date` 列 (表自带 date 列时补 `data_date`), 消费方能区分每行归属日期;
- `filter=字段:值1|值2` 等值过滤 (值间 OR、多条件 AND, 可重复参数), 免拉全量自行筛;
- `sort=字段` / `sort=字段:desc` 排序; `offset`+`limit` 分页 (响应含 `total` 总数);
- `columns=a,b` 字段裁剪; 响应自带 `fields` schema 与当前数据日期。
配套枚举端点 `GET /api/ext-data/{id}/values?field=X`: 字段去重取值 + 出现次数 (降序), 日期语义与 `/rows` 相同 —— 二开方先取候选值做筛选下拉, 再带 `filter` 查明细。
上传/回补/删除等写端点属于管理操作, 不对外; 读端点即数据出口。
##### 市场级扩展表 (无标的列)
创建时勾选**市场级数据** (`market_level`): 行 = 全市场每日一条, 无 symbol/code 列 —— 适用于市场环境状态、情绪指数、择时信号等序列。上传/JSON 写入/接口拉取均跳过标的关联强制, 去重按首列; 通过 `/rows` (范围查询 + 自动 date 列) 消费完整时序。这是把"市场环境类数据发布为扩展数据"的数据底座, 策略侧的市场级过滤契约按需在此之上建设。
#### 扩展字段接入信号/因子
扩展表的数值字段(int/float)可直接用作**自定义信号条件**与**因子**——无需任何额外配置:
+189
View File
@@ -0,0 +1,189 @@
# 核心能力开放与插件化改造设计方案
> 状态: 设计稿(未实现)。本文档定义「核心域 / 扩展域」边界、对外 API 的认证与契约形态、以及分期改造路线。
> 遵循 [CONTRIBUTING.md](../CONTRIBUTING.md) 的数据口径、插件化与测试矩阵要求; 扩展机制现状见 [secondary-development.md](./secondary-development.md)。
## 1. 目标与非目标
**目标**
- 项目自身保留**核心域能力**(数据、策略/回测、监控推送、扩展数据存储), 并以稳定 HTTP API 对外开放, 二次开发者不修改本项目源码即可在其上构建应用。
- 外围能力**扩展域化**: 通过既有插件机制(数据源 plugin.yaml / 前端插槽 / 后端 custom / 扩展数据表)接入, 而不是长在核心里。
- 开放有护栏: Token + scope 授权、按 Token 限流、CORS 白名单; 读开放、写受控。
**非目标**
- 不做多用户/多租户(数据层全局单用户是既定取舍; 以「单用户 + 多 Token + scope」满足外部调用方隔离需求)。
- 不做前端运行时插件(remote module/微前端)与第三方代码沙箱——等真实第三方插件作者出现再立项。
- 不做热更新; 桌面版更新走「下载安装包 + 静默覆盖安装 + 重启」路径。
## 2. 核心域 / 扩展域边界
**划线原则**: 按「数据与语义的归属 + 依赖箭头方向」划分, 不按页面外观划分。核心域 = 被多个模块消费其数据/语义、且自身不依赖外围的; 扩展域 = 只消费核心接口、可整体替换或移除而不伤核心的。
### 2.1 核心域(保留在主程序, 代码不拆出)
| 域 | 内容 | 现有承载 |
|---|---|---|
| **数据底座** | 日K/除权/分钟/实时/财务的同步管道、TickFlow 与数据源插件路由、enriched 存储、复权与单位口径 | `services/kline_sync` `services/financial_sync`、`plugins/*`、DuckDB/parquet |
| **订阅锚点** | 自选与分组(监控范围、持仓提醒、看板的数据订阅范围锚点) | `api/watchlist`、`data/watchlist` |
| **策略与回测引擎** | 策略注册/运行、回测矩阵引擎、优化器、挖掘运行时、**环境过滤语义**(regime_filter 的 T-1 掩码)、回测候选缓存 | `strategy/`、`backtest/*`、`api/strategies` `api/screener` `api/backtest` |
| **环境数据生产** | regime_builder 的市场状态计算与盘后增量(回测/挖掘/因子归因/AI 四处消费同一口径) | `services/regime_builder`、盘后管道 |
| **监控与推送管道** | 规则引擎、SSE/语音/系统通知/Webhook/留痕五通道 | `services/quote_service`、`api/monitor-rules` `api/alerts` |
| **扩展数据存储** | 扩展表 CRUD、时序分区/快照、市场级表(market_level)、rows/values 查询出口 | `services/ext_data` `ext_pull`、`api/ext-data` |
| **认证与配置** | 访问密码(本机/内网 UI)、**API Token 与 scope(本方案新增)**、secrets | `services/auth`、`api/settings` |
### 2.2 扩展域(通过既有机制接入, 核心只暴露契约)
| 能力 | 接入机制 | 状态 |
|---|---|---|
| 第三方行情数据源 | `backend/app/plugins/<name>/plugin.yaml` + Provider | ✅ 已有 |
| 无代码 HTTP 数据源 | `data/data_sources/*.yaml` | ✅ 已有 |
| 用户自注册数据表(含市场级时序) | 扩展数据配置 + 上传/拉取/回补 | ✅ 已有(本方案补齐市场级与查询出口) |
| 自定义声明式页面 | `analysis_menus` 配置 + 三模板 | ✅ 已有 |
| 前端页面/导航/插槽 | `frontend/src/custom/<ns>/extension.tsx` | ✅ 已有(AI 助手即范例) |
| 后端路由/启动钩子/通知格式化 | `backend/app/custom/<module>.py` | ✅ 已有 |
| 自定义/AI/叠加策略 | `data/strategies/` 目录 | ✅ 已有 |
| AI 分析、大盘复盘、概念/行业/个股/财务分析、监控中心/异动/连板/持仓提醒、模拟盘、品牌页等**展示与外围业务页** | 核心数据的消费端; 二开可用扩展页面平行实现替代 UI | 页面可替换; 模拟盘拆分为官方插件的候选(见 §6 V3) |
## 3. 对外开放形态总览
```
┌─ 二次开发者的应用 ────────────────────────────────┐
│ 浏览器 SPA / 脚本 / 自建服务 │
└──────────────┬────────────────────────────────────┘
│ HTTPS + Bearer Token (scope 授权)
▼
┌─ 开放网关层(新增, 薄) ────────────────────────────┐
│ Token 认证 → scope 校验 → 限流(按 Token) → CORS │
├─ 核心 API(Tier A 稳定契约, 见 §5) ────────────────┤
│ 行情数据 / 扩展数据 / 策略与回测 / 监控事件 / 模拟盘 │
├─ 管理接口(Tier B, 永不开放给 Token 的写 scope) ────┤
│ 数据同步、扩展表 CRUD、配置、密码 │
└────────────────────────────────────────────────────┘
```
## 4. 认证、授权与护栏
### 4.1 Token 体系
- 存储沿用 secrets 模式: `data/user_data/tokens.json`(chmod 0600): `{id, name, token_hash, scopes, created_at, last_used_at, revoked}`; 明文 Token 只在创建响应中出现一次(前缀 `tsp_` + 32 字节随机 hex)。
- 认证通道与现有单密码**并行不替代**: 本机/内网 UI 登录照旧; `Authorization: Bearer tsp_...` 是外部调用方专用第二通道。
- 中间件顺序: 访问密码门 → Token 识别(有 Bearer 头则走 Token 路径, 免密码会话)→ scope 校验 → 限流。
### 4.2 Scope 矩阵(最小够用, 宁少勿滥)
| Scope | 允许 | 默认签发 |
|---|---|---|
| `read:market` | 行情/指数/标的搜索/分时/日K 读取 | ✅ 新 Token 默认 |
| `read:analysis` | 策略列表与运行结果、回测报告与候选、市场环境、监控事件读取 | ❌ 显式勾选 |
| `read:ext` | 扩展数据 rows/values/schema 读取 | ❌ 显式勾选 |
| `run:backtest` | 触发回测/挖掘任务(受重活并发器约束) | ❌ 显式勾选 |
| `paper:trade` | 模拟盘下单/撤单/建户(写操作, 最高敏感级) | ❌ 显式勾选 |
| `admin` | 管理接口 | ❌ **永不签发给 Token**, 仅 UI 会话可用 |
写敏感端点(扩展表 upload/ingest/backfill/delete、数据同步、设置)一律不进 Token scope —— 外部只读 + 指定的两类写(回测任务、模拟盘)。
### 4.3 限流与 CORS
- 按 Token 滑动窗口: 默认 120 req/min; 触发 429 + `Retry-After`。重活端点(回测/挖掘提交)另受既有 `heavy_job_limiter` 并发约束。
- CORS: `OPEN_API_CORS_ORIGINS` 配置(逗号分隔白名单), 默认空 = 不放行跨域; 放行后浏览器端二开可直接调用。
- 响应头: `X-RateLimit-Remaining` / `X-RateLimit-Limit`。
## 5. 核心 API 清单(Tier A 契约)
**稳定性分层**: Tier A = 对外承诺稳定(字段只增不改, 破坏性变更需 v2 前缀 + 一个大版本过渡期); Tier B = 内部实现细节, 随时变化, 不出现在开放文档。以 OpenAPI `x-stability: stable` 标注 + `/openapi.json` 过滤视图 `?tier=a` 实现, 不搬迁路由前缀。
### 5.1 行情数据(read:market)
| 端点 | 说明 |
|---|---|
| `GET /api/kline/instruments/search` | 标的搜索(代码/名称/拼音首字母, 含 ETF/指数) |
| `GET /api/kline/daily?symbol=&start=&end=` | 日K(前复权/不复权口径参数) |
| `GET /api/intraday/...` | 分时/多日分时 |
| `GET /api/index/...` | 指数行情与估值截面 |
| `GET /api/overview/market-snapshot` | 全市场当日快照(一次拉取, 二开图表直供) |
### 5.2 扩展数据(read:ext)
| 端点 | 说明 |
|---|---|
| `GET /api/ext-data/{id}/rows` | 明细查询五件套: date 单日 / start_date+end_date 范围(自动行级 date 列)/ filter 等值 / sort / offset+limit; columns 裁剪 |
| `GET /api/ext-data/{id}/values?field=` | 字段取值枚举(filter 下拉配套) |
| `GET /api/ext-data/schema/{id}` | 表结构 |
| `GET /api/ext-data`(列表) | 只返回 id/label/mode/market_level/字段名, 不含拉取配置等管理细节 |
市场级表(market_level)通过同一组端点消费 —— 这是「市场环境类数据发布为扩展数据」的出口。
### 5.3 策略与回测(read:analysis / run:backtest)
| 端点 | 说明 |
|---|---|
| `GET /api/strategies` / `GET /api/strategies/{id}` | 策略清单与详情(参数定义、数据依赖) |
| `POST /api/screener/run` | 运行策略选股(异步任务) |
| `GET /api/screener/result...` | 运行结果(含扩展列) |
| `POST /api/backtest` | 提交回测(含 regime_filter) → 任务 id |
| `GET /api/backtest/{job}/status` / `GET .../report` | 任务状态 / 回测报告(指标、回合、净值序列) |
| `GET /api/backtest/candidates` | 持久化回测候选摘要 |
| `GET /api/regime/states` 等 | 市场环境状态序列(内置 regime 的读取出口) |
### 5.4 监控与事件(read:analysis)
| 端点 | 说明 |
|---|---|
| `GET /api/alerts` | 告警/事件留痕查询 |
| `GET /api/events/stream`(SSE) | 实时事件流(告警/成交/系统事件); Token 以 query 参数短期票据接入(SSE 无法带 header) |
| Webhook 订阅注册 | **V2 再做**: 按 Token 注册回调 URL + 事件类型 + 签名密钥(复用现有 custom webhook 通道能力) |
### 5.5 模拟盘(paper:trade)
| 端点 | 说明 |
|---|---|
| `GET /api/paper/accounts` `/overview` `/positions` `/nav` `/stats` `/compare` | 账户与账务读取 |
| `POST /api/paper/orders` `DELETE /api/paper/orders/{id}` | 下单/撤单(写) |
| `GET /api/paper/trades` | 成交台账(append-only 事实源) |
### 5.6 系统
| 端点 | 说明 |
|---|---|
| `GET /api/health` | 探活(无鉴权, 白名单已有) |
| `GET /api/auth/api-tokens`(UI 会话) | Token 管理: 列表/创建/吊销(管理页, 不属于开放面) |
| `GET /api/openapi.json?tier=a` | Tier A 契约视图(机器可读) |
### 5.7 通用约定
- **分页**: `offset` + `limit`, 响应含 `total`; 列表端点统一 `items`/`rows` 字段名在契约化时逐一固定。
- **错误**: FastAPI 默认 `{detail}`; 契约化后统一补 `code` 字段(Tier A 端点)。
- **日期**: 全部 `YYYY-MM-DD` / ISO 时间戳, 北京时间口径(与内部一致, 不做时区换算)。
- **金额单位**: 元; 比例: 百分数数值(与各域现状一致, 契约文档逐一标注, 不强行统一)。
## 6. 分期路线
### V1 开门(Token + 护栏 + 最小契约)— 建议立即
- Token 存储/创建/吊销 + 设置页管理 UI(「系统 → API Token」)。
- 中间件: Bearer 识别 → scope 校验 → 限流 → CORS 配置。
- Tier A 标注: 先覆盖 `read:market` + `read:ext` 两组(行情与扩展数据, 风险最低、二开最先用)。
- 测试矩阵: 认证(无/错/过期 Token 401、scope 不足 403)、限流 429、CORS 头。
### V2 契约化 + 示例仓库
- Tier A 扩到策略/回测/监控事件/模拟盘; `?tier=a` OpenAPI 视图。
- 组织下建 `api-examples` 示例仓库: 「拉行情 → 跑策略 → 取结果 → 模拟盘跟单」三个可运行脚本 + README。
- SSE 短期票据(query token)方案落地。
### V3 钩子契约 + 模拟盘拆分试点
- 核心提供 `盘中快照钩子` `盘后结算钩子` 事件契约, 模拟盘改为通过钩子接入 → 验证「核心 + 官方插件」拆分工程可行性(当前反向 import 30 处, 先立契约再搬)。
### V4 市场级过滤契约(按需)
- 引擎把 regime_filter 泛化为「date 对齐的市场级扩展表过滤」: 官方 regime 表迁移为第一个实现, 第三方择时数据集(市场级扩展表)可直接选入回测过滤。触发条件: 第一个真实的第三方择时数据集出现。
## 7. 风险与守住的边界
- **Token 泄漏 = 面板数据泄漏**: 默认只读 scope、可随时吊销、创建时明文只显一次; paper:trade 永不默认。
- **契约冻结过早**: Tier A 从最小两组起步, 按二开反馈逐域升级, 未验证的端点留在 Tier B。
- **限流误伤本机 UI**: Token 限流只作用于 Bearer 路径; UI 会话不受 Token 桶约束。
- **不因开放破坏内部演进**: 开放面是「视图 + 网关」, 不重写内部路由; 内部 API 与开放 API 解耦, 由标注层衔接。
@@ -33,6 +33,8 @@ export function CreateExtDialog({ onClose }: { onClose: () => void }) {
const [label, setLabel] = useState('')
const [description, setDescription] = useState('')
const [mode, setMode] = useState<'snapshot' | 'timeseries'>('snapshot')
// 市场级表: 行 = 全市场每日一条, 无标的列 (市场环境/情绪指数类), 跳过标的关联
const [marketLevel, setMarketLevel] = useState(false)
const [fields, setFields] = useState<ExtDataField[]>([])
const [detectedSourceNames, setDetectedSourceNames] = useState<string[]>([])
const [error, setError] = useState('')
@@ -114,14 +116,17 @@ export function CreateExtDialog({ onClose }: { onClose: () => void }) {
id,
label,
mode,
fields: [
{ name: 'symbol', dtype: 'string', label: '标的代码' },
{ name: 'code', dtype: 'string', label: '代码' },
...userF,
],
fields: marketLevel
? userF
: [
{ name: 'symbol', dtype: 'string', label: '标的代码' },
{ name: 'code', dtype: 'string', label: '代码' },
...userF,
],
description: description.trim() || undefined,
symbol_map: submittedSymbolMap,
code_map: submittedCodeMap,
symbol_map: marketLevel ? {} : submittedSymbolMap,
code_map: marketLevel ? {} : submittedCodeMap,
market_level: marketLevel,
})
if (sourceMode === 'url' && (savePull || importNow || enablePull)) {
@@ -172,7 +177,7 @@ export function CreateExtDialog({ onClose }: { onClose: () => void }) {
id.trim()
&& label.trim()
&& fields.some((f) => f.name.trim())
&& matchStatus !== 'none'
&& (marketLevel || matchStatus !== 'none')
&& (sourceMode !== 'url' || url.trim()),
)
@@ -187,6 +192,15 @@ export function CreateExtDialog({ onClose }: { onClose: () => void }) {
setDetectedSourceNames(detected.map(f => f.name))
// 市场级表: 不做标的识别, 直接采认全部检测字段
if (marketLevel) {
setFields(detected)
setSymbolMap({})
setCodeMap({})
setMatchStatus('full')
return
}
if (symCands.length === 1 && codeCands.length === 1) {
sm = { type: 'mapped', col: symCands[0] }
cm = { type: 'mapped', col: codeCands[0] }
@@ -395,6 +409,31 @@ export function CreateExtDialog({ onClose }: { onClose: () => void }) {
)
})}
</div>
{/* 市场级: 行=全市场一条, 无标的列 (择时状态/情绪指数类) */}
<label className="mt-2 flex cursor-pointer items-center gap-2 rounded-lg border border-border/60 bg-elevated/40 px-2.5 py-1.5">
<input
type="checkbox"
checked={marketLevel}
onChange={e => {
setMarketLevel(e.target.checked)
if (e.target.checked) {
setSymbolMap({})
setCodeMap({})
setMatchStatus('full')
} else {
setMatchStatus('none')
}
}}
className="h-3.5 w-3.5 shrink-0"
/>
<span className="text-[10px] text-secondary">
市场级数据
<span className="ml-1 text-muted" title="行 = 全市场每日一条, 无标的代码列。适用于市场环境/情绪指数/择时状态等序列, 跳过标的关联; 通过 /rows 与 /values 接口消费">
每行是全市场一条 (无标的列), 跳过标的关联
</span>
</span>
</label>
</div>
<div className="grid grid-cols-[1fr_1fr] gap-3">
@@ -32,6 +32,11 @@ export function ExtDataPullPanel({ config, onSaved }: {
const [dateFormat, setDateFormat] = useState(pull?.date_format ?? 'iso')
const [timeField, setTimeField] = useState(pull?.time_field ?? '')
const [timeoutSec, setTimeoutSec] = useState(pull?.timeout_seconds ?? 30)
// 分页协议 (仅 GET): 页码参数名配置后按页循环拉取
const [pageParam, setPageParam] = useState(pull?.page_param ?? '')
const [pageSizeParam, setPageSizeParam] = useState(pull?.page_size_param ?? '')
const [pageSize, setPageSize] = useState(pull?.page_size ?? 0)
const [maxPages, setMaxPages] = useState(pull?.max_pages ?? 20)
const [enabled, setEnabled] = useState(pull?.enabled ?? false)
// 接口鉴权: 方式入 pull 配置; Key 本体只存后端 secrets.json
@@ -92,6 +97,11 @@ export function ExtDataPullPanel({ config, onSaved }: {
date_format: dateFormat,
time_field: timeField.trim() || null,
timeout_seconds: effTimeoutSec,
page_param: pageParam.trim() || null,
page_size_param: pageSizeParam.trim() || null,
page_size: pageSize > 0 ? pageSize : 0,
page_start: pull?.page_start ?? 1, // UI 不暴露, 保留已有值 (从 0 计数的接口手改 config.json)
max_pages: maxPages >= 1 && maxPages <= 200 ? maxPages : 20,
}
}
@@ -383,6 +393,42 @@ export function ExtDataPullPanel({ config, onSaved }: {
/>
</div>
<div>
<div className="text-[10px] text-muted mb-1">
分页 (接口数据量大时分页拉取 · 仅 GET · 留空页码参数=单次请求)
</div>
<div className="grid grid-cols-2 gap-1.5">
<input
value={pageParam} onChange={e => setPageParam(e.target.value)}
placeholder="页码参数名 · 如 page"
title="配置后按页循环拉取, 直到空页/最后一页/达上限"
className="rounded-btn border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40"
/>
<input
value={pageSizeParam} onChange={e => setPageSizeParam(e.target.value)}
placeholder="每页条数参数名 · 如 pageSize"
title="配合下方每页条数一起发送; 也用于短页判停"
className="rounded-btn border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40"
/>
<input
type="number" min={1}
value={pageSize}
onChange={e => setPageSize(Number(e.target.value))}
placeholder="每页条数"
title="每页条数值 (>0 且已填参数名才发送)"
className="rounded-btn border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40"
/>
<input
type="number" min={1} max={200}
value={maxPages}
onChange={e => setMaxPages(Number(e.target.value))}
placeholder="最多页数 · 默认 20"
title="安全上限, 防止接口永远返回数据拖死拉取循环"
className="rounded-btn border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40"
/>
</div>
</div>
<div>
<div className="text-[10px] text-muted mb-1">字段映射 (外部名 → 内部名,JSON,可选)</div>
+29 -2
View File
@@ -3223,18 +3223,30 @@ export const api = {
analysisMenuDelete: (id: string) =>
request<{ status: string }>(`/api/analysis-menus/${encodeURIComponent(id)}`, { method: 'DELETE' }),
extDataCreate: (body: { id: string; label: string; mode: 'snapshot' | 'timeseries'; fields: { name: string; dtype: string; label: string }[]; description?: string; symbol_map?: Record<string, string>; code_map?: Record<string, string> }) =>
extDataCreate: (body: { id: string; label: string; mode: 'snapshot' | 'timeseries'; fields: { name: string; dtype: string; label: string }[]; description?: string; symbol_map?: Record<string, string>; code_map?: Record<string, string>; market_level?: boolean }) =>
request<ExtDataConfig>('/api/ext-data', {
method: 'POST',
body: JSON.stringify(body),
}),
extDataUpdate: (id: string, body: { label?: string; fields?: { name: string; dtype: string; label: string }[]; description?: string }) =>
extDataUpdate: (id: string, body: { label?: string; fields?: { name: string; dtype: string; label: string }[]; description?: string; symbol_map?: Record<string, string>; code_map?: Record<string, string>; market_level?: boolean }) =>
request<ExtDataConfig>(`/api/ext-data/${id}`, {
method: 'PUT',
body: JSON.stringify(body),
}),
/** 字段取值枚举 (filter 配套): 去重 + 计数, 按出现次数降序 */
extDataValues: (id: string, field: string, opts?: { date?: string; start_date?: string; end_date?: string; limit?: number }) => {
const qs = new URLSearchParams({ field })
if (opts?.date) qs.set('date', opts.date)
if (opts?.start_date) qs.set('start_date', opts.start_date)
if (opts?.end_date) qs.set('end_date', opts.end_date)
if (opts?.limit) qs.set('limit', String(opts.limit))
return request<{ id: string; field: string; date: string | null; total: number; distinct: number; values: { value: string | number | null; count: number }[] }>(
`/api/ext-data/${id}/values?${qs.toString()}`,
)
},
extDataDelete: (id: string) =>
request<{ status: string }>(`/api/ext-data/${id}`, { method: 'DELETE' }),
@@ -3265,6 +3277,11 @@ export const api = {
time_field?: string | null;
auth?: ExtPullAuth;
timeout_seconds?: number;
page_param?: string | null;
page_size_param?: string | null;
page_size?: number;
page_start?: number;
max_pages?: number;
}) =>
request<{ status: string; pull: PullConfig }>(
`/api/ext-data/${id}/pull`,
@@ -4110,6 +4127,16 @@ export interface PullConfig {
auth?: ExtPullAuth | null
/** 单次拉取请求超时 (秒), 默认 30 */
timeout_seconds?: number
/** 分页协议 (仅 GET): 页码参数名 (如 "page"), 配置后按页循环拉取 */
page_param?: string | null
/** 每页条数参数名 (如 "pageSize"), 配合 page_size 一起发送 */
page_size_param?: string | null
/** 每页条数值 (>0 且配置 page_size_param 才发送); 也用于短页判停 */
page_size?: number
/** 起始页码 (有的接口从 0 计数), 默认 1 */
page_start?: number
/** 分页安全上限, 默认 20 (防接口永远返回数据拖死循环) */
max_pages?: number
}
export interface ExtDataBackfillResult {