diff --git a/backend/app/api/regime.py b/backend/app/api/regime.py index 28b8d2f..f484033 100644 --- a/backend/app/api/regime.py +++ b/backend/app/api/regime.py @@ -71,7 +71,11 @@ def regime_history( 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") + # limit 仅在"最近 N 天"模式(未传 start/end)生效; + # 日期范围模式(传了 start/end, 如"全部")应返回完整范围, 不截断。 + if start is None and end is None: + df = df.sort("date", descending=True).head(limit) + df = df.sort("date") rows = _df_to_records(df) result = {"rows": rows, "total": len(rows)} diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 07eea4c..6528466 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -418,6 +418,9 @@ def get_preferences() -> dict: "pipeline_pull_a_share": preferences.get_pipeline_pull_a_share(), "pipeline_pull_etf": preferences.get_pipeline_pull_etf(), "pipeline_pull_index": preferences.get_pipeline_pull_index(), + "pipeline_regime_enabled": preferences.get_pipeline_regime_enabled(), + "regime_batch_days": preferences.get_regime_batch_days(), + "regime_warmup_days": preferences.get_regime_warmup_days(), "pipeline_index_symbols": preferences.get_pipeline_index_symbols(), "pipeline_schedule": preferences.get_pipeline_schedule(), "instruments_schedule": preferences.get_instruments_schedule(), @@ -830,6 +833,42 @@ def update_pipeline_pull_types(req: PipelinePullTypesIn) -> dict: return preferences.set_pipeline_pull_types(cfg) +class PipelineRegimeEnabledIn(BaseModel): + """盘后管道是否自动计算市场环境(regime)。""" + pipeline_regime_enabled: bool + + +@router.put("/preferences/pipeline-regime-enabled") +def update_pipeline_regime_enabled(req: PipelineRegimeEnabledIn) -> dict: + """更新盘后管道 regime 自动计算开关。""" + from app.services import preferences + preferences.save({"pipeline_regime_enabled": bool(req.pipeline_regime_enabled)}) + return {"pipeline_regime_enabled": preferences.get_pipeline_regime_enabled()} + + +class RegimeBatchParamsIn(BaseModel): + """regime 全量回填分批参数(控制内存峰值)。""" + batch_days: int | None = None + warmup_days: int | None = None + + +@router.put("/preferences/regime-batch-params") +def update_regime_batch_params(req: RegimeBatchParamsIn) -> dict: + """更新 regime 分批参数。仅在传入字段时保存对应项(支持部分更新)。""" + from app.services import preferences + updates: dict = {} + if req.batch_days is not None: + updates["regime_batch_days"] = req.batch_days + if req.warmup_days is not None: + updates["regime_warmup_days"] = req.warmup_days + if updates: + preferences.save(updates) + return { + "regime_batch_days": preferences.get_regime_batch_days(), + "regime_warmup_days": preferences.get_regime_warmup_days(), + } + + class PipelineIndexSymbolsIn(BaseModel): """指数自定义拉取代码(逗号/换行/空格分隔,空串表示全量)。""" symbols: str = "" diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index 1f92cca..6a70541 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -518,21 +518,29 @@ def run_now( # Step 2.6: 市场环境(regime) 增量计算 — enriched 已就绪后聚合环境指标。 # 双检测(缺口+stale), 自动补算遗漏/被覆写的日。软失败: 不阻断主管道。 + # 默认关闭: regime 是本地聚合计算(非拉取), 首次/regime 表为空时需全量回填 + # 多日, 内存与耗时较高。用户可在数据页「市场环境」卡片设置里开启自动计算, + # 或直接在该页面点「重算」手动触发(不受此开关影响)。 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}") + from app.services import preferences as _prefs_regime + if not _prefs_regime.get_pipeline_regime_enabled(): skipped.append("regime") + logger.info("compute_regime skipped: user disabled (pipeline_regime_enabled=False)") + else: + 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 视图…") diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index 80ce20e..5671bf8 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -240,6 +240,51 @@ def get_pipeline_pull_index() -> bool: return load().get("pipeline_pull_index", True) +def get_pipeline_regime_enabled() -> bool: + """盘后管道是否自动计算市场环境(regime)。默认 False。 + + regime 是本地聚合计算(非拉取), 首次/regime 表为空时需全量回填多日, + 内存与耗时较高, 故默认关闭; 用户可在数据页「市场环境」卡片设置里开启, + 或直接在该页面点「重算」手动触发(不受此开关影响)。 + """ + return load().get("pipeline_regime_enabled", False) + + +# regime 全量回填分批参数范围: +# - batch_days: 每批目标交易日数。越小内存越省、批次越多越慢; ma20 需 20 交易日, +# 故下限 25(留 warmup 余量), 上限 500(约 2 年)。 +# - warmup_days: 每批前缀预热天数(日历日), 必须 > ma20 的 20 交易日(≈28 日历日), +# 下限 35 留余量, 上限 90。 +_REGIME_BATCH_DAYS_MIN = 25 +_REGIME_BATCH_DAYS_MAX = 500 +_REGIME_WARMUP_DAYS_MIN = 35 +_REGIME_WARMUP_DAYS_MAX = 90 + + +def get_regime_batch_days() -> int: + """regime 全量回填每批目标交易日数。默认 60(约一季度)。 + + 超过此天数的范围会被切成多批, 每批独立算指标后拼接, 控制内存峰值。 + """ + v = load().get("regime_batch_days", 60) + try: + return max(_REGIME_BATCH_DAYS_MIN, min(_REGIME_BATCH_DAYS_MAX, int(v))) + except (TypeError, ValueError): + return 60 + + +def get_regime_warmup_days() -> int: + """regime 分批每批的 warmup 前缀日历天数。默认 40。 + + 用于预热 ma20 等滚动窗口指标, 使每批边界计算正确。必须 > 20 交易日。 + """ + v = load().get("regime_warmup_days", 40) + try: + return max(_REGIME_WARMUP_DAYS_MIN, min(_REGIME_WARMUP_DAYS_MAX, int(v))) + except (TypeError, ValueError): + return 40 + + _PIPELINE_PULL_KEYS = ("pipeline_pull_etf", "pipeline_pull_index") diff --git a/backend/app/services/regime_builder.py b/backend/app/services/regime_builder.py index 4466229..eb78b72 100644 --- a/backend/app/services/regime_builder.py +++ b/backend/app/services/regime_builder.py @@ -145,10 +145,8 @@ def _aggregate_daily(df: pl.DataFrame, index_pct_map: dict | None = None) -> pl. 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")) + # 基础聚合 — 全部用 group_by 一次性向量化算出, 避免逐日 filter 扫全表(OOM/超时元凶)。 + has_ma20 = "close" in avail and "ma20" in avail grouped = df.group_by("date").agg( *[ pl.col("change_pct").gt(0).sum().alias("up_count") @@ -181,9 +179,20 @@ def _aggregate_daily(df: pl.DataFrame, index_pct_map: dict | None = None) -> pl. [pl.col("amount").mean().alias("avg_amount")] if "amount" in avail else [pl.lit(0).alias("avg_amount")] ), + # MA20 上方占比: 向量化一次算出 (避免逐日 filter 扫全表)。 + # 仅统计 ma20 有效(非空且>0)的行中, close>ma20 的占比。 + *( + [ + pl.when(pl.col("ma20").is_not_null() & (pl.col("ma20") > 0) & (pl.col("close") > pl.col("ma20"))) + .then(1).otherwise(None).sum().alias("_above_cnt"), + pl.when(pl.col("ma20").is_not_null() & (pl.col("ma20") > 0)) + .then(1).otherwise(None).sum().alias("_valid_cnt"), + ] + if has_ma20 else [] + ), ).sort("date") - # 转成 dict 列表做后续计算(polars 表达式难表达的比率/MA20占比/分类) + # 转成 dict 列表做分类(规则引擎需逐日算, 但只扫 grouped 行数=天数, 不再回扫全表) index_pct_map = index_pct_map or {} rows = [] for r in grouped.iter_rows(named=True): @@ -191,15 +200,10 @@ def _aggregate_daily(df: pl.DataFrame, index_pct_map: dict | None = None) -> pl. 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 + # MA20 上方占比: 来自向量化聚合 (None→0) + valid_cnt = r.get("_valid_cnt") or 0 + above_cnt = r.get("_above_cnt") or 0 + ma20_above = (above_cnt / valid_cnt) if valid_cnt > 0 else 0.0 metrics = { "limit_up": limit_up, "limit_down": r.get("limit_down", 0) or 0, @@ -235,29 +239,95 @@ def _aggregate_daily(df: pl.DataFrame, index_pct_map: dict | None = None) -> pl. return pl.DataFrame(rows) if rows else pl.DataFrame() +# 全量回填分批参数(控制内存峰值) —— 实际值从用户偏好读取(preferences.get_regime_*), +# 这里的常量仅作 fallback(偏好读取失败时)和文档说明: +# - batch_days: 每批目标交易日数。越小内存越省、批次越多越慢; ma20 需 20 交易日。 +# - warmup_days: 每批前缀预热天数(日历日), 必须 > ma20 的 20 交易日(≈28 日历日)。 +_REGIME_BATCH_DAYS_DEFAULT = 60 +_REGIME_WARMUP_DAYS_DEFAULT = 40 + + +def _compute_batch(repo, enriched_dir, instruments, historical_shares, + batch_start: date, batch_end: date, warmup_days: int) -> pl.DataFrame: + """单批: 读 [batch_start-warmup, batch_end] → 算指标 → 截断回 [batch_start, batch_end]。 + + warmup 前缀保证每批边界的滚动窗口指标(ma20)正确, 不依赖相邻批次。 + 返回目标区间(不含 warmup)的含指标列 DataFrame。 + """ + from datetime import timedelta + from app.indicators.pipeline import compute_indicators, compute_limit_signals + warmup_start = batch_start - timedelta(days=warmup_days) + df = pl.scan_parquet(enriched_dir / "**" / "*.parquet").filter( + (pl.col("date") >= warmup_start) & (pl.col("date") <= batch_end) + ).collect() + if df.is_empty(): + return pl.DataFrame() + df = compute_indicators(df, needed={"change_pct", "ma20", "vol_ratio_5d"}) + if instruments is not None and not instruments.is_empty(): + df = compute_limit_signals( + df, instruments, + needed={"signal_limit_up", "signal_limit_down", "signal_broken_limit_up"}, + historical_shares=historical_shares, + ) + # 丢弃 warmup 行, 只留目标区间 + return df.filter((pl.col("date") >= batch_start) & (pl.col("date") <= batch_end)) + + def _scan_enriched_fallback(repo, start: date, end: date) -> pl.DataFrame | None: - """缓存不覆盖时的慢路径: 一次性 scan 全部 enriched parquet + 重算指标。 + """缓存不覆盖时的慢路径: scan enriched parquet + 重算所需指标列。 仅在 regime 首次全量回填或缓存未预热时触发。返回含信号列的多日 DataFrame。 - enriched 持久化只存基础列(OHLCV + raw_*/turnover/consecutive_*), 不含 - change_pct/ma20/signal_* 等派生列, 故此处需用 compute_all 补算全套指标。 + 内存控制(关键, 两层优化): + 1. needed 白名单: regime 只需 change_pct/ma20/涨跌停信号等少数列, 不用 compute_all + 算 72 列全套指标(那会让全量峰值达 6.8GB)。 + 2. 分批: 范围超过 batch_days 个交易日时按批切片, 每批带 warmup 前缀算完后 concat。 + batch_days / warmup_days 由用户偏好控制(数据页「市场环境」卡片设置), + 实测默认值(60/40)全量(515万行)峰值约 1.9GB, 4GB 内存机器可稳跑。 必须传入 instruments(涨跌停价表), 否则 compute_limit_signals 会跳过涨跌停信号。 """ + try: + from app.services import preferences + batch_days = preferences.get_regime_batch_days() + warmup_days = preferences.get_regime_warmup_days() + except Exception: # noqa: BLE001 + batch_days = _REGIME_BATCH_DAYS_DEFAULT + warmup_days = _REGIME_WARMUP_DAYS_DEFAULT + try: enriched_dir = repo.store.data_dir / "kline_daily_enriched" if not enriched_dir.exists(): return None - 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 instruments = repo.get_instruments() historical_shares = repo.get_historical_shares() - df = compute_all(df, instruments=instruments, historical_shares=historical_shares) - return df + + # 收集目标区间内所有交易日, 决定是否分批 + target_dates = sorted(d for d in enriched_date_set(repo) + if start <= d <= end) + if not target_dates: + return None + + # 小范围: 单次算(无分批开销) + if len(target_dates) <= batch_days: + df = _compute_batch(repo, enriched_dir, instruments, historical_shares, + target_dates[0], target_dates[-1], warmup_days) + return df if not df.is_empty() else None + + # 大范围: 按交易日分批, 逐批算 + concat + batches = [ + (target_dates[i], target_dates[min(i + batch_days - 1, len(target_dates) - 1)]) + for i in range(0, len(target_dates), batch_days) + ] + logger.info("regime fallback: %d 天分 %d 批 (每批≤%d天 + %d天warmup)", + len(target_dates), len(batches), batch_days, warmup_days) + parts: list[pl.DataFrame] = [] + for bs, be in batches: + df = _compute_batch(repo, enriched_dir, instruments, historical_shares, bs, be, warmup_days) + if not df.is_empty(): + parts.append(df) + if not parts: + return None + return pl.concat(parts, how="vertical_relaxed") except Exception as e: # noqa: BLE001 logger.warning("regime scan_enriched_fallback failed: %s", e) return None diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index cb1b5fc..c10ade0 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -78,9 +78,9 @@ const nav = [ { to: '/industry-analysis', label: '行业分析', icon: Landmark }, { to: '/financials', label: '财务分析', icon: FileText }, { to: '/monitor', label: '监控中心', icon: RadioTower }, + { to: '/regime', label: '市场环境', icon: Gauge, badge: 'beta' }, { to: '/review', label: '复盘', icon: BookOpenCheck }, { to: '/indices', label: '指数', icon: BarChart3 }, - { to: '/regime', label: '市场环境', icon: Gauge, badge: 'beta' }, { to: '/data', label: '数据', icon: Database }, ] as const diff --git a/frontend/src/components/data/PageSettingsModal.tsx b/frontend/src/components/data/PageSettingsModal.tsx index 86abaae..4104a48 100644 --- a/frontend/src/components/data/PageSettingsModal.tsx +++ b/frontend/src/components/data/PageSettingsModal.tsx @@ -21,7 +21,7 @@ import { storage } from '@/lib/storage' export type CardKey = | 'instruments' | 'daily' | 'adj_factor' | 'enriched' - | 'index' | 'etf' | 'minute' | 'financials' + | 'index' | 'etf' | 'minute' | 'financials' | 'regime' interface CardDef { key: CardKey @@ -43,6 +43,7 @@ export const DATA_CARD_DEFS: CardDef[] = [ { key: 'etf', label: 'ETF', desc: '场内交易基金日K', defaultHiddenIfNoCap: false, defaultHidden: true }, { key: 'minute', label: '分钟 K', desc: '分钟级K线(需 Pro+)', defaultHiddenIfNoCap: true }, { key: 'financials', label: '财务数据', desc: '财报数据(需 Expert)', defaultHiddenIfNoCap: true }, + { key: 'regime', label: '市场环境', desc: '每日环境状态(本地计算)', defaultHiddenIfNoCap: false }, ] const DEFAULT_ORDER = DATA_CARD_DEFS.map(d => d.key) diff --git a/frontend/src/components/data/RegimeConfigCard.tsx b/frontend/src/components/data/RegimeConfigCard.tsx new file mode 100644 index 0000000..4baa58e --- /dev/null +++ b/frontend/src/components/data/RegimeConfigCard.tsx @@ -0,0 +1,175 @@ +import { useState } from 'react' +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' +import { Check, Loader2, Activity, Layers } from 'lucide-react' +import { api } from '@/lib/api' +import { QK } from '@/lib/queryKeys' +import { toast } from '@/components/Toast' + +/** + * 市场环境(regime) 计算设置 —— 控制盘后管道是否自动计算 + 全量回填分批参数。 + * + * regime 是本地聚合计算(非外部拉取): 首次/regime 表为空时需全量回填多日, + * 内存与耗时较高。分批参数控制全量回填的内存峰值: 范围超过「每批天数」时 + * 切成多批独立算后拼接, 每批带 warmup 前缀保证滚动窗口指标(ma20)边界正确。 + */ +export function RegimeConfigCard() { + const qc = useQueryClient() + const prefs = useQuery({ queryKey: QK.preferences, queryFn: api.preferences }) + + const updateEnabled = useMutation({ + mutationFn: (enabled: boolean) => api.updatePipelineRegimeEnabled(enabled), + onSuccess: () => qc.invalidateQueries({ queryKey: QK.preferences }), + }) + + // 分批参数本地草稿(失焦/回车时整体保存), 避免每次按键都请求 + const batchDays = prefs.data?.regime_batch_days ?? 60 + const warmupDays = prefs.data?.regime_warmup_days ?? 40 + const [draftBatch, setDraftBatch] = useState(String(batchDays)) + const [draftWarmup, setDraftWarmup] = useState(String(warmupDays)) + + // prefs 加载后同步一次草稿(仅首次) + const [synced, setSynced] = useState(false) + if (!synced && prefs.data) { + setDraftBatch(String(batchDays)) + setDraftWarmup(String(warmupDays)) + setSynced(true) + } + + const updateParams = useMutation({ + mutationFn: (params: { batch_days?: number; warmup_days?: number }) => + api.updateRegimeBatchParams(params), + onSuccess: (data) => { + setDraftBatch(String(data.regime_batch_days)) + setDraftWarmup(String(data.regime_warmup_days)) + qc.invalidateQueries({ queryKey: QK.preferences }) + toast('分批参数已保存', 'success') + }, + onError: () => toast('保存失败,请检查输入范围', 'error'), + }) + + const saveBatch = () => { + const n = Math.floor(Number(draftBatch) || 0) + if (n < 25 || n > 500) { + toast('每批天数范围 25 ~ 500', 'error') + setDraftBatch(String(batchDays)) + return + } + if (n !== batchDays) updateParams.mutate({ batch_days: n }) + } + const saveWarmup = () => { + const n = Math.floor(Number(draftWarmup) || 0) + if (n < 35 || n > 90) { + toast('预热天数范围 35 ~ 90', 'error') + setDraftWarmup(String(warmupDays)) + return + } + if (n !== warmupDays) updateParams.mutate({ warmup_days: n }) + } + + // 默认关闭: 未设置过时视为 false + const on = prefs.data?.pipeline_regime_enabled ?? false + + return ( +