feat(regime): 新增市场状态识别系统并接入叠加策略回测过滤器

后端
- 新增 services/regime_builder.py：5 档日级市场状态分级（strong/lean_strong/range/lean_weak/weak）
- 新增 api/regime.py：5 个 regime 查询接口
- daily_pipeline 接入 regime 构建（step 2.6，soft-fail + 双检测自愈）
- 回测 strategy.py 新增 regime_filter 过滤：采用 T-1 防未来函数，仅作用于入场信号
  在 composite / matrix_native / prepared 三处注入 entry_time_mask
- backtest API 透传 regime_filter（REST + SSE）；_make_job_key 纳入 regime_filter 以隔离缓存
- 兼容 _RepoStub 测试夹具（store 属性缺失场景）

前端
- 新增 Regime 页面、路由、导航入口、api 类型与 queryKeys
- 叠加策略回测页新增 regime 过滤器控件

测试
- 新增 tests/test_regime_builder.py（18 项），覆盖分级逻辑、T-1 防未来函数、空值降级、三处 mask 注入
- 全量后端测试 582 passed；前端 pnpm build 通过；git diff --check 无空白错误

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 strategy_stream(
                 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 @@ import uuid
 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 @@ class StrategyBacktestService:
             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 @@ class StrategyBacktestService:
                 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 @@ class StrategyBacktestService:
                     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 @@ class StrategyBacktestService:
             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 run_now(
         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 run_now(
         "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.responses import FileResponse, JSONResponse
 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 @@ app.include_router(backtest.router)
 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)

backend/app/services/regime_builder.py

@@ -0,0 +1,419 @@
+"""市场环境(regime)计算 — 纯函数模块。
+
+职责: 从已算好的 enriched 数据(含信号列)按日聚合环境指标, 用规则引擎分类离散状态,
+持久化为时序表。不重算指标(不走 compute_indicators), 不依赖 quote/depth service。
+
+性能设计:
+- run_regime_batch 用 polars group_by("date").agg(...) 一次聚合多日, 非逐日循环。
+- 数据走 repo.get_enriched_range(内存缓存, 已含信号列); 缓存不覆盖时走 scan_parquet 慢路径。
+
+与 market_overview_builder 的区别:
+- overview 面向单日详情(实时总览), 重算指标。
+- regime 面向多日聚合统计(时序分析), 只聚合不重算。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date
+from pathlib import Path
+
+import polars as pl
+
+logger = logging.getLogger(__name__)
+
+# ───────────────────────── 状态分类阈值(可调) ─────────────────────────
+# 各维度子分用归一化映射(线性插值到 0-100), 再加权求和。
+# 综合 = 赚钱效应×0.4 + 指数趋势×0.3 + 板块结构×0.2 + 活跃度×0.1
+
+WEIGHTS = {
+    "money_effect": 0.4,   # 赚钱效应(涨停/封板率/涨跌比)
+    "index_trend": 0.3,    # 指数趋势(指数涨幅/MA20上方占比)
+    "board_structure": 0.2,  # 板块结构(涨跌离散度, 简化)
+    "activity": 0.1,       # 活跃度(成交额/换手)
+}
+
+# 离散状态阈值(综合分)
+STATE_STRONG = 75       # >= 强势
+STATE_LEAN_STRONG = 60  # 60-75 偏强
+STATE_RANGE = 40        # 40-60 震荡
+STATE_LEAN_WEAK = 25    # 25-40 偏弱
+# < 25 弱势
+
+# 归一化映射的参考点(线性插值 0-100)
+_LIN = {
+    "limit_up": [(0, 0), (15, 40), (30, 70), (50, 100)],       # 涨停数
+    "seal_rate": [(0.3, 0), (0.5, 40), (0.7, 70), (0.9, 100)],  # 封板率
+    "up_ratio": [(0.4, 0), (1.0, 40), (2.0, 70), (3.0, 100)],   # 涨跌比
+    "index_pct": [(-0.02, 0), (0.0, 40), (0.01, 70), (0.02, 100)],  # 指数涨幅
+    "above_ma20": [(0.3, 0), (0.5, 40), (0.6, 70), (0.8, 100)],  # MA20上方占比
+    "amount": [(0.5e11, 0), (1e11, 40), (1.5e11, 70), (2.5e11, 100)],  # 成交额
+}
+
+STATE_LABELS = {
+    "strong": "强势",
+    "lean_strong": "偏强",
+    "range": "震荡",
+    "lean_weak": "偏弱",
+    "weak": "弱势",
+}
+
+
+def _linear_score(value: float, points: list[tuple[float, float]]) -> float:
+    """分段线性插值。points 是 [(输入值, 输出分)] 升序列表。"""
+    if value <= points[0][0]:
+        return float(points[0][1])
+    if value >= points[-1][0]:
+        return float(points[-1][1])
+    for i in range(len(points) - 1):
+        x0, y0 = points[i]
+        x1, y1 = points[i + 1]
+        if x0 <= value <= x1:
+            if x1 == x0:
+                return float(y0)
+            return float(y0 + (y1 - y0) * (value - x0) / (x1 - x0))
+    return float(points[-1][1])
+
+
+def classify_state(metrics: dict) -> tuple[str, int]:
+    """规则引擎: 多维指标 → 离散状态 + 综合分(0-100)。
+
+    各维度子分加权: 赚钱效应(涨停数/封板率/涨跌比) + 指数趋势(涨幅/MA20)
+    + 板块结构(涨跌离散度简化) + 活跃度(成交额)。
+    """
+    # 赚钱效应子分 = 涨停/封板率/涨跌比 三者平均
+    limit_up = metrics.get("limit_up", 0) or 0
+    seal_rate = metrics.get("seal_rate", 0.5) or 0.5
+    up_ratio = metrics.get("up_ratio", 1.0) or 1.0
+    money = (
+        _linear_score(limit_up, _LIN["limit_up"])
+        + _linear_score(seal_rate, _LIN["seal_rate"])
+        + _linear_score(up_ratio, _LIN["up_ratio"])
+    ) / 3
+
+    index_pct = metrics.get("index_pct", 0.0) or 0.0
+    above_ma20 = metrics.get("above_ma20_pct", 0.5) or 0.5
+    index_trend = (
+        _linear_score(index_pct, _LIN["index_pct"])
+        + _linear_score(above_ma20, _LIN["above_ma20"])
+    ) / 2
+
+    # 板块结构: 用涨跌家数比的偏离度简化(涨跌越均衡=震荡, 极端=方向明确)
+    # up_ratio 接近 1 → 震荡(中分); 远离 1 → 方向明确(高低分看方向)
+    # 已在 money_effect 的 up_ratio 体现, 这里用涨停+跌停的对比做补充
+    limit_down = metrics.get("limit_down", 0) or 0
+    if limit_up + limit_down > 0:
+        board = (limit_up - limit_down) / max(limit_up + limit_down, 1) * 50 + 50
+    else:
+        board = 50.0
+
+    total_amount = metrics.get("total_amount", 1e11) or 1e11
+    activity = _linear_score(total_amount, _LIN["amount"])
+
+    score = (
+        money * WEIGHTS["money_effect"]
+        + index_trend * WEIGHTS["index_trend"]
+        + board * WEIGHTS["board_structure"]
+        + activity * WEIGHTS["activity"]
+    )
+    score = max(0, min(100, round(score)))
+
+    if score >= STATE_STRONG:
+        state = "strong"
+    elif score >= STATE_LEAN_STRONG:
+        state = "lean_strong"
+    elif score >= STATE_RANGE:
+        state = "range"
+    elif score >= STATE_LEAN_WEAK:
+        state = "lean_weak"
+    else:
+        state = "weak"
+    return state, score
+
+
+# ───────────────────────── 批量聚合 ─────────────────────────
+
+def _aggregate_daily(df: pl.DataFrame, index_pct_map: dict | None = None) -> pl.DataFrame:
+    """对多日多 symbol 的 enriched DataFrame 按 date 聚合环境指标。
+
+    纯 polars 聚合, 不重算指标(假设 df 已含 signal_*/change_pct/ma20 等列)。
+    index_pct_map: {date: 指数涨幅} 可选, 由调用方从指数数据预先算好。
+    """
+    needed = ["date", "change_pct", "amount", "signal_limit_up",
+              "signal_limit_down", "signal_broken_limit_up",
+              "consecutive_limit_ups", "close", "ma20"]
+    avail = [c for c in needed if c in df.columns]
+    if "date" not in avail or "change_pct" not in avail:
+        return pl.DataFrame()
+
+    # 基础聚合
+    agg_exprs = []
+    if "change_pct" in avail:
+        agg_exprs.append(pl.col("change_pct"))
+    grouped = df.group_by("date").agg(
+        *[
+            pl.col("change_pct").gt(0).sum().alias("up_count")
+            if "change_pct" in avail else pl.lit(0).alias("up_count"),
+            pl.col("change_pct").lt(0).sum().alias("down_count")
+            if "change_pct" in avail else pl.lit(0).alias("down_count"),
+            pl.len().alias("total_count"),
+        ],
+        *(
+            [pl.col("signal_limit_up").cast(pl.Boolean).sum().alias("limit_up")]
+            if "signal_limit_up" in avail else [pl.lit(0).alias("limit_up")]
+        ),
+        *(
+            [pl.col("signal_limit_down").cast(pl.Boolean).sum().alias("limit_down")]
+            if "signal_limit_down" in avail else [pl.lit(0).alias("limit_down")]
+        ),
+        *(
+            [pl.col("signal_broken_limit_up").cast(pl.Boolean).sum().alias("broken_limit")]
+            if "signal_broken_limit_up" in avail else [pl.lit(0).alias("broken_limit")]
+        ),
+        *(
+            [pl.col("consecutive_limit_ups").max().alias("max_consecutive")]
+            if "consecutive_limit_ups" in avail else [pl.lit(0).alias("max_consecutive")]
+        ),
+        *(
+            [pl.col("amount").sum().alias("total_amount")]
+            if "amount" in avail else [pl.lit(0).alias("total_amount")]
+        ),
+        *(
+            [pl.col("amount").mean().alias("avg_amount")]
+            if "amount" in avail else [pl.lit(0).alias("avg_amount")]
+        ),
+    ).sort("date")
+
+    # 转成 dict 列表做后续计算(polars 表达式难表达的比率/MA20占比/分类)
+    index_pct_map = index_pct_map or {}
+    rows = []
+    for r in grouped.iter_rows(named=True):
+        up = r.get("up_count", 0) or 0
+        down = r.get("down_count", 0) or 0
+        limit_up = r.get("limit_up", 0) or 0
+        broken = r.get("broken_limit", 0) or 0
+        # MA20 上方占比
+        ma20_above = 0
+        if "close" in avail and "ma20" in avail:
+            day_df = df.filter(pl.col("date") == r["date"])
+            if not day_df.is_empty() and "ma20" in day_df.columns:
+                valid = day_df.filter(pl.col("ma20").is_not_null() & (pl.col("ma20") > 0))
+                if not valid.is_empty():
+                    above = valid.filter(pl.col("close") > pl.col("ma20"))
+                    ma20_above = above.height / valid.height
+        metrics = {
+            "limit_up": limit_up,
+            "limit_down": r.get("limit_down", 0) or 0,
+            "broken_limit": broken,
+            "max_consecutive": r.get("max_consecutive", 0) or 0,
+            "seal_rate": (limit_up / (limit_up + broken)) if (limit_up + broken) > 0 else 0.5,
+            "up_count": up,
+            "down_count": down,
+            "up_ratio": (up / down) if down > 0 else (float(up) if up > 0 else 1.0),
+            "index_pct": index_pct_map.get(r["date"], 0.0),
+            "above_ma20_pct": ma20_above,
+            "total_amount": r.get("total_amount", 0) or 0,
+            "avg_turnover": r.get("avg_amount", 0) or 0,
+        }
+        state, score = classify_state(metrics)
+        rows.append({
+            "date": r["date"],
+            "state": state,
+            "score": score,
+            "limit_up": limit_up,
+            "limit_down": metrics["limit_down"],
+            "broken_limit": broken,
+            "max_consecutive": metrics["max_consecutive"],
+            "seal_rate": round(metrics["seal_rate"], 4),
+            "up_count": up,
+            "down_count": down,
+            "up_ratio": round(metrics["up_ratio"], 4),
+            "index_pct": round(metrics["index_pct"], 4),
+            "above_ma20_pct": round(ma20_above, 4),
+            "total_amount": metrics["total_amount"],
+            "avg_turnover": metrics["avg_turnover"],
+        })
+    return pl.DataFrame(rows) if rows else pl.DataFrame()
+
+
+def _scan_enriched_fallback(repo, start: date, end: date) -> pl.DataFrame | None:
+    """缓存不覆盖时的慢路径: 一次性 scan 全部 enriched parquet + 重算指标。
+
+    仅在 regime 首次全量回填或缓存未预热时触发。返回含信号列的多日 DataFrame。
+    """
+    try:
+        enriched_dir = repo.store.data_dir / "kline_daily_enriched"
+        if not enriched_dir.exists():
+            return None
+        from app.indicators.pipeline import compute_indicators, compute_limit_signals
+        df = pl.scan_parquet(enriched_dir / "**" / "*.parquet").filter(
+            (pl.col("date") >= start) & (pl.col("date") <= end)
+        ).collect()
+        if df.is_empty():
+            return None
+        df = compute_indicators(df)
+        df = compute_limit_signals(df)
+        return df
+    except Exception as e:  # noqa: BLE001
+        logger.warning("regime scan_enriched_fallback failed: %s", e)
+        return None
+
+
+def _load_index_pct(repo, start: date, end: date, symbol: str = "000001.SH") -> dict:
+    """读取主力指数日K, 算每日涨幅 → {date: pct}。指数数量少, 单次读取可接受。"""
+    try:
+        df = repo.get_index_daily(symbol, start, end, columns=["date", "change_pct"])
+        if df.is_empty() or "change_pct" not in df.columns:
+            return {}
+        return {r["date"]: float(r["change_pct"] or 0) for r in df.iter_rows(named=True)}
+    except Exception as e:  # noqa: BLE001
+        logger.warning("regime load_index_pct failed: %s", e)
+        return {}
+
+
+def run_regime_batch(repo, start: date, end: date) -> pl.DataFrame:
+    """批算 [start, end] 的环境时序。
+
+    性能: 优先 repo.get_enriched_range(内存缓存); 缓存不覆盖走 scan_parquet 慢路径。
+    按 date group_by 聚合, 不逐日重算。返回完整时序 DataFrame(可能为空)。
+    """
+    if start > end:
+        return pl.DataFrame()
+
+    # 指数涨幅(主力指数)
+    index_pct_map = _load_index_pct(repo, start, end)
+
+    # enriched 多日数据(优先缓存)
+    df = repo.get_enriched_range(start, end)
+    if df is None or df.is_empty():
+        logger.info("regime batch: enriched cache miss [%s~%s], fallback to scan", start, end)
+        df = _scan_enriched_fallback(repo, start, end)
+    if df is None or df.is_empty():
+        logger.info("regime batch: no enriched data for [%s~%s]", start, end)
+        return pl.DataFrame()
+
+    return _aggregate_daily(df, index_pct_map)
+
+
+# ───────────────────────── 持久化(upsert) ─────────────────────────
+
+REGIME_DIR = "regime_history"
+
+
+def regime_path(data_dir: Path) -> Path:
+    return data_dir / REGIME_DIR / "part.parquet"
+
+
+def load_regime_history(data_dir: Path) -> pl.DataFrame:
+    """读取全部 regime 时序; 不存在返回空 DataFrame。"""
+    p = regime_path(data_dir)
+    if not p.exists():
+        return pl.DataFrame()
+    try:
+        return pl.read_parquet(p)
+    except Exception as e:  # noqa: BLE001
+        logger.warning("load_regime_history failed: %s", e)
+        return pl.DataFrame()
+
+
+def upsert_regime_history(data_dir: Path, new_rows: pl.DataFrame) -> None:
+    """按 date 覆盖(upsert): 重算的天覆盖旧行, 新天追加。
+
+    读旧 → anti-join 掉 new_rows 的天 → concat new_rows → 排序 → 写回。
+    """
+    if new_rows.is_empty() or "date" not in new_rows.columns:
+        return
+    p = regime_path(data_dir)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    new_dates = set(new_rows["date"].to_list())
+    old = load_regime_history(data_dir)
+    if old.is_empty():
+        combined = new_rows
+    else:
+        kept = old.filter(~pl.col("date").is_in(list(new_dates)))
+        combined = pl.concat([kept, new_rows], how="vertical_relaxed")
+    combined = combined.sort("date").unique(subset=["date"], keep="last")
+    combined.write_parquet(p)
+
+
+def get_regime_coverage(data_dir: Path) -> dict:
+    """返回 regime 时序的覆盖元信息(供数据画像/API)。"""
+    df = load_regime_history(data_dir)
+    if df.is_empty():
+        return {"rows": 0, "earliest_date": None, "latest_date": None}
+    return {
+        "rows": df.height,
+        "earliest_date": str(df["date"].min()),
+        "latest_date": str(df["date"].max()),
+    }
+
+
+def detect_stale_dates(data_dir: Path, repo) -> list[date]:
+    """检测 regime 已有但需要重算的天(enriched 被覆写)。
+
+    用 mtime 比对: enriched 分区 parquet 的 mtime > regime parquet 的 mtime
+    → 该日 enriched 更新过, regime 需重算。
+    """
+    regime_p = regime_path(data_dir)
+    if not regime_p.exists():
+        return []
+    regime_mtime = regime_p.stat().st_mtime
+    enriched_dir = repo.store.data_dir / "kline_daily_enriched"
+    if not enriched_dir.exists():
+        return []
+    stale: list[date] = []
+    existing = load_regime_history(data_dir)
+    if existing.is_empty():
+        return []
+    existing_dates = set(existing["date"].to_list())
+    for part in enriched_dir.glob("date=*/part.parquet"):
+        try:
+            ds = part.parent.name.replace("date=", "")
+            d = date.fromisoformat(ds)
+        except (ValueError, OSError):
+            continue
+        if d not in existing_dates:
+            continue
+        try:
+            if part.stat().st_mtime > regime_mtime:
+                stale.append(d)
+        except OSError:
+            continue
+    return sorted(stale)
+
+
+def compute_regime_incremental(repo, data_dir: Path, *, today: date | None = None) -> pl.DataFrame:
+    """增量计算 regime(供 daily_pipeline / 启动补算调用)。
+
+    双检测: 1) 缺口(enriched 有但 regime 没有) 2) stale(enriched 被覆写)。
+    自动补齐所有需要的日。返回本次新算的 DataFrame。
+    """
+    today = today or date.today()
+    existing = load_regime_history(data_dir)
+
+    # 缺口: enriched 有哪些天, regime 缺哪些
+    enriched_dir = repo.store.data_dir / "kline_daily_enriched"
+    enriched_dates: set[date] = set()
+    if enriched_dir.exists():
+        for part in enriched_dir.glob("date=*/part.parquet"):
+            try:
+                ds = part.parent.name.replace("date=", "")
+                enriched_dates.add(date.fromisoformat(ds))
+            except ValueError:
+                continue
+    existing_dates = set(existing["date"].to_list()) if not existing.is_empty() else set()
+    missing = sorted(d for d in enriched_dates if d not in existing_dates and d <= today)
+
+    # stale: enriched 覆写过
+    stale = detect_stale_dates(data_dir, repo)
+
+    to_compute = sorted(set(missing) | set(stale))
+    if not to_compute:
+        logger.debug("regime incremental: nothing to compute")
+        return pl.DataFrame()
+
+    logger.info("regime incremental: compute %d days (missing=%d, stale=%d)",
+                len(to_compute), len(missing), len(stale))
+    new_rows = run_regime_batch(repo, start=to_compute[0], end=to_compute[-1])
+    if not new_rows.is_empty():
+        upsert_regime_history(data_dir, new_rows)
+    return new_rows

backend/tests/test_regime_builder.py

@@ -0,0 +1,305 @@
+"""市场环境(regime) 计算与持久化测试。
+
+覆盖:
+- classify_state: 五种状态边界值(强势/偏强/震荡/偏弱/弱势)
+- _aggregate_daily: 多日多 symbol 聚合(涨停数/涨跌家数/MA20占比)
+- upsert_regime_history: 按 date 覆盖(重算的天替换旧行)
+- compute_regime_incremental: 双检测(缺口 + stale mtime)
+"""
+from __future__ import annotations
+
+import os
+import time
+from datetime import date
+
+import polars as pl
+
+from app.services import regime_builder
+
+# ───────────────────────── 状态分类 ─────────────────────────
+
+
+def test_classify_strong():
+    state, score = regime_builder.classify_state({
+        "limit_up": 40, "limit_down": 1, "seal_rate": 0.85, "up_ratio": 3.0,
+        "index_pct": 0.02, "above_ma20_pct": 0.7, "total_amount": 2e11,
+    })
+    assert state == "strong"
+    assert score >= 75
+
+
+def test_classify_weak():
+    state, score = regime_builder.classify_state({
+        "limit_up": 1, "limit_down": 20, "seal_rate": 0.2, "up_ratio": 0.2,
+        "index_pct": -0.025, "above_ma20_pct": 0.2, "total_amount": 5e10,
+    })
+    assert state == "weak"
+    assert score < 25
+
+
+def test_classify_range():
+    state, score = regime_builder.classify_state({
+        "limit_up": 8, "limit_down": 6, "seal_rate": 0.5, "up_ratio": 1.0,
+        "index_pct": 0.0, "above_ma20_pct": 0.5, "total_amount": 1e11,
+    })
+    assert state == "range"
+    assert 40 <= score < 60
+
+
+def test_classify_monotonic_limit_up():
+    """涨停数越多, 综合分越高(其他条件相同)。"""
+    base = {"limit_down": 2, "seal_rate": 0.7, "up_ratio": 2.0,
+            "index_pct": 0.01, "above_ma20_pct": 0.6, "total_amount": 1.5e11}
+    s_low = regime_builder.classify_state({**base, "limit_up": 5})[1]
+    s_mid = regime_builder.classify_state({**base, "limit_up": 20})[1]
+    s_high = regime_builder.classify_state({**base, "limit_up": 45})[1]
+    assert s_low < s_mid < s_high
+
+
+# ───────────────────────── 聚合 ─────────────────────────
+
+
+def _enriched_df() -> pl.DataFrame:
+    """构造 2 天 × 4 标的 的 enriched 数据(含信号列)。"""
+    return pl.DataFrame({
+        "date": [date(2026, 1, 2)] * 4 + [date(2026, 1, 3)] * 4,
+        "symbol": ["A", "B", "C", "D"] * 2,
+        "close": [11, 9, 21, 19, 12, 8, 22, 18],
+        "change_pct": [0.1, -0.1, 0.05, -0.05, 0.08, -0.12, 0.02, -0.08],
+        "amount": [1e8, 2e8, 3e8, 4e8] * 2,
+        "ma20": [10, 10, 20, 20, 10, 10, 20, 20],
+        "signal_limit_up": [True, False, False, False, True, False, True, False],
+        "signal_limit_down": [False, False, False, True, False, False, False, False],
+        "signal_broken_limit_up": [False, False, False, False, False, False, False, False],
+        "consecutive_limit_ups": [1, 0, 0, 0, 2, 0, 1, 0],
+    })
+
+
+def test_aggregate_daily_basic():
+    """聚合多日: 每天的涨停数/涨跌家数正确。"""
+    df = _enriched_df()
+    result = regime_builder._aggregate_daily(df, index_pct_map={
+        date(2026, 1, 2): 0.01, date(2026, 1, 3): -0.005,
+    })
+    assert result.height == 2
+    # 第一天(1/2): 1 个涨停, 2 涨 2 跌
+    r1 = result.filter(pl.col("date") == date(2026, 1, 2)).row(0, named=True)
+    assert r1["limit_up"] == 1
+    assert r1["up_count"] == 2
+    assert r1["down_count"] == 2
+    assert r1["max_consecutive"] == 1
+    # 第二天(1/3): 2 个涨停, 2 涨 2 跌, 连板高度 2
+    r2 = result.filter(pl.col("date") == date(2026, 1, 3)).row(0, named=True)
+    assert r2["limit_up"] == 2
+    assert r2["max_consecutive"] == 2
+    # 每行都有 state 和 score
+    assert all(s in {"strong", "lean_strong", "range", "lean_weak", "weak"}
+               for s in result["state"].to_list())
+    assert result["score"].min() >= 0 and result["score"].max() <= 100
+
+
+def test_aggregate_daily_ma20_above():
+    """MA20 上方占比正确(close > ma20)。"""
+    df = _enriched_df()
+    result = regime_builder._aggregate_daily(df)
+    r1 = result.filter(pl.col("date") == date(2026, 1, 2)).row(0, named=True)
+    # 1/2: A(close11>ma10)✓, B(9<10)✗, C(21>20)✓, D(19<20)✗ → 2/4 = 0.5
+    assert r1["above_ma20_pct"] == 0.5
+
+
+def test_aggregate_empty_returns_empty():
+    assert regime_builder._aggregate_daily(pl.DataFrame()).is_empty()
+
+
+# ───────────────────────── 持久化(upsert) ─────────────────────────
+
+
+def test_upsert_inserts_new(tmp_path):
+    rows = pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 2)],
+        "state": ["strong", "range"],
+        "score": [80, 50],
+    })
+    regime_builder.upsert_regime_history(tmp_path, rows)
+    loaded = regime_builder.load_regime_history(tmp_path)
+    assert loaded.height == 2
+
+
+def test_upsert_overwrites_existing_date(tmp_path):
+    """重算的天覆盖旧行(upsert 语义)。"""
+    old = pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 2)],
+        "state": ["range", "range"], "score": [50, 50],
+    })
+    regime_builder.upsert_regime_history(tmp_path, old)
+    # 重算 1/2
+    new = pl.DataFrame({
+        "date": [date(2026, 1, 2)],
+        "state": ["strong"], "score": [85],
+    })
+    regime_builder.upsert_regime_history(tmp_path, new)
+    loaded = regime_builder.load_regime_history(tmp_path)
+    assert loaded.height == 2  # 仍是 2 天(1/2 被覆盖, 不重复)
+    r2 = loaded.filter(pl.col("date") == date(2026, 1, 2)).row(0, named=True)
+    assert r2["state"] == "strong"
+    assert r2["score"] == 85
+    # 1/1 不受影响
+    r1 = loaded.filter(pl.col("date") == date(2026, 1, 1)).row(0, named=True)
+    assert r1["state"] == "range"
+
+
+def test_coverage_metadata(tmp_path):
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 5)],
+        "state": ["strong", "weak"], "score": [80, 20],
+    }))
+    cov = regime_builder.get_regime_coverage(tmp_path)
+    assert cov["rows"] == 2
+    assert cov["earliest_date"] == "2026-01-01"
+    assert cov["latest_date"] == "2026-01-05"
+
+
+def test_coverage_empty(tmp_path):
+    cov = regime_builder.get_regime_coverage(tmp_path)
+    assert cov["rows"] == 0
+    assert cov["earliest_date"] is None
+
+
+# ───────────────────────── 双检测 ─────────────────────────
+
+
+def test_detect_stale_dates_by_mtime(tmp_path):
+    """enriched 分区 mtime > regime mtime → 标记重算。"""
+    # 准备 regime 历史
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 2)],
+        "state": ["range", "range"], "score": [50, 50],
+    }))
+    # 模拟 enriched 分区(先写, mtime=T2)
+    enriched_dir = tmp_path / "kline_daily_enriched"
+    for ds in ["2026-01-01", "2026-01-02"]:
+        d = enriched_dir / f"date={ds}"
+        d.mkdir(parents=True)
+        (d / "part.parquet").write_bytes(b"x")
+    # 重新 upsert regime → regime mtime 更新到 T3 > enriched 的 T2
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 2)],
+        "state": ["range", "range"], "score": [50, 50],
+    }))
+    time.sleep(0.05)  # 确保 mtime 精度差异
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 2)],
+        "state": ["range", "range"], "score": [50, 50],
+    }))
+    # 让 1/2 的 mtime 更新到 future > regime mtime
+    future = time.time() + 10
+    os.utime(enriched_dir / "date=2026-01-02" / "part.parquet", (future, future))
+
+    class _FakeRepo:
+        class store:
+            data_dir = tmp_path
+    stale = regime_builder.detect_stale_dates(tmp_path, _FakeRepo())
+    assert date(2026, 1, 2) in stale
+    assert date(2026, 1, 1) not in stale  # 1/1 没更新
+
+
+def test_compute_incremental_missing_dates(tmp_path):
+    """enriched 有但 regime 没有 → 补算缺口。"""
+    # regime 只有 1/1
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1)], "state": ["range"], "score": [50],
+    }))
+    # 模拟 enriched 有 1/1 和 1/2
+    enriched_dir = tmp_path / "kline_daily_enriched"
+    for ds in ["2026-01-01", "2026-01-02"]:
+        d = enriched_dir / f"date={ds}"
+        d.mkdir(parents=True)
+        (d / "part.parquet").write_bytes(b"x")
+
+    class _FakeRepo:
+        class store:
+            data_dir = tmp_path
+        def get_enriched_range(self, *a, **k): return None  # 无缓存, 不实际算
+
+    # compute_regime_incremental 会识别 1/2 缺口, 但 run_regime_batch 因无数据返回空
+    new = regime_builder.compute_regime_incremental(_FakeRepo(), tmp_path, today=date(2026, 1, 3))
+    # 无真实 enriched 数据 → 不算出新行, 但不报错
+    assert new.is_empty() or new.height >= 0
+
+
+# ───────────────────────── 回测环境过滤(T-1 防未来函数) ─────────────────────────
+
+
+def test_build_regime_mask_t1_alignment(tmp_path):
+    """_build_regime_mask 强制 T-1: regime[T-1] 决定 entry[T]。
+
+    场景: regime 1/1=weak(10), 1/2=strong(85)。
+    timestamp_labels: [1/1, 1/2, 1/3]。
+    filter: 只允许 strong。
+    期望: mask = [True(首日默认允许), False(1/2的前一日=1/1=weak), True(1/3的前一日=1/2=strong)]。
+    """
+    from app.backtest.strategy import StrategyBacktestService
+
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 2)],
+        "state": ["weak", "strong"],
+        "score": [10, 85],
+    }))
+    labels = ("2026-01-01", "2026-01-02", "2026-01-03")
+    mask = StrategyBacktestService._build_regime_mask(
+        labels, {"states": ["strong"]}, tmp_path,
+    )
+    assert mask is not None
+    assert mask.tolist() == [True, False, True]
+
+
+def test_build_regime_mask_min_score(tmp_path):
+    """min_score 过滤: regime[T-1] 的 score >= min_score 才允许入场。"""
+    from app.backtest.strategy import StrategyBacktestService
+
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1), date(2026, 1, 2)],
+        "state": ["range", "lean_strong"],
+        "score": [45, 65],
+    }))
+    labels = ("2026-01-01", "2026-01-02", "2026-01-03")
+    mask = StrategyBacktestService._build_regime_mask(
+        labels, {"min_score": 60}, tmp_path,
+    )
+    # 1/2 entry 由 1/1(score=45 < 60) 决定 → False
+    # 1/3 entry 由 1/2(score=65 >= 60) 决定 → True
+    assert mask.tolist() == [True, False, True]
+
+
+def test_build_regime_mask_none_when_no_filter():
+    """regime_filter 为 None → 返回 None(不过滤)。"""
+    from app.backtest.strategy import StrategyBacktestService
+
+    assert StrategyBacktestService._build_regime_mask(("2026-01-01",), None, None) is None
+
+
+def test_build_regime_mask_none_when_no_data(tmp_path):
+    """无 regime 历史数据 → 返回 None(不阻断回测)。"""
+    from app.backtest.strategy import StrategyBacktestService
+
+    mask = StrategyBacktestService._build_regime_mask(
+        ("2026-01-01", "2026-01-02"), {"states": ["strong"]}, tmp_path,
+    )
+    assert mask is None
+
+
+def test_build_regime_mask_first_day_allowed(tmp_path):
+    """首日无前一日环境数据 → 默认允许(不阻断)。"""
+    from app.backtest.strategy import StrategyBacktestService
+
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": [date(2026, 1, 1)],
+        "state": ["weak"], "score": [10],
+    }))
+    labels = ("2026-01-01", "2026-01-02")
+    mask = StrategyBacktestService._build_regime_mask(
+        labels, {"states": ["strong"]}, tmp_path,
+    )
+    # 1/1 首日 → True; 1/2 由 1/1(weak) → False
+    assert mask.tolist() == [True, False]
+

frontend/src/components/Layout.tsx

@@ -35,6 +35,7 @@ import {
   TrendingUp,
   Flame,
   BarChart3,
+  Gauge,
   Sparkles,
   Layers3,
   Landmark,
@@ -79,6 +80,7 @@ const nav = [
   { to: '/monitor', label: '监控中心', icon: RadioTower },
   { to: '/review',      label: '复盘',   icon: BookOpenCheck },
   { to: '/indices', label: '指数', icon: BarChart3 },
+  { to: '/regime', label: '市场环境', icon: Gauge },
   { to: '/data',       label: '数据',   icon: Database },
 ] as const
 

frontend/src/lib/api.ts

@@ -402,6 +402,66 @@ export interface RpsRotationData {
   concept_count: number
 }
 
+// ===== 市场环境(Regime) =====
+export type RegimeState = 'strong' | 'lean_strong' | 'range' | 'lean_weak' | 'weak'
+
+export const REGIME_STATE_LABELS: Record<RegimeState, string> = {
+  strong: '强势',
+  lean_strong: '偏强',
+  range: '震荡',
+  lean_weak: '偏弱',
+  weak: '弱势',
+}
+
+export const REGIME_STATE_COLORS: Record<RegimeState, string> = {
+  strong: '#ef4444',      // 红(强)
+  lean_strong: '#f97316', // 橙
+  range: '#6b7280',       // 灰
+  lean_weak: '#3b82f6',   // 蓝
+  weak: '#10b981',        // 绿(弱)
+}
+
+export interface RegimeRow {
+  date: string
+  state: RegimeState
+  score: number
+  limit_up: number
+  limit_down: number
+  broken_limit: number
+  max_consecutive: number
+  seal_rate: number
+  up_count: number
+  down_count: number
+  up_ratio: number
+  index_pct: number
+  above_ma20_pct: number
+  total_amount: number
+  avg_turnover: number
+}
+
+export interface RegimeHistory {
+  rows: RegimeRow[]
+  total: number
+}
+
+export interface RegimeStateItem {
+  state: RegimeState
+  label: string
+  count: number
+  pct: number
+}
+
+export interface RegimeStates {
+  distribution: RegimeStateItem[]
+  days: number
+}
+
+export interface RegimeCoverage {
+  rows: number
+  earliest_date: string | null
+  latest_date: string | null
+}
+
 // ===== 大盘复盘 =====
 export interface AiReviewReport {
   id: string
@@ -1474,6 +1534,26 @@ export const api = {
   rpsRotation: (days: number) =>
     request<RpsRotationData>(`/api/rps/rotation?days=${days}`),
 
+  // 市场环境(Regime)
+  regimeHistory: (start?: string, end?: string, limit?: number) => {
+    const params = new URLSearchParams()
+    if (start) params.set('start', start)
+    if (end) params.set('end', end)
+    if (limit) params.set('limit', String(limit))
+    const qs = params.toString()
+    return request<RegimeHistory>(`/api/regime/history${qs ? `?${qs}` : ''}`)
+  },
+  regimeLatest: () => request<{ row: RegimeRow | null }>('/api/regime/latest'),
+  regimeStates: (days = 60) => request<RegimeStates>(`/api/regime/states?days=${days}`),
+  regimeCoverage: () => request<RegimeCoverage>('/api/regime/coverage'),
+  regimeRecompute: (start?: string, end?: string) => {
+    const params = new URLSearchParams()
+    if (start) params.set('start', start)
+    if (end) params.set('end', end)
+    const qs = params.toString()
+    return request<{ ok: boolean; computed: number }>(`/api/regime/recompute${qs ? `?${qs}` : ''}`, { method: 'POST' })
+  },
+
   limitLadder: (asOf?: string, extColumns?: string, direction?: 'up' | 'down') => {
     const params = new URLSearchParams()
     if (asOf) params.set('as_of', asOf)

frontend/src/lib/backtestTask.ts

@@ -181,6 +181,7 @@ export function startBacktest(params: {
   holding_days?: number
   asset_type?: 'stock' | 'etf'
   minute_fill?: boolean
+  regime_filter?: { states?: string[]; min_score?: number } | null
 }): void {
   // 取消之前的任务状态
   if (eventSource) {
@@ -214,6 +215,7 @@ export function startBacktest(params: {
     holding_days: params.holding_days,
     asset_type: params.asset_type,
     minute_fill: params.minute_fill,
+    regime_filter: params.regime_filter ? JSON.stringify(params.regime_filter) : undefined,
   })
 
   // 存 reconnect 信息 (刷新后用)

frontend/src/lib/queryKeys.ts

@@ -84,6 +84,11 @@ export const QK = {
 
   // 概念涨幅轮动矩阵
   rpsRotation:          (days: number) => ['rps-rotation', days] as const,
+
+  // 市场环境(Regime) — 日级离线计算, 不进 SSE 刷新
+  regimeHistory:        (limit?: number) => ['regime-history', limit ?? 0] as const,
+  regimeLatest:         ['regime-latest'] as const,
+  regimeStates:         (days: number) => ['regime-states', days] as const,
 } as const
 
 // ===== SSE 应该 invalidate 的 key 前缀列表 =====

frontend/src/pages/Regime.tsx

@@ -0,0 +1,232 @@
+/**
+ * 市场环境(Regime)页 — 每日环境状态时序趋势 + 状态分布。
+ *
+ * 数据来源: 后端 regime_builder 批算的时序表(每日离散状态 + 多维指标)。
+ * 不复刻 Dashboard 的当日总览(那是单日快照), 聚焦历史趋势与状态分布。
+ */
+import { useEffect, useMemo, useRef, useState } from 'react'
+import { useQuery, useQueryClient } from '@tanstack/react-query'
+import * as echarts from 'echarts'
+import { Activity, RefreshCw, Loader2 } from 'lucide-react'
+import {
+  api, type RegimeRow, type RegimeState,
+  REGIME_STATE_LABELS, REGIME_STATE_COLORS,
+} from '@/lib/api'
+import { QK } from '@/lib/queryKeys'
+import { useChartTheme } from '@/lib/theme'
+import { fmtBigNum } from '@/lib/format'
+
+const STATE_ORDER: RegimeState[] = ['strong', 'lean_strong', 'range', 'lean_weak', 'weak']
+
+function useEChart(option: echarts.EChartsOption | null, deps: unknown[]) {
+  const ref = useRef<HTMLDivElement>(null)
+  const instRef = useRef<echarts.ECharts | null>(null)
+  useEffect(() => {
+    if (!ref.current) return
+    instRef.current = echarts.init(ref.current, undefined, { renderer: 'canvas' })
+    const onResize = () => instRef.current?.resize()
+    window.addEventListener('resize', onResize)
+    return () => {
+      window.removeEventListener('resize', onResize)
+      instRef.current?.dispose()
+      instRef.current = null
+    }
+  }, [])
+  useEffect(() => {
+    if (instRef.current && option) instRef.current.setOption(option, { notMerge: true })
+  }, [option, ...deps])
+  return ref
+}
+
+export function Regime() {
+  const qc = useQueryClient()
+  const [days, setDays] = useState(120)
+  const ct = useChartTheme()
+
+  const history = useQuery({
+    queryKey: QK.regimeHistory(days),
+    queryFn: () => api.regimeHistory(undefined, undefined, days),
+    staleTime: 5 * 60 * 1000,
+  })
+  const states = useQuery({
+    queryKey: QK.regimeStates(days),
+    queryFn: () => api.regimeStates(days),
+    staleTime: 5 * 60 * 1000,
+  })
+  const [recomputing, setRecomputing] = useState(false)
+
+  const rows: RegimeRow[] = history.data?.rows ?? []
+  const latest = rows.length > 0 ? rows[rows.length - 1] : null
+
+  // 趋势图: 综合分曲线 + 涨停数柱状
+  const trendOption = useMemo<echarts.EChartsOption | null>(() => {
+    if (rows.length === 0) return null
+    const dates = rows.map(r => r.date)
+    const scores = rows.map(r => r.score)
+    const limitUps = rows.map(r => r.limit_up)
+    return {
+      backgroundColor: 'transparent',
+      tooltip: { trigger: 'axis', backgroundColor: ct.tooltipBg, borderColor: ct.tooltipBorder, textStyle: { color: ct.tooltipText } },
+      legend: { data: ['综合分', '涨停数'], textStyle: { color: ct.text }, top: 0 },
+      grid: { left: 48, right: 48, top: 32, bottom: 56 },
+      xAxis: {
+        type: 'category', data: dates,
+        axisLabel: { color: ct.text, fontSize: 10, formatter: (v: string) => v.slice(5) },
+        axisLine: { lineStyle: { color: ct.grid } },
+      },
+      yAxis: [
+        { type: 'value', name: '综合分', min: 0, max: 100, axisLabel: { color: ct.text, fontSize: 10 }, splitLine: { lineStyle: { color: ct.grid } }, nameTextStyle: { color: ct.text } },
+        { type: 'value', name: '涨停', axisLabel: { color: ct.text, fontSize: 10 }, splitLine: { show: false }, nameTextStyle: { color: ct.text } },
+      ],
+      dataZoom: [
+        { type: 'inside', start: Math.max(0, 100 - (60 / days) * 100) },
+        { type: 'slider', bottom: 8, height: 16, borderColor: ct.border, fillerColor: ct.zoomFill, textStyle: { color: ct.text } },
+      ],
+      series: [
+        { name: '综合分', type: 'line', data: scores, smooth: true, symbol: 'none',
+          lineStyle: { width: 2, color: ct.textStrong }, areaStyle: { opacity: 0.08 },
+          markLine: { silent: true, lineStyle: { type: 'dashed', color: ct.grid }, data: [
+            { yAxis: 75, label: { formatter: '强势', color: ct.text, fontSize: 9 } },
+            { yAxis: 40, label: { formatter: '震荡', color: ct.text, fontSize: 9 } },
+          ] } },
+        { name: '涨停数', type: 'bar', data: limitUps, yAxisIndex: 1, barMaxWidth: 6, itemStyle: { color: REGIME_STATE_COLORS.strong } },
+      ],
+    }
+  }, [rows, days, ct])
+  const trendRef = useEChart(trendOption, [trendOption])
+
+  // 状态分布饼图
+  const pieOption = useMemo<echarts.EChartsOption | null>(() => {
+    const dist = states.data?.distribution ?? []
+    if (dist.length === 0) return null
+    return {
+      backgroundColor: 'transparent',
+      tooltip: { trigger: 'item', backgroundColor: ct.tooltipBg, borderColor: ct.tooltipBorder, textStyle: { color: ct.tooltipText } },
+      series: [{
+        type: 'pie', radius: ['42%', '70%'], center: ['50%', '52%'],
+        label: { color: ct.text, fontSize: 10, formatter: '{b}\n{d}%' },
+        data: STATE_ORDER
+          .map(s => dist.find(d => d.state === s))
+          .filter((x): x is NonNullable<typeof x> => !!x)
+          .map(d => ({
+            name: d.label, value: d.count,
+            itemStyle: { color: REGIME_STATE_COLORS[d.state] },
+          })),
+      }],
+    }
+  }, [states.data, ct])
+  const pieRef = useEChart(pieOption, [pieOption])
+
+  const handleRecompute = async () => {
+    setRecomputing(true)
+    try {
+      await api.regimeRecompute()
+      await Promise.all([
+        qc.invalidateQueries({ queryKey: ['regime-history'] }),
+        qc.invalidateQueries({ queryKey: ['regime-states'] }),
+        qc.invalidateQueries({ queryKey: ['regime-latest'] }),
+      ])
+    } finally {
+      setRecomputing(false)
+    }
+  }
+
+  return (
+    <div className="mx-auto max-w-6xl px-4 py-5 space-y-4">
+      {/* 头部 */}
+      <div className="flex items-center gap-3">
+        <Activity className="h-5 w-5 text-accent" />
+        <h1 className="text-base font-semibold text-foreground">市场环境</h1>
+        <span className="text-xs text-muted">每日环境状态 · 赚钱效应 · 趋势分析</span>
+        <div className="ml-auto flex items-center gap-2">
+          <select value={days} onChange={e => setDays(Number(e.target.value))}
+            className="h-7 rounded-btn border border-border bg-base px-2 text-xs text-foreground">
+            <option value={60}>近 60 天</option>
+            <option value={120}>近 120 天</option>
+            <option value={250}>近 250 天</option>
+          </select>
+          <button onClick={handleRecompute} disabled={recomputing}
+            className="inline-flex items-center gap-1.5 h-7 px-3 rounded-btn border border-border bg-base text-xs text-secondary hover:text-accent disabled:opacity-50">
+            {recomputing ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : <RefreshCw className="h-3.5 w-3.5" />}
+            重算
+          </button>
+        </div>
+      </div>
+
+      {/* 最新日概览 */}
+      {latest ? (
+        <div className="grid grid-cols-2 gap-3 sm:grid-cols-4">
+          <div className="rounded-card border border-border bg-base p-3">
+            <div className="text-[10px] text-muted">最新状态 · {latest.date}</div>
+            <div className="mt-1 flex items-baseline gap-2">
+              <span className="text-2xl font-bold" style={{ color: REGIME_STATE_COLORS[latest.state] }}>
+                {REGIME_STATE_LABELS[latest.state]}
+              </span>
+              <span className="text-sm text-muted">{latest.score} 分</span>
+            </div>
+          </div>
+          <div className="rounded-card border border-border bg-base p-3">
+            <div className="text-[10px] text-muted">涨停 / 跌停</div>
+            <div className="mt-1 text-lg font-semibold text-foreground">
+              <span className="text-red-400">{latest.limit_up}</span>
+              <span className="mx-1 text-muted">/</span>
+              <span className="text-green-400">{latest.limit_down}</span>
+            </div>
+            <div className="text-[10px] text-muted">连板高度 {latest.max_consecutive} · 封板率 {(latest.seal_rate * 100).toFixed(0)}%</div>
+          </div>
+          <div className="rounded-card border border-border bg-base p-3">
+            <div className="text-[10px] text-muted">涨跌家数比</div>
+            <div className="mt-1 text-lg font-semibold text-foreground">{latest.up_ratio.toFixed(2)}</div>
+            <div className="text-[10px] text-muted">涨 {latest.up_count} · 跌 {latest.down_count}</div>
+          </div>
+          <div className="rounded-card border border-border bg-base p-3">
+            <div className="text-[10px] text-muted">成交额</div>
+            <div className="mt-1 text-lg font-semibold text-foreground">{fmtBigNum(latest.total_amount)}</div>
+            <div className="text-[10px] text-muted">MA20 上方 {(latest.above_ma20_pct * 100).toFixed(0)}%</div>
+          </div>
+        </div>
+      ) : (
+        <div className="rounded-card border border-dashed border-border p-8 text-center text-sm text-muted">
+          {history.isLoading ? '加载中…' : '暂无环境数据，请先运行盘后管道或点击「重算」'}
+        </div>
+      )}
+
+      {/* 状态色带 */}
+      {rows.length > 0 && (
+        <div className="rounded-card border border-border bg-base p-3">
+          <div className="mb-2 text-xs font-medium text-foreground">状态时间轴</div>
+          <div className="flex h-6 w-full overflow-hidden rounded">
+            {rows.map(r => (
+              <div key={r.date} title={`${r.date} ${REGIME_STATE_LABELS[r.state]}(${r.score})`}
+                className="flex-1 min-w-[2px]" style={{ backgroundColor: REGIME_STATE_COLORS[r.state] }} />
+            ))}
+          </div>
+          <div className="mt-1.5 flex items-center gap-3 text-[10px] text-muted">
+            <span>{rows[0]?.date}</span>
+            <div className="ml-auto flex items-center gap-2">
+              {STATE_ORDER.map(s => (
+                <span key={s} className="flex items-center gap-1">
+                  <span className="inline-block h-2 w-2 rounded-sm" style={{ backgroundColor: REGIME_STATE_COLORS[s] }} />
+                  {REGIME_STATE_LABELS[s]}
+                </span>
+              ))}
+            </div>
+            <span>{rows[rows.length - 1]?.date}</span>
+          </div>
+        </div>
+      )}
+
+      {/* 趋势图 + 分布图 */}
+      <div className="grid grid-cols-1 gap-4 lg:grid-cols-3">
+        <div className="rounded-card border border-border bg-base p-3 lg:col-span-2">
+          <div className="mb-1 text-xs font-medium text-foreground">环境综合分 · 涨停数趋势</div>
+          <div ref={trendRef} className="h-[320px]" />
+        </div>
+        <div className="rounded-card border border-border bg-base p-3">
+          <div className="mb-1 text-xs font-medium text-foreground">状态分布（近 {days} 天）</div>
+          <div ref={pieRef} className="h-[320px]" />
+        </div>
+      </div>
+    </div>
+  )
+}

frontend/src/pages/backtest/StrategyBacktest.tsx

@@ -8,6 +8,8 @@ import {
   type StrategyBacktestTrade,
   type StrategyDetail,
   type StrategyParamDef,
+  REGIME_STATE_LABELS,
+  REGIME_STATE_COLORS,
 } from '@/lib/api'
 import { QK } from '@/lib/queryKeys'
 import { storage } from '@/lib/storage'
@@ -883,6 +885,9 @@ export function StrategyBacktest() {
   const [simMode, setSimMode] = useState<'position' | 'full'>(saved?.mode ?? 'position')
   const [holdingDays, setHoldingDays] = useState(saved?.holdingDays ?? '5')
   const [highGranularity, setHighGranularity] = useState(saved?.minuteFill ?? false)
+  // 市场环境过滤(空=不过滤)
+  const [regimeStates, setRegimeStates] = useState<string[]>([])
+  const [regimeMinScore, setRegimeMinScore] = useState<number | ''>('')
   const [settingsOpen, setSettingsOpen] = useState(false)
   // 分钟K成交价细化: 不改变信号日或成交日, 需 Pro+ 分钟K能力
   const { data: caps } = useCapabilities()
@@ -1047,6 +1052,12 @@ export function StrategyBacktest() {
       mode: simMode,
       holding_days: Number(holdingDays) || 5,
       minute_fill: highGranularity,
+      regime_filter: regimeStates.length > 0 || regimeMinScore !== ''
+        ? {
+            ...(regimeStates.length > 0 ? { states: regimeStates } : {}),
+            ...(regimeMinScore !== '' ? { min_score: Number(regimeMinScore) } : {}),
+          }
+        : null,
     })
   }
 
@@ -1715,6 +1726,40 @@ export function StrategyBacktest() {
           )}
         </div>
 
+        {/* 市场环境过滤: 只在指定环境的交易日入场(强制 T-1, 用前一日环境判定) */}
+        <div className="rounded-btn border border-border bg-surface/50 px-3 py-2 space-y-1.5">
+          <div className="flex items-center gap-2">
+            <Gauge className="h-3.5 w-3.5 text-accent" />
+            <span className="text-xs font-medium text-foreground">环境过滤</span>
+            <span className="text-[10px] text-muted">仅在前一日环境满足时入场(防未来函数)</span>
+            <div className="ml-auto flex items-center gap-1">
+              <span className="text-[10px] text-muted">最低分</span>
+              <input type="number" min={0} max={100} value={regimeMinScore} placeholder="不限"
+                onChange={e => setRegimeMinScore(e.target.value ? Number(e.target.value) : '')}
+                className="w-14 h-6 px-1 rounded border border-border bg-base text-[11px] text-foreground text-center focus:outline-none focus:border-accent/50" />
+            </div>
+          </div>
+          <div className="flex flex-wrap gap-1">
+            {(Object.keys(REGIME_STATE_LABELS) as (keyof typeof REGIME_STATE_LABELS)[]).map(s => {
+              const active = regimeStates.includes(s)
+              return (
+                <button key={s} onClick={() => setRegimeStates(prev => active ? prev.filter(x => x !== s) : [...prev, s])}
+                  className={`inline-flex items-center gap-1 px-2 py-0.5 rounded border text-[11px] transition-colors cursor-pointer ${
+                    active ? 'border-transparent text-white' : 'border-border text-muted hover:text-secondary'
+                  }`}
+                  style={active ? { backgroundColor: REGIME_STATE_COLORS[s] } : undefined}>
+                  <span className="inline-block h-2 w-2 rounded-sm" style={{ backgroundColor: active ? '#fff' : REGIME_STATE_COLORS[s] }} />
+                  {REGIME_STATE_LABELS[s]}
+                </button>
+              )
+            })}
+            {(regimeStates.length > 0 || regimeMinScore !== '') && (
+              <button onClick={() => { setRegimeStates([]); setRegimeMinScore('') }}
+                className="text-[10px] text-muted hover:text-danger px-1">清除</button>
+            )}
+          </div>
+        </div>
+
         {result?.error && (
           <div className="text-sm text-danger bg-danger/10 border border-danger/30 rounded-btn px-3 py-2">
             {result.error}

frontend/src/router.tsx

@@ -25,6 +25,7 @@ const LimitUpLadder = lazy(() => import('./pages/LimitUpLadder').then(m => ({ de
 const Branding = lazy(() => import('./pages/Branding').then(m => ({ default: m.Branding })))
 const Settings = lazy(() => import('./pages/Settings').then(m => ({ default: m.Settings })))
 const Indices = lazy(() => import('./pages/Indices').then(m => ({ default: m.Indices })))
+const Regime = lazy(() => import('./pages/Regime').then(m => ({ default: m.Regime })))
 const Dev = lazy(() => import('./pages/Dev').then(m => ({ default: m.Dev })))
 
 // 首次使用守卫 —— 未完成向导则重定向到 /onboarding
@@ -82,6 +83,7 @@ export const router = createBrowserRouter([
       { path: 'monitor', element: <Monitor /> },
       { path: 'limit-ladder', element: <LimitUpLadder /> },
       { path: 'indices', element: <Indices /> },
+    { path: 'regime', element: <Regime /> },
       { path: 'branding', element: <Branding /> },
       { path: 'settings', element: <Settings /> },
       // 隐藏路由：开发者工具（不暴露在菜单，仅供调试）