From 46ebcd89c8292d56dafd4e26b41324f1a6a41e4d Mon Sep 17 00:00:00 2001 From: shy3130 Date: Sun, 2 Aug 2026 12:22:48 +0800 Subject: [PATCH] =?UTF-8?q?fix(regime):=20=E4=BF=AE=E5=A4=8D=E9=87=8D?= =?UTF-8?q?=E7=AE=97=E6=97=A0=E4=BA=A7=E5=87=BA=E5=B9=B6=E8=A1=A5=E5=85=A8?= =?UTF-8?q?=20beta=20=E6=A0=87=E7=AD=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端 - /recompute 全量分支原本误调 compute_regime_incremental(增量补差), 无缺口/stale 时直接返回 0 行; 改为用 earliest_enriched_date ~ 今天 走 run_regime_batch + upsert 强制覆盖, 真正实现"全量重算" - _scan_enriched_fallback 原本漏调 compute_limit_signals 的 instruments 参数致慢路径必抛异常 → enriched 缓存未预热时拿不到含信号列数据 → 聚合 返回空; 改用主管道同款 compute_all(df, instruments, historical_shares) 一站式补算, 与 indicators/pipeline.py 对齐 - 新增 enriched_date_set / earliest_enriched_date 辅助函数(抽出复用) 前端 - Regime 重算按钮: 加"重算中…"文字反馈 + 完成 toast(区分有/无新数据) + catch 捕获错误(原异常被静默吞掉) - 市场环境菜单加 beta 胶囊标签(对齐 Settings 数据源标签样式) 验证 - 后端 582 passed; 真实数据全量重算产出 986 行(2022-07-08 ~ 2026-07-31) - 前端 tsc + pnpm build 通过 --- backend/app/api/regime.py | 23 +++++++++----- backend/app/services/regime_builder.py | 42 ++++++++++++++++++-------- frontend/src/components/Layout.tsx | 14 ++++++--- frontend/src/pages/Regime.tsx | 9 ++++-- 4 files changed, 63 insertions(+), 25 deletions(-) diff --git a/backend/app/api/regime.py b/backend/app/api/regime.py index 7019142..28b8d2f 100644 --- a/backend/app/api/regime.py +++ b/backend/app/api/regime.py @@ -132,16 +132,25 @@ def regime_coverage(request: Request): @router.post("/recompute") def regime_recompute(request: Request, start: date | None = None, end: date | None = None): - """手动触发重算(全量或指定区间)。管理员操作。""" + """手动触发重算(全量或指定区间)。管理员操作。 + + - 不传 start: 强制全量重算(enriched 最早日 ~ 今天), 覆盖所有已有行。 + 与 daily_pipeline 的增量补差(compute_regime_incremental)不同 —— 此接口面向 + 人工「我要重新算一遍」的预期, 必须真正重算而非增量补缺口。 + - 传 start: 仅重算 [start, end] 区间。 + """ 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) + # 全量: 从 enriched 最早日强制重算到今天 + earliest = regime_builder.earliest_enriched_date(repo) + if earliest is None: + invalidate_regime_cache() + return {"ok": True, "computed": 0} + start = earliest + 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/services/regime_builder.py b/backend/app/services/regime_builder.py index c7ba921..4466229 100644 --- a/backend/app/services/regime_builder.py +++ b/backend/app/services/regime_builder.py @@ -239,19 +239,24 @@ def _scan_enriched_fallback(repo, start: date, end: date) -> pl.DataFrame | None """缓存不覆盖时的慢路径: 一次性 scan 全部 enriched parquet + 重算指标。 仅在 regime 首次全量回填或缓存未预热时触发。返回含信号列的多日 DataFrame。 + + enriched 持久化只存基础列(OHLCV + raw_*/turnover/consecutive_*), 不含 + change_pct/ma20/signal_* 等派生列, 故此处需用 compute_all 补算全套指标。 + 必须传入 instruments(涨跌停价表), 否则 compute_limit_signals 会跳过涨跌停信号。 """ 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 + from app.indicators.pipeline import compute_all 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) + instruments = repo.get_instruments() + historical_shares = repo.get_historical_shares() + df = compute_all(df, instruments=instruments, historical_shares=historical_shares) return df except Exception as e: # noqa: BLE001 logger.warning("regime scan_enriched_fallback failed: %s", e) @@ -391,15 +396,7 @@ def compute_regime_incremental(repo, data_dir: Path, *, today: date | None = Non 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 + enriched_dates = enriched_date_set(repo) 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) @@ -417,3 +414,24 @@ def compute_regime_incremental(repo, data_dir: Path, *, today: date | None = Non if not new_rows.is_empty(): upsert_regime_history(data_dir, new_rows) return new_rows + + +def enriched_date_set(repo) -> set[date]: + """扫描 kline_daily_enriched 分区目录, 返回所有已有日期集合。""" + enriched_dir = repo.store.data_dir / "kline_daily_enriched" + dates: set[date] = set() + if not enriched_dir.exists(): + return dates + for part in enriched_dir.glob("date=*/part.parquet"): + try: + ds = part.parent.name.replace("date=", "") + dates.add(date.fromisoformat(ds)) + except ValueError: + continue + return dates + + +def earliest_enriched_date(repo) -> date | None: + """返回 enriched 最早日期(供全量重算定起点)。无数据返回 None。""" + dates = enriched_date_set(repo) + return min(dates) if dates else None diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index 04352b0..cb1b5fc 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -80,7 +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: '/regime', label: '市场环境', icon: Gauge, badge: 'beta' }, { to: '/data', label: '数据', icon: Database }, ] as const @@ -381,11 +381,12 @@ export function Layout() { }, [alertsTotal]) // 合并内置页面 + 可见的扩展分析菜单 - const analysisNav = (analysisMenus?.items ?? []) + type NavItem = { to: string; label: string; icon: typeof Gauge; badge?: string } + const analysisNav: NavItem[] = (analysisMenus?.items ?? []) .filter(m => m.visible) .map(m => ({ to: `/analysis/${m.id}`, label: m.label, icon: m.icon === 'tags' ? Tags : BarChart3 })) - const allNav = [...nav, ...analysisNav] + const allNav: NavItem[] = [...nav, ...analysisNav] const savedOrder = prefs?.nav_order ?? [] const navItems = savedOrder.length > 0 @@ -463,7 +464,7 @@ export function Layout() {