Merge pull request #156 from shy3130/feat/regime-filter

feat(regime): 市场状态识别系统 + 叠加策略回测过滤器

Author: shy3130

backend/app/api/backtest.py

@@ -207,6 +207,7 @@ class StrategyBacktestRequest(BaseModel):
     holding_days: int = 5
     asset_type: str = "stock"
     minute_fill: bool = False
+    regime_filter: dict | None = None
 
 
 @router.post("/strategy/run")
@@ -241,6 +242,7 @@ def strategy_run(req: StrategyBacktestRequest, request: Request):
         holding_days=req.holding_days,
         asset_type=req.asset_type,
         minute_fill=req.minute_fill,
+        regime_filter=req.regime_filter,
     )
     task = make_worker_task("backtest", settings.data_dir, cfg)
     return run_worker_task(task)
@@ -315,8 +317,9 @@ def _make_job_key(
     commission_pct: float | None = None, stamp_tax_pct: float | None = None,
     asset_type: str = "stock",
     minute_fill: bool = False,
+    regime_filter: str | None = None,
 ) -> str:
-    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}|{minute_fill}"
+    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}|{minute_fill}|{regime_filter}"
     return hashlib.md5(raw.encode()).hexdigest()[:12]
 
 
@@ -344,6 +347,7 @@ async def strategy_stream(
     holding_days: int = 5,
     asset_type: str = "stock",
     minute_fill: bool = False,
+    regime_filter: str | None = None,
 ):
     """SSE 流式策略回测: 实时推送进度, 完成后推送结果, 支持重连 (刷新/切页后恢复)。
 
@@ -383,6 +387,7 @@ async def strategy_stream(
         commission_pct, stamp_tax_pct,
         asset_type=asset_type,
         minute_fill=minute_fill,
+        regime_filter=regime_filter,
     )
 
     _cleanup_stale_jobs()
@@ -443,6 +448,7 @@ async def event_generator():
                 holding_days=int(holding_days),
                 asset_type=asset_type,
                 minute_fill=minute_fill,
+                regime_filter=json.loads(regime_filter) if regime_filter else None,
             )
 
             def _run_backtest():

backend/app/api/regime.py

@@ -0,0 +1,147 @@
+"""市场环境(regime) API — 时序查询 + 手动重算。
+
+装配逻辑在 app.services.regime_builder(纯函数), API 层薄壳 + TTL 缓存。
+"""
+from __future__ import annotations
+
+import threading
+import time
+from datetime import date
+from typing import Any
+
+from fastapi import APIRouter, Query, Request
+
+from app.services import regime_builder
+
+router = APIRouter(prefix="/api/regime", tags=["regime"])
+
+_CACHE_TTL = 5.0
+_cache: dict[str, Any] | None = None
+_cache_ts: float = 0.0
+_cache_lock = threading.Lock()
+
+
+def invalidate_regime_cache() -> None:
+    """清空 regime 查询缓存。批算/重算后调用。"""
+    global _cache, _cache_ts
+    with _cache_lock:
+        _cache = None
+        _cache_ts = 0.0
+
+
+def _data_dir(request: Request) -> Any:
+    return request.app.state.repo.store.data_dir
+
+
+def _df_to_records(df) -> list[dict]:
+    """polars DataFrame → JSON 安全的 list[dict](date 转 ISO 字符串)。"""
+    if df is None or df.is_empty():
+        return []
+    records = []
+    for r in df.to_dicts():
+        if "date" in r and r["date"] is not None:
+            r["date"] = str(r["date"])
+        records.append(r)
+    return records
+
+
+@router.get("/history")
+def regime_history(
+    request: Request,
+    start: date | None = Query(None),
+    end: date | None = Query(None),
+    limit: int = Query(120, ge=1, le=1000),
+):
+    """历史环境时序(含状态/指标)。默认最近 N 天。"""
+    global _cache, _cache_ts
+    cache_key = f"hist|{start}|{end}|{limit}"
+    with _cache_lock:
+        if (
+            _cache is not None
+            and _cache.get("key") == cache_key
+            and (time.time() - _cache_ts) < _CACHE_TTL
+        ):
+            return _cache["data"]
+
+    df = regime_builder.load_regime_history(_data_dir(request))
+    if df.is_empty():
+        result: dict = {"rows": [], "total": 0}
+    else:
+        if start:
+            df = df.filter(pl_col_date(df, ">=", start))
+        if end:
+            df = df.filter(pl_col_date(df, "<=", end))
+        df = df.sort("date", descending=True).head(limit).sort("date")
+        rows = _df_to_records(df)
+        result = {"rows": rows, "total": len(rows)}
+
+    with _cache_lock:
+        _cache = {"key": cache_key, "data": result}
+        _cache_ts = time.time()
+    return result
+
+
+def pl_col_date(df, op: str, value: date):
+    """polars 日期过滤辅助(避免重复 import)。"""
+    import polars as pl
+
+    col = pl.col("date")
+    return col >= value if op == ">=" else col <= value
+
+
+@router.get("/latest")
+def regime_latest(request: Request):
+    """最新一日环境(轻量)。"""
+    df = regime_builder.load_regime_history(_data_dir(request))
+    if df.is_empty():
+        return {"row": None}
+    latest = df.sort("date", descending=True).head(1)
+    rows = _df_to_records(latest)
+    return {"row": rows[0] if rows else None}
+
+
+@router.get("/states")
+def regime_states(
+    request: Request,
+    days: int = Query(60, ge=1, le=1000),
+):
+    """状态分布统计(各状态天数/占比)。"""
+    df = regime_builder.load_regime_history(_data_dir(request))
+    if df.is_empty():
+        return {"distribution": [], "days": 0}
+    df = df.sort("date", descending=True).head(days)
+    total = df.height
+    counts = df.group_by("state").len().sort("len", descending=True)
+    distribution = [
+        {
+            "state": r["state"],
+            "label": regime_builder.STATE_LABELS.get(r["state"], r["state"]),
+            "count": r["len"],
+            "pct": round(r["len"] / total * 100, 1) if total else 0,
+        }
+        for r in counts.to_dicts()
+    ]
+    return {"distribution": distribution, "days": total}
+
+
+@router.get("/coverage")
+def regime_coverage(request: Request):
+    """regime 数据覆盖元信息(供数据画像)。"""
+    return regime_builder.get_regime_coverage(_data_dir(request))
+
+
+@router.post("/recompute")
+def regime_recompute(request: Request, start: date | None = None, end: date | None = None):
+    """手动触发重算(全量或指定区间)。管理员操作。"""
+    repo = request.app.state.repo
+    data_dir = _data_dir(request)
+    end = end or date.today()
+    if start is None:
+        # 全量: 从 enriched 最早日算到今天
+        new_rows = regime_builder.compute_regime_incremental(repo, data_dir, today=end)
+    else:
+        new_rows = regime_builder.run_regime_batch(repo, start=start, end=end)
+        if not new_rows.is_empty():
+            regime_builder.upsert_regime_history(data_dir, new_rows)
+    invalidate_regime_cache()
+    return {"ok": True, "computed": new_rows.height if not new_rows.is_empty() else 0}

backend/app/backtest/strategy.py

@@ -13,6 +13,7 @@
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass, field
 from datetime import date, timedelta
+from pathlib import Path
 from typing import Literal
 
 import numpy as np
@@ -473,6 +474,9 @@ class StrategyBacktestConfig:
     holding_days: int = 5
     # 分钟K精确成交: 开启后用当日分钟K确定穿越价/VWAP (需 Pro+ 分钟K能力)
     minute_fill: bool = False
+    # 市场环境过滤: {"states": ["strong",...], "min_score": 60}。
+    # 强制 T-1: regime[T-1] 决定 entry[T](防未来函数)。None=不过滤。
+    regime_filter: dict | None = None
 
     def __post_init__(self) -> None:
         if self.entry_fill is None:
@@ -846,6 +850,13 @@ def prepare_matrix_optimization(
             first.start,
             first.end,
         )
+        # 市场环境过滤(优化器共享, 用首个 config 的 regime_filter)
+        _rm = self._build_regime_mask(
+            market_data.timestamp_labels, first.regime_filter,
+            getattr(getattr(self.engine.repo, "store", None), "data_dir", None),
+        )
+        if _rm is not None:
+            entry_time_mask = entry_time_mask & _rm
         exit_time_mask = self._matrix_date_range_mask(
             market_data.timestamp_labels,
             first.start,
@@ -1127,6 +1138,13 @@ def _err(msg: str) -> StrategyBacktestResult:
                 config.start,
                 config.end,
             )
+            # 市场环境过滤(强制 T-1): 只叠加 entry, 不影响 exit
+            _rm = self._build_regime_mask(
+                market_data.timestamp_labels, config.regime_filter,
+                getattr(getattr(self.engine.repo, "store", None), "data_dir", None),
+            )
+            if _rm is not None:
+                entry_time_mask = entry_time_mask & _rm
             exit_time_mask = self._matrix_date_range_mask(
                 market_data.timestamp_labels,
                 config.start,
@@ -1220,6 +1238,12 @@ def _err(msg: str) -> StrategyBacktestResult:
                     config.start,
                     config.end,
                 )
+                _rm = self._build_regime_mask(
+                    market_data.timestamp_labels, config.regime_filter,
+                    getattr(getattr(self.engine.repo, "store", None), "data_dir", None),
+                )
+                if _rm is not None:
+                    entry_time_mask = entry_time_mask & _rm
                 exit_time_mask = self._matrix_date_range_mask(
                     market_data.timestamp_labels,
                     config.start,
@@ -1620,6 +1644,56 @@ def _matrix_date_range_mask(
             count=len(timestamp_labels),
         )
 
+    @staticmethod
+    def _build_regime_mask(
+        timestamp_labels: tuple[str, ...],
+        regime_filter: dict | None,
+        data_dir: Path | None,
+    ) -> np.ndarray | None:
+        """构造逐日 regime mask。强制 T-1 防未来函数: regime[T-1] 决定 entry[T]。
+
+        timestamp_labels[i] 的入场资格 = 它的"前一交易日"的 regime 是否满足条件。
+        "前一交易日"用 timestamp_labels 自身的顺序确定(回测时间轴上的前一天)。
+        边界: 首日无前一日环境 → 默认允许(不阻断)。
+        regime_filter 为 None 或无 regime 数据时返回 None(不过滤)。
+        """
+        if not regime_filter or data_dir is None:
+            return None
+        allowed_states = set(regime_filter.get("states") or [])
+        min_score = regime_filter.get("min_score")
+        if not allowed_states and min_score is None:
+            return None
+
+        from app.services import regime_builder
+        regime_df = regime_builder.load_regime_history(data_dir)
+        if regime_df.is_empty():
+            return None
+
+        # 构建 date(ISO) → (state, score) 映射
+        regime_map: dict[str, tuple[str, int]] = {}
+        for r in regime_df.iter_rows(named=True):
+            d = r.get("date")
+            ds = str(d)[:10] if d is not None else None
+            if ds:
+                regime_map[ds] = (str(r.get("state", "")), int(r.get("score", 0) or 0))
+
+        # 对每个 label, 找它的前一交易日的 regime(timestamp_labels 顺序里的前一天)
+        n = len(timestamp_labels)
+        mask = np.ones(n, dtype=bool)  # 默认允许
+        for i in range(1, n):
+            prev_label = timestamp_labels[i - 1][:10]
+            entry = regime_map.get(prev_label)
+            if entry is None:
+                continue  # 无前一日环境数据 → 允许(不阻断)
+            state, score = entry
+            ok = True
+            if allowed_states and state not in allowed_states:
+                ok = False
+            if min_score is not None and score < min_score:
+                ok = False
+            mask[i] = ok
+        return mask
+
     def _build_candidate_filter_mask(
         self,
         panel: pl.DataFrame,

backend/app/jobs/daily_pipeline.py

@@ -516,6 +516,24 @@ def _minute_chunk_progress(cur: int, tot: int, seg_label: str = "") -> None:
         else:
             logger.info("sync_minute skipped: user disabled")
 
+    # Step 2.6: 市场环境(regime) 增量计算 — enriched 已就绪后聚合环境指标。
+    # 双检测(缺口+stale), 自动补算遗漏/被覆写的日。软失败: 不阻断主管道。
+    regime_days = 0
+    try:
+        emit("compute_regime", 90, "计算市场环境…")
+        from app.services import regime_builder
+        from app.api.regime import invalidate_regime_cache
+        new_regime = regime_builder.compute_regime_incremental(repo, repo.store.data_dir)
+        regime_days = new_regime.height if not new_regime.is_empty() else 0
+        if regime_days:
+            invalidate_regime_cache()
+            logger.info("compute_regime: %d days", regime_days)
+        emit("compute_regime", 92, f"市场环境 {regime_days} 天")
+    except Exception as e:  # noqa: BLE001
+        logger.warning("compute_regime failed (soft): %s", e)
+        stage_errors.append(f"compute_regime: {e}")
+        skipped.append("regime")
+
     # Step 3: 刷新视图
     emit("refresh_views", 95, "刷新 DuckDB 视图…")
     _refresh_views(repo)
@@ -534,6 +552,7 @@ def _minute_chunk_progress(cur: int, tot: int, seg_label: str = "") -> None:
         "etf_daily_rows": written_etf_daily,
         "etf_adj_factor_symbols": etf_adj_symbols,
         "minute_rows": written_minute,
+        "regime_days": regime_days,
         "lagging_symbols": len(lagging_symbols),
         "skipped_stages": skipped,
         "stage_errors": stage_errors,

backend/app/main.py

@@ -12,7 +12,7 @@
 from fastapi.staticfiles import StaticFiles
 
 from app import __version__
-from app.api import analysis, auth as auth_api, backtest, data, ext_data, financials, indices, intraday, kline, market_recap, monitor_rules, alerts, overview, pipeline, rps, screener, settings as settings_api, signals, stock_analysis, strategy, watchlist
+from app.api import analysis, auth as auth_api, backtest, data, ext_data, financials, indices, intraday, kline, market_recap, monitor_rules, alerts, overview, pipeline, regime, rps, screener, settings as settings_api, signals, stock_analysis, strategy, watchlist
 from app.api.routes import router as core_router
 from app.config import settings
 from app.jobs import daily_pipeline
@@ -338,6 +338,7 @@ async def auth_middleware(request: Request, call_next):
 app.include_router(intraday.router)
 app.include_router(indices.router)
 app.include_router(overview.router)
+app.include_router(regime.router)
 app.include_router(analysis.router)
 app.include_router(pipeline.router)
 app.include_router(data.router)