Merge PR #498: fix(pipeline): 分钟同步门控与 api/kline 对齐, 自定义源按 dataset 放行

backend/app/jobs/daily_pipeline.py

@@ -637,7 +637,13 @@ def run_now(
     minute_on = preferences.get_minute_sync_enabled()
     minute_days = preferences.get_minute_sync_days()
     written_minute = 0
-    if minute_on and capset.has(Cap.KLINE_MINUTE_BATCH):
+    # 门控口径与 api/kline._minute_allowed 一致: TickFlow 需 KLINE_MINUTE_BATCH
+    # 能力; 自定义源声明 minute 数据集即放行 (resolver 异常按降级 TickFlow 处理)。
+    _, minute_fallback, _minute_resolve_err = kline_sync._resolve_minute_provider(
+        preferences.get_minute_data_provider()
+    )
+    minute_allowed = (not minute_fallback) or capset.has(Cap.KLINE_MINUTE_BATCH)
+    if minute_on and minute_allowed:
         minute_start = today - _td(days=minute_days)
         emit("sync_minute", 90, f"获取分钟K [{minute_start} ~ {today}]…")
         logger.info("sync_minute: [%s ~ %s] start", minute_start, today)
@@ -658,7 +664,7 @@ def run_now(
     else:
         skipped.append("sync_minute")
         if minute_on:
-            logger.info("sync_minute skipped: no KLINE_MINUTE_BATCH capability")
+            logger.info("sync_minute skipped: minute provider has no capability/dataset")
         else:
             logger.info("sync_minute skipped: user disabled")
 

backend/tests/test_daily_pipeline_minute_gate.py

@@ -0,0 +1,81 @@
+"""盘后管道分钟K阶段门控: 自定义 minute 源无需 TickFlow KLINE_MINUTE_BATCH 能力。
+
+门控口径与 api/kline._minute_allowed 一致 — minute_data_provider 解析为自定义源
+(声明 minute 数据集) 即放行, TickFlow 档位无能力时才跳过。
+"""
+from __future__ import annotations
+
+from datetime import date
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from app.jobs import daily_pipeline
+from app.services import data_integrity, preferences
+from app.tickflow.capabilities import CapabilitySet
+
+
+def _run_pipeline(tmp_path: Path, monkeypatch, minute_provider: str) -> tuple[dict, list[dict]]:
+    """跑 run_now 到分钟阶段, 返回 (result, sync_and_persist_minute 调用记录)。"""
+    monkeypatch.setattr(daily_pipeline.settings, "data_dir", tmp_path)
+    monkeypatch.setattr(preferences, "load", lambda: {
+        "pipeline_pull_a_share": False, "pipeline_pull_etf": False,
+        "pipeline_regime_enabled": False, "minute_sync_enabled": True,
+        "minute_sync_days": 5, "minute_data_provider": minute_provider,
+        "adj_factor_provider": "tickflow",
+    })
+    monkeypatch.setattr(daily_pipeline.instrument_sync, "sync_instruments", lambda *_: 0)
+    monkeypatch.setattr(daily_pipeline, "_resolve_universe", lambda *_: [])
+    monkeypatch.setattr(daily_pipeline, "_invalidate", lambda *_: None)
+    monkeypatch.setattr(daily_pipeline, "_refresh_views", lambda *_: None)
+    monkeypatch.setattr(daily_pipeline, "_refresh_single_view", lambda *_: None)
+    monkeypatch.setattr(data_integrity, "scan_recent_integrity", lambda *_, **__: [])
+    if minute_provider == "tickflow":
+        monkeypatch.setattr(
+            daily_pipeline.kline_sync, "_resolve_minute_provider",
+            lambda name: (None, True, None),
+        )
+    else:
+        # 真 resolver + monkeypatch 自定义源注册表: 该源声明 minute 数据集
+        from app.data_providers import custom as custom_sources
+
+        # 绕开 provider 白名单过滤, 让门控拿到的就是被测的源名
+        monkeypatch.setattr(
+            preferences, "get_minute_data_provider", lambda: minute_provider,
+        )
+
+        monkeypatch.setattr(
+            custom_sources, "provider_has_dataset",
+            lambda name, ds: name == minute_provider and ds == "minute",
+        )
+        monkeypatch.setattr(custom_sources, "get_provider", lambda name: object())
+
+    calls: list[dict] = []
+
+    def fake_sync_minute(symbols, *args, **kwargs):
+        calls.append({"symbols": symbols, **kwargs})
+        return 1
+
+    monkeypatch.setattr(
+        daily_pipeline.kline_sync, "sync_and_persist_minute", fake_sync_minute,
+    )
+    repo = SimpleNamespace(
+        store=SimpleNamespace(data_dir=tmp_path),
+        latest_daily_date=lambda: date(2026, 9, 28),
+    )
+    return daily_pipeline.run_now(repo, CapabilitySet(set())), calls
+
+
+@pytest.mark.parametrize("provider,expect_called", [
+    ("customsrc", True),  # 自定义源声明 minute 数据集 → 放行 (本次修复的行为)
+    ("tickflow", False),  # TickFlow 无 KLINE_MINUTE_BATCH → 跳过
+])
+def test_minute_stage_gate_routes_by_provider(tmp_path, monkeypatch, provider, expect_called):
+    result, calls = _run_pipeline(tmp_path, monkeypatch, provider)
+    assert bool(calls) is expect_called
+    if expect_called:
+        assert calls[0]["days"] == 5
+        assert "sync_minute" not in result["skipped_stages"]
+    else:
+        assert "sync_minute" in result["skipped_stages"]