From 010c484289fd57047d3e93f8a541ec606bfbec23 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 31 Aug 2026 22:53:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20=E5=85=A8=E9=87=8F=E5=88=86?= =?UTF-8?q?=E9=92=9F=20(full=5Fminute)=20=E6=95=B0=E6=8D=AE=E9=9B=86?= =?UTF-8?q?=E5=BC=80=E6=94=BE=E6=8F=92=E4=BB=B6/=E8=87=AA=E5=AE=9A?= =?UTF-8?q?=E4=B9=89=E6=BA=90=E6=8E=A5=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 像其他能力一样可路由: 声明 full_minute 数据集并在设置页路由, 即接入 盘中全市场分钟落盘服务 (与 TickFlow Expert 同一能力键 intraday.universe)。 - policy: _DATASET_CAP_MAP 增补 full_minute → INTRADAY_UNIVERSE, 自定义源 声明且被路由时自动补授能力, 服务门控对两类源统一口径 - capabilities 注册表: full_minute 从不可路由 (field=None) 改为 full_minute_data_provider 偏好路由; preferences/settings API 收发新字段 - 插件契约: get_intraday_batch (修复轮, 未实现回退 get_minute 当日窗口) + get_intraday_latest (稳态增量, 可选; 未实现降级仅修复轮, 节奏下限 60s) - YAML 声明式源: full_minute 数据集与 minute 同形, loader 白名单放行, 源编辑器 UI 可配置; 仅修复轮语义 - kline_sync 边界: fetch_intraday_custom_batch / _fetch_intraday_custom_latest, 帧过北京墙钟守卫, 与 TickFlow 边界函数同纪律 - minute_refresh: 删「自定义分钟源让位」逻辑, 改按路由取数; status 增 provider / provider_effective / repair_only - 前端: 监控页卡片按能力驱动, 数据源页路由矩阵, api.ts 类型 - 测试: 门控/路由/降级/增广新增用例, 后端 1332 全过 - 文档: plugin-development / custom-data-source / configuration / features / CONTRIBUTING 契约细化 (AI 可照文档接入) --- CONTRIBUTING.md | 4 +- backend/app/api/settings.py | 3 + backend/app/data_providers/capabilities.py | 6 +- backend/app/data_providers/custom/loader.py | 2 +- backend/app/data_providers/custom/provider.py | 39 ++++- backend/app/services/kline_sync.py | 94 ++++++++++ backend/app/services/minute_refresh.py | 128 ++++++++++---- backend/app/services/preferences.py | 5 + backend/app/tickflow/policy.py | 2 + backend/tests/test_capability_augment.py | 26 ++- backend/tests/test_capability_matrix.py | 34 ++-- backend/tests/test_minute_refresh.py | 165 ++++++++++++++++-- backend/tests/test_trading_day.py | 1 - docs/configuration.md | 11 +- docs/custom-data-source.md | 9 +- docs/features.md | 2 +- docs/plugin-development.md | 31 +++- frontend/src/lib/api.ts | 9 +- .../src/pages/settings/DataSourceEditor.tsx | 6 +- frontend/src/pages/settings/DataSources.tsx | 13 +- frontend/src/pages/settings/Monitoring.tsx | 8 +- 21 files changed, 503 insertions(+), 95 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 0b44732..724f6ab 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -122,7 +122,7 @@ 数据源已经插件化。任何通用功能都必须通过 provider 能力和标准化数据集访问数据，不能把 TickFlow SDK 调用硬编码到策略、监控、回测、API 或前端流程中。 - 使用现有的 `get_provider()`、`provider_has_dataset()` 和 preferences 路由能力。 -- 支持的数据集包括但不限于 `daily`、`adj_factor`、`minute`、`realtime`、`financial`；新增数据集应先定义清晰的输入输出契约。 +- 支持的数据集包括但不限于 `daily`、`adj_factor`、`minute`、`full_minute`（盘中全市场分钟落盘）、`realtime`、`financial`；新增数据集应先定义清晰的输入输出契约。 - provider 负责把供应商字段、单位、日期和代码格式转换为内部标准格式。 - 上层服务依赖标准字段和能力声明，不依赖供应商响应结构。 - 只有明确标注为 TickFlow 专属的功能才可以直接依赖 TickFlow，并且不得影响其他 provider。 @@ -138,7 +138,7 @@ - 各页面能力门控统一以矩阵的 `usable` 为准（生效源当前能否真正提供该能力），不是 TickFlow 套餐视角；缺能力提示统一引导到数据源配置。 - 能力层中立：通用界面（侧栏徽章、能力路由卡、各页门控提示）不得出现 TickFlow 档位/订阅词汇；档位信息只在 TickFlow 专属详情卡展示。provider 名称作为路由事实可以出现。 - 每个能力独立路由，禁止跟随/派生特殊值（`same_as_daily` 已下线）；存量非法偏好值由 preferences getter 回退默认自愈，不做迁移。 -- 边界注记：分时监控由分钟能力兜底（`intraday_monitor_support`），不单设分时能力；`depth5` 已进矩阵但插件数据集白名单暂未开放，当前仅 TickFlow 提供。 +- 边界注记：分时监控由分钟能力兜底（`intraday_monitor_support`），不单设分时能力；`full_minute`（全量分钟）数据集已开放插件/自定义源声明；`depth5` 已进矩阵但插件数据集白名单暂未开放，当前仅 TickFlow 提供。 - 实时指数为产品级固定契约，不走路由矩阵：展示层（侧栏指数条、市场总览）固定核心四只（`backend/app/services/index_const.py` 单一权威：上证/深成/创业板/科创综指），后端各消费方与前端 Layout 引用同一份定义不建副本；指数页保留但标的固定为核心四只（无全指数搜索/浏览，`/api/index/list`、`/api/index/search` 已下线）；侧栏指数多选配置已下线，相关偏好（`realtime_index_symbols`/`sidebar_index_symbols`/`indices_nav_pinned`/`realtime_pull_index`/`realtime_index_mode`）已删除。监控规则的指数标的不受限——quote_service 把核心四只 + 启用规则的指数并入显式拉取。 - 自定义源指数补充协议：A 股快照普遍不含指数（fuyao 实测无指数，指数在其独立端点）。provider 可实现可选方法 `get_realtime_indices(symbols) -> list[dict]`（record 结构与 realtime 一致），quote_service 在自定义源分支鸭子类型调用补拉；未实现的源指数缓存为空，由本地日K兜底接管。fuyao 指数快照有连坐语义——请求混入未知代码整批失败，插件侧必须先行过滤不支持的后缀（如 `.BJ`）。 diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index a78ed30..69f48c2 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -412,6 +412,7 @@ class DataProvidersIn(BaseModel): daily_data_provider: str | None = None adj_factor_provider: str | None = None minute_data_provider: str | None = None + full_minute_data_provider: str | None = None depth5_data_provider: str | None = None realtime_data_provider: str | None = None financial_data_provider: str | None = None @@ -498,6 +499,7 @@ def get_preferences() -> dict: "daily_data_provider": preferences.get_daily_data_provider(), "adj_factor_provider": preferences.get_adj_factor_provider(), "minute_data_provider": preferences.get_minute_data_provider(), + "full_minute_data_provider": preferences.get_full_minute_data_provider(), "minute_history_days": _minute_history_days(), "depth5_data_provider": preferences.get_depth5_data_provider(), "realtime_data_provider": preferences.get_realtime_data_provider(), @@ -776,6 +778,7 @@ def update_data_providers(req: DataProvidersIn, request: Request) -> dict: "daily_data_provider": preferences.get_daily_data_provider(), "adj_factor_provider": preferences.get_adj_factor_provider(), "minute_data_provider": preferences.get_minute_data_provider(), + "full_minute_data_provider": preferences.get_full_minute_data_provider(), "depth5_data_provider": preferences.get_depth5_data_provider(), "realtime_data_provider": preferences.get_realtime_data_provider(), "financial_data_provider": preferences.get_financial_provider(), diff --git a/backend/app/data_providers/capabilities.py b/backend/app/data_providers/capabilities.py index c6bb1d1..f778eed 100644 --- a/backend/app/data_providers/capabilities.py +++ b/backend/app/data_providers/capabilities.py @@ -81,11 +81,11 @@ CAPABILITY_REGISTRY: list[dict] = [ "id": "full_minute", "label": "全量分钟", "desc": "盘中全市场当日分钟落盘 (冷启动全天 + 标的池增量)", - "field": None, + "field": "full_minute_data_provider", "default": "tickflow", "tf_tier": "expert", - # intraday.universe 能力 (TickFlow Expert 专有): 插件契约不开放此数据集, - # 生效源恒为 TickFlow — 不可路由, 无对应 provider 偏好字段 + # TickFlow 侧需 Expert 档; 插件/自定义源声明 full_minute 数据集即可提供 + # (插件实现 get_intraday_batch / 可选 get_intraday_latest, YAML 仅修复轮) }, ] diff --git a/backend/app/data_providers/custom/loader.py b/backend/app/data_providers/custom/loader.py index 6f71d83..b08893c 100644 --- a/backend/app/data_providers/custom/loader.py +++ b/backend/app/data_providers/custom/loader.py @@ -423,7 +423,7 @@ def _sanitize_for_yaml(config: dict) -> dict: datasets_out: dict = {} for ds_name, ds_cfg in (config.get("datasets") or {}).items(): - if ds_name not in {"daily", "adj_factor", "realtime", "minute", "financial"}: + if ds_name not in {"daily", "adj_factor", "realtime", "minute", "full_minute", "financial"}: continue if not isinstance(ds_cfg, dict): continue diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py index 346e894..2e62148 100644 --- a/backend/app/data_providers/custom/provider.py +++ b/backend/app/data_providers/custom/provider.py @@ -30,6 +30,8 @@ _REQUIRED = { "adj_factor": {"symbol", "trade_date", "ex_factor"}, "realtime": {"symbol", "last_price", "prev_close", "open", "high", "low", "volume"}, "minute": {"symbol", "datetime", "open", "high", "low", "close", "volume", "amount"}, + # full_minute (全量分钟) 与 minute 同形: 当日窗口批量拉取, 字段映射一致 + "full_minute": {"symbol", "datetime", "open", "high", "low", "close", "volume", "amount"}, # financial 字段由数据源决定, 只要求能映射出 symbol "financial": {"symbol"}, } @@ -116,7 +118,7 @@ class GenericHTTPProvider: errors.append(f"{dataset}: pct_unit 必须是 percent 或 decimal") if dataset != "realtime": request_params = [cfg.symbols_param, cfg.start_param, cfg.end_param] - if dataset == "minute": + if dataset in {"minute", "full_minute"}: request_params.extend( name for name in (cfg.asset_type_param, cfg.freq_param) if name ) @@ -205,7 +207,38 @@ class GenericHTTPProvider: 配置的参数名注入请求 (GET → params, POST → body), 用于上游需区分 stock/ETF/index 或固定频率的场景。 """ - cfg = self._dataset("minute") + return self._fetch_minute_dataset( + "minute", symbols, start_time, end_time, asset_type, freq, on_chunk_done, + ) + + def get_intraday_batch( + self, + symbols: list[str], + count: int = 300, # noqa: ARG002 — 与插件契约对齐, YAML 源按时间窗口取全天 + asset_type: AssetType = "stock", + ) -> pl.DataFrame: + """全量分钟修复轮: 按当日窗口批量拉取 full_minute 数据集 (chunked + rpm 限速)。 + + 与 get_minute 同形 (字段映射/归一一致), 区别仅在数据集名与窗口由调用方 + 传当日值。稳态增量 (get_intraday_latest) YAML 声明式源不提供 — 服务自动 + 降级为仅修复轮模式并放慢节奏。 + """ + start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + return self._fetch_minute_dataset( + "full_minute", symbols, start, datetime.now(), asset_type, "1m", None, + ) + + def _fetch_minute_dataset( + self, + ds_name: str, + symbols: list[str], + start_time: datetime | None, + end_time: datetime | None, + asset_type: AssetType = "stock", + freq: str = "1m", + on_chunk_done: Callable[[int, int], None] | None = None, + ) -> pl.DataFrame: + cfg = self._dataset(ds_name) override: dict[str, Any] = {} if cfg.asset_type_param: override[cfg.asset_type_param] = asset_type @@ -280,7 +313,7 @@ class GenericHTTPProvider: start_time = end_time - timedelta(days=7) if dataset == "realtime": rows = self._request_rows(cfg) - elif dataset == "minute": + elif dataset in {"minute", "full_minute"}: override: dict[str, Any] = {} if cfg.asset_type_param: override[cfg.asset_type_param] = "stock" diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index aa04fa4..d634479 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -1064,6 +1064,100 @@ def fetch_intraday_universe_increment( return (_normalize_minute(seg), 1) +def _resolve_full_minute_provider( + provider_name: str, +) -> tuple[object | None, bool, str | None]: + """解析全量分钟生效的自定义源。返回 (provider, should_use_tickflow, error_msg): + + - provider_name == "tickflow" / 未配 full_minute dataset → (None, True, None) + - resolver 异常 (registry 损坏 / 插件失效 / 源不存在) → (None, True, str(e)) + - 成功 → (provider, False, None) + + 与 _resolve_minute_provider 同构, 仅数据集名不同。 + """ + if provider_name == "tickflow": + return (None, True, None) + from app.data_providers import custom as custom_sources + try: + if not custom_sources.provider_has_dataset(provider_name, "full_minute"): + return (None, True, None) + provider = custom_sources.get_provider(provider_name) + return (provider, False, None) + except Exception as e: # noqa: BLE001 + return (None, True, str(e)) + + +def fetch_intraday_custom_batch( + provider: object, + provider_name: str, + symbols: list[str], +) -> tuple[pl.DataFrame, int]: + """自定义源全量分钟修复轮: 当日窗口全市场批量拉取 (不落盘)。 + + provider 契约 (见 docs/plugin-development.md): + - get_intraday_batch(symbols, count, asset_type) 优先 — 源自管批量端点; + - 未实现则回退 get_minute(symbols, 当日窗口) — 复用逐标的分钟K机制, + 请求数按 chunk 回调统计。 + 帧统一过北京墙钟守卫 (与 _try_custom_minute 同纪律)。 + 返回 (当日分钟K, 请求数); 失败返回空 df 由调用方按空轮处理。 + """ + try: + method = getattr(provider, "get_intraday_batch", None) + if callable(method): + df = method(symbols) + requests = 1 + else: + counted = {"requests": 0} + + def _count_requests(cur: int, total: int) -> None: + counted["requests"] = max(counted["requests"], int(total)) + + end = cn_now() + start = end.replace(hour=0, minute=0, second=0, microsecond=0) + df = provider.get_minute( + symbols, start_time=start, end_time=end, + asset_type="stock", freq="1m", on_chunk_done=_count_requests, + ) + requests = counted["requests"] + except Exception as e: # noqa: BLE001 + logger.warning("custom full_minute batch via %s failed: %s", provider_name, e) + return (pl.DataFrame(), 0) + try: + df = _enforce_minute_beijing_wallclock(df, source=provider_name) + except Exception as e: # noqa: BLE001 + logger.warning("custom full_minute datetime 契约校验失败 (%s): %s", provider_name, e) + return (pl.DataFrame(), 0) + return (df, max(requests, 1)) + + +def fetch_intraday_custom_latest( + provider: object, + provider_name: str, + *, + count: int = 3, +) -> tuple[pl.DataFrame, int] | None: + """自定义源全量分钟稳态增量轮 (不落盘)。 + + provider 可选实现 get_intraday_latest(symbols=None, count) → 每只标的最新 + count 根分钟K, 尽量单请求/低请求量 (TickFlow 的 intraday.universe 同义)。 + 未实现返回 None — 调用方降级为仅修复轮模式。失败返回 (空 df, 0) 按空轮处理。 + """ + method = getattr(provider, "get_intraday_latest", None) + if not callable(method): + return None + try: + df = method(count=count) + except Exception as e: # noqa: BLE001 + logger.warning("custom full_minute latest via %s failed: %s", provider_name, e) + return (pl.DataFrame(), 0) + try: + df = _enforce_minute_beijing_wallclock(df, source=provider_name) + except Exception as e: # noqa: BLE001 + logger.warning("custom full_minute datetime 契约校验失败 (%s): %s", provider_name, e) + return (pl.DataFrame(), 0) + return (df, 1) + + def fetch_minute_single( symbol: str, trade_date: date, diff --git a/backend/app/services/minute_refresh.py b/backend/app/services/minute_refresh.py index cc809d8..829cce0 100644 --- a/backend/app/services/minute_refresh.py +++ b/backend/app/services/minute_refresh.py @@ -1,34 +1,41 @@ -"""盘中分钟K增量落盘服务 (Expert 专有)。 +"""盘中分钟K增量落盘服务 (全量分钟能力, 按能力路由)。 两段式拉取 (见 feat/minute-strategy 方案): -- 全天修复轮: intraday.batch (日内分时批量) 并发脉冲一次拉全市场当日全部 - 分钟K — 冷启动 (如 10 点才开服务, 补 9:30 起缺口) / 覆盖滞后超阈值 / - 连续空轮自愈时触发。 -- 稳态增量轮: intraday.universe 传 CN_Equity_A 标的池, 单请求返回全市场 - 每只最新 3 根 (服务端上限), 靠 _write_minute_partition 的 - unique(symbol,datetime) 幂等合并滚出全天。 +- 全天修复轮: 全市场日内分时批量并发脉冲一次拉当日全部分钟K — 冷启动 + (如 10 点才开服务, 补 9:30 起缺口) / 覆盖滞后超阈值 / 连续空轮自愈时触发。 +- 稳态增量轮: 传标的池/universe 单请求返回全市场每只最新 3 根, 靠 + _write_minute_partition 的 unique(symbol,datetime) 幂等合并滚出全天。 单轮合并写入当日 kline_minute 分区, 供分钟策略 (minute_filter) 读到新鲜数据。 +数据源路由 (像其他能力一样可接入): +- 生效源由偏好 full_minute_data_provider 决定 (设置 → 数据源 → 全量分钟): + TickFlow (Expert 档, intraday.batch + intraday.universe) 或声明 full_minute + 数据集的插件/自定义源。 +- 自定义源契约 (docs/plugin-development.md): get_intraday_batch(symbols) 修复轮 + (未实现自动回退 get_minute 当日窗口); get_intraday_latest(count) 稳态增量 + (可选, 未实现自动降级为仅修复轮并放慢节奏至 >=60s)。 +- 能力门控 Cap.INTRADAY_UNIVERSE: TickFlow 按档位探测, 自定义源声明数据集且 + 被路由时由 policy._augment_custom_sources 补授 — 门控口径对两类源统一。 + 设计约束: -- Expert 专有: 能力门控 Cap.INTRADAY_UNIVERSE (全量分钟) — 仅 TickFlow - Expert 档具备, 天然排他 (自定义分钟源无此能力, 且服务本就让位插件)。 -- 修复轮并发脉冲: 全市场按 batch_size 分块 (5546/200 = 28 块) 一次打出。 - 任何 60s 滑动窗口至多一个脉冲 (28 < 48 安全 rpm); 单块失败不拖垮整轮, - 失败块单独重试一次 (见 fetch_intraday_full_market_burst)。 +- 修复轮并发脉冲: TickFlow 路径全市场按 batch_size 分块 (5546/200 = 28 块) + 一次打出。任何 60s 滑动窗口至多一个脉冲 (28 < 48 安全 rpm); 单块失败不拖垮 + 整轮, 失败块单独重试一次 (见 fetch_intraday_full_market_burst)。自定义源 + 由 provider 自管批量与限速 (rpm 配置/内部并发), 服务不代限。 - 稳态轮单请求: 无脉冲并发, 间隔可低至 3s; 实际节奏 = max(间隔, 单轮完成), 服务端响应 ~5s 时自动退化为响应节奏, 不会重叠请求。 -- 固定节奏: 默认 6s 一轮 (clamp [3, 300]), 不补跑 (missed 轮次直接跳过)。 +- 固定节奏: 默认 6s 一轮 (clamp [3, 300]), 不补跑 (missed 轮次直接跳过); + 仅修复轮的自定义源下限抬到 60s (全天批量打不住 6s 节奏)。 - 仅连续竞价时段运行 (9:30-11:30 / 13:00-15:00), 午休/收盘自动暂停与恢复; 午休后恢复因覆盖滞后会多跑一次修复轮, 幂等无害。 - 不与其他分钟能力冲突: 与 盘后分钟同步 (kline.minute.batch) / 分时监控路径 分属不同限流池; 落盘走 _write_minute_partition 的 unique(symbol,datetime) 合并, 与盘后同步写同一分区安全幂等。 -- 数据源插件化让位: 配置了自定义分钟源 (minute_data_provider != tickflow) 时 - 服务不启动 — 盘中增量交由插件自管, 本服务不抢占。 -分层: 本模块只做调度/落盘/状态; TickFlow SDK 调用全部在 kline_sync 边界层 -(fetch_intraday_full_market_burst / fetch_intraday_universe_increment), +分层: 本模块只做调度/落盘/状态; 取数全部经 kline_sync 边界层 +(TickFlow: fetch_intraday_full_market_burst / fetch_intraday_universe_increment; +自定义: fetch_intraday_custom_batch / fetch_intraday_custom_latest), 保持插件化边界不泄漏。 """ from __future__ import annotations @@ -58,6 +65,9 @@ _LOOP_STEP_S = 2.0 _REPAIR_LAG_MINUTES = 3.0 # 连续空轮达到该次数 → 强制全天修复轮 (自愈 universe 端点持续异常)。 _EMPTY_ROUNDS_TO_REPAIR = 2 +# 仅修复轮的自定义源 (无 get_intraday_latest 廉价增量): 全天批量节奏下限, +# 6s 一轮全市场拉取会把绝大多数源打爆。 +_REPAIR_ONLY_MIN_INTERVAL_S = 60 def _in_continuous_session(now=None) -> bool: @@ -124,8 +134,10 @@ class MinuteRefreshService: def capability_ok(self) -> bool: """Cap.INTRADAY_UNIVERSE (全量分钟) 存在。能力探测结果缓存在 app.state。 - 门控挂在稳态增量的主能力上; 全天修复轮用的 intraday.batch 与其 - 同属 Expert 档 (tiers.yaml), 目前两者必然同时持有。 + 门控挂在稳态增量的主能力上; TickFlow 路径全天修复轮用的 intraday.batch + 与其同属 Expert 档 (tiers.yaml), 两者必然同时持有。自定义源声明 + full_minute 数据集且被路由时, 由 policy._augment_custom_sources 补授 + 同一能力键 — 门控口径对两类源统一。 """ capset = getattr(self._app_state, "capabilities", None) if self._app_state else None if capset is None: @@ -137,19 +149,54 @@ class MinuteRefreshService: except Exception: return False - def custom_provider_active(self) -> bool: - """配置了自定义分钟源 → 让位插件, 本服务不启动。""" + def active_provider(self) -> str: + """全量分钟生效源名 ("tickflow" 或自定义源名, 偏好 full_minute_data_provider)。""" try: - return preferences.get_minute_data_provider() != "tickflow" - except Exception: - return False + return preferences.get_full_minute_data_provider() + except Exception: # noqa: BLE001 — 偏好文件异常按 TickFlow 处理 + return "tickflow" + + def _resolve_custom(self) -> tuple[object | None, str]: + """解析自定义源。返回 (provider_or_None, effective_name): + + - 偏好 tickflow / 源未声明 full_minute 数据集 / 解析异常 → (None, "tickflow") + (与 minute 数据集同纪律: 静默降级 TickFlow, 能力门控决定能否真正运行) + - 成功 → (provider, name) + """ + from app.services import kline_sync + + name = self.active_provider() + if name == "tickflow": + return (None, "tickflow") + provider, use_tickflow, err = kline_sync._resolve_full_minute_provider(name) + if use_tickflow: + if err is not None: + logger.warning( + "full_minute provider %s 解析失败, 本轮降级 TickFlow: %s", name, err, + ) + return (None, "tickflow") + return (provider, name) + + def _custom_supports_increment(self, provider: object) -> bool: + """自定义源是否实现 get_intraday_latest (稳态增量); 否则仅修复轮。""" + return callable(getattr(provider, "get_intraday_latest", None)) + + def repair_only(self) -> bool: + """当前生效源只能全天修复轮 (无廉价增量端点) — 节奏下限抬到 60s。""" + provider, name = self._resolve_custom() + return provider is not None and not self._custom_supports_increment(provider) + + def _effective_interval(self) -> int: + """本轮间隔: 偏好值; 仅修复轮的自定义源下限 60s (全天批量打不住 6s 节奏)。""" + interval = preferences.get_minute_refresh_interval() + if self.repair_only(): + return max(interval, _REPAIR_ONLY_MIN_INTERVAL_S) + return interval def _gate_reason(self) -> str | None: """返回本轮不执行的原因 (None = 放行)。""" if not preferences.get_minute_refresh_enabled(): return "disabled" - if self.custom_provider_active(): - return "custom_minute_provider" if not self.capability_ok(): return "capability" if not _in_continuous_session(): @@ -171,7 +218,7 @@ class MinuteRefreshService: try: reason = self._gate_reason() if reason is None: - interval = preferences.get_minute_refresh_interval() + interval = self._effective_interval() started = time.time() self._run_round() # 固定节奏: 下一轮 = max(本轮起点+间隔, 本轮完成), 不补跑 @@ -234,13 +281,21 @@ class MinuteRefreshService: from app.services import kline_sync t0 = time.perf_counter() + custom, provider_name = self._resolve_custom() mode = self._select_mode() + if mode == "increment" and custom is not None and not self._custom_supports_increment(custom): + # 无廉价增量端点的源: 增量轮退化为修复轮 (全天批量幂等覆盖, 数据不丢) + mode = "full" mode_label = "增量" if mode == "increment" else "全天修复" with self._round_lock: if mode == "increment": # fetch 计时只覆盖网络取数; full 分支的 universe 维表读取不计入 fetch_started = time.perf_counter() - df, requests = kline_sync.fetch_intraday_universe_increment() + if custom is not None: + latest = kline_sync.fetch_intraday_custom_latest(custom, provider_name) + df, requests = latest if latest is not None else (pl.DataFrame(), 0) + else: + df, requests = kline_sync.fetch_intraday_universe_increment() self._state.last_symbols = ( df["symbol"].n_unique() if not df.is_empty() else 0 ) @@ -251,9 +306,14 @@ class MinuteRefreshService: self._state.last_error = "empty universe (instruments 未加载)" logger.warning("全量分钟[%s] 本轮中止: 标的池为空 (instruments 未加载)", mode_label) return - capset = getattr(self._app_state, "capabilities", None) if self._app_state else None fetch_started = time.perf_counter() - df, requests = kline_sync.fetch_intraday_full_market_burst(symbols, capset) + if custom is not None: + df, requests = kline_sync.fetch_intraday_custom_batch( + custom, provider_name, symbols, + ) + else: + capset = getattr(self._app_state, "capabilities", None) if self._app_state else None + df, requests = kline_sync.fetch_intraday_full_market_burst(symbols, capset) fetch_ms = (time.perf_counter() - fetch_started) * 1000 self._state.last_requests = requests if df.is_empty(): @@ -325,9 +385,11 @@ class MinuteRefreshService: return { "enabled": enabled, "running": running, + "provider": self.active_provider(), + "provider_effective": self._resolve_custom()[1], + "repair_only": self.repair_only(), "interval_seconds": preferences.get_minute_refresh_interval(), "capability_ok": self.capability_ok(), - "custom_provider_active": self.custom_provider_active(), "in_trading_hours": _in_continuous_session(), "gate_reason": gate if (enabled and running) else (gate or "disabled"), "rounds": self._state.rounds, @@ -342,8 +404,8 @@ class MinuteRefreshService: } def trigger_manual_round(self) -> dict[str, Any]: - """手动触发一轮 (无视时段门控, 但仍受能力/插件门控); 供状态页「立即刷新」。""" - if self.custom_provider_active() or not self.capability_ok(): + """手动触发一轮 (无视时段门控, 但仍受能力门控); 供状态页「立即刷新」。""" + if not self.capability_ok(): return {"ok": False, "reason": self._gate_reason() or "capability"} threading.Thread(target=self._run_round, daemon=True, name="minute-refresh-manual").start() return {"ok": True} diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index f5daaa6..ef07406 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -262,6 +262,11 @@ def get_minute_data_provider() -> str: return provider if provider in _allowed_data_providers() else "tickflow" +def get_full_minute_data_provider() -> str: + provider = str(load().get("full_minute_data_provider", "tickflow") or "tickflow").lower() + return provider if provider in _allowed_data_providers() else "tickflow" + + def get_depth5_data_provider() -> str: provider = str(load().get("depth5_data_provider", "tickflow") or "tickflow").lower() return provider if provider in _allowed_data_providers() else "tickflow" diff --git a/backend/app/tickflow/policy.py b/backend/app/tickflow/policy.py index a54802f..8c1b286 100644 --- a/backend/app/tickflow/policy.py +++ b/backend/app/tickflow/policy.py @@ -311,6 +311,7 @@ _DATASET_CAP_MAP: tuple[tuple[str, Cap], ...] = ( ("adj_factor", Cap.ADJ_FACTOR), ("minute", Cap.KLINE_MINUTE_BATCH), ("financial", Cap.FINANCIAL), + ("full_minute", Cap.INTRADAY_UNIVERSE), ) @@ -327,6 +328,7 @@ def _augment_custom_sources(capset: CapabilitySet) -> None: "adj_factor": adj_provider, "minute": preferences.get_minute_data_provider(), "financial": preferences.get_financial_provider(), + "full_minute": preferences.get_full_minute_data_provider(), } for dataset, cap in _DATASET_CAP_MAP: provider = active_providers[dataset] diff --git a/backend/tests/test_capability_augment.py b/backend/tests/test_capability_augment.py index 0044649..e2ca9e0 100644 --- a/backend/tests/test_capability_augment.py +++ b/backend/tests/test_capability_augment.py @@ -1,6 +1,6 @@ """能力标准统一: 自定义/插件数据源能力增广回归测试。 -对应 _augment_custom_sources 的数据集→能力映射 (daily/adj_factor/minute/financial): +对应 _augment_custom_sources 的数据集→能力映射 (daily/adj_factor/minute/financial/full_minute): 某数据集的当前 provider 非 tickflow 且声明了该数据集 → grant 对应能力; 取数路由仍按 preferences 分流, 不会误调 TickFlow。 """ @@ -13,13 +13,14 @@ from app.tickflow.policy import _augment_custom_sources def _set_providers(monkeypatch, *, daily="tickflow", adj="tickflow", - minute="tickflow", financial="tickflow") -> None: + minute="tickflow", financial="tickflow", full_minute="tickflow") -> None: """mock preferences 各数据集 provider getter。""" from app.services import preferences monkeypatch.setattr(preferences, "get_daily_data_provider", lambda: daily) monkeypatch.setattr(preferences, "get_adj_factor_provider", lambda: adj) monkeypatch.setattr(preferences, "get_minute_data_provider", lambda: minute) monkeypatch.setattr(preferences, "get_financial_provider", lambda: financial) + monkeypatch.setattr(preferences, "get_full_minute_data_provider", lambda: full_minute) def _set_datasets(monkeypatch, datasets: set[str]) -> None: @@ -51,6 +52,27 @@ def test_adj_custom_source_grants_adj_factor(monkeypatch): assert capset.has(Cap.ADJ_FACTOR) +def test_full_minute_custom_source_grants_intraday_universe(monkeypatch): + """全量分钟: 声明 full_minute 数据集且被路由 → 补授 INTRADAY_UNIVERSE, + minute_refresh 服务门控与 TickFlow Expert 口径统一。""" + _set_providers(monkeypatch, full_minute="mock_src") + _set_datasets(monkeypatch, {"full_minute"}) + capset = CapabilitySet() + _augment_custom_sources(capset) + assert capset.has(Cap.INTRADAY_UNIVERSE) + # 声明了 full_minute 不等于声明 minute → 不补逐标的分钟K能力 + assert not capset.has(Cap.KLINE_MINUTE_BATCH) + + +def test_full_minute_dataset_without_routing_not_granted(monkeypatch): + """源声明了 full_minute 但路由仍是 tickflow → 不增广 (TickFlow 档位自决)。""" + _set_providers(monkeypatch, full_minute="tickflow") + _set_datasets(monkeypatch, {"full_minute"}) + capset = CapabilitySet() + _augment_custom_sources(capset) + assert not capset.has(Cap.INTRADAY_UNIVERSE) + + def test_minute_custom_source_grants_minute_batch(monkeypatch): """原有 minute 增广行为保持。""" _set_providers(monkeypatch, minute="mock_src") diff --git a/backend/tests/test_capability_matrix.py b/backend/tests/test_capability_matrix.py index 351de72..3fc7b00 100644 --- a/backend/tests/test_capability_matrix.py +++ b/backend/tests/test_capability_matrix.py @@ -17,6 +17,7 @@ DEFAULT_CURRENT = { "daily_data_provider": "tickflow", "adj_factor_provider": "tickflow", "minute_data_provider": "tickflow", + "full_minute_data_provider": "tickflow", "depth5_data_provider": "tickflow", "realtime_data_provider": "tickflow", "financial_data_provider": "tickflow", @@ -42,7 +43,7 @@ def test_registry_covers_all_routing_fields(): "realtime", "daily", "minute", "full_minute", "depth5", "adj_factor", "financial", } full_minute = next(c for c in CAPABILITY_REGISTRY if c["id"] == "full_minute") - assert full_minute["field"] is None + assert full_minute["field"] == "full_minute_data_provider" assert full_minute["tf_tier"] == "expert" for cap in CAPABILITY_REGISTRY: assert cap["default"] == "tickflow" @@ -237,20 +238,33 @@ def test_unknown_current_display_falls_back_to_name(monkeypatch): assert caps["realtime"]["effective_display"] == "ghost" -def test_full_minute_row_is_non_routable_expert_only(monkeypatch): - """全量分钟行: field=None 不可路由, 生效源恒为 TickFlow, 按 expert 档判定可用。""" - _fake_sources(monkeypatch, []) - # 即使有插件声明别的数据集也不会成为全量分钟候选 (契约不开放该数据集) +def test_full_minute_routable_like_other_capabilities(monkeypatch): + """全量分钟行: 与其他能力同样可路由 — 声明 full_minute 数据集的源进候选, + 路由到它则 usable=True (TickFlow 档位不足也不拦); 未路由且档位不足才不可用。""" + _fake_sources( + monkeypatch, + [], + [{"name": "myfm", "display_name": "MyFM", "datasets": ["full_minute"]}], + ) caps = _by_id(build_capability_matrix(dict(DEFAULT_CURRENT), tickflow_tier="expert")) fm = caps["full_minute"] - assert fm["field"] is None - assert [c["name"] for c in fm["candidates"]] == ["tickflow"] + assert fm["field"] == "full_minute_data_provider" + assert [c["name"] for c in fm["candidates"]] == ["tickflow", "myfm"] assert fm["usable"] is True assert fm["tf_available"] is True assert fm["effective"] == "tickflow" - caps_pro = _by_id(build_capability_matrix(dict(DEFAULT_CURRENT), tickflow_tier="pro")) + # TickFlow 档位不足 (pro) 但路由到声明该数据集的自定义源 → 同样可用 + routed = dict(DEFAULT_CURRENT, full_minute_data_provider="myfm") + caps_pro = _by_id(build_capability_matrix(routed, tickflow_tier="pro")) fm_pro = caps_pro["full_minute"] - assert fm_pro["candidates"] == [] - assert fm_pro["usable"] is False + assert [c["name"] for c in fm_pro["candidates"]] == ["myfm"] + assert fm_pro["usable"] is True assert fm_pro["tf_available"] is False + assert fm_pro["effective"] == "myfm" + + # 档位不足且未路由 → 不可用 + caps_pro_default = _by_id(build_capability_matrix(dict(DEFAULT_CURRENT), tickflow_tier="pro")) + fm_default = caps_pro_default["full_minute"] + assert [c["name"] for c in fm_default["candidates"]] == ["myfm"] + assert fm_default["usable"] is False diff --git a/backend/tests/test_minute_refresh.py b/backend/tests/test_minute_refresh.py index 7aada3e..0d775bb 100644 --- a/backend/tests/test_minute_refresh.py +++ b/backend/tests/test_minute_refresh.py @@ -2,13 +2,15 @@ 覆盖: - 连续竞价时段判定 (含边界) -- 门控链: 开关关闭 / 自定义分钟源让位 / 能力缺失 / 时段外 / 放行 +- 门控链: 开关关闭 / 能力缺失 / 时段外 / 放行 (自定义源与 TickFlow 统一口径) +- 数据源路由: full_minute 偏好 → 自定义源 (get_intraday_batch / get_intraday_latest / + get_minute 回退) 或降级 TickFlow; 仅修复轮源 60s 节奏下限 - 单轮: mock 边界层脉冲 + 落盘, 校验状态字段与 universe 来源 -- 偏好读写: 默认关闭、间隔 clamp [60, 300] +- 偏好读写: 默认关闭、间隔 clamp [3, 120] - API: /minute-refresh/status 无服务时 available=false -不发起真实网络请求: fetch_intraday_full_market_burst 与 _write_minute_partition -均 monkeypatch 替换。 +不发起真实网络请求: TickFlow 侧边界函数与 _write_minute_partition 均 +monkeypatch 替换; 自定义源侧用内存 fake provider 走真实边界包装。 """ from __future__ import annotations @@ -70,12 +72,20 @@ def test_continuous_session_boundaries(): # ── 门控链 ────────────────────────────────────────────────────────── -def _svc(tmp_path, monkeypatch, *, enabled=True, custom_provider=False, capability=True, in_hours=True): +def _svc(tmp_path, monkeypatch, *, enabled=True, capability=True, in_hours=True, + full_minute_provider="tickflow", custom=None): + """custom 非 None 时模拟「full_minute 路由到声明该数据集的自定义源」: + resolver 直接返回 fake provider (注册表在测试环境未加载)。""" _isolated_prefs(tmp_path, monkeypatch) preferences.save({"minute_refresh_enabled": enabled}) - if custom_provider: - # 模拟已注册的自定义分钟源 (真实注册表在测试环境未加载) - monkeypatch.setattr(preferences, "get_minute_data_provider", lambda: "a-stock-data") + monkeypatch.setattr( + preferences, "get_full_minute_data_provider", lambda: full_minute_provider, + ) + if custom is not None: + monkeypatch.setattr( + "app.services.kline_sync._resolve_full_minute_provider", + lambda name: (custom, False, None), + ) svc = MinuteRefreshService(_FakeRepo(["600000.SH"])) svc.set_app_state(_FakeAppState(capability)) monkeypatch.setattr(minute_refresh, "_in_continuous_session", lambda now=None: in_hours) @@ -90,9 +100,15 @@ def test_gate_disabled(tmp_path, monkeypatch): assert _svc(tmp_path, monkeypatch, enabled=False)._gate_reason() == "disabled" -def test_gate_custom_provider_yields(tmp_path, monkeypatch): - svc = _svc(tmp_path, monkeypatch, custom_provider=True) - assert svc._gate_reason() == "custom_minute_provider" +def test_gate_full_minute_custom_provider_runs(tmp_path, monkeypatch): + """全量分钟路由到自定义源: 不再让位 — 能力口径统一 (capset 增广后放行)。""" + class _P: # noqa: D401 — fake provider + def get_intraday_batch(self, symbols, count=300, asset_type="stock"): + return pl.DataFrame() + svc = _svc(tmp_path, monkeypatch, full_minute_provider="myfm", custom=_P()) + assert svc._gate_reason() is None + assert svc.active_provider() == "myfm" + assert svc.status()["provider"] == "myfm" def test_gate_capability_missing(tmp_path, monkeypatch): @@ -108,7 +124,7 @@ def test_gate_outside_trading_hours(tmp_path, monkeypatch): def test_gate_pass(tmp_path, monkeypatch): svc = _svc(tmp_path, monkeypatch) assert svc._gate_reason() is None - assert svc.capability_ok() and not svc.custom_provider_active() + assert svc.capability_ok() and svc.active_provider() == "tickflow" # ── 单轮 ──────────────────────────────────────────────────────────── @@ -253,6 +269,131 @@ def test_consecutive_empty_rounds_escalate_to_full(tmp_path, monkeypatch): assert svc.status()["last_mode"] == "full" +# ── 自定义源 (full_minute 路由) 轮次 ──────────────────────────────── + + +class _FakeCustomProvider: + """内存 fake: 按 flags 暴露 get_intraday_batch / get_intraday_latest / get_minute。""" + + def __init__(self, *, batch=False, latest=False, minute=False): + self.calls: dict = {} + # 未启用的方法置 None (实例属性遮蔽类方法) — getattr 返回 None → 边界 + # 包装按"未实现"处理, 与真实 provider 只暴露部分方法的形状一致 + if not batch: + self.get_intraday_batch = None + if not latest: + self.get_intraday_latest = None + if not minute: + self.get_minute = None + + def get_intraday_batch(self, symbols, count=300, asset_type="stock"): + self.calls["batch"] = list(symbols) + return _full_df() + + def get_intraday_latest(self, symbols=None, count=3): + self.calls["latest"] = count + return _inc_df() + + def get_minute(self, symbols, start_time=None, end_time=None, + asset_type="stock", freq="1m", on_chunk_done=None): + self.calls["minute"] = { + "symbols": list(symbols), + "start": start_time, "end": end_time, + } + if on_chunk_done: + on_chunk_done(1, 4) + return _full_df() + + +def _patch_write(monkeypatch) -> dict: + calls = {"rows": None} + monkeypatch.setattr( + "app.services.kline_sync._write_minute_partition", + lambda df, minute_dir: calls.update(rows=df.height) or df.height, + ) + return calls + + +def test_custom_provider_full_round_via_intraday_batch(tmp_path, monkeypatch): + """修复轮走 provider.get_intraday_batch (真实边界包装含时区守卫)。""" + provider = _FakeCustomProvider(batch=True) + svc = _svc(tmp_path, monkeypatch, full_minute_provider="myfm", custom=provider) + monkeypatch.setattr( + minute_refresh.MinuteRefreshService, "_today_coverage_lag_minutes", + lambda self: None, # 冷启动 → 全天修复 + ) + write_calls = _patch_write(monkeypatch) + svc._run_round() + assert provider.calls["batch"] == ["600000.SH"] + assert write_calls["rows"] == 2 + st = svc.status() + assert st["last_mode"] == "full" + assert st["provider"] == "myfm" + assert st["provider_effective"] == "myfm" + + +def test_custom_provider_increment_round_via_latest(tmp_path, monkeypatch): + """实现 get_intraday_latest 的源: 覆盖新鲜时走稳态增量, 不打全天批量。""" + provider = _FakeCustomProvider(batch=True, latest=True) + svc = _svc(tmp_path, monkeypatch, full_minute_provider="myfm", custom=provider) + monkeypatch.setattr( + minute_refresh.MinuteRefreshService, "_today_coverage_lag_minutes", + lambda self: 0.2, + ) + _patch_write(monkeypatch) + svc._run_round() + assert provider.calls == {"latest": 3} + st = svc.status() + assert st["last_mode"] == "increment" + assert st["last_requests"] == 1 + assert st["repair_only"] is False + + +def test_custom_provider_latest_missing_forces_full_with_minute_fallback(tmp_path, monkeypatch): + """无 get_intraday_latest: 增量轮退化为修复轮, 批量未实现时回退 get_minute + (当日窗口), 节奏下限抬到 60s。""" + provider = _FakeCustomProvider(minute=True) + svc = _svc(tmp_path, monkeypatch, full_minute_provider="myfm", custom=provider) + monkeypatch.setattr( + minute_refresh.MinuteRefreshService, "_today_coverage_lag_minutes", + lambda self: 0.2, # 覆盖新鲜 → 本应增量 + ) + _patch_write(monkeypatch) + svc._run_round() + # 增量退化为修复: 走 get_minute 回退 (chunk 回调统计请求数) + assert "minute" in provider.calls + assert provider.calls["minute"]["symbols"] == ["600000.SH"] + assert svc.status()["last_mode"] == "full" + assert svc.status()["last_requests"] == 4 + assert svc.repair_only() is True + assert svc._effective_interval() == 60 + + +def test_custom_provider_unresolved_degrades_to_tickflow(tmp_path, monkeypatch): + """路由指向未声明 full_minute 数据集的源 (真实 resolver 判定) → 降级 + TickFlow 路径, 门控口径不变。""" + svc = _svc(tmp_path, monkeypatch, full_minute_provider="not-registered") + monkeypatch.setattr( + minute_refresh.MinuteRefreshService, "_today_coverage_lag_minutes", + lambda self: 0.2, + ) + _patch_write(monkeypatch) + modes: list[str] = [] + monkeypatch.setattr( + "app.services.kline_sync.fetch_intraday_universe_increment", + lambda *a, **k: (modes.append("tf-inc"), (_inc_df(), 1))[1], + ) + monkeypatch.setattr( + "app.services.kline_sync.fetch_intraday_full_market_burst", + lambda symbols, capset, *, count=300: (modes.append("tf-full"), (_full_df(), 28))[1], + ) + svc._run_round() + assert modes == ["tf-inc"] + st = svc.status() + assert st["provider_effective"] == "tickflow" + assert st["last_mode"] == "increment" + + def test_status_reports_gate_reason_when_stopped(tmp_path, monkeypatch): svc = _svc(tmp_path, monkeypatch, enabled=False) st = svc.status() diff --git a/backend/tests/test_trading_day.py b/backend/tests/test_trading_day.py index baa6322..586c8a0 100644 --- a/backend/tests/test_trading_day.py +++ b/backend/tests/test_trading_day.py @@ -205,7 +205,6 @@ def _minute_service(monkeypatch): "app.services.minute_refresh.preferences.get_minute_refresh_enabled", lambda: True, ) - monkeypatch.setattr(svc, "custom_provider_active", lambda: False) monkeypatch.setattr(svc, "capability_ok", lambda: True) return svc diff --git a/docs/configuration.md b/docs/configuration.md index 631798f..c5fdcd0 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -33,16 +33,19 @@ TickFlow 是内置默认数据源;同时支持插件化接入第三方数据源( ### 全量分钟 (full_minute) -「全量分钟」是一项**独立能力**(能力键 `full_minute`,探测名 `intraday.universe`),不是某个档位的封闭功能:盘中把全市场当日 1 分钟 K 持续增量落盘到本地 `data/kline_minute/` 当日分区,分钟策略(`minute_filter`)与分时视图即可读到新鲜数据。**当前接入方式为 TickFlow Expert** — 配置 Expert 档 Key 即可使用。 +「全量分钟」是一项**独立能力**(能力键 `full_minute`,探测名 `intraday.universe`),与其他能力同样**可路由**:盘中把全市场当日 1 分钟 K 持续增量落盘到本地 `data/kline_minute/` 当日分区,分钟策略(`minute_filter`)与分时视图即可读到新鲜数据。接入方式二选一: + +- **TickFlow Expert**:配置 Expert 档 Key,零配置即用(修复轮 `intraday.batch` + 稳态 `intraday.universe` 单请求增量) +- **插件/自定义源**:声明 `full_minute` 数据集并在 **设置 → 数据源 → 全量分钟** 路由到该源 — Python 插件实现 `get_intraday_batch`(必需)/`get_intraday_latest`(可选,未实现自动降级仅修复轮、节奏下限 60s);YAML 声明式源数据集配置与 `minute` 同形(仅修复轮语义)。契约细节见 [plugin-development.md](./plugin-development.md) 与 [custom-data-source.md](./custom-data-source.md) 接入步骤: -1. **设置 → 凭据与能力** 配置 TickFlow API Key(Expert 档),点「重新检测」,能力列表出现「全量分钟」 +1. **设置 → 凭据与能力**(TickFlow 路径)配置 API Key(Expert 档),或在 **设置 → 数据源** 声明/安装提供 `full_minute` 的源并路由;点「重新检测」后能力列表出现「全量分钟」 2. **开启实时行情**后落盘服务自动启动;仅连续竞价时段(9:30–11:30 / 13:00–15:00)运行,午休/收盘自动暂停与恢复 -3. 冷启动(如 10 点才开服务)自动触发**全天修复轮**,一次并发脉冲补齐 9:30 起的全部缺口;稳态走**增量轮**(默认 6 秒一轮,可配 3–120 秒),全市场每只取最新 3 根幂等合并滚出全天 +3. 冷启动(如 10 点才开服务)自动触发**全天修复轮**,一次批量补齐 9:30 起的全部缺口;稳态走**增量轮**(默认 6 秒一轮,可配 3–120 秒),幂等合并滚出全天 4. 与盘后分钟同步写同一分区(`unique(symbol, datetime)` 幂等合并),互不冲突 -说明:标的池为 A 股股票(CN_Equity_A),ETF 不在内(分时走批量补拉路径);覆盖滞后超阈值或连续空轮会自动再跑修复轮自愈。该数据集暂未纳入插件 loader 白名单;但配置了自定义分钟源(`minute_data_provider ≠ tickflow`)时,内置服务自动让位不启动,盘中分时由你的数据源按需供给(见 [plugin-development.md](./plugin-development.md))。 +说明:标的池为 A 股股票(CN_Equity_A),ETF 不在内(分时走批量补拉路径);覆盖滞后超阈值或连续空轮会自动再跑修复轮自愈。 --- diff --git a/docs/custom-data-source.md b/docs/custom-data-source.md index 104d701..83720ee 100644 --- a/docs/custom-data-source.md +++ b/docs/custom-data-source.md @@ -4,7 +4,7 @@ ## 支持范围 -当前自定义源支持五类数据: +当前自定义源支持六类数据: | 数据集 | 配置名 | 说明 | | --- | --- | --- | @@ -12,10 +12,15 @@ | 除权因子 | `adj_factor` | 批量返回一组股票的复权因子 | | 实时行情 | `realtime` | 返回全市场快照,用于盘中 enriched 增量计算 | | 分钟K | `minute` | 返回 1m 分钟K(需映射出 symbol / datetime / OHLC / 量额) | +| 全量分钟 | `full_minute` | 与 `minute` 同形;声明后可被路由为「全量分钟」生效源,内置服务盘中按当日窗口全市场批量落盘(仅修复轮语义,节奏下限 60s) | | 财务数据 | `financial` | 一个配置覆盖全部财务表,请求时把表名作为参数传给上游;字段由数据源决定,仅需映射出 symbol | 深度盘口(depth5)暂无数据集契约,仍由 TickFlow 提供。 +`full_minute` 声明式源只提供修复轮(当日窗口批量);廉价增量端点 +(`get_intraday_latest`)是 Python 插件契约,见 +[plugin-development.md](./plugin-development.md)。 + ## 配置位置 把 YAML 放到运行数据目录下: @@ -294,7 +299,7 @@ cp docs/examples/custom-data-source/mock_source.yaml data/data_sources/mock_sour # 上游若返回百分数值 (3.66 表示 3.66%), 在 realtime 数据集声明 pct_unit: percent, # 不要依赖数值自动识别; 逐列转换也可用 transforms: turnover_rate: "value / 100" -分钟K (minute): +分钟K (minute) 与 全量分钟 (full_minute, 字段同 minute): symbol = 股票代码 # datetime 必须是北京时间墙钟 (如 2026-08-28 09:35:00), 不要返回 UTC; # 入口守卫会自动纠偏 UTC 特征帧, 但契约仍要求源头写对 diff --git a/docs/features.md b/docs/features.md index 409cba4..d417550 100644 --- a/docs/features.md +++ b/docs/features.md @@ -134,7 +134,7 @@ ### 全量分钟(盘中增量落盘) -「全量分钟」是独立能力(`full_minute`):盘中将全市场当日 1 分钟 K 持续增量落盘 `data/kline_minute/`,分钟策略与分时视图即取即用。**当前通过 TickFlow Expert 接入** — 配 Key 检测出能力后,开启实时行情即自动运行(冷启动全天修复轮 + 稳态增量轮);配置自定义分钟源时内置服务自动让位。 +「全量分钟」是独立能力(`full_minute`),**与其他能力同样可路由**:盘中将全市场当日 1 分钟 K 持续增量落盘 `data/kline_minute/`,分钟策略与分时视图即取即用。接入方式:TickFlow Expert,或声明 `full_minute` 数据集的插件/自定义源(设置 → 数据源 → 全量分钟 路由);配好能力后开启实时行情即自动运行(冷启动全天修复轮 + 稳态增量轮)。 详见 [configuration.md → 全量分钟](./configuration.md#全量分钟-full_minute)。 diff --git a/docs/plugin-development.md b/docs/plugin-development.md index 252a1e7..1d7d422 100644 --- a/docs/plugin-development.md +++ b/docs/plugin-development.md @@ -149,6 +149,14 @@ class MyProvider: on_chunk_done=None, freq="1m") -> pl.DataFrame: """分钟K: [symbol, datetime(北京墙钟), open, high, low, close, volume, amount]""" + def get_intraday_batch(self, symbols, count=300, asset_type="stock") -> pl.DataFrame: + """(声明 full_minute 数据集时实现) 全量分钟修复轮: 给定标的当日 1 分钟K, + canonical 8 列同 get_minute; 内部自行分块/限速。""" + + def get_intraday_latest(self, symbols=None, count=3) -> pl.DataFrame: + """(可选, full_minute 稳态增量轮) 尽量单请求返回全市场每只最新 count 根; + 未实现则服务降级为仅修复轮 (节奏下限 60s)。""" + def get_realtime(self) -> list[dict]: """全市场实时快照 → list[dict]。失败软返回 [], 不抛异常(不阻断轮询线程)。""" @@ -184,10 +192,25 @@ class MyProvider: 深历史（TickFlow 基准）。浅源（如 stock-sdk 免费分时仅保留最近 5 个交易日）声明后， 个股分时档位自动收窄为可行选项并默认 5 日，深源默认 20 日。 -> **全量分钟与插件源的关系**:「全量分钟」(`intraday.universe`,盘中全市场分钟增量 -> 落盘)是独立能力,当前经 TickFlow Expert 接入、由内置服务(`minute_refresh`)执行, -> 该数据集暂未纳入插件 loader 白名单。但配置了自定义分钟源(`minute_data_provider -> != tickflow`)时,内置服务自动让位不启动,盘中分时由你的源经 `get_minute` 按需供给。 +> **全量分钟 (full_minute) 数据集契约**:声明 `full_minute` 数据集并把 +> `full_minute_data_provider` 路由到你的源,即接入「全量分钟」能力(盘中全市场 +> 当日分钟K增量落盘,由内置服务 `minute_refresh` 调度,与 TickFlow Expert 同一 +> 能力键 `intraday.universe`)。需实现: +> +> - `get_intraday_batch(symbols, count=300, asset_type="stock") -> pl.DataFrame` +> — **必须**(或已有 `get_minute` 自动回退,但强烈建议实现批量端点)。 +> 返回给定标的当日 1 分钟K,canonical 8 列 +> `[symbol, datetime(北京墙钟 naive), open, high, low, close, volume, amount]`, +> 内部自行分块/限速。服务在冷启动、覆盖断档、连续空轮时调用(修复轮)。 +> - `get_intraday_latest(symbols=None, count=3) -> pl.DataFrame` — **可选**, +> 稳态增量轮专用:尽量单请求返回全市场每只标的最新 `count` 根。未实现时 +> 服务自动降级为仅修复轮,节奏下限抬到 60s(全天批量打不住 6s 节奏)。 +> +> 两个方法的返回帧都过 `_enforce_minute_beijing_wallclock` 时区守卫(与 +> `get_minute` 同纪律);失败抛异常或返回空 df 均按空轮处理,连续空轮触发 +> 修复轮自愈。声明方式:插件在 `plugin.yaml` 的 `datasets:` 列表加入 +> `full_minute`。YAML 声明式源同样支持(数据集配置与 `minute` 同形,仅提供 +> 修复轮语义,见 [custom-data-source.md](./custom-data-source.md))。 ### 异常语义 diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 03dd8b8..e469e4e 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1485,6 +1485,7 @@ export type ProviderField = | 'daily_data_provider' | 'adj_factor_provider' | 'minute_data_provider' + | 'full_minute_data_provider' | 'depth5_data_provider' | 'realtime_data_provider' | 'financial_data_provider' @@ -1604,6 +1605,8 @@ export interface Preferences { daily_data_provider?: string adj_factor_provider?: string minute_data_provider?: string + /** 全量分钟 (盘中全市场分钟落盘) 生效源; 默认 tickflow (需 Expert 档) */ + full_minute_data_provider?: string /** 分钟源 1 分钟历史深度(交易日); null/缺省 = 深历史 (如 tickflow)。分时档位据此收窄 */ minute_history_days?: number | null depth5_data_provider?: string @@ -1797,15 +1800,17 @@ export const api = { }), }), - /** 全量分钟 (盘中全市场分钟落盘) 服务状态 (TickFlow Expert 专有) */ + /** 全量分钟 (盘中全市场分钟落盘) 服务状态 (按能力路由: TickFlow Expert 或声明 full_minute 的插件/自定义源) */ minuteRefreshStatus: () => request<{ available: boolean enabled?: boolean running?: boolean + provider?: string + provider_effective?: string + repair_only?: boolean interval_seconds?: number capability_ok?: boolean - custom_provider_active?: boolean in_trading_hours?: boolean gate_reason?: string | null rounds?: number diff --git a/frontend/src/pages/settings/DataSourceEditor.tsx b/frontend/src/pages/settings/DataSourceEditor.tsx index 26a0fdb..1596394 100644 --- a/frontend/src/pages/settings/DataSourceEditor.tsx +++ b/frontend/src/pages/settings/DataSourceEditor.tsx @@ -9,7 +9,7 @@ import { toast } from '@/components/Toast' const INPUT_CLS = 'w-full h-9 px-2.5 rounded-lg bg-base border-0 ring-1 ring-border/40 text-xs text-foreground placeholder:text-muted/30 focus:outline-none focus:ring-2 focus:ring-accent/40 transition-shadow' -const DATASETS = ['daily', 'adj_factor', 'realtime', 'minute'] as const +const DATASETS = ['daily', 'adj_factor', 'realtime', 'minute', 'full_minute'] as const type DatasetKey = typeof DATASETS[number] const DATASET_LABEL: Record = { @@ -17,6 +17,7 @@ const DATASET_LABEL: Record = { adj_factor: '除权因子', realtime: '实时行情', minute: '分钟K', + full_minute: '全量分钟', } const TARGET_FIELDS: Record = { @@ -24,6 +25,7 @@ const TARGET_FIELDS: Record = { adj_factor: ['symbol', 'trade_date', 'ex_factor'], realtime: ['symbol', 'name', 'last_price', 'prev_close', 'open', 'high', 'low', 'volume', 'amount', 'change_pct', 'change_amount', 'amplitude', 'turnover_rate', 'timestamp', 'session'], minute: ['symbol', 'datetime', 'open', 'high', 'low', 'close', 'volume', 'amount'], + full_minute: ['symbol', 'datetime', 'open', 'high', 'low', 'close', 'volume', 'amount'], } // 内部字段的中文说明 (下拉选项展示用) @@ -486,7 +488,7 @@ function DatasetDetail({ )} - {datasetKey === 'minute' && ( + {(datasetKey === 'minute' || datasetKey === 'full_minute') && ( <> = { daily_data_provider: 'tickflow', adj_factor_provider: 'tickflow', minute_data_provider: 'tickflow', + full_minute_data_provider: 'tickflow', depth5_data_provider: 'tickflow', realtime_data_provider: 'tickflow', financial_data_provider: 'tickflow', @@ -612,19 +613,13 @@ export function SettingsDataSourcesPanel({ highlight }: { highlight?: string } = daily: dailyPref, adj_factor: adjPref === 'same_as_daily' ? dailyPref : adjPref, minute: prefs.data?.minute_data_provider || 'tickflow', + full_minute: prefs.data?.full_minute_data_provider || 'tickflow', realtime: prefs.data?.realtime_data_provider || 'tickflow', depth5: prefs.data?.depth5_data_provider || 'tickflow', financial: prefs.data?.financial_data_provider || 'tickflow', } - const servingDatasets = (name: string) => { - const ids = Object.entries(effProvider).filter(([, v]) => v === name).map(([k]) => k) - if (name === 'tickflow') { - // 不可路由能力 (field=null, 如全量分钟): 仅 TickFlow 提供, usable 即服务中 - ids.push(...(matrix.data?.capabilities ?? []) - .filter(c => c.field == null && c.usable).map(c => c.id)) - } - return ids - } + const servingDatasets = (name: string) => + Object.entries(effProvider).filter(([, v]) => v === name).map(([k]) => k) const servingSetOf = (name: string) => new Set(servingDatasets(name)) const matrixCaps = matrix.data?.capabilities ?? [] diff --git a/frontend/src/pages/settings/Monitoring.tsx b/frontend/src/pages/settings/Monitoring.tsx index 5d012b0..392d017 100644 --- a/frontend/src/pages/settings/Monitoring.tsx +++ b/frontend/src/pages/settings/Monitoring.tsx @@ -366,19 +366,19 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = {/* ========== 右列 ========== */}
- {/* 全量分钟 (TickFlow Expert 专有): 盘中全市场分钟落盘, intraday.universe 单请求增量 */} + {/* 全量分钟: 盘中全市场分钟落盘, 按能力路由 (TickFlow Expert 或声明 full_minute 的插件/自定义源) */} save({ minute_refresh_enabled: v })} - disabled={!hasFullMinuteCap || !!rs?.custom_provider_active} + disabled={!hasFullMinuteCap} />