diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index 0dfe412..8c03b5c 100644 --- a/backend/app/api/backtest.py +++ b/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(): diff --git a/backend/app/api/regime.py b/backend/app/api/regime.py new file mode 100644 index 0000000..7019142 --- /dev/null +++ b/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} diff --git a/backend/app/backtest/strategy.py b/backend/app/backtest/strategy.py index b938d1a..edb4c30 100644 --- a/backend/app/backtest/strategy.py +++ b/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, diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index f5edef1..1f92cca 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/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, diff --git a/backend/app/main.py b/backend/app/main.py index 13232fc..9a6b988 100644 --- a/backend/app/main.py +++ b/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) diff --git a/backend/app/services/regime_builder.py b/backend/app/services/regime_builder.py new file mode 100644 index 0000000..c7ba921 --- /dev/null +++ b/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 diff --git a/backend/tests/test_regime_builder.py b/backend/tests/test_regime_builder.py new file mode 100644 index 0000000..e5b11eb --- /dev/null +++ b/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] + diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index b7ce57e..04352b0 100644 --- a/frontend/src/components/Layout.tsx +++ b/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 diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 26d6d52..912b666 100644 --- a/frontend/src/lib/api.ts +++ b/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 = { + strong: '强势', + lean_strong: '偏强', + range: '震荡', + lean_weak: '偏弱', + weak: '弱势', +} + +export const REGIME_STATE_COLORS: Record = { + 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(`/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(`/api/regime/history${qs ? `?${qs}` : ''}`) + }, + regimeLatest: () => request<{ row: RegimeRow | null }>('/api/regime/latest'), + regimeStates: (days = 60) => request(`/api/regime/states?days=${days}`), + regimeCoverage: () => request('/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) diff --git a/frontend/src/lib/backtestTask.ts b/frontend/src/lib/backtestTask.ts index 3001215..01ae568 100644 --- a/frontend/src/lib/backtestTask.ts +++ b/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 信息 (刷新后用) diff --git a/frontend/src/lib/queryKeys.ts b/frontend/src/lib/queryKeys.ts index fbc06a0..ae65cdc 100644 --- a/frontend/src/lib/queryKeys.ts +++ b/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 前缀列表 ===== diff --git a/frontend/src/pages/Regime.tsx b/frontend/src/pages/Regime.tsx new file mode 100644 index 0000000..c2f7343 --- /dev/null +++ b/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(null) + const instRef = useRef(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(() => { + 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(() => { + 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 => !!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 ( +
+ {/* 头部 */} +
+ +

市场环境

+ 每日环境状态 · 赚钱效应 · 趋势分析 +
+ + +
+
+ + {/* 最新日概览 */} + {latest ? ( +
+
+
最新状态 · {latest.date}
+
+ + {REGIME_STATE_LABELS[latest.state]} + + {latest.score} 分 +
+
+
+
涨停 / 跌停
+
+ {latest.limit_up} + / + {latest.limit_down} +
+
连板高度 {latest.max_consecutive} · 封板率 {(latest.seal_rate * 100).toFixed(0)}%
+
+
+
涨跌家数比
+
{latest.up_ratio.toFixed(2)}
+
涨 {latest.up_count} · 跌 {latest.down_count}
+
+
+
成交额
+
{fmtBigNum(latest.total_amount)}
+
MA20 上方 {(latest.above_ma20_pct * 100).toFixed(0)}%
+
+
+ ) : ( +
+ {history.isLoading ? '加载中…' : '暂无环境数据，请先运行盘后管道或点击「重算」'} +
+ )} + + {/* 状态色带 */} + {rows.length > 0 && ( +
+
状态时间轴
+
+ {rows.map(r => ( +
+ ))} +
+
+ {rows[0]?.date} +
+ {STATE_ORDER.map(s => ( + + + {REGIME_STATE_LABELS[s]} + + ))} +
+ {rows[rows.length - 1]?.date} +
+
+ )} + + {/* 趋势图 + 分布图 */} +
+
+
环境综合分 · 涨停数趋势
+
+
+
+
状态分布（近 {days} 天）
+
+
+
+
+ ) +} diff --git a/frontend/src/pages/backtest/StrategyBacktest.tsx b/frontend/src/pages/backtest/StrategyBacktest.tsx index b650520..dc38aef 100644 --- a/frontend/src/pages/backtest/StrategyBacktest.tsx +++ b/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([]) + const [regimeMinScore, setRegimeMinScore] = useState('') 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() { )}
+ {/* 市场环境过滤: 只在指定环境的交易日入场(强制 T-1, 用前一日环境判定) */} +
+
+ + 环境过滤 + 仅在前一日环境满足时入场(防未来函数) +
+ 最低分 + 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" /> +
+
+
+ {(Object.keys(REGIME_STATE_LABELS) as (keyof typeof REGIME_STATE_LABELS)[]).map(s => { + const active = regimeStates.includes(s) + return ( + + ) + })} + {(regimeStates.length > 0 || regimeMinScore !== '') && ( + + )} +
+
+ {result?.error && (
{result.error} diff --git a/frontend/src/router.tsx b/frontend/src/router.tsx index 474eecb..41e6785 100644 --- a/frontend/src/router.tsx +++ b/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: }, { path: 'limit-ladder', element: }, { path: 'indices', element: }, + { path: 'regime', element: }, { path: 'branding', element: }, { path: 'settings', element: }, // 隐藏路由：开发者工具（不暴露在菜单，仅供调试）