From 81daf018c555ddc10f3213b416d03fb937b00484 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Fri, 14 Aug 2026 13:04:04 +0800 Subject: [PATCH] =?UTF-8?q?perf(indicators,kline):=20=E7=83=AD=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E4=BF=A1=E5=8F=B7=E7=BC=93=E5=AD=98=20+=20=E6=97=A5K?= =?UTF-8?q?=E5=BF=AB=E8=B7=AF=E5=BE=84=E8=B7=B3=E8=BF=87=E5=85=A8=E9=87=8F?= =?UTF-8?q?=E6=8C=87=E6=A0=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三处性能优化 (后端内部, 不改 API 契约): 1. 增量热路径自定义信号缓存 (pipeline.py) compute_enriched_today 每秒级执行, 原每轮 glob + 读所有 JSON + 重编译 表达式。新增 _custom_signal_exprs_today 模块级缓存 (allow_shift=False), 与全量版分开缓存, invalidate_custom_signals 统一失效。 2. ETF/指数迷你蜡烛图跳过全量指标 (repository.py) get_index_daily/get_etf_daily 增加快路径: 请求列全是 parquet 存储列时 scan + 列下推直接返回, 跳过 150 天 warmup + 全套指标计算。缺列回退完整路径。 3. 单股日K跳过全量指标 (repository.py) get_daily 同样加快路径并保留缓存覆盖最新日。采用最保守的列下推方案, 不引入 needed 裁剪, 避免依赖闭包漏算金融数据的风险。 验证: pytest 全套 582 passed; Ruff 无新增告警; 快路径功能验证通过。 --- backend/app/indicators/pipeline.py | 37 ++++++++++++++++++++++-------- backend/app/tickflow/repository.py | 32 ++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 9 deletions(-) diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index dab7777..24ade27 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -36,12 +36,16 @@ logger = logging.getLogger(__name__) # ── 自定义信号缓存 ───────────────────────────────────── # 从 data/user_data/custom_signals/*.json 加载并编译为 Polars 表达式。 +# 两套表达式分别用于全量路径 (allow_shift=True, 支持日期偏移条件) +# 和盘中增量热路径 (allow_shift=False, 跳过偏移条件)。 # 模块级缓存：首次调用时加载，invalidate_custom_signals() 后下次重载。 +# 增量路径每秒级执行, 若不缓存则每轮 glob + 读所有 JSON + 重编译表达式。 _custom_signal_exprs: dict[str, pl.Expr] | None = None +_custom_signal_exprs_today: dict[str, pl.Expr] | None = None def _get_custom_signal_exprs() -> dict[str, pl.Expr]: - """懒加载自定义信号表达式（带模块级缓存）。""" + """懒加载自定义信号表达式（带模块级缓存，allow_shift=True）。""" global _custom_signal_exprs if _custom_signal_exprs is None: from app.strategy import custom_signals @@ -54,10 +58,29 @@ def _get_custom_signal_exprs() -> dict[str, pl.Expr]: return _custom_signal_exprs +def _get_custom_signal_exprs_today() -> dict[str, pl.Expr]: + """盘中增量热路径专用 (allow_shift=False, 跳过日期偏移条件)。 + + 与全量版分开缓存：盘中单日快照上 .shift 跨 symbol 语义不正确, + build_expressions(allow_shift=False) 会跳过带偏移的信号, 结果集不同。 + """ + global _custom_signal_exprs_today + if _custom_signal_exprs_today is None: + from app.strategy import custom_signals + try: + sigs = custom_signals.load_all(settings.data_dir) + _custom_signal_exprs_today = custom_signals.build_expressions(sigs, allow_shift=False) + except Exception as e: + logger.warning("custom signals load failed (today): %s", e) + _custom_signal_exprs_today = {} + return _custom_signal_exprs_today + + def invalidate_custom_signals() -> None: """失效自定义信号缓存（保存/删除信号后调用，下次计算重新加载）。""" - global _custom_signal_exprs + global _custom_signal_exprs, _custom_signal_exprs_today _custom_signal_exprs = None + _custom_signal_exprs_today = None # enriched parquet 仅存储的列 (14 列) @@ -1652,14 +1675,10 @@ def compute_enriched_today( df = df.drop([c for c in drop_cols if c in df.columns]) # 自定义信号（日级实时路径同样注入, 但不支持日期偏移条件 → allow_shift=False） + # 复用模块级缓存 _custom_signal_exprs_today: 增量热路径每秒级执行, + # 不缓存则每轮 glob + 读所有 JSON + 重编译表达式。失效由 invalidate_custom_signals 统一管理。 from app.strategy import custom_signals - try: - sigs = custom_signals.load_all(settings.data_dir) - today_exprs = custom_signals.build_expressions(sigs, allow_shift=False) - except Exception as e: - logger.warning("custom signals load failed (today): %s", e) - today_exprs = {} - df = custom_signals.inject(df, today_exprs) + df = custom_signals.inject(df, _get_custom_signal_exprs_today()) # 清理 NaN / Inf float_cols = [c for c in df.columns if df[c].dtype.is_float()] diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index e61c0d9..10c874f 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -1335,6 +1335,23 @@ class KlineRepository: """单股日K查询 — 从14列parquet读取后即时计算指标。""" from datetime import timedelta + # 快路径: 请求的列全是 parquet 直接存储的列 (如迷你蜡烛图只要 OHLCV) → + # scan + 列下推直接返回, 跳过 warmup(150天) 与 _compute_enriched_range 全套指标计算。 + # 仍用 enriched_latest 缓存覆盖最新日 (盘中更准), 只保留请求列。 + # 试探 scan 仅读请求的列; 缺列时回退到下方完整计算路径 (代价仅一次轻量 scan)。 + if columns: + df = self._scan_daily_symbol(symbol, start, end, columns) + if not df.is_empty() and all(c in df.columns for c in columns): + cached, cache_date = self.get_enriched_latest() + if cached is not None and not cached.is_empty() and cache_date: + if start <= cache_date <= end: + cached_part = self._filter_cached(cached, symbol, columns) + if not cached_part.is_empty(): + df = df.filter(pl.col("date") != cache_date) + common_cols = [c for c in df.columns if c in cached_part.columns] + df = pl.concat([df.select(common_cols), cached_part.select(common_cols)]) + return df + # 扩展范围用于指标预热 (MA60 需要 ~60 交易日 ≈ 120 日历日) warmup_start = start - timedelta(days=150) @@ -1389,6 +1406,14 @@ class KlineRepository: """指数日K查询 — 从独立指数 enriched parquet 读取后即时计算通用指标。""" from datetime import timedelta + # 快路径: 若请求的列全部是 parquet 直接存储的列 (如迷你蜡烛图只要 OHLCV), + # 直接 scan + 列下推返回, 跳过 warmup(150天) 与 _compute_index_enriched_range 全套指标计算。 + # 试探 scan 仅读请求的列, 缺列时回退到下方完整计算路径 (代价仅一次轻量 scan)。 + if columns: + df = self._scan_index_daily_symbol(symbol, start, end, columns) + if not df.is_empty() and all(c in df.columns for c in columns): + return df + warmup_start = start - timedelta(days=150) df = self._scan_index_daily_symbol(symbol, warmup_start, end, None) if not df.is_empty(): @@ -1409,6 +1434,13 @@ class KlineRepository: """ETF 日K查询 — 优先读独立 ETF enriched，兼容旧版 index enriched 中的 ETF。""" from datetime import timedelta + # 快路径: 独立 ETF 数据 + 请求列全是 parquet 存储列 → 直接 scan 列下推, 跳过 warmup + compute。 + # 无独立 ETF 数据 (旧版存入 index enriched) 时 df 为空, 自然回退到下方完整路径。 + if columns: + df = self._scan_etf_daily_symbol(symbol, start, end, columns) + if not df.is_empty() and all(c in df.columns for c in columns): + return df + warmup_start = start - timedelta(days=150) df = self._scan_etf_daily_symbol(symbol, warmup_start, end, None) if df.is_empty():