From 9aa96edbd72ea406535f9fa90b01bec96a83182a Mon Sep 17 00:00:00 2001 From: Jinfeng Sun <86536994+Tendo33@users.noreply.github.com> Date: Wed, 8 Jul 2026 18:10:19 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E8=BF=9B:=20=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E9=9F=A7=E6=80=A7=20+=20=E6=95=B0=E6=8D=AE=E6=80=A7=E8=83=BD?= =?UTF-8?q?=20+=20=E6=AD=BB=E4=BB=A3=E7=A0=81=E6=B8=85=E7=90=86=20+=20?= =?UTF-8?q?=E5=89=8D=E7=AB=AF=20UX=20+=20ST/Sharpe=20=E4=BF=AE=E5=A4=8D=20?= =?UTF-8?q?(#78)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(concurrency): 共享缓存/任务表加锁, 全局限速, depth 原子写, 认证热路径缓存 修复多线程下的竞态与阻塞: - overview/strategy_cache/PanelCache/StrategyMonitor._watching 四处共享状态加锁, 消除 "dict/OrderedDict mutated" 与丢更新/半写读取 - strategy_cache/depth parquet 改临时文件 + os.replace 原子写 - rate_limits 改进程级共享时间轴限速, 并发同步不再聚合超过单能力 rpm; scheduler 令牌账目与 sleep 分离, sleep 不再独占锁串行化其他请求 - auth.is_configured() 内存缓存, 认证中间件不再每请求读盘阻塞事件循环 - api/backtest 任务清理/取消全程持 _jobs_lock, 并用 Semaphore(2) 限并发重回测 Co-Authored-By: Claude Opus 4.8 (1M context) * perf(data): limit_ladder 去 N+1 全市场重算, 指标裁剪, factor 向量化 - limit_ladder 前一日 consecutive 改窄读单日 parquet 存储列 (谓词/投影下推), 替代 range(1,10) 逐日 _load_enriched_for_date 全市场指标重算 (最坏 9x) - compute_indicators 新增可选 needed 裁剪 (默认 None 行为逐位不变, 已对照验证), factor 只算所需因子列 - factor._calc_period_return 用 Polars join 替代 Python 逐行 price_map 循环, _add_groups 去 map_elements 改纯表达式 (输出逐位一致) - screener ext value_map 按 parquet mtime 记忆化, 免每请求磁盘重读 (DuckDB 过滤仍用隔离 :memory: 连接, 不扩大注入面) Co-Authored-By: Claude Opus 4.8 (1M context) * refactor(backend): 报表存储去重, 删死代码, DuckDB 视图重建收敛, 管道失败如实标记 - 三份近乎逐字复制的 *_reports.py 收敛到共享 JsonReportStore (原子写 + 锁), 各模块公有 API/id 格式/上限/落盘 schema 完全保持不变 - 删除 ext_pull.py 中字节相同的死 _run_loop (Python 只绑第二个) 及无用 import - 13 张 DuckDB 视图重建收敛为唯一权威 repository.rebuild_views(), daily_pipeline 与 /api/data/clear 改为调用 (修好 clear 路径漏挂视图的漂移) - daily_pipeline 累积 stage_errors 并在末尾抛出, 部分失败不再误报成功; free/None 模式的能力门控跳过不计入失败 Co-Authored-By: Claude Opus 4.8 (1M context) * feat(frontend): SSE 连接态, 路由代码分割, 查询失效修复, 三态与无障碍 - 实时行情 SSE: 连接态 store + 指数退避 + 断线徽标/toast (避免静默丢告警); 回测 SSE 断线有界重连 + 可重试, 不再永久卡住进度条 - router 全部 React.lazy + Suspense, vite manualChunks 拆图表库 (echarts 变独立 1MB 懒加载 chunk, 首屏包显著减小) - 修 Data 清库后其它页显示旧数据 (改回广域失效); 修 Watchlist kline 失效键 永不匹配; query key 收敛到 QK 工厂 (新增 strategyDetail) - Monitor/Analysis/StockAnalysis/ExtPages/CustomSignals 补 loading 门控与 error/empty 三态区分 - 新增共享 Modal 原语 (焦点陷阱/ESC/焦点还原/aria), 改造 3 个高频弹窗; Toast/AlertToast 加 aria-live 与键盘可达; Watchlist/LimitUpLadder 卡片 memo 修复本轮 review 发现的缺陷: - Modal 焦点 effect 依赖 onClose 致每次输入抢焦点 → 改 ref 只装一次 - StrategySettingsDialog 删除确认框被 Modal 面板裁剪 → 移出作兄弟节点 Co-Authored-By: Claude Opus 4.8 (1M context) * fix(quant): 修正 ST 板块限价套错 与 因子 Sharpe 年化频率 两个不报错但会算错数的领域 bug: 1. ST 5% 涨跌停限幅被无条件套到创业板/科创板 ST 股: 注册制改革后 创业板(300/301)、科创板(688/689) 的风险警示股仍执行 20%, 北交所 30%, 只有主板 ST 才是 5%。原代码 _is_st 先判且覆盖板块限幅, 导致 创业板/科创板 ST 的涨停价按 5% 计算 → +5% 被误报涨停、真 +20% 涨停被漏报, 污染 signal_limit_up / consecutive_limit_ups / 连板梯队 / near_limit_up。 修正: ST 5% 仅在 ~(创业板|科创板|北交所) 时生效 (EOD + 盘中两条路径 + near_limit_up)。 2. 因子回测 Sharpe 一律乘 √252, 但 group_nav 每点是一个调仓周期收益: 月频调仓下是月收益, 乘 √252 会把 Sharpe 高估 √(252/12) ≈ 4.6x (周频 ≈2.2x), 使无效因子显示成明星因子, 废掉"先筛无效指标"的用途。 修正: 年化系数按 config.rebalance 取 √252/√52/√12。 新增 tests/test_st_limit_and_sharpe.py (5 例) 覆盖两处修正。 Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- backend/app/api/backtest.py | 22 +- backend/app/api/data.py | 27 +- backend/app/api/overview.py | 25 +- backend/app/api/screener.py | 54 ++- backend/app/backtest/engine.py | 29 +- backend/app/backtest/factor.py | 102 +++--- backend/app/indicators/pipeline.py | 324 ++++++++++++------ backend/app/jobs/daily_pipeline.py | 66 ++-- backend/app/services/ai_reports.py | 69 +--- backend/app/services/auth.py | 19 +- backend/app/services/depth_service.py | 25 +- backend/app/services/ext_pull.py | 52 --- backend/app/services/json_report_store.py | 126 +++++++ backend/app/services/market_recap_reports.py | 66 +--- backend/app/services/screener.py | 49 ++- backend/app/services/stock_reports.py | 61 +--- backend/app/services/strategy_cache.py | 34 +- backend/app/strategy/builtin/near_limit_up.py | 11 +- backend/app/strategy/monitor.py | 23 +- backend/app/tickflow/rate_limits.py | 42 ++- backend/app/tickflow/repository.py | 35 ++ backend/app/tickflow/scheduler.py | 12 +- backend/tests/test_st_limit_and_sharpe.py | 88 +++++ frontend/src/components/AlertToast.tsx | 20 +- frontend/src/components/Layout.tsx | 27 +- frontend/src/components/Modal.tsx | 137 ++++++++ frontend/src/components/RpsRotationDialog.tsx | 36 +- frontend/src/components/Toast.tsx | 7 +- .../screener/StrategyBuilderDialog.tsx | 22 +- .../screener/StrategySettingsDialog.tsx | 34 +- frontend/src/lib/backtestTask.ts | 59 +++- frontend/src/lib/queryKeys.ts | 1 + frontend/src/lib/useQuoteStream.ts | 65 +++- frontend/src/pages/Analysis.tsx | 11 +- frontend/src/pages/Data.tsx | 2 + frontend/src/pages/Dev.tsx | 3 +- frontend/src/pages/LimitUpLadder.tsx | 18 +- frontend/src/pages/Monitor.tsx | 17 +- frontend/src/pages/StockAnalysis.tsx | 12 +- frontend/src/pages/Watchlist.tsx | 39 ++- .../src/pages/backtest/StrategyBacktest.tsx | 16 +- frontend/src/pages/settings/CustomSignals.tsx | 11 +- frontend/src/pages/settings/ExtPages.tsx | 11 +- frontend/src/router.tsx | 41 ++- frontend/vite.config.js | 14 + frontend/vite.config.ts | 12 + 46 files changed, 1352 insertions(+), 624 deletions(-) create mode 100644 backend/app/services/json_report_store.py create mode 100644 backend/tests/test_st_limit_and_sharpe.py create mode 100644 frontend/src/components/Modal.tsx diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index 720d852..6d06d6b 100644 --- a/backend/app/api/backtest.py +++ b/backend/app/api/backtest.py @@ -271,13 +271,18 @@ _running_jobs: dict[str, _BacktestJob] = {} _jobs_lock = threading.Lock() _JOB_TTL = 300 # 完成后保留 5 分钟 +# 并发回测上限: 多个重回测同时跑会 OOM (服务器内存约 1.8GB)。用信号量限并发, +# 超出的任务在 _run_backtest 里排队, SSE 连接照常保持, run 一开始就有进度。 +_backtest_semaphore = threading.Semaphore(2) + def _cleanup_stale_jobs(): - """清理过期任务 (完成超过 TTL 的)。""" + """清理过期任务 (完成超过 TTL 的)。全程持 _jobs_lock: 迭代+pop 与其他访问互斥。""" now = time.time() - stale = [k for k, j in _running_jobs.items() if j.done and now - j.finish_ts > _JOB_TTL] - for k in stale: - _running_jobs.pop(k, None) + with _jobs_lock: + stale = [k for k, j in _running_jobs.items() if j.done and now - j.finish_ts > _JOB_TTL] + for k in stale: + _running_jobs.pop(k, None) def _make_job_key( @@ -404,6 +409,9 @@ async def strategy_stream( ) def _run_backtest(): + # 信号量限并发: 超额任务在此阻塞排队, 不并发吃满内存 (等待期间 cancel_event + # 仍可置位, svc.run 会据此提前返回 cancelled)。持槽跑完在 finally 释放。 + _backtest_semaphore.acquire() try: result = svc.run(cfg, lambda d: job.progress.append(d), job.cancel_event) job.result = result @@ -413,6 +421,8 @@ async def strategy_stream( job.error = str(e) job.done = True job.finish_ts = time.time() + finally: + _backtest_semaphore.release() # 启动后台线程 (不阻塞事件循环) threading.Thread(target=_run_backtest, daemon=True).start() @@ -493,7 +503,9 @@ async def strategy_cancel(request: Request): stamp_tax_pct=_get_opt_float("stamp_tax_pct"), asset_type=_get("asset_type", "stock"), ) - job = _running_jobs.get(job_key) + # 持锁读任务表: 与 _cleanup_stale_jobs 的 pop、stream 的写入互斥 + with _jobs_lock: + job = _running_jobs.get(job_key) if job and not job.done: job.cancel_event.set() return {"ok": True} diff --git a/backend/app/api/data.py b/backend/app/api/data.py index 4aaaf39..b03933b 100644 --- a/backend/app/api/data.py +++ b/backend/app/api/data.py @@ -680,30 +680,9 @@ def clear_data(request: Request): from app.api.overview import invalidate_overview_cache invalidate_overview_cache() - # 刷新 DuckDB 视图（空 parquet 目录也需要重新挂载） - d = data_dir.as_posix() - for name, path in { - "kline_daily": f"{d}/kline_daily/**/*.parquet", - "kline_enriched": f"{d}/kline_daily_enriched/**/*.parquet", - "kline_index_daily": f"{d}/kline_index_daily/**/*.parquet", - "kline_index_enriched": f"{d}/kline_index_enriched/**/*.parquet", - "kline_etf_daily": f"{d}/kline_etf_daily/**/*.parquet", - "kline_etf_enriched": f"{d}/kline_etf_enriched/**/*.parquet", - "kline_etf_minute": f"{d}/kline_etf_minute/**/*.parquet", - "kline_minute": f"{d}/kline_minute/**/*.parquet", - "adj_factor": f"{d}/adj_factor/**/*.parquet", - "adj_factor_etf": f"{d}/adj_factor_etf/**/*.parquet", - "instruments": f"{d}/instruments/**/*.parquet", - "instruments_index": f"{d}/instruments_index/**/*.parquet", - "instruments_etf": f"{d}/instruments_etf/**/*.parquet", - }.items(): - try: - repo.db.execute( - f"CREATE OR REPLACE VIEW {name} AS " - f"SELECT * FROM read_parquet('{path}', union_by_name=true)" - ) - except Exception: - pass + # 刷新 DuckDB 视图（空 parquet 目录也需要重新挂载）—— + # 委托给 repository 的唯一权威实现, 覆盖全部视图 (此前这里内联的副本漏了几张)。 + repo.rebuild_views() logger.info("数据已清除: 删除 %d 个 parquet 文件", deleted) invalidate_data_cache(None) diff --git a/backend/app/api/overview.py b/backend/app/api/overview.py index 3bd7ede..4cce44d 100644 --- a/backend/app/api/overview.py +++ b/backend/app/api/overview.py @@ -3,6 +3,7 @@ from __future__ import annotations import math import re +import threading import time from datetime import date from typing import Any @@ -19,6 +20,9 @@ _CACHE_TTL = 5.0 _cache: dict[str, Any] | None = None _cache_key: str | None = None _cache_ts: float = 0.0 +# 缓存跨线程读写锁: market_overview 在 FastAPI 线程池读, invalidate 在数据刷新线程清, +# 无锁会读到撕裂/过期状态。用模块级 Lock 守护 check-then-set 与 clear。 +_cache_lock = threading.Lock() def invalidate_overview_cache() -> None: @@ -27,9 +31,10 @@ def invalidate_overview_cache() -> None: 清除数据后调用, 避免看板在 TTL 窗口内继续返回旧的聚合结果。 """ global _cache, _cache_key, _cache_ts - _cache = None - _cache_key = None - _cache_ts = 0.0 + with _cache_lock: + _cache = None + _cache_key = None + _cache_ts = 0.0 CORE_INDEX_NAMES = { @@ -363,10 +368,14 @@ def market_overview(request: Request, as_of: date | None = None): global _cache, _cache_key, _cache_ts now = time.time() cache_key = as_of.isoformat() if as_of else "latest" - if _cache is not None and _cache_key == cache_key and (now - _cache_ts) < _CACHE_TTL: - return _cache + # 读缓存持锁, 避免与 invalidate 的 clear 竞态读到撕裂状态 + with _cache_lock: + if _cache is not None and _cache_key == cache_key and (now - _cache_ts) < _CACHE_TTL: + return _cache + # 装配在锁外进行 (耗时), 允许并发未命中时各自构建, 不长时间持锁串行化请求 data = _build_overview(request, as_of) - _cache = data - _cache_key = cache_key - _cache_ts = now + with _cache_lock: + _cache = data + _cache_key = cache_key + _cache_ts = now return data diff --git a/backend/app/api/screener.py b/backend/app/api/screener.py index daf1be7..d81c962 100644 --- a/backend/app/api/screener.py +++ b/backend/app/api/screener.py @@ -1,8 +1,10 @@ """Screener API。""" from __future__ import annotations +import glob as _glob import logging import math +import os import re import time from dataclasses import asdict @@ -64,11 +66,34 @@ def _quote_ident(name: str) -> str: return '"' + name.replace('"', '""') + '"' +# ── 扩展列 value_map 缓存 ──────────────────────────────────────────── +# 每次请求 _load_ext_value_maps 都会重新从磁盘读 ext parquet 并重建 {symbol: value}。 +# 用底层 parquet 文件的 (路径, mtime) 签名做 memoize: 文件未变则复用上次的 map, +# parquet 被重写 (mtime 变化) 时自动失效重算。仅缓存基于 config 的快照/时序路径, +# 无 config 的 DuckDB view 回退路径不缓存 (少见)。 +_ext_value_map_cache: dict[tuple[str, str], tuple[Any, dict[str, Any]]] = {} + + +def _ext_parquet_signature(cfg, data_dir) -> Optional[tuple]: + """该扩展配置底层 parquet 文件的 (路径, mtime) 签名; 出错返回 None (禁用缓存)。""" + try: + from app.api.ext_data import _parquet_glob + pattern = _parquet_glob(cfg, data_dir) + files = sorted(_glob.glob(pattern, recursive=True)) + if not files: + return None + return tuple((f, os.path.getmtime(f)) for f in files) + except Exception: # noqa: BLE001 + return None + + def _load_ext_value_maps(repo, ext_columns: Optional[str]) -> dict[str, dict[str, Any]]: """按请求加载扩展列，返回 {输出列名: {symbol: value}}。 策略结果缓存是共享文件，不能被不同 ext_columns 组合污染；因此扩展列只在 返回前通过该投影映射追加到结果副本中。 + + 基于 config 的路径按 parquet 文件 mtime 签名 memoize, 文件未变时跳过磁盘重读。 """ ext_specs = _parse_ext_columns(ext_columns) if ext_columns else [] if not ext_specs: @@ -87,8 +112,15 @@ def _load_ext_value_maps(repo, ext_columns: Optional[str]) -> dict[str, dict[str for config_id, field_name in ext_specs: out_col = f"{config_id}__{field_name}" cfg = configs.get(config_id) + cache_key = (config_id, field_name) + sig = _ext_parquet_signature(cfg, data_dir) if cfg else None try: if cfg: + # 命中缓存 (文件签名一致) → 复用, 免去磁盘重读 + cached = _ext_value_map_cache.get(cache_key) + if cached is not None and sig is not None and cached[0] == sig: + value_maps[out_col] = cached[1] + continue # 时序扩展表只取最新分区，避免历史分区把同一 symbol JOIN 放大。 ext_df, _ = _read_ext_dataframe(cfg, data_dir) else: @@ -101,11 +133,14 @@ def _load_ext_value_maps(repo, ext_columns: Optional[str]) -> dict[str, dict[str continue ext_df = ext_df.select(["symbol", field_name]).unique(subset=["symbol"], keep="last") - value_maps[out_col] = { + vmap = { str(row["symbol"]): _safe_ext_value(row.get(field_name)) for row in ext_df.to_dicts() if row.get("symbol") } + value_maps[out_col] = vmap + if cfg and sig is not None: + _ext_value_map_cache[cache_key] = (sig, vmap) except Exception as e: # noqa: BLE001 logger.debug("screener ext column join skipped for %s.%s: %s", config_id, field_name, e) @@ -479,8 +514,6 @@ def limit_ladder( ext_columns: 动态 JOIN 扩展数据, 如 "concept.concept,industry.industry" """ - from datetime import timedelta - import polars as pl is_down = direction == "down" @@ -540,17 +573,10 @@ def limit_ladder( sealed_counts_up = _count_sealed(up_map, sealed_up_ready) sealed_counts_down = _count_sealed(down_map, sealed_down_ready) - # 加载前一日数据获取 prev consecutive_limit_ups/downs - prev_consec: pl.DataFrame = pl.DataFrame() - for delta in range(1, 10): - candidate = as_of - timedelta(days=delta) - df_prev = svc._load_enriched_for_date(candidate) - if not df_prev.is_empty() and consec_col in df_prev.columns: - prev_consec = df_prev.select( - "symbol", - pl.col(consec_col).alias("prev_consec"), - ) - break + # 加载前一日的 prev consecutive_limit_ups/downs + # 窄读: 仅取前一交易日的 [symbol, consec_col] 两列 (存储列, 直接谓词下推读 parquet), + # 替代旧的 range(1,10) 循环逐日 _load_enriched_for_date 全量指标重算 (最坏 9× 全市场重算)。 + prev_consec: pl.DataFrame = svc.load_prior_consecutive(as_of, consec_col) if not prev_consec.is_empty(): df = df.join(prev_consec, on="symbol", how="left") diff --git a/backend/app/backtest/engine.py b/backend/app/backtest/engine.py index 53568a9..8af1b6f 100644 --- a/backend/app/backtest/engine.py +++ b/backend/app/backtest/engine.py @@ -6,6 +6,7 @@ from __future__ import annotations import hashlib import logging +import threading import time from collections import OrderedDict from dataclasses import dataclass @@ -126,6 +127,10 @@ class PanelCache: self._cache: OrderedDict[str, _CacheEntry] = OrderedDict() self._max_size = max_size self._ttl = ttl_seconds + # 跨请求单例, SSE 回测在各自 daemon 线程并发访问 OrderedDict。 + # 无锁的 move_to_end/del/popitem check-then-act 会抛 "OrderedDict mutated"。 + # 用实例锁守护所有 OrderedDict 变更; compute_fn (重扫盘) 放锁外避免串行化。 + self._lock = threading.Lock() def get_or_compute( self, @@ -139,21 +144,25 @@ class PanelCache: key = self._make_key(symbols, start, end, columns, asset_type) now = time.monotonic() - if key in self._cache: - entry = self._cache[key] - if now - entry.ts < self._ttl: - self._cache.move_to_end(key) - return entry.df - del self._cache[key] + with self._lock: + if key in self._cache: + entry = self._cache[key] + if now - entry.ts < self._ttl: + self._cache.move_to_end(key) + return entry.df + del self._cache[key] + # 计算在锁外 (可能重扫 parquet, 耗时); 并发相同 key 至多重复算一次, 不会崩 df = compute_fn(symbols, start, end, columns, asset_type) - self._cache[key] = _CacheEntry(df=df, ts=now) - if len(self._cache) > self._max_size: - self._cache.popitem(last=False) + with self._lock: + self._cache[key] = _CacheEntry(df=df, ts=now) + if len(self._cache) > self._max_size: + self._cache.popitem(last=False) return df def invalidate(self) -> None: - self._cache.clear() + with self._lock: + self._cache.clear() @staticmethod def _make_key(symbols: list[str] | None, start: date, end: date, columns: list[str] | None, asset_type: str = "stock") -> str: diff --git a/backend/app/backtest/factor.py b/backend/app/backtest/factor.py index e1df935..2b02b76 100644 --- a/backend/app/backtest/factor.py +++ b/backend/app/backtest/factor.py @@ -174,7 +174,7 @@ class FactorBacktestService: # ── 2. 分层回测 ── panel = self._add_groups(panel, factor_col, config.n_groups) group_nav = self._calc_group_nav(panel, config) - group_stats = self._calc_group_stats(group_nav, config.start, config.end) + group_stats = self._calc_group_stats(group_nav, config.start, config.end, config.rebalance) # ── 3. 多空组合 ── long_short_nav, long_short_stats = self._calc_long_short(group_nav, config) @@ -207,7 +207,8 @@ class FactorBacktestService: from app.indicators.pipeline import compute_indicators - computed = compute_indicators(panel) + # 只需要单个因子列 → 用 needed 裁剪, 跳过无关的 EMA/KDJ/RSI 等计算 pass + computed = compute_indicators(panel, needed={factor_col}) if factor_col not in computed.columns: return panel return computed.select(["symbol", "date", "close", factor_col]) @@ -242,7 +243,6 @@ class FactorBacktestService: import datetime as _dt all_dates = sorted(panel["date"].unique().to_list()) - date_set = set(all_dates) if rebalance == "weekly": # 调仓日 = 每周一 @@ -268,45 +268,58 @@ class FactorBacktestService: panel = panel.with_columns(pl.lit(None).cast(pl.Float64).alias("_next_return")) return panel - # 对每个调仓日，找到下一个调仓日 + # 对每个调仓日，找到下一个调仓日 (仅在 unique 日期上做, 成本极低) sorted_rebalance = sorted(rebalance_dates) - next_rebalance_map: dict = {} + reb_dates: list = [] + next_dates: list = [] for i, d in enumerate(sorted_rebalance): if i + 1 < len(sorted_rebalance): - next_rebalance_map[d] = sorted_rebalance[i + 1] + reb_dates.append(d) + next_dates.append(sorted_rebalance[i + 1]) # 最后一个调仓日没有下一个，不计算收益 - # 构建 (date, symbol) → next_rebalance_date 的 close 价格映射 + if not reb_dates: + panel = panel.with_columns(pl.lit(None).cast(pl.Float64).alias("_next_return")) + return panel + panel = panel.sort(["symbol", "date"]) - dates_col = panel["date"].to_list() - close_col = panel["close"].to_list() - symbol_col = panel["symbol"].to_list() + date_dtype = panel.schema["date"] - # 先找下个调仓日的 close - # 建立 (date, symbol) → close 的快速查找 - price_map: dict[tuple, float] = {} - for i in range(len(dates_col)): - price_map[(str(dates_col[i]), symbol_col[i])] = close_col[i] + # 调仓日 → 下一调仓日 的映射表 (向量化 JOIN, 替代 Python 逐行 price_map 循环) + rebal_df = pl.DataFrame( + {"date": reb_dates, "_next_reb_date": next_dates} + ).with_columns( + pl.col("date").cast(date_dtype), + pl.col("_next_reb_date").cast(date_dtype), + ) - next_returns = [None] * len(panel) - for i in range(len(panel)): - d = dates_col[i] - d_val = d if isinstance(d, _dt.date) else _dt.date.fromisoformat(str(d)) - if d not in rebalance_dates: - continue - next_d = next_rebalance_map.get(d) - if next_d is None: - continue - next_d_str = str(next_d)[:10] - d_str = str(d)[:10] - sym = symbol_col[i] - next_close = price_map.get((next_d_str, sym)) - cur_close = close_col[i] - if next_close is not None and cur_close and cur_close > 0: - next_returns[i] = (next_close / cur_close - 1.0) + # (symbol, 下一调仓日) → 该日 close 的查找表 (等价于原 price_map, 重复取 last) + price_lookup = ( + panel.select( + pl.col("symbol"), + pl.col("date").alias("_next_reb_date"), + pl.col("close").alias("_next_close"), + ) + .unique(subset=["symbol", "_next_reb_date"], keep="last") + ) - panel = panel.with_columns( - pl.Series("_next_return", next_returns, dtype=pl.Float64) + # 只在调仓日标记行有效: 下一调仓日该股 close / 当日 close - 1; 缺价或非调仓日为 null + panel = ( + panel.join(rebal_df, on="date", how="left") + .join(price_lookup, on=["symbol", "_next_reb_date"], how="left") + .with_columns( + pl.when( + pl.col("_next_reb_date").is_not_null() + & pl.col("_next_close").is_not_null() + & (pl.col("close") > 0) + ) + .then(pl.col("_next_close") / pl.col("close") - 1.0) + .otherwise(None) + .cast(pl.Float64) + .alias("_next_return") + ) + .drop(["_next_reb_date", "_next_close"]) + .sort(["symbol", "date"]) ) return panel @@ -323,14 +336,16 @@ class FactorBacktestService: ) .with_columns( ( - ((pl.col("_factor_ord") * n_groups) / pl.col("_factor_count")) - .floor() - .cast(pl.Int64) - + 1 + pl.lit("Q") + + ( + ((pl.col("_factor_ord") * n_groups) / pl.col("_factor_count")) + .floor() + .cast(pl.Int64) + + 1 + ) + .clip(1, n_groups) + .cast(pl.Utf8) ) - .clip(1, n_groups) - .cast(pl.Utf8) - .map_elements(lambda v: f"Q{v}", return_dtype=pl.Utf8) .alias("_group") ) .drop(["_factor_ord", "_factor_count"]) @@ -384,6 +399,7 @@ class FactorBacktestService: @staticmethod def _calc_group_stats( group_nav: list[dict], start: date, end: date, + rebalance: str = "monthly", ) -> list[dict]: if not group_nav: return [] @@ -417,10 +433,12 @@ class FactorBacktestService: if values[j - 1] > 0: daily_rets.append(values[j] / values[j - 1] - 1) - # 夏普 + # 夏普 — 年化系数必须匹配 group_nav 的调仓频率 (每个净值点 = 一个调仓周期收益); + # 周/月频收益若乘 √252 会把 Sharpe 高估 √(252/期数) 倍 (月频 ≈4.6x, 周频 ≈2.2x)。 if daily_rets: arr = np.array(daily_rets) - sharpe = float(np.mean(arr) / np.std(arr)) * np.sqrt(252) if np.std(arr) > 0 else 0.0 + _ann = {"daily": 252, "weekly": 52, "monthly": 12}.get(rebalance, 252) + sharpe = float(np.mean(arr) / np.std(arr)) * np.sqrt(_ann) if np.std(arr) > 0 else 0.0 win_rate = float(np.mean(arr > 0)) else: sharpe = 0.0 diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index dd3e89b..5143842 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -281,11 +281,67 @@ def _apply_adj_factor(raw: pl.DataFrame, factors: pl.DataFrame) -> pl.DataFrame: # 技术指标计算 (从 OHLCV 计算) # ================================================================ -def compute_indicators(df: pl.DataFrame) -> pl.DataFrame: +# ── compute_indicators 的列依赖关系 (供 needed 裁剪时求闭包) ──────────── +# target -> 其计算所依赖的中间/指标列 (仅列出依赖非原始 OHLCV 的项) +_INDICATOR_DEPS: dict[str, set[str]] = { + "macd_dif": {"_ema12", "_ema26"}, + "boll_upper": {"ma20", "_boll_std"}, + "boll_lower": {"ma20", "_boll_std"}, + "macd_dea": {"macd_dif"}, + "macd_hist": {"macd_dif", "macd_dea"}, + "kdj_k": {"_kdj_ln", "_kdj_hn"}, + "kdj_d": {"kdj_k"}, + "kdj_j": {"kdj_k", "kdj_d"}, + "atr_14": {"_tr"}, + "vol_ratio_5d": {"_vol_ma5"}, + "annual_vol_20d": {"_daily_pct"}, + "rsi_6": {"_delta", "_gain", "_loss"}, + "rsi_14": {"_delta", "_gain", "_loss"}, + "rsi_24": {"_delta", "_gain", "_loss"}, +} + +# compute_indicators 可产出的全部指标/临时列 (needed=None 时即为此全集, 行为不变) +_ALL_INDICATOR_COLS: frozenset[str] = frozenset({ + "prev_close", "ma5", "ma10", "ma20", "ma30", "ma60", + "ema5", "ema10", "ema20", "ema30", "ema60", "_ema12", "_ema26", + "_boll_std", "_kdj_ln", "_kdj_hn", "_tr", "vol_ma5", "vol_ma10", + "_vol_ma5", "high_60d", "low_60d", + "macd_dif", "boll_upper", "boll_lower", "macd_dea", "macd_hist", + "kdj_k", "kdj_d", "kdj_j", + "atr_14", "vol_ratio_5d", + "momentum_5d", "momentum_10d", "momentum_20d", "momentum_30d", "momentum_60d", + "change_pct", "change_amount", "amplitude", "_daily_pct", "annual_vol_20d", + "rsi_6", "rsi_14", "rsi_24", +}) + + +def _resolve_needed(needed: set[str] | None) -> set[str]: + """把 needed 展开为闭包 (含所依赖的中间列)。needed=None → 全集。""" + if needed is None: + return set(_ALL_INDICATOR_COLS) + want = set(needed) + changed = True + while changed: + changed = False + for target in list(want): + deps = _INDICATOR_DEPS.get(target) + if deps and not deps <= want: + want |= deps + changed = True + return want + + +def compute_indicators(df: pl.DataFrame, needed: set[str] | None = None) -> pl.DataFrame: """从 OHLCV 数据计算全套技术指标。 输入必须包含: symbol, date, open, high, low, close, volume 返回添加了所有指标列的 DataFrame。 + + needed: + None (默认) — 计算全部指标, 行为与历史逐位一致 (所有 gate 为真, 表达式/顺序不变)。 + 列名集合 — 仅计算这些列及其依赖闭包, 跳过无关的 EMA/KDJ/RSI 等 pass; + 输出保留输入列 + 所需指标列。被保留列的数值与全量计算逐位一致 + (逐列 window/rolling 相互独立, 跳过其它列不影响保留列)。 """ if df.is_empty(): return df @@ -293,133 +349,179 @@ def compute_indicators(df: pl.DataFrame) -> pl.DataFrame: import time as _time _t0 = _time.perf_counter() + want = _resolve_needed(needed) + df = df.sort(["symbol", "date"]) # Pass 1: 均线 + EMA + MACD 基础 + BOLL 基础 + KDJ 基础 + ATR 基础 + 量价 + 极值 prev_close = pl.col("close").shift(1).over("symbol") - df = df.with_columns([ - # 前收盘价 - prev_close.alias("prev_close"), - # MA (最大 MA60) - pl.col("close").rolling_mean(5).over("symbol").alias("ma5"), - pl.col("close").rolling_mean(10).over("symbol").alias("ma10"), - pl.col("close").rolling_mean(20).over("symbol").alias("ma20"), - pl.col("close").rolling_mean(30).over("symbol").alias("ma30"), - pl.col("close").rolling_mean(60).over("symbol").alias("ma60"), - # EMA (不含 ema12/ema26, MACD 内部自算) - pl.col("close").ewm_mean(alpha=_ema_alpha(5), adjust=False).over("symbol").alias("ema5"), - pl.col("close").ewm_mean(alpha=_ema_alpha(10), adjust=False).over("symbol").alias("ema10"), - pl.col("close").ewm_mean(alpha=_ema_alpha(20), adjust=False).over("symbol").alias("ema20"), - pl.col("close").ewm_mean(alpha=_ema_alpha(30), adjust=False).over("symbol").alias("ema30"), - pl.col("close").ewm_mean(alpha=_ema_alpha(60), adjust=False).over("symbol").alias("ema60"), - # MACD base (内部计算, 不存 ema12/ema26) - pl.col("close").ewm_mean(alpha=_ema_alpha(12), adjust=False).over("symbol").alias("_ema12"), - pl.col("close").ewm_mean(alpha=_ema_alpha(26), adjust=False).over("symbol").alias("_ema26"), - # BOLL base - pl.col("close").rolling_std(20).over("symbol").alias("_boll_std"), - # KDJ base - pl.col("low").rolling_min(9).over("symbol").alias("_kdj_ln"), - pl.col("high").rolling_max(9).over("symbol").alias("_kdj_hn"), - # ATR base - pl.max_horizontal( + _p1: list[pl.Expr] = [] + if "prev_close" in want: + _p1.append(prev_close.alias("prev_close")) + if "ma5" in want: + _p1.append(pl.col("close").rolling_mean(5).over("symbol").alias("ma5")) + if "ma10" in want: + _p1.append(pl.col("close").rolling_mean(10).over("symbol").alias("ma10")) + if "ma20" in want: + _p1.append(pl.col("close").rolling_mean(20).over("symbol").alias("ma20")) + if "ma30" in want: + _p1.append(pl.col("close").rolling_mean(30).over("symbol").alias("ma30")) + if "ma60" in want: + _p1.append(pl.col("close").rolling_mean(60).over("symbol").alias("ma60")) + if "ema5" in want: + _p1.append(pl.col("close").ewm_mean(alpha=_ema_alpha(5), adjust=False).over("symbol").alias("ema5")) + if "ema10" in want: + _p1.append(pl.col("close").ewm_mean(alpha=_ema_alpha(10), adjust=False).over("symbol").alias("ema10")) + if "ema20" in want: + _p1.append(pl.col("close").ewm_mean(alpha=_ema_alpha(20), adjust=False).over("symbol").alias("ema20")) + if "ema30" in want: + _p1.append(pl.col("close").ewm_mean(alpha=_ema_alpha(30), adjust=False).over("symbol").alias("ema30")) + if "ema60" in want: + _p1.append(pl.col("close").ewm_mean(alpha=_ema_alpha(60), adjust=False).over("symbol").alias("ema60")) + if "_ema12" in want: + _p1.append(pl.col("close").ewm_mean(alpha=_ema_alpha(12), adjust=False).over("symbol").alias("_ema12")) + if "_ema26" in want: + _p1.append(pl.col("close").ewm_mean(alpha=_ema_alpha(26), adjust=False).over("symbol").alias("_ema26")) + if "_boll_std" in want: + _p1.append(pl.col("close").rolling_std(20).over("symbol").alias("_boll_std")) + if "_kdj_ln" in want: + _p1.append(pl.col("low").rolling_min(9).over("symbol").alias("_kdj_ln")) + if "_kdj_hn" in want: + _p1.append(pl.col("high").rolling_max(9).over("symbol").alias("_kdj_hn")) + if "_tr" in want: + _p1.append(pl.max_horizontal( pl.col("high") - pl.col("low"), (pl.col("high") - prev_close).abs(), (pl.col("low") - prev_close).abs(), - ).alias("_tr"), - # 量价 base - pl.col("volume").rolling_mean(5).over("symbol").alias("vol_ma5"), - pl.col("volume").rolling_mean(10).over("symbol").alias("vol_ma10"), - pl.col("volume").rolling_mean(5).over("symbol").alias("_vol_ma5"), - # 极值 - pl.col("close").rolling_max(60).over("symbol").alias("high_60d"), - pl.col("close").rolling_min(60).over("symbol").alias("low_60d"), - ]) + ).alias("_tr")) + if "vol_ma5" in want: + _p1.append(pl.col("volume").rolling_mean(5).over("symbol").alias("vol_ma5")) + if "vol_ma10" in want: + _p1.append(pl.col("volume").rolling_mean(10).over("symbol").alias("vol_ma10")) + if "_vol_ma5" in want: + _p1.append(pl.col("volume").rolling_mean(5).over("symbol").alias("_vol_ma5")) + if "high_60d" in want: + _p1.append(pl.col("close").rolling_max(60).over("symbol").alias("high_60d")) + if "low_60d" in want: + _p1.append(pl.col("close").rolling_min(60).over("symbol").alias("low_60d")) + if _p1: + df = df.with_columns(_p1) # Pass 2: MACD + BOLL (基于 Pass 1 基础列) - df = df.with_columns([ - (pl.col("_ema12") - pl.col("_ema26")).alias("macd_dif"), - (pl.col("ma20") + 2 * pl.col("_boll_std")).alias("boll_upper"), - (pl.col("ma20") - 2 * pl.col("_boll_std")).alias("boll_lower"), - ]).with_columns( - pl.col("macd_dif").ewm_mean(alpha=_ema_alpha(9), adjust=False).over("symbol").alias("macd_dea"), - ).with_columns( - ((pl.col("macd_dif") - pl.col("macd_dea")) * 2).alias("macd_hist"), - ) + _p2: list[pl.Expr] = [] + if "macd_dif" in want: + _p2.append((pl.col("_ema12") - pl.col("_ema26")).alias("macd_dif")) + if "boll_upper" in want: + _p2.append((pl.col("ma20") + 2 * pl.col("_boll_std")).alias("boll_upper")) + if "boll_lower" in want: + _p2.append((pl.col("ma20") - 2 * pl.col("_boll_std")).alias("boll_lower")) + if _p2: + df = df.with_columns(_p2) + if "macd_dea" in want: + df = df.with_columns( + pl.col("macd_dif").ewm_mean(alpha=_ema_alpha(9), adjust=False).over("symbol").alias("macd_dea"), + ) + if "macd_hist" in want: + df = df.with_columns( + ((pl.col("macd_dif") - pl.col("macd_dea")) * 2).alias("macd_hist"), + ) # Pass 3: KDJ - _kdj_rsv = ( - 100 * (pl.col("close") - pl.col("_kdj_ln")) - / (pl.col("_kdj_hn") - pl.col("_kdj_ln")).fill_null(1e-12) - ) - df = df.with_columns([ - _kdj_rsv.ewm_mean(alpha=1.0 / 3, adjust=False).over("symbol").alias("kdj_k"), - ]).with_columns([ - pl.col("kdj_k").ewm_mean(alpha=1.0 / 3, adjust=False).over("symbol").alias("kdj_d"), - ]).with_columns([ - (3 * pl.col("kdj_k") - 2 * pl.col("kdj_d")).alias("kdj_j"), - ]) + if "kdj_k" in want: + _kdj_rsv = ( + 100 * (pl.col("close") - pl.col("_kdj_ln")) + / (pl.col("_kdj_hn") - pl.col("_kdj_ln")).fill_null(1e-12) + ) + df = df.with_columns([ + _kdj_rsv.ewm_mean(alpha=1.0 / 3, adjust=False).over("symbol").alias("kdj_k"), + ]) + if "kdj_d" in want: + df = df.with_columns([ + pl.col("kdj_k").ewm_mean(alpha=1.0 / 3, adjust=False).over("symbol").alias("kdj_d"), + ]) + if "kdj_j" in want: + df = df.with_columns([ + (3 * pl.col("kdj_k") - 2 * pl.col("kdj_d")).alias("kdj_j"), + ]) # Pass 4: ATR + 量比 + 动量 + 波动 + 涨跌幅 + 涨跌额 + 振幅 - df = df.with_columns( - pl.col("_tr").ewm_mean(alpha=1.0 / 14, adjust=False).over("symbol").alias("atr_14"), - ).with_columns( - (pl.col("volume") / pl.col("_vol_ma5")).alias("vol_ratio_5d"), - ).with_columns([ - # 动量: 5d/10d/20d/30d/60d - (pl.col("close") / pl.col("close").shift(5).over("symbol") - 1).alias("momentum_5d"), - (pl.col("close") / pl.col("close").shift(10).over("symbol") - 1).alias("momentum_10d"), - (pl.col("close") / pl.col("close").shift(20).over("symbol") - 1).alias("momentum_20d"), - (pl.col("close") / pl.col("close").shift(30).over("symbol") - 1).alias("momentum_30d"), - (pl.col("close") / pl.col("close").shift(60).over("symbol") - 1).alias("momentum_60d"), - # 日涨跌幅 - (pl.col("close") / pl.col("close").shift(1).over("symbol") - 1).alias("change_pct"), - ]).with_columns( - # 涨跌额 - (pl.col("close") - pl.col("close").shift(1).over("symbol")).alias("change_amount"), - ).with_columns( - # 振幅 = (high - low) / prev_close - pl.when(pl.col("close").shift(1).over("symbol") > 0) - .then((pl.col("high") - pl.col("low")) / pl.col("close").shift(1).over("symbol")) - .otherwise(None) - .alias("amplitude"), - ).with_columns( - # 日涨跌幅 (用于波动率) - pl.col("close").pct_change().over("symbol").alias("_daily_pct"), - ).with_columns( - # 年化波动率 - (pl.col("_daily_pct").rolling_std(20).over("symbol") * (252 ** 0.5)) - .alias("annual_vol_20d"), - ) + if "atr_14" in want: + df = df.with_columns( + pl.col("_tr").ewm_mean(alpha=1.0 / 14, adjust=False).over("symbol").alias("atr_14"), + ) + if "vol_ratio_5d" in want: + df = df.with_columns( + (pl.col("volume") / pl.col("_vol_ma5")).alias("vol_ratio_5d"), + ) + _p4mom: list[pl.Expr] = [] + if "momentum_5d" in want: + _p4mom.append((pl.col("close") / pl.col("close").shift(5).over("symbol") - 1).alias("momentum_5d")) + if "momentum_10d" in want: + _p4mom.append((pl.col("close") / pl.col("close").shift(10).over("symbol") - 1).alias("momentum_10d")) + if "momentum_20d" in want: + _p4mom.append((pl.col("close") / pl.col("close").shift(20).over("symbol") - 1).alias("momentum_20d")) + if "momentum_30d" in want: + _p4mom.append((pl.col("close") / pl.col("close").shift(30).over("symbol") - 1).alias("momentum_30d")) + if "momentum_60d" in want: + _p4mom.append((pl.col("close") / pl.col("close").shift(60).over("symbol") - 1).alias("momentum_60d")) + if "change_pct" in want: + _p4mom.append((pl.col("close") / pl.col("close").shift(1).over("symbol") - 1).alias("change_pct")) + if _p4mom: + df = df.with_columns(_p4mom) + if "change_amount" in want: + df = df.with_columns( + (pl.col("close") - pl.col("close").shift(1).over("symbol")).alias("change_amount"), + ) + if "amplitude" in want: + df = df.with_columns( + pl.when(pl.col("close").shift(1).over("symbol") > 0) + .then((pl.col("high") - pl.col("low")) / pl.col("close").shift(1).over("symbol")) + .otherwise(None) + .alias("amplitude"), + ) + if "_daily_pct" in want: + df = df.with_columns( + pl.col("close").pct_change().over("symbol").alias("_daily_pct"), + ) + if "annual_vol_20d" in want: + df = df.with_columns( + (pl.col("_daily_pct").rolling_std(20).over("symbol") * (252 ** 0.5)) + .alias("annual_vol_20d"), + ) # Pass 5: RSI - df = df.with_columns( - pl.col("close").diff().over("symbol").alias("_delta"), - ).with_columns([ - pl.when(pl.col("_delta") > 0).then(pl.col("_delta")).otherwise(0.0).alias("_gain"), - pl.when(pl.col("_delta") < 0).then(-pl.col("_delta")).otherwise(0.0).alias("_loss"), - ]) - for n in (6, 14, 24): - a = 1.0 / n - df = df.with_columns([ - pl.col("_gain").ewm_mean(alpha=a, adjust=False).over("symbol").alias(f"_rsi_avg_gain_{n}"), - pl.col("_loss").ewm_mean(alpha=a, adjust=False).over("symbol").alias(f"_rsi_avg_loss_{n}"), - ]).with_columns( - (100 - 100 / (1 + pl.col(f"_rsi_avg_gain_{n}") / - pl.when(pl.col(f"_rsi_avg_loss_{n}") == 0) - .then(1e-12) - .otherwise(pl.col(f"_rsi_avg_loss_{n}")) - )).alias(f"rsi_{n}"), - ) + if want & {"rsi_6", "rsi_14", "rsi_24"}: + df = df.with_columns( + pl.col("close").diff().over("symbol").alias("_delta"), + ).with_columns([ + pl.when(pl.col("_delta") > 0).then(pl.col("_delta")).otherwise(0.0).alias("_gain"), + pl.when(pl.col("_delta") < 0).then(-pl.col("_delta")).otherwise(0.0).alias("_loss"), + ]) + for n in (6, 14, 24): + if f"rsi_{n}" not in want: + continue + a = 1.0 / n + df = df.with_columns([ + pl.col("_gain").ewm_mean(alpha=a, adjust=False).over("symbol").alias(f"_rsi_avg_gain_{n}"), + pl.col("_loss").ewm_mean(alpha=a, adjust=False).over("symbol").alias(f"_rsi_avg_loss_{n}"), + ]).with_columns( + (100 - 100 / (1 + pl.col(f"_rsi_avg_gain_{n}") / + pl.when(pl.col(f"_rsi_avg_loss_{n}") == 0) + .then(1e-12) + .otherwise(pl.col(f"_rsi_avg_loss_{n}")) + )).alias(f"rsi_{n}"), + ) # Pass 6: 换手率 (需要 float_shares, 后续在 compute_all 中 JOIN instruments 后补充) - # 清理临时列 - df = df.drop(["_boll_std", "_tr", "_ema12", "_ema26", + # 清理临时列 (只丢弃实际存在的临时列) + _temp_cols = ["_boll_std", "_tr", "_ema12", "_ema26", "_kdj_ln", "_kdj_hn", "_vol_ma5", "_daily_pct", "_delta", "_gain", "_loss", "_rsi_avg_gain_6", "_rsi_avg_loss_6", "_rsi_avg_gain_14", "_rsi_avg_loss_14", - "_rsi_avg_gain_24", "_rsi_avg_loss_24"]) + "_rsi_avg_gain_24", "_rsi_avg_loss_24"] + df = df.drop([c for c in _temp_cols if c in df.columns]) _elapsed = (_time.perf_counter() - _t0) * 1000 import logging as _logging @@ -541,10 +643,11 @@ def compute_limit_signals(df: pl.DataFrame, instruments: pl.DataFrame) -> pl.Dat .alias("_board_pct") ) - # ST → 5%（覆盖板块默认值） + # ST → 5%, 但仅限主板风险警示股; 创业板/科创板/北交所 ST 保留各自板块限幅 + # (注册制改革后 创业板 300/301、科创板 688/689 的 ST 仍执行 20%, 北交所 30%)。 if "_is_st" in df.columns: df = df.with_columns( - pl.when(pl.col("_is_st").fill_null(False)) + pl.when(pl.col("_is_st").fill_null(False) & ~(is_chinext | is_star | is_bj)) .then(0.05) .otherwise(pl.col("_board_pct")) .alias("_limit_pct") @@ -1447,7 +1550,10 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> .otherwise(0.10) ) if "_is_st" in df.columns: - limit_pct = pl.when(pl.col("_is_st").fill_null(False)).then(0.05).otherwise(limit_pct) + # ST 5% 仅主板生效; 创业板/科创板/北交所 ST 保留板块限幅 (同 compute_limit_signals) + limit_pct = pl.when( + pl.col("_is_st").fill_null(False) & ~(is_chinext | is_star | is_bj) + ).then(0.05).otherwise(limit_pct) limit_pct = limit_pct.alias("_limit_pct") limit_up_price = _limit_price(prev_raw, limit_pct, up=True) diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index 0ff9b0f..af992d0 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -31,6 +31,19 @@ logger = logging.getLogger(__name__) ProgressCb = Callable[..., None] +class PipelineStageError(RuntimeError): + """管道有阶段软失败(数据可能陈旧)时抛出, 让上层 job_store 把任务标记为 failed。 + + 这些阶段单独 try/except 吞掉异常以不中断整条管道, 但一旦失败即代表对应数据陈旧。 + 抛出前进度协议已走完(done/100), 故前端进度条正常收尾, 仅终态如实反映为 failed —— + 不再"部分失败却报成功"。 + """ + + def __init__(self, errors: list[str]) -> None: + self.errors = errors + super().__init__("盘后管道部分阶段失败: " + "; ".join(errors)) + + def _noop(stage: str, pct: int, msg: str, **kwargs) -> None: # noqa: ARG001 pass @@ -97,6 +110,9 @@ def run_now( """ emit = on_progress or _noop skipped: list[str] = [] + # 阶段软失败累积: 下列阶段 try/except 吞异常以不中断管道, 但失败即代表数据可能陈旧。 + # 管道末尾若非空则抛 PipelineStageError, 让任务终态如实标记为 failed(而非误报成功)。 + stage_errors: list[str] = [] # Step 0: 先同步个股维表, 再解析标的池 — 确保标的池基于最新 instruments emit("sync_instruments", 2, "同步个股维表…") @@ -194,6 +210,7 @@ def run_now( len(lagging_symbols), lagging_symbols[:10]) except Exception as e: # noqa: BLE001 logger.warning("laggard detection failed: %s", e) + stage_errors.append(f"laggard detection: {e}") # Step 1.5: 同步除权因子 — 范围与日K拉取方式对齐 # 日K范围拉取(补缺口/首次) → 除权用日K范围 [daily_range_start, now] @@ -395,6 +412,7 @@ def run_now( emit("sync_index", 88, f"ETF 除权因子完成,{etf_adj_symbols} 只") except Exception as e: # noqa: BLE001 logger.warning("ETF adj_factor skipped: %s", e) + stage_errors.append(f"ETF adj_factor: {e}") etf_dir = repo.store.data_dir / "kline_etf_enriched" etf_dates = sorted( d.name[5:] for d in etf_dir.glob("date=*") @@ -427,6 +445,7 @@ def run_now( except Exception as e: # noqa: BLE001 logger.warning("sync_index/etf failed: %s", e) emit("sync_index", 89, f"指数/ETF同步失败:{e}") + stage_errors.append(f"index/etf sync: {e}") else: skipped.append("sync_index") @@ -466,7 +485,7 @@ def run_now( emit("done", 100, "完成") _invalidate(None) # 兜底:全清 - return { + result = { "universe_size": len(universe), "daily_days": new_daily_days, "adj_factor_symbols": len(affected_symbols), @@ -479,36 +498,20 @@ def run_now( "minute_rows": written_minute, "lagging_symbols": len(lagging_symbols), "skipped_stages": skipped, + "stage_errors": stage_errors, } + # 有阶段软失败: 进度协议已走完(done/100, 前端进度条正常收尾), 但数据可能陈旧, + # 抛出让上层 job_store 把终态标记为 failed —— 不再"部分失败却报成功"。 + if stage_errors: + raise PipelineStageError(stage_errors) + + return result + def _refresh_views(repo: KlineRepository) -> None: - """刷新所有 DuckDB 视图。""" - d = repo.store.data_dir.as_posix() - views = { - "kline_daily": f"{d}/kline_daily/**/*.parquet", - "kline_enriched": f"{d}/kline_daily_enriched/**/*.parquet", - "kline_index_daily": f"{d}/kline_index_daily/**/*.parquet", - "kline_index_enriched": f"{d}/kline_index_enriched/**/*.parquet", - "kline_etf_daily": f"{d}/kline_etf_daily/**/*.parquet", - "kline_etf_enriched": f"{d}/kline_etf_enriched/**/*.parquet", - "kline_etf_minute": f"{d}/kline_etf_minute/**/*.parquet", - "kline_minute": f"{d}/kline_minute/**/*.parquet", - "adj_factor": f"{d}/adj_factor/**/*.parquet", - "adj_factor_etf": f"{d}/adj_factor_etf/**/*.parquet", - "instruments": f"{d}/instruments/**/*.parquet", - "instruments_index": f"{d}/instruments_index/**/*.parquet", - "instruments_etf": f"{d}/instruments_etf/**/*.parquet", - } - for name, path in views.items(): - try: - repo.db.execute( - f"CREATE OR REPLACE VIEW {name} AS " - f"SELECT * FROM read_parquet('{path}', union_by_name=true)" - ) - except Exception as e: # noqa: BLE001 - logger.warning("refresh view %s failed: %s", name, e) - repo.store._register_unified_views() + """刷新所有 DuckDB 视图 —— 委托给 repository 的唯一权威实现 rebuild_views()。""" + repo.rebuild_views() def _refresh_single_view(repo: KlineRepository, name: str) -> None: @@ -832,8 +835,13 @@ def start_scheduler(repo: KlineRepository, capset: CapabilitySet) -> AsyncIOSche # 旧 capset —— 否则 Key 中途过期/续费后, 调度管道仍按旧档位打端点。 app_state = _get_app_state() capset_live = getattr(app_state, "capabilities", None) or capset - result = run_now(repo, capset_live, on_progress=on_progress) - repo.refresh_cache() + try: + result = run_now(repo, capset_live, on_progress=on_progress) + finally: + # 即便有阶段软失败(run_now 末尾抛 PipelineStageError), 已落盘的日K/enriched + # 仍需刷进内存缓存, 否则 live_agg 基准列停留在旧交易日。放 finally 保证部分 + # 成功也生效; 随后异常继续上抛, 由 _run_tracked 标记任务 failed。 + repo.refresh_cache() return result scheduler.add_job( diff --git a/backend/app/services/ai_reports.py b/backend/app/services/ai_reports.py index caf4115..13c74c6 100644 --- a/backend/app/services/ai_reports.py +++ b/backend/app/services/ai_reports.py @@ -3,6 +3,9 @@ 存储位置: data/user_data/ai_reports.json (数组,按 created_at 降序) 保留最近 MAX_REPORTS 条;超出自动裁剪最旧的。 +存储机制统一委托给 JsonReportStore(原子写 + 实例锁), 本模块只固化 +财务分析报告 的文件名 / 上限 / id 前缀, 对外保持原有函数签名不变。 + 每条报告结构: { "id": "rpt_xxx", # 唯一 id @@ -17,46 +20,16 @@ """ from __future__ import annotations -import json -import logging -import time -from pathlib import Path - -logger = logging.getLogger(__name__) +from app.services.json_report_store import JsonReportStore MAX_REPORTS = 20 - -def _path() -> Path: - from app.config import settings - p = settings.data_dir / "user_data" / "ai_reports.json" - p.parent.mkdir(parents=True, exist_ok=True) - return p +_store = JsonReportStore("ai_reports.json", MAX_REPORTS, id_prefix="rpt") def list_reports() -> list[dict]: """返回全部报告(按 created_at 降序)。""" - p = _path() - if not p.exists(): - return [] - try: - data = json.loads(p.read_text(encoding="utf-8")) - if isinstance(data, list): - return sorted(data, key=lambda r: r.get("created_at", ""), reverse=True) - except Exception as e: # noqa: BLE001 - logger.warning("ai_reports.json malformed: %s", e) - return [] - - -def _save_all(reports: list[dict]) -> None: - """全量写入(裁剪到 MAX_REPORTS)。""" - # 保持降序 - reports.sort(key=lambda r: r.get("created_at", ""), reverse=True) - if len(reports) > MAX_REPORTS: - reports = reports[:MAX_REPORTS] - _path().write_text( - json.dumps(reports, indent=2, ensure_ascii=False), encoding="utf-8", - ) + return _store.list_reports() def save_report(report: dict) -> dict: @@ -64,38 +37,14 @@ def save_report(report: dict) -> dict: 自动补全 id 与 created_at(若缺),并裁剪到上限。 """ - reports = list_reports() - if not report.get("id"): - report["id"] = f"rpt_{int(time.time() * 1000)}_{report.get('symbol', 'x')}" - if not report.get("created_at"): - report["created_at"] = _now_iso() - reports.append(report) - _save_all(reports) - logger.info("AI report saved: %s (%s), total %d", report.get("symbol"), report.get("id"), len(reports)) - return report + return _store.save_report(report) def delete_report(report_id: str) -> bool: """删除指定报告。返回是否删除成功。""" - reports = list_reports() - before = len(reports) - reports = [r for r in reports if r.get("id") != report_id] - if len(reports) < before: - _save_all(reports) - return True - return False + return _store.delete_report(report_id) def clear_reports() -> int: """清空全部报告。返回删除数量。""" - reports = list_reports() - n = len(reports) - if n > 0: - _save_all([]) - return n - - -def _now_iso() -> str: - """当前本地时间 ISO 字符串(带秒精度,前端 toLocaleString 友好)。""" - from datetime import datetime - return datetime.now().isoformat(timespec="seconds") + return _store.clear_reports() diff --git a/backend/app/services/auth.py b/backend/app/services/auth.py index 77c7b35..dbfcbad 100644 --- a/backend/app/services/auth.py +++ b/backend/app/services/auth.py @@ -36,6 +36,10 @@ _lock = threading.Lock() # 内存中的有效会话: { token: expire_ts }。进程重启后从磁盘恢复。 _sessions: dict[str, float] = {} +# 「是否已设密码」缓存: 每个 /api/ 请求都要判定, auth_middleware 原先每次 read_text +# 磁盘 (阻塞事件循环)。此处懒加载缓存, set_password 后失效重算 (仍返回最新真值)。 +_configured_cache: bool | None = None + def _path() -> Path: from app.config import settings @@ -87,13 +91,21 @@ def _verify_password(password: str, salt_hex: str, hash_hex: str) -> bool: # ================================================================ def is_configured() -> bool: - """是否已设置访问密码。""" - d = _load() - return bool(d.get("password_hash")) + """是否已设置访问密码 (带缓存)。 + + 热路径 (auth_middleware 每请求调用): 命中缓存则不碰磁盘, 不阻塞事件循环。 + 首次或 set_password 失效后, 懒加载重读一次 auth.json, 保证返回最新真值。 + """ + global _configured_cache + if _configured_cache is None: + d = _load() + _configured_cache = bool(d.get("password_hash")) + return _configured_cache def set_password(password: str) -> None: """设置/修改访问密码。清空所有现有会话(强制重新登录)。""" + global _configured_cache if len(password) < 6: raise ValueError("密码至少 6 位") salt_hex, hash_hex = _hash_password(password) @@ -105,6 +117,7 @@ def set_password(password: str) -> None: "updated_at": int(time.time()), "sessions": {}, # 清空持久化会话 }) + _configured_cache = None # 失效缓存, 下次 is_configured 重读最新真值 logger.info("access password set") diff --git a/backend/app/services/depth_service.py b/backend/app/services/depth_service.py index 51e503f..89ff9b2 100644 --- a/backend/app/services/depth_service.py +++ b/backend/app/services/depth_service.py @@ -22,6 +22,7 @@ from __future__ import annotations import logging import math +import os import threading import time from datetime import date, time as dt_time @@ -57,6 +58,10 @@ class DepthService: def __init__(self) -> None: self._lock = threading.Lock() + # 拉取+定版串行锁 (镜像 quote_service._fetch_lock): _fetch_and_seal 可能同时被 + # 请求线程 (run_once persist=True)、轮询线程、盘后 finalize 触发, 都写同一 parquet, + # 无锁会交叉写坏文件。_lock 只护内存缓存, 此锁护整段 fetch+seal。 + self._fetch_lock = threading.Lock() self._running = False self._thread: threading.Thread | None = None self._repo = None # 延迟注入(KlineRepository) @@ -138,14 +143,16 @@ class DepthService: def start_polling(self) -> None: """启动盘中轮询线程(连板梯队监控开启 + 有能力 + 交易时段)。""" - if self._running: - return if not self._has_capability(): return from app.services import preferences if not preferences.get_limit_ladder_monitor_enabled(): return - self._running = True + # check-then-act 加锁: 两个线程同时 start_polling 不会各起一个轮询线程 + with self._lock: + if self._running: + return + self._running = True self._thread = threading.Thread(target=self._poll_loop, daemon=True) self._thread.start() logger.info("depth sealed 盘中轮询已启动") @@ -191,7 +198,14 @@ class DepthService: persist=True: 盘后定版, 写 depth5 parquet persist=False: 盘中轮询, 只更新内存缓存 + + 全程持 _fetch_lock: 请求线程 (run_once)、轮询线程、finalize 不会交叉写 parquet。 """ + with self._fetch_lock: + self._fetch_and_seal_locked(persist) + + def _fetch_and_seal_locked(self, persist: bool = False) -> None: + """_fetch_and_seal 的实际逻辑, 须在持有 _fetch_lock 时调用。""" if not self._repo: return @@ -331,7 +345,10 @@ class DepthService: ds = today.isoformat() out = self._repo.store.data_dir / "depth5" / f"date={ds}" / "part.parquet" out.parent.mkdir(parents=True, exist_ok=True) - df.write_parquet(out) + # 原子写: 先写临时文件再 os.replace, 避免读侧 (get_sealed_map) 读到半写 parquet + tmp = out.with_name(out.name + ".tmp") + df.write_parquet(tmp) + os.replace(tmp, out) self._persisted_date = today logger.info("depth sealed 落盘: %d 行 → %s", df.height, out) diff --git a/backend/app/services/ext_pull.py b/backend/app/services/ext_pull.py index cfd2b99..f9528b7 100644 --- a/backend/app/services/ext_pull.py +++ b/backend/app/services/ext_pull.py @@ -14,7 +14,6 @@ import httpx from app.services.ext_data import ( ExtConfig, ExtConfigStore, - PullConfig, rows_to_parquet, ) @@ -289,57 +288,6 @@ class PullScheduler: except asyncio.CancelledError: pass - async def _run_loop(self, config: ExtConfig) -> None: - """单个配置的定时拉取循环。 - - 策略: 启用后立即执行一次, 之后按 interval 循环。 - 每次循环重读最新配置 (fresh), interval 取自 fresh.pull.schedule_minutes, - 这样用户中途修改间隔也能立即生效 (无需重启)。 - """ - try: - while self._running: - # 每轮重读最新配置 — 用户可能修改了 url / interval / enabled - store = ExtConfigStore(self._data_dir) - fresh = store.get(config.id) - if not fresh or not fresh.pull or not fresh.pull.enabled: - break - pull = fresh.pull - - # 先执行一次 (启用即拉取, 让用户立刻看到生效) - try: - n, d = await fetch_and_ingest(fresh, self._data_dir) - fresh.pull.last_run = datetime.now(timezone.utc).isoformat() - fresh.pull.last_status = "success" - fresh.pull.last_message = f"{n} rows @ {d}" - fresh.pull.last_rows = n - store.upsert(fresh) - logger.info("PullScheduler: %s success, %d rows", config.id, n) - except Exception as e: - fresh2 = store.get(config.id) - if fresh2 and fresh2.pull: - fresh2.pull.last_run = datetime.now(timezone.utc).isoformat() - fresh2.pull.last_status = "error" - fresh2.pull.last_message = str(e)[:200] - store.upsert(fresh2) - logger.warning("PullScheduler: %s error: %s", config.id, e) - - # 间隔取自最新配置 (每次重新读取, 修复改间隔不生效) - interval = max(pull.schedule_minutes * 60, 60) # 至少 60s - # 预告下次运行时间, 供前端展示 - next_dt = datetime.now(timezone.utc).timestamp() + interval - latest = store.get(config.id) - if latest and latest.pull: - latest.pull.next_run = datetime.fromtimestamp( - next_dt, tz=timezone.utc - ).isoformat() - store.upsert(latest) - - await asyncio.sleep(interval) - if not self._running: - break - except asyncio.CancelledError: - pass - # 全局单例 pull_scheduler = PullScheduler() diff --git a/backend/app/services/json_report_store.py b/backend/app/services/json_report_store.py new file mode 100644 index 0000000..95570db --- /dev/null +++ b/backend/app/services/json_report_store.py @@ -0,0 +1,126 @@ +"""AI 报告类 JSON 存储的共享底座。 + +财务分析 / 个股分析 / 大盘复盘三类报告的持久化机制此前是近乎逐字的三份拷贝 +(_path / list_reports / _save_all / save_report / delete_report / _now_iso 完全一致, +只差文件名、上限、id 前缀)。这里抽出唯一实现, 三个模块各自实例化并委托, 对外仍保持 +原有函数签名不变。 + +差异通过构造参数固化: filename(存储文件名) / max_reports(保留上限) / id_prefix(id 前缀) / +id_with_symbol(id 是否带 symbol 后缀 —— 大盘复盘无 symbol)。 + +存储文件: data/user_data/{filename} (数组, 按 created_at 降序), 保留最近 max_reports 条, +超出自动裁剪最旧的。写入走临时文件 + os.replace 原子替换, 避免进程中断留下半截 JSON。 +读写同时可能来自请求线程与调度线程, 故加实例锁串行化写路径。 +""" +from __future__ import annotations + +import json +import logging +import os +import threading +import time +from pathlib import Path + +logger = logging.getLogger(__name__) + + +class JsonReportStore: + """一类 AI 报告的 JSON 存储 (原子写 + 实例锁)。""" + + def __init__( + self, + filename: str, + max_reports: int, + id_prefix: str, + id_with_symbol: bool = True, + ) -> None: + self.filename = filename + self.max_reports = max_reports + self.id_prefix = id_prefix + self.id_with_symbol = id_with_symbol + # 请求线程 + 调度线程可能并发写, 用实例锁串行化读-改-写 + self._lock = threading.Lock() + + def _path(self) -> Path: + from app.config import settings + p = settings.data_dir / "user_data" / self.filename + p.parent.mkdir(parents=True, exist_ok=True) + return p + + def list_reports(self) -> list[dict]: + """返回全部报告(按 created_at 降序)。""" + p = self._path() + if not p.exists(): + return [] + try: + data = json.loads(p.read_text(encoding="utf-8")) + if isinstance(data, list): + return sorted(data, key=lambda r: r.get("created_at", ""), reverse=True) + except Exception as e: # noqa: BLE001 + logger.warning("%s malformed: %s", self.filename, e) + return [] + + def _save_all(self, reports: list[dict]) -> None: + """全量写入(裁剪到 max_reports, 原子替换)。""" + # 保持降序 + reports.sort(key=lambda r: r.get("created_at", ""), reverse=True) + if len(reports) > self.max_reports: + reports = reports[:self.max_reports] + self._atomic_write(reports) + + def _atomic_write(self, reports: list[dict]) -> None: + """先写临时文件再 os.replace 原子替换, 避免进程中断留下损坏的 JSON。""" + p = self._path() + text = json.dumps(reports, indent=2, ensure_ascii=False) + tmp = p.with_suffix(p.suffix + ".tmp") + tmp.write_text(text, encoding="utf-8") + os.replace(tmp, p) + + def _make_id(self, report: dict) -> str: + base = f"{self.id_prefix}_{int(time.time() * 1000)}" + if self.id_with_symbol: + return f"{base}_{report.get('symbol', 'x')}" + return base + + def save_report(self, report: dict) -> dict: + """新增一条报告并持久化。返回保存后的报告(含 id / created_at)。 + + 自动补全 id 与 created_at(若缺),并裁剪到上限。 + """ + with self._lock: + reports = self.list_reports() + if not report.get("id"): + report["id"] = self._make_id(report) + if not report.get("created_at"): + report["created_at"] = self._now_iso() + reports.append(report) + self._save_all(reports) + total = min(len(reports), self.max_reports) + logger.info("report saved: %s → %s, total %d", self.filename, report.get("id"), total) + return report + + def delete_report(self, report_id: str) -> bool: + """删除指定报告。返回是否删除成功。""" + with self._lock: + reports = self.list_reports() + before = len(reports) + reports = [r for r in reports if r.get("id") != report_id] + if len(reports) < before: + self._save_all(reports) + return True + return False + + def clear_reports(self) -> int: + """清空全部报告。返回删除数量。""" + with self._lock: + reports = self.list_reports() + n = len(reports) + if n > 0: + self._save_all([]) + return n + + @staticmethod + def _now_iso() -> str: + """当前本地时间 ISO 字符串(带秒精度,前端 toLocaleString 友好)。""" + from datetime import datetime + return datetime.now().isoformat(timespec="seconds") diff --git a/backend/app/services/market_recap_reports.py b/backend/app/services/market_recap_reports.py index 06be304..e4902f6 100644 --- a/backend/app/services/market_recap_reports.py +++ b/backend/app/services/market_recap_reports.py @@ -1,8 +1,9 @@ """AI 大盘复盘报告持久化存储。 与 stock_reports.py(个股分析报告)/ ai_reports.py(财务分析报告)完全独立 —— -单独的文件、字段、上限,互不影响。刻意不复用,避免引入 kind 判别字段与分支 -(解耦 > 抽象)。 +单独的文件、字段、上限,互不影响。存储机制委托给共享的 JsonReportStore +(原子写 + 实例锁), 本模块只固化 大盘复盘报告 的文件名 / 上限 / id 前缀 +(复盘无 symbol, id 不带 symbol 后缀), 对外保持原有函数签名不变。 存储位置: data/user_data/ai_market_recaps.json (数组,按 created_at 降序) 保留最近 MAX_REPORTS 条;超出自动裁剪最旧的。 @@ -21,72 +22,25 @@ """ from __future__ import annotations -import json -import logging -import time -from pathlib import Path - -logger = logging.getLogger(__name__) +from app.services.json_report_store import JsonReportStore MAX_REPORTS = 20 - -def _path() -> Path: - from app.config import settings - p = settings.data_dir / "user_data" / "ai_market_recaps.json" - p.parent.mkdir(parents=True, exist_ok=True) - return p +_store = JsonReportStore( + "ai_market_recaps.json", MAX_REPORTS, id_prefix="mkr", id_with_symbol=False, +) def list_reports() -> list[dict]: """返回全部报告(按 created_at 降序)。""" - p = _path() - if not p.exists(): - return [] - try: - data = json.loads(p.read_text(encoding="utf-8")) - if isinstance(data, list): - return sorted(data, key=lambda r: r.get("created_at", ""), reverse=True) - except Exception as e: # noqa: BLE001 - logger.warning("ai_market_recaps.json malformed: %s", e) - return [] - - -def _save_all(reports: list[dict]) -> None: - """全量写入(裁剪到 MAX_REPORTS)。""" - reports.sort(key=lambda r: r.get("created_at", ""), reverse=True) - if len(reports) > MAX_REPORTS: - reports = reports[:MAX_REPORTS] - _path().write_text( - json.dumps(reports, indent=2, ensure_ascii=False), encoding="utf-8", - ) + return _store.list_reports() def save_report(report: dict) -> dict: """新增一条报告并持久化。返回保存后的报告(含 id / created_at)。""" - reports = list_reports() - if not report.get("id"): - report["id"] = f"mkr_{int(time.time() * 1000)}" - if not report.get("created_at"): - report["created_at"] = _now_iso() - reports.append(report) - _save_all(reports) - logger.info("Market recap saved: %s (as_of=%s), total %d", - report.get("id"), report.get("as_of"), len(reports)) - return report + return _store.save_report(report) def delete_report(report_id: str) -> bool: """删除指定报告。返回是否删除成功。""" - reports = list_reports() - before = len(reports) - reports = [r for r in reports if r.get("id") != report_id] - if len(reports) < before: - _save_all(reports) - return True - return False - - -def _now_iso() -> str: - from datetime import datetime - return datetime.now().isoformat(timespec="seconds") + return _store.delete_report(report_id) diff --git a/backend/app/services/screener.py b/backend/app/services/screener.py index 763bbf3..fac6731 100644 --- a/backend/app/services/screener.py +++ b/backend/app/services/screener.py @@ -267,6 +267,44 @@ class ScreenerService: df_full = self._compute_enriched_full(df, target_date) return df_full + def load_prior_consecutive(self, as_of: date, consec_col: str) -> pl.DataFrame: + """窄读: 仅取前一交易日的 [symbol, consec_col] 两列 (谓词下推到单日 parquet)。 + + consecutive_limit_ups / consecutive_limit_downs 是 enriched 的存储列, + 可直接从 parquet 读取, 无需 _load_enriched_for_date 的全量指标重算 + (历史日期该慢路径最坏会触发 9 次全市场 compute_enriched_full)。 + + 选取逻辑与旧循环等价: 在 as_of 前 1~9 天内找到第一个存在的日分区 + (即前一交易日), 读取其 symbol + consec_col。存储列的值与重算值逐位一致 + (连板计数为 run-length, 150 天 warmup 完全覆盖 A 股最长连板, 二者相等)。 + + 返回列: symbol, prev_consec。找不到前一交易日时返回空 DataFrame。 + """ + enriched_dir = self.repo.store.data_dir / self._enriched_dirname + for delta in range(1, 10): + candidate = as_of - timedelta(days=delta) + target_parquet = enriched_dir / f"date={candidate.isoformat()}" / "part.parquet" + if not target_parquet.exists(): + continue + try: + lf = pl.scan_parquet(target_parquet) + cols = lf.collect_schema().names() + except Exception as e: # noqa: BLE001 + logger.warning("load_prior_consecutive scan failed for %s: %s", candidate, e) + return pl.DataFrame() + # 存储列理论上必含 consec_col; 若该分区缺列则继续向前找 (与旧循环一致) + if "symbol" not in cols or consec_col not in cols: + continue + try: + return lf.select( + "symbol", + pl.col(consec_col).alias("prev_consec"), + ).collect() + except Exception as e: # noqa: BLE001 + logger.warning("load_prior_consecutive read failed for %s: %s", candidate, e) + return pl.DataFrame() + return pl.DataFrame() + def _compute_enriched_full(self, df_target: pl.DataFrame, target_date: date) -> pl.DataFrame: """从 14 列基础数据即时计算完整 enriched (含全部指标和信号)。 @@ -437,6 +475,10 @@ class ScreenerService: df = df.filter(pl.col("symbol").is_in(pool)) # 用 DuckDB 做 SQL 过滤 (注册临时视图) + # 用独立的 :memory: 连接 (而非复用 repo 共享连接的 cursor): conditions 是用户 + # 传入的 SQL 片段, 隔离连接下注入至多能碰 read_csv/read_parquet 文件; 若复用共享 + # 连接则会把 app 已注册的真实业务表也暴露给注入, 扩大攻击面。隔离连接创建开销极低。 + con = None try: import duckdb con = duckdb.connect(database=":memory:") @@ -448,10 +490,15 @@ class ScreenerService: if limit: sql += f" LIMIT {limit}" df_result = con.execute(sql).pl() - con.close() except Exception as e: # noqa: BLE001 logger.warning("screener SQL query failed: %s", e) df_result = pl.DataFrame() + finally: + if con is not None: + try: + con.close() + except Exception: # noqa: BLE001 + pass rows = df_result.to_dicts() if not df_result.is_empty() else [] elapsed = (time.perf_counter() - t0) * 1000 diff --git a/backend/app/services/stock_reports.py b/backend/app/services/stock_reports.py index d522d23..af4b552 100644 --- a/backend/app/services/stock_reports.py +++ b/backend/app/services/stock_reports.py @@ -1,7 +1,8 @@ """AI 个股分析报告持久化存储。 与 ai_reports.py(财务分析报告)完全独立 —— 单独的文件、字段、上限, -互不影响。刻意不复用,避免引入 kind 判别字段与分支(解耦 > 抽象)。 +互不影响。存储机制委托给共享的 JsonReportStore(原子写 + 实例锁), +本模块只固化 个股分析报告 的文件名 / 上限 / id 前缀, 对外保持原有函数签名不变。 存储位置: data/user_data/ai_stock_reports.json (数组,按 created_at 降序) 保留最近 MAX_REPORTS 条;超出自动裁剪最旧的。 @@ -21,71 +22,23 @@ """ from __future__ import annotations -import json -import logging -import time -from pathlib import Path - -logger = logging.getLogger(__name__) +from app.services.json_report_store import JsonReportStore MAX_REPORTS = 50 - -def _path() -> Path: - from app.config import settings - p = settings.data_dir / "user_data" / "ai_stock_reports.json" - p.parent.mkdir(parents=True, exist_ok=True) - return p +_store = JsonReportStore("ai_stock_reports.json", MAX_REPORTS, id_prefix="sar") def list_reports() -> list[dict]: """返回全部报告(按 created_at 降序)。""" - p = _path() - if not p.exists(): - return [] - try: - data = json.loads(p.read_text(encoding="utf-8")) - if isinstance(data, list): - return sorted(data, key=lambda r: r.get("created_at", ""), reverse=True) - except Exception as e: # noqa: BLE001 - logger.warning("ai_stock_reports.json malformed: %s", e) - return [] - - -def _save_all(reports: list[dict]) -> None: - """全量写入(裁剪到 MAX_REPORTS)。""" - reports.sort(key=lambda r: r.get("created_at", ""), reverse=True) - if len(reports) > MAX_REPORTS: - reports = reports[:MAX_REPORTS] - _path().write_text( - json.dumps(reports, indent=2, ensure_ascii=False), encoding="utf-8", - ) + return _store.list_reports() def save_report(report: dict) -> dict: """新增一条报告并持久化。返回保存后的报告(含 id / created_at)。""" - reports = list_reports() - if not report.get("id"): - report["id"] = f"sar_{int(time.time() * 1000)}_{report.get('symbol', 'x')}" - if not report.get("created_at"): - report["created_at"] = _now_iso() - reports.append(report) - _save_all(reports) - logger.info("Stock report saved: %s (%s), total %d", report.get("symbol"), report.get("id"), len(reports)) - return report + return _store.save_report(report) def delete_report(report_id: str) -> bool: """删除指定报告。返回是否删除成功。""" - reports = list_reports() - before = len(reports) - reports = [r for r in reports if r.get("id") != report_id] - if len(reports) < before: - _save_all(reports) - return True - return False - - -def _now_iso() -> str: - from datetime import datetime - return datetime.now().isoformat(timespec="seconds") + return _store.delete_report(report_id) diff --git a/backend/app/services/strategy_cache.py b/backend/app/services/strategy_cache.py index 201134d..37c9251 100644 --- a/backend/app/services/strategy_cache.py +++ b/backend/app/services/strategy_cache.py @@ -15,6 +15,8 @@ from __future__ import annotations import json import logging +import os +import threading import time from datetime import date, datetime from pathlib import Path @@ -34,6 +36,11 @@ logger = logging.getLogger(__name__) _CACHE_FILENAME = "strategy_cache.json" +# 读写同一 JSON 文件的进程内锁: write_cache 的 read-modify-write 与并发 read_cache +# 无锁会丢更新/读到半写文件。read_cache 与 write_cache 共用此锁; write 内部复用 +# _read_cache_unlocked 避免自死锁。写入用临时文件 + os.replace 做到原子替换。 +_file_lock = threading.Lock() + def _cache_path(data_dir: Path) -> Path: return data_dir / "user_data" / _CACHE_FILENAME @@ -62,6 +69,12 @@ def read_cache(data_dir: Path) -> dict | None: 保护价值有限。故移除: 盘后缓存总能读出, 实时新鲜度由 /api/screener/cached 端点叠加监控引擎的内存实时结果 (latest_strategy_results) 来保证。 """ + with _file_lock: + return _read_cache_unlocked(data_dir) + + +def _read_cache_unlocked(data_dir: Path) -> dict | None: + """实际读取逻辑 (不持锁)。供 read_cache 与 write_cache 复用, 避免重入死锁。""" path = _cache_path(data_dir) if not path.exists(): return None @@ -100,8 +113,20 @@ def write_cache( path = _cache_path(data_dir) path.parent.mkdir(parents=True, exist_ok=True) - # 读取旧缓存 - old = read_cache(data_dir) + # 整个 read-modify-write 持锁: 避免并发 write 丢更新, 也避免与 read_cache 撕裂 + with _file_lock: + _write_cache_locked(path, data_dir, as_of, results) + + +def _write_cache_locked( + path: Path, + data_dir: Path, + as_of: str, + results: dict[str, Any], +) -> None: + """持 _file_lock 后的实际写入逻辑 (read-merge-write + 原子替换)。""" + # 读取旧缓存 (已持锁, 走不重入的 _read_cache_unlocked) + old = _read_cache_unlocked(data_dir) old_as_of = old.get("as_of") if old else None old_ever_rows: dict[str, dict[str, dict]] = old.get("today_ever_rows", {}) if old else {} @@ -141,7 +166,10 @@ def write_cache( "updated_at": int(time.time() * 1000), } try: - path.write_text(json.dumps(payload, ensure_ascii=False, default=_json_default), encoding="utf-8") + # 原子写: 先写临时文件再 os.replace, 避免读侧读到半写的 JSON + tmp = path.with_name(path.name + ".tmp") + tmp.write_text(json.dumps(payload, ensure_ascii=False, default=_json_default), encoding="utf-8") + os.replace(tmp, path) total_rows = sum(len(r.get("rows", [])) for r in results.values()) total_ever = sum(len(v) for v in today_ever_matched.values()) logger.info("策略缓存已写入: %s, %d 策略, %d 命中, %d 曾命中", as_of, len(results), total_rows, total_ever) diff --git a/backend/app/strategy/builtin/near_limit_up.py b/backend/app/strategy/builtin/near_limit_up.py index fec68e2..1b933cf 100644 --- a/backend/app/strategy/builtin/near_limit_up.py +++ b/backend/app/strategy/builtin/near_limit_up.py @@ -4,19 +4,20 @@ import polars as pl def _limit_pct() -> pl.Expr: """根据板块和 ST 动态计算涨跌幅限制 (小数)。 - 创业板(300/301)/科创板(688): 20% + 创业板(300/301)/科创板(688): 20% (含其 ST) 北交所(.BJ): 30% - ST: 5% - 主板: 10% + 主板 ST: 5% ← ST 5% 仅主板生效, 创业板/科创板 ST 仍是 20% + 主板普通: 10% """ is_st = pl.col("name").str.contains("(?i)ST").fill_null(False) is_cyb = pl.col("symbol").str.starts_with("300") | pl.col("symbol").str.starts_with("301") is_kcb = pl.col("symbol").str.starts_with("688") is_bj = pl.col("symbol").str.contains(r"\.BJ$") return ( - pl.when(is_st).then(0.05) - .when(is_cyb | is_kcb).then(0.20) + # 板块判定优先于 ST: 创业板/科创板 ST 保留 20%, 北交所 30%; ST 5% 只剩主板 + pl.when(is_cyb | is_kcb).then(0.20) .when(is_bj).then(0.30) + .when(is_st).then(0.05) .otherwise(0.10) ) diff --git a/backend/app/strategy/monitor.py b/backend/app/strategy/monitor.py index 9082430..c407410 100644 --- a/backend/app/strategy/monitor.py +++ b/backend/app/strategy/monitor.py @@ -13,6 +13,7 @@ from __future__ import annotations import datetime as _dt import logging +import threading import time from dataclasses import dataclass, field from typing import Any, Callable @@ -88,6 +89,10 @@ class StrategyMonitorService: self._alert_handler = alert_handler # strategy_id → 监控配置 self._watching: dict[str, dict] = {} + # _watching 跨线程锁: on_quote_update 跑在行情轮询线程迭代 _watching, + # API 线程同时 start/stop 增删会抛 "dict changed size during iteration"。 + # 增删与迭代前的快照都持此锁 (镜像 MonitorRuleEngine.evaluate 的 list 快照)。 + self._watching_lock = threading.Lock() def start(self, strategy_id: str, config: dict) -> None: """开始监控一个策略 @@ -98,19 +103,23 @@ class StrategyMonitorService: "alerts": [{"field": "rsi_14", "op": ">", "value": 80, "message": "..."}], } """ - self._watching[strategy_id] = config + with self._watching_lock: + self._watching[strategy_id] = config logger.info("strategy monitor started: %s", strategy_id) def stop(self, strategy_id: str) -> None: - self._watching.pop(strategy_id, None) + with self._watching_lock: + self._watching.pop(strategy_id, None) logger.info("strategy monitor stopped: %s", strategy_id) def stop_all(self) -> None: - self._watching.clear() + with self._watching_lock: + self._watching.clear() @property def watching(self) -> dict[str, dict]: - return dict(self._watching) + with self._watching_lock: + return dict(self._watching) def on_quote_update(self, df: pl.DataFrame) -> list[StrategyAlert]: """行情更新后调用。向量化检查所有监控策略。 @@ -125,7 +134,11 @@ class StrategyMonitorService: all_alerts: list[StrategyAlert] = [] - for strategy_id, cfg in self._watching.items(): + # 迭代前持锁快照, 避免行情线程迭代时 API 线程 start/stop 改变字典大小 + with self._watching_lock: + watching_items = list(self._watching.items()) + + for strategy_id, cfg in watching_items: # 买入信号 entry_sigs = cfg.get("entry_signals", []) if entry_sigs: diff --git a/backend/app/tickflow/rate_limits.py b/backend/app/tickflow/rate_limits.py index 86c13e3..f68546e 100644 --- a/backend/app/tickflow/rate_limits.py +++ b/backend/app/tickflow/rate_limits.py @@ -5,6 +5,7 @@ TickFlow-backed services. It intentionally does not manage custom data sources. """ from __future__ import annotations +import threading import time from dataclasses import dataclass from typing import TypeVar @@ -13,6 +14,28 @@ from app.tickflow.capabilities import Cap, CapabilitySet T = TypeVar("T") +# 进程级共享限速器: 原先每个调用方各自本地 sleep(60/rpm), 并发同步 (kline/index/ +# depth/watchlist/custom) 时聚合请求速率会成倍超过单能力 rpm → 429。 +# 这里用一张按 rpm 分桶的「下一个可用时刻」表 (Lock 守护), 所有调用方按同一时间轴 +# 排队, 使跨调用方的聚合发包间隔 >= 60/rpm。以 rpm 为键 (调用方签名只带 rpm, 不带 cap; +# rpm 是各能力速率的代理); 恰好同 rpm 的不同能力会共享一队, 偏保守但绝不超速。 +_slot_lock = threading.Lock() +_next_slot: dict[int, float] = {} + + +def _reserve_slot(rpm: int, interval: float) -> float: + """在共享时间轴上为一次请求预约一个发包槽, 返回需等待的秒数 (>=0)。 + + interval = 60/rpm。now 早于该 rpm 桶的 next_slot 时排到 next_slot, 否则排到 now; + 随后把该桶 next_slot 后移 interval。持锁仅做时间账目, 不在锁内 sleep。 + """ + key = rpm if rpm and rpm > 0 else -1 + with _slot_lock: + now = time.monotonic() + scheduled = max(now, _next_slot.get(key, now)) + _next_slot[key] = scheduled + interval + return scheduled - now + @dataclass(frozen=True) class ResolvedLimit: @@ -51,12 +74,21 @@ def chunked(items: list[T], batch_size: int | None) -> list[list[T]]: def sleep_between_batches(index: int, rpm: int | None, *, default_interval: float = 0.0) -> None: - """Sleep before every batch after the first, using the existing interval formula.""" - if index <= 0: - return + """Sleep before every batch after the first, using the existing interval formula. + + 内部改用进程级共享限速器 (_reserve_slot): 保持「首批不 sleep, 后续每批间隔 60/rpm」 + 的单调用方观感, 同时让并发调用方按同一时间轴排队, 聚合速率不再超过单能力 rpm。 + """ interval = batch_interval(rpm, default=default_interval) - if interval > 0: - time.sleep(interval) + if interval <= 0: + return + if index <= 0: + # 首批不 sleep, 但登记一个占位槽, 让后续/并发调用方在同一时间轴上排队 + _reserve_slot(rpm or -1, interval) + return + wait = _reserve_slot(rpm or -1, interval) + if wait > 0: + time.sleep(wait) def min_batch(preferred: int, limit: ResolvedLimit) -> int: diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index 342b9b6..d24f224 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -1636,6 +1636,41 @@ class KlineRepository: with self._lock: self.store._register_unified_views() + def rebuild_views(self) -> None: + """重建全部 13 张 parquet 视图并重挂 unified 视图 —— 唯一权威实现。 + + 原先 daily_pipeline._refresh_views(盘后管道) 与 /api/data/clear(清库) 各自 + 内联了同一份视图重建 SQL, 清库那份还漏了几张视图导致漂移。此处收敛为单一入口: + 覆盖全部 13 张视图 (二者的超集), 空目录 (清库后) 也能安全重挂。 + """ + d = self.store.data_dir.as_posix() + views = { + "kline_daily": f"{d}/kline_daily/**/*.parquet", + "kline_enriched": f"{d}/kline_daily_enriched/**/*.parquet", + "kline_index_daily": f"{d}/kline_index_daily/**/*.parquet", + "kline_index_enriched": f"{d}/kline_index_enriched/**/*.parquet", + "kline_etf_daily": f"{d}/kline_etf_daily/**/*.parquet", + "kline_etf_enriched": f"{d}/kline_etf_enriched/**/*.parquet", + "kline_etf_minute": f"{d}/kline_etf_minute/**/*.parquet", + "kline_minute": f"{d}/kline_minute/**/*.parquet", + "adj_factor": f"{d}/adj_factor/**/*.parquet", + "adj_factor_etf": f"{d}/adj_factor_etf/**/*.parquet", + "instruments": f"{d}/instruments/**/*.parquet", + "instruments_index": f"{d}/instruments_index/**/*.parquet", + "instruments_etf": f"{d}/instruments_etf/**/*.parquet", + } + for name, path in views.items(): + try: + with self._lock: + self.db.execute( + f"CREATE OR REPLACE VIEW {name} AS " + f"SELECT * FROM read_parquet('{path}', union_by_name=true)" + ) + except Exception as e: # noqa: BLE001 + logger.warning("rebuild view %s failed: %s", name, e) + with self._lock: + self.store._register_unified_views() + @staticmethod def _atomic_write_parquet(df: pl.DataFrame, out: Path) -> None: """先写临时文件再原子替换, 避免进程中断留下损坏的 parquet。 diff --git a/backend/app/tickflow/scheduler.py b/backend/app/tickflow/scheduler.py index 588b67c..9d0ad89 100644 --- a/backend/app/tickflow/scheduler.py +++ b/backend/app/tickflow/scheduler.py @@ -58,9 +58,11 @@ class Scheduler: if bucket is None: return lock = self._locks[cap] - async with lock: - while True: + # 只把令牌账目放锁内; sleep 放锁外, 否则一个协程在 sleep 期间独占锁, + # 会把同 capability 下其他有令牌可用的请求也串行阻塞。 + while True: + async with lock: wait = bucket.consume(n) - if wait == 0: - return - await asyncio.sleep(wait) + if wait == 0: + return + await asyncio.sleep(wait) diff --git a/backend/tests/test_st_limit_and_sharpe.py b/backend/tests/test_st_limit_and_sharpe.py new file mode 100644 index 0000000..55722a0 --- /dev/null +++ b/backend/tests/test_st_limit_and_sharpe.py @@ -0,0 +1,88 @@ +"""回归测试: + +1. ST 5% 涨跌停限幅仅适用于主板风险警示股; 创业板/科创板 ST 仍执行 20%。 + (修正前 _is_st 无条件套 5%, 会误报/漏报这批股的涨停。) +2. 因子回测 Sharpe 的年化系数须匹配调仓频率 (月频 √12 / 周频 √52 / 日频 √252); + (修正前一律 √252, 月频 Sharpe 被高估 √21 ≈ 4.6 倍。) +""" +from __future__ import annotations + +from datetime import date + +import polars as pl + +from app.backtest.factor import FactorBacktestService +from app.indicators.pipeline import compute_limit_signals +from app.strategy.builtin.near_limit_up import _limit_pct + + +def test_near_limit_pct_st_only_on_main_board(): + df = pl.DataFrame({ + "symbol": ["300001", "688001", "600001", "000001", "830001.BJ"], + "name": ["*ST创业", "科创ST", "*ST主板", "平安银行", "北交ST"], + }) + lp = df.with_columns(_limit_pct().alias("lp"))["lp"].to_list() + assert lp[0] == 0.20 # 创业板 ST → 20% (不再是 5%) + assert lp[1] == 0.20 # 科创板 ST → 20% + assert lp[2] == 0.05 # 主板 ST → 5% + assert lp[3] == 0.10 # 主板普通 → 10% + assert lp[4] == 0.30 # 北交所 → 30% + + +def _two_day(symbol: str, prev_close: float, today_close: float) -> pl.DataFrame: + """2 日最小输入: 首日平收, 次日收于 today_close。""" + return pl.DataFrame({ + "symbol": [symbol, symbol], + "date": [date(2024, 1, 2), date(2024, 1, 3)], + "raw_close": [prev_close, today_close], + "close": [prev_close, today_close], + "raw_high": [prev_close, today_close], + "open": [prev_close, today_close], + "high": [prev_close, today_close], + "low": [prev_close, today_close], + "change_pct": [0.0, today_close / prev_close - 1], + "vol_ratio_5d": [1.0, 1.0], + }) + + +def _last_limit_up(symbol: str, name: str, prev_close: float, today_close: float): + df = _two_day(symbol, prev_close, today_close) + inst = pl.DataFrame({"symbol": [symbol], "name": [name]}) + out = compute_limit_signals(df, inst).sort("date") + return out["signal_limit_up"].to_list()[-1], out["consecutive_limit_ups"].to_list()[-1] + + +def test_st_chinext_limit_up_detected_at_20pct(): + # 创业板 *ST 昨收 10.00 → 今日 +20% 至 12.00 应识别为涨停 (修正前按 5% 会漏) + sig, consec = _last_limit_up("300001", "*ST创业", 10.0, 12.0) + assert sig is True + assert consec == 1 + + +def test_st_chinext_plus5pct_is_not_a_false_limit_up(): + # 同股仅 +5% 至 10.50 不应误报涨停 (修正前按 5% 会误报) + sig, _ = _last_limit_up("300001", "*ST创业", 10.0, 10.5) + assert sig is False + + +def test_st_main_board_still_limits_at_5pct(): + # 主板 *ST 昨收 10.00 → 今日 +5% 至 10.50 仍应识别为涨停 + sig, _ = _last_limit_up("600001", "*ST主板", 10.0, 10.5) + assert sig is True + + +def test_sharpe_annualization_matches_rebalance_frequency(): + nav = [ + {"date": "2024-01-31", "Q1": 1.00}, + {"date": "2024-02-29", "Q1": 1.02}, + {"date": "2024-03-29", "Q1": 1.01}, + {"date": "2024-04-30", "Q1": 1.05}, + {"date": "2024-05-31", "Q1": 1.04}, + {"date": "2024-06-28", "Q1": 1.08}, + ] + start, end = date(2024, 1, 1), date(2024, 6, 30) + m = FactorBacktestService._calc_group_stats(nav, start, end, "monthly")[0]["sharpe"] + d = FactorBacktestService._calc_group_stats(nav, start, end, "daily")[0]["sharpe"] + assert m != 0.0 and d != 0.0 + # 同一净值曲线, daily(√252) / monthly(√12) 的比值应为 √21 ≈ 4.58 + assert abs((d / m) - (252 / 12) ** 0.5) < 0.05 diff --git a/frontend/src/components/AlertToast.tsx b/frontend/src/components/AlertToast.tsx index bb9fc1a..c511976 100644 --- a/frontend/src/components/AlertToast.tsx +++ b/frontend/src/components/AlertToast.tsx @@ -102,7 +102,12 @@ export function AlertToastContainer() { if (!items.length) return null return ( -
+
{items .filter(item => !(item.alert.source === 'strategy' && !item.alert.symbol)) @@ -125,7 +130,16 @@ export function AlertToastContainer() { exit={{ opacity: 0, x: 60, scale: 0.9 }} transition={{ duration: 0.25, ease: [0.16, 1, 0.3, 1] }} onClick={() => handleClick(item.id)} - className="pointer-events-auto relative overflow-hidden rounded-xl border border-border/60 bg-surface/95 backdrop-blur-md shadow-2xl pl-3 pr-2 py-2.5 cursor-pointer hover:border-accent/40 hover:shadow-accent/10 transition-all" + role="button" + tabIndex={0} + aria-label={`查看监控通知${ev.name ? ` ${ev.name}` : ''}${ev.symbol ? ` ${ev.symbol}` : ''}`} + onKeyDown={(e) => { + if (e.key === 'Enter' || e.key === ' ') { + e.preventDefault() + handleClick(item.id) + } + }} + className="pointer-events-auto relative overflow-hidden rounded-xl border border-border/60 bg-surface/95 backdrop-blur-md shadow-2xl pl-3 pr-2 py-2.5 cursor-pointer hover:border-accent/40 hover:shadow-accent/10 transition-all focus:outline-none focus-visible:ring-2 focus-visible:ring-accent/60" > {/* 左侧色条 */}
@@ -143,7 +157,7 @@ export function AlertToastContainer() { {fmtPct(pct)} )} -
diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index d2bd162..25bef41 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -1,8 +1,8 @@ -import { useEffect, useRef, useState } from 'react' +import { useEffect, useRef, useState, Suspense } from 'react' import { NavLink, Outlet, useNavigate } from 'react-router-dom' import { useQuery, useQueryClient } from '@tanstack/react-query' import { motion } from 'framer-motion' -import { useQuoteStream } from '@/lib/useQuoteStream' +import { useQuoteStream, useQuoteStreamStatus } from '@/lib/useQuoteStream' import { ToastContainer } from '@/components/Toast' import { AlertToastContainer } from '@/components/AlertToast' import { AiAnalysisHost } from '@/components/financials/AiAnalysisHost' @@ -46,6 +46,7 @@ import { Sun, Moon, X, + WifiOff, } from 'lucide-react' import { Logo } from './Logo' import { api, type IndexQuote } from '@/lib/api' @@ -337,6 +338,8 @@ export function Layout() { // SSE: 行情更新时自动刷新相关 queries + 告警通知 useQuoteStream(realtimeEnabled, prefs?.sse_refresh_pages) + // 实时 SSE 连接状态 — 断开时顶部显示徽标, 提示可能漏策略告警 + const streamStatus = useQuoteStreamStatus() const toggleQuote = useToggleRealtimeQuotes() const isRunning = quoteStatus?.running ?? false @@ -659,7 +662,25 @@ export function Layout() { transition={{ duration: 0.25, ease: [0.16, 1, 0.3, 1] }} className="h-full overflow-auto scrollbar-gutter-stable" > - + {streamStatus === 'reconnecting' && ( +
+ + 实时连接断开 · 重连中 +
+ )} + + +
+ } + > + + diff --git a/frontend/src/components/Modal.tsx b/frontend/src/components/Modal.tsx new file mode 100644 index 0000000..7ca8332 --- /dev/null +++ b/frontend/src/components/Modal.tsx @@ -0,0 +1,137 @@ +import { useEffect, useRef, type ReactNode } from 'react' + +/** + * 共享模态对话框原语 — 统一处理可访问性: + * - role="dialog" + aria-modal + aria-labelledby / aria-label + * - ESC 关闭 + * - 打开时把焦点移入对话框 (initialFocusRef 或首个可聚焦元素) + * - Tab / Shift+Tab 焦点陷阱 (焦点不会跑出对话框) + * - 关闭时把焦点还给打开前的元素 + * - 点击遮罩关闭 (可用 closeOnBackdrop 关闭) + * + * 视觉: 提供居中遮罩 + 面板容器, 面板样式由 panelClassName 定制。 + */ +export interface ModalProps { + onClose: () => void + children: ReactNode + /** 对话框标题元素 id (用于 aria-labelledby) */ + labelledBy?: string + /** 无可见标题时的无障碍名称 */ + ariaLabel?: string + /** 面板 className (尺寸/背景/圆角等) */ + panelClassName?: string + /** 遮罩 className (覆盖默认居中/背景) */ + overlayClassName?: string + /** 打开时聚焦的元素; 不传则聚焦面板内首个可聚焦元素 */ + initialFocusRef?: React.RefObject + /** 点击遮罩是否关闭 (默认 true) */ + closeOnBackdrop?: boolean +} + +const FOCUSABLE = [ + 'a[href]', + 'button:not([disabled])', + 'textarea:not([disabled])', + 'input:not([disabled])', + 'select:not([disabled])', + '[tabindex]:not([tabindex="-1"])', +].join(',') + +export function Modal({ + onClose, + children, + labelledBy, + ariaLabel, + panelClassName = 'w-[92vw] max-w-lg bg-surface border border-border rounded-card shadow-xl', + overlayClassName = 'fixed inset-0 z-50 flex items-center justify-center bg-black/50 backdrop-blur-sm', + initialFocusRef, + closeOnBackdrop = true, +}: ModalProps) { + const panelRef = useRef(null) + // onClose 存 ref: 焦点陷阱/ESC effect 只在挂载时装一次。否则父级每次重渲染 (或未 memo 的 + // onClose) 都让 effect 重跑, requestAnimationFrame(focusFirst) 会在每次输入后把焦点抢回 + // 面板首个元素, 导致对话框内文本框无法输入。 + const onCloseRef = useRef(onClose) + onCloseRef.current = onClose + + useEffect(() => { + // 记住打开前的焦点, 关闭时还原 + const prevActive = document.activeElement as HTMLElement | null + + // 初始聚焦 + const focusFirst = () => { + if (initialFocusRef?.current) { + initialFocusRef.current.focus() + return + } + const panel = panelRef.current + if (!panel) return + const first = panel.querySelector(FOCUSABLE) + ;(first ?? panel).focus() + } + // 等一帧确保内容已挂载 + const raf = requestAnimationFrame(focusFirst) + + const onKeyDown = (e: KeyboardEvent) => { + if (e.key === 'Escape') { + e.stopPropagation() + onCloseRef.current() + return + } + if (e.key !== 'Tab') return + const panel = panelRef.current + if (!panel) return + const nodes = Array.from(panel.querySelectorAll(FOCUSABLE)) + .filter(el => el.offsetParent !== null || el === document.activeElement) + if (nodes.length === 0) { + e.preventDefault() + panel.focus() + return + } + const first = nodes[0] + const last = nodes[nodes.length - 1] + const active = document.activeElement as HTMLElement | null + if (e.shiftKey) { + if (active === first || !panel.contains(active)) { + e.preventDefault() + last.focus() + } + } else { + if (active === last || !panel.contains(active)) { + e.preventDefault() + first.focus() + } + } + } + + document.addEventListener('keydown', onKeyDown, true) + return () => { + cancelAnimationFrame(raf) + document.removeEventListener('keydown', onKeyDown, true) + // 还原焦点 + prevActive?.focus?.() + } + // 只在挂载时装一次: onClose 走 ref, initialFocusRef 为稳定 ref 对象, 无需进依赖。 + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []) + + return ( +
{ if (e.target === e.currentTarget) onClose() } : undefined} + > +
e.stopPropagation()} + > + {children} +
+
+ ) +} diff --git a/frontend/src/components/RpsRotationDialog.tsx b/frontend/src/components/RpsRotationDialog.tsx index be669f8..f8e2fd0 100644 --- a/frontend/src/components/RpsRotationDialog.tsx +++ b/frontend/src/components/RpsRotationDialog.tsx @@ -7,6 +7,7 @@ import { QK } from '@/lib/queryKeys' import { cn } from '@/lib/cn' import { fmtPct } from '@/lib/format' import { MarkdownRenderer } from '@/components/financials/MarkdownRenderer' +import { Modal } from '@/components/Modal' interface Props { onClose: () => void @@ -128,12 +129,6 @@ export function RpsRotationDialog({ onClose }: Props) { handleScroll() }, [handleScroll, rowCount]) - // ESC 关闭 - useEffect(() => { - const onKey = (e: KeyboardEvent) => { if (e.key === 'Escape') onClose() } - window.addEventListener('keydown', onKey) - return () => window.removeEventListener('keydown', onKey) - }, [onClose]) // 选中概念的追踪行: 找出它在每个日期列的(排名, 涨幅)。 // 每列已按涨幅降序排好, 故排名 = 该概念在数组里的索引 + 1。 @@ -204,31 +199,22 @@ export function RpsRotationDialog({ onClose }: Props) { }, [visibleRange, getRowIndex, dates, columns, selected]) return ( - - { if (e.target === e.currentTarget) onClose() }} - > - + {/* 标题栏 */}
- 概念涨幅轮动 + 概念涨幅轮动 {conceptCount > 0 ? `${dates.length} 天 · ${conceptCount} 个概念` : '暂无数据'}
-
@@ -438,8 +424,6 @@ export function RpsRotationDialog({ onClose }: Props) { 每列各自按当日涨幅排序 · 点击单元格追踪概念在各日的排名变化
- - - + ) } diff --git a/frontend/src/components/Toast.tsx b/frontend/src/components/Toast.tsx index 389514d..2439b59 100644 --- a/frontend/src/components/Toast.tsx +++ b/frontend/src/components/Toast.tsx @@ -31,7 +31,12 @@ export function ToastContainer() { if (!items.length) return null return ( -
+
{items.map(t => (
{/* 中间：标题 */} - + {strategyId ? '修改策略' : '创建策略'} {/* 右侧：步骤 + 关闭 */} @@ -284,7 +284,7 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create 2
)} - +
@@ -506,8 +506,6 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create )} - - - + ) } diff --git a/frontend/src/components/screener/StrategySettingsDialog.tsx b/frontend/src/components/screener/StrategySettingsDialog.tsx index 24b235c..4405c69 100644 --- a/frontend/src/components/screener/StrategySettingsDialog.tsx +++ b/frontend/src/components/screener/StrategySettingsDialog.tsx @@ -6,6 +6,7 @@ import { BUILTIN_COLUMNS } from '@/lib/watchlist-columns' import { color } from '@/lib/colors' import { SignalPicker } from './SignalPicker' import { SignalTriggerActions } from '@/components/signals/SignalTriggerActions' +import { Modal } from '@/components/Modal' // 内置列名 → 中文标签 const FIELD_LABEL: Record = {} @@ -320,30 +321,21 @@ export function StrategySettingsDialog({ strategyId, onClose, onSaved, onAiModif return ( <> - - { if (e.target === e.currentTarget) onClose() }} - > - + {/* 标题 */}
- {detail?.name ?? strategyId} + {detail?.name ?? strategyId} {detail && {{ builtin: '内置', custom: '自定义', ai: 'AI' }[detail.source] ?? detail.source}} {strategyId}
- +
{/* 内容 */} @@ -568,11 +560,10 @@ export function StrategySettingsDialog({ strategyId, onClose, onSaved, onAiModif - - - + - {/* 删除确认弹窗 */} + {/* 删除确认弹窗 — 必须放 Modal 外: Modal 面板有 backdrop-blur (为 fixed 后代建立定位上下文) + + overflow-hidden, 放里面会导致本应全屏居中的确认框相对面板定位并被裁剪/错位。 */} {showDeleteConfirm && ( )} + ) } diff --git a/frontend/src/lib/backtestTask.ts b/frontend/src/lib/backtestTask.ts index 16c25e6..b6bc3c3 100644 --- a/frontend/src/lib/backtestTask.ts +++ b/frontend/src/lib/backtestTask.ts @@ -25,8 +25,13 @@ export interface BacktestTask { result: StrategyBacktestResult | null progress: BacktestProgress | null error: string | null + /** 连接中断、正在有界重连中 (UI 显示"连接中断，重试中") */ + reconnecting: boolean } +// 连接断开后最多自动重连次数, 超过则放弃并进入可重试的错误态 +const MAX_RECONNECT_ATTEMPTS = 5 + let current: BacktestTask | null = null const listeners = new Set<() => void>() let taskSeq = 0 @@ -73,11 +78,28 @@ function connectSSE(url: string): void { const es = new EventSource(url) eventSource = es + // 本次连接的重连计数 (EventSource 断开会自动重连并再次触发 onerror) + let reconnectAttempts = 0 + + const clearReconnecting = () => { + if (current?.id === id && current.reconnecting) { + current = { ...current, reconnecting: false } + emit() + } + reconnectAttempts = 0 + } + + es.onopen = () => { + clearReconnecting() + } + es.addEventListener('progress', (e: MessageEvent) => { if (current?.id !== id) return + // 收到数据说明连接恢复正常 + reconnectAttempts = 0 try { const prog = JSON.parse(e.data) as BacktestProgress - current = { ...current, progress: prog } + current = { ...current, progress: prog, reconnecting: false } emit() } catch { /* ignore */ } }) @@ -86,10 +108,10 @@ function connectSSE(url: string): void { if (current?.id !== id) return try { const result = JSON.parse(e.data) as StrategyBacktestResult - current = { ...current, isPending: false, result, error: null } + current = { ...current, isPending: false, result, error: null, reconnecting: false } emit() } catch { - current = { ...current, isPending: false, error: '结果解析失败' } + current = { ...current, isPending: false, error: '结果解析失败', reconnecting: false } emit() } es.close() @@ -103,17 +125,36 @@ function connectSSE(url: string): void { if (e.data) { try { const msg = JSON.parse(e.data)?.message ?? '回测出错' - current = { ...current, isPending: false, error: msg } + current = { ...current, isPending: false, error: msg, reconnecting: false } emit() } catch { - current = { ...current, isPending: false, error: '回测出错' } + current = { ...current, isPending: false, error: '回测出错', reconnecting: false } emit() } es.close() eventSource = null localStorage.removeItem(RECONNECT_KEY) + return } - // 无 data: 连接异常断开, EventSource 会自动重连, 不改变状态 + // 无 data: 连接异常断开。EventSource 会自动重连, 但需给出可见状态并有界放弃, + // 避免进度条永久冻结、isPending 永远 true。 + reconnectAttempts += 1 + if (reconnectAttempts > MAX_RECONNECT_ATTEMPTS) { + // 放弃: 停止自动重连, 进入可重试的错误态 (用户可重新发起回测) + es.close() + eventSource = null + current = { + ...current, + isPending: false, + reconnecting: false, + error: '连接中断，请重试', + } + emit() + return + } + // 仍在重试窗口内: 标记 reconnecting, 让 UI 显示"连接中断，重试中" + current = { ...current, reconnecting: true } + emit() }) } @@ -147,7 +188,7 @@ export function startBacktest(params: { } const id = ++taskSeq - current = { id, isPending: true, result: null, progress: null, error: null } + current = { id, isPending: true, result: null, progress: null, error: null, reconnecting: false } emit() const qs = buildQuery({ @@ -203,7 +244,7 @@ export async function stopBacktest(): Promise { eventSource = null } if (current?.isPending) { - current = { ...current, isPending: false, error: '已取消' } + current = { ...current, isPending: false, error: '已取消', reconnecting: false } emit() } localStorage.removeItem(RECONNECT_KEY) @@ -221,7 +262,7 @@ export function tryReconnect(): boolean { if (!qs) return false // 有未完成的任务, 重连 const id = ++taskSeq - current = { id, isPending: true, result: null, progress: null, error: null } + current = { id, isPending: true, result: null, progress: null, error: null, reconnecting: false } emit() connectSSE(`/api/backtest/strategy/stream?${qs}`) return true diff --git a/frontend/src/lib/queryKeys.ts b/frontend/src/lib/queryKeys.ts index b83a07a..d2dadba 100644 --- a/frontend/src/lib/queryKeys.ts +++ b/frontend/src/lib/queryKeys.ts @@ -41,6 +41,7 @@ export const QK = { // Backtest backtestStatus: ['backtest-status'] as const, + strategyDetail: (id: string) => ['strategy-detail', id] as const, // Data / Pipeline dataStatus: ['data-status'] as const, diff --git a/frontend/src/lib/useQuoteStream.ts b/frontend/src/lib/useQuoteStream.ts index 483eced..95acce7 100644 --- a/frontend/src/lib/useQuoteStream.ts +++ b/frontend/src/lib/useQuoteStream.ts @@ -1,4 +1,4 @@ -import { useEffect, useRef, useCallback } from 'react' +import { useEffect, useRef, useCallback, useSyncExternalStore } from 'react' import { useQueryClient } from '@tanstack/react-query' import { SSE_INVALIDATE_PREFIXES, QK } from './queryKeys' import { getQueryConfig } from './useQueryConfig' @@ -7,6 +7,45 @@ import { pushAlertToasts } from '@/components/AlertToast' import { feedReviewEvent } from './reviewStore' import type { StrategyAlertEvent } from './api' +// ===== 全局 SSE 连接状态 (模块级 store, 仿 AlertToast.tsx 模式) ===== +// 实时行情 SSE 断开时 UI 无感知 → 会漏掉策略告警。这里暴露连接状态, +// 供 Layout 顶部渲染徽标、连续失败 N 次后弹一次 toast。 +export type QuoteStreamStatus = 'connected' | 'reconnecting' | 'disconnected' + +let _streamStatus: QuoteStreamStatus = 'disconnected' +const _statusListeners = new Set<() => void>() + +// 连续失败到达该阈值后弹一次 toast (只弹一次, 恢复后重置) +const FAILS_BEFORE_TOAST = 3 +// 指数退避上限 +const BACKOFF_CAP_MS = 60_000 + +function _emitStatus() { + _statusListeners.forEach((fn) => fn()) +} + +function _setStatus(s: QuoteStreamStatus) { + if (_streamStatus === s) return + _streamStatus = s + _emitStatus() +} + +function _subscribeStatus(fn: () => void) { + _statusListeners.add(fn) + return () => { + _statusListeners.delete(fn) + } +} + +function _getStatus() { + return _streamStatus +} + +/** React hook: 读取全局实时行情 SSE 连接状态 (供 Layout 徽标使用) */ +export function useQuoteStreamStatus(): QuoteStreamStatus { + return useSyncExternalStore(_subscribeStatus, _getStatus, () => 'disconnected' as const) +} + /** * 全局 SSE hook: 监听后端行情更新推送 + 策略监控通知。 * @@ -54,10 +93,22 @@ export function useQuoteStream( // SSE 始终连接 — 监控告警不依赖实时行情开关 // (quotes_updated 行情刷新受 enabled 控制, strategy_alert 始终处理) + // 连续失败计数 (用于指数退避 + 到阈值弹一次 toast) + let failCount = 0 + let toastFired = false + const connect = () => { + _setStatus(failCount > 0 ? 'reconnecting' : _streamStatus) const es = new EventSource('/api/intraday/stream') esRef.current = es + es.onopen = () => { + // 连接成功: 重置退避与 toast 标记 + failCount = 0 + toastFired = false + _setStatus('connected') + } + // sse-starlette ping 心跳走 SSE comment，不会到达这里 es.addEventListener('quotes_updated', () => { @@ -129,7 +180,16 @@ export function useQuoteStream( es.onerror = () => { es.close() esRef.current = null - const delay = getQueryConfig().sse.reconnectDelay + failCount += 1 + _setStatus('reconnecting') + // 连续失败到阈值 → 弹一次 toast (漏行情=可能漏策略告警, 需明确告知) + if (failCount >= FAILS_BEFORE_TOAST && !toastFired) { + toastFired = true + toast('实时连接已断开，正在重连…', 'error') + } + // 指数退避 (base * 2^(n-1), 上限 60s), 替代原来固定 5s + const base = getQueryConfig().sse.reconnectDelay + const delay = Math.min(base * 2 ** (failCount - 1), BACKOFF_CAP_MS) retryRef.current = setTimeout(connect, delay) } } @@ -142,6 +202,7 @@ export function useQuoteStream( esRef.current.close() esRef.current = null } + _setStatus('disconnected') } }, [qc, handleAlerts]) } diff --git a/frontend/src/pages/Analysis.tsx b/frontend/src/pages/Analysis.tsx index 0975060..2f226dc 100644 --- a/frontend/src/pages/Analysis.tsx +++ b/frontend/src/pages/Analysis.tsx @@ -3,6 +3,7 @@ import { Link } from 'react-router-dom' import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import { BarChart3, ChevronDown, ChevronUp, Plus, Save, Trash2 } from 'lucide-react' import { PageHeader } from '@/components/PageHeader' +import { Skeleton } from '@/components/data/Skeleton' import { api, type AnalysisColumn, type ExtDataConfig, type ExtDataField } from '@/lib/api' import { QK } from '@/lib/queryKeys' @@ -281,7 +282,15 @@ export function Analysis() { ))} - {menuItems.length === 0 && ( + {menus.isLoading && + Array.from({ length: 3 }).map((_, i) => ( +
+ + + +
+ ))} + {!menus.isLoading && menuItems.length === 0 && (
暂无分析菜单，点击右上角新建。
)} diff --git a/frontend/src/pages/Data.tsx b/frontend/src/pages/Data.tsx index e710347..b84b92d 100644 --- a/frontend/src/pages/Data.tsx +++ b/frontend/src/pages/Data.tsx @@ -96,6 +96,8 @@ export function Data() { const clearData = useMutation({ mutationFn: api.dataClear, onSuccess: () => { + // 清库删除全部 parquet + alerts, 各页 (看板/自选/选股/指数/财务/个股/连板/监控) 缓存 + // 数据均已失效, 故广域失效所有 query 令其重取 —— 数据已清空, 全量刷新是正确行为而非误伤。 qc.invalidateQueries() setShowClearConfirm(false) }, diff --git a/frontend/src/pages/Dev.tsx b/frontend/src/pages/Dev.tsx index 81c4e4a..5862419 100644 --- a/frontend/src/pages/Dev.tsx +++ b/frontend/src/pages/Dev.tsx @@ -4,6 +4,7 @@ import { Loader2, Search, AlertTriangle, CheckCircle2, XCircle, FlaskConical, Ac import { PageHeader } from '@/components/PageHeader' import { api } from '@/lib/api' import { cn } from '@/lib/cn' +import { QK } from '@/lib/queryKeys' import { resetBadge } from '@/lib/monitorBadge' // ── 分钟K探测 (迁移自 MinuteDataProbe) ───────────────── @@ -216,7 +217,7 @@ function SeedPanel() { mutationFn: () => api.monitorRuleSeed(), onSuccess: (data) => { setMsg(`已生成 ${data.generated} 条监控规则`) - qc.invalidateQueries({ queryKey: ['monitor-rules'] }) + qc.invalidateQueries({ queryKey: QK.monitorRules }) setTimeout(() => setMsg(''), 4000) }, onError: () => { diff --git a/frontend/src/pages/LimitUpLadder.tsx b/frontend/src/pages/LimitUpLadder.tsx index 1d8ae20..6c6ef5d 100644 --- a/frontend/src/pages/LimitUpLadder.tsx +++ b/frontend/src/pages/LimitUpLadder.tsx @@ -1,4 +1,4 @@ -import { useState, useCallback, useMemo } from 'react' +import React, { useState, useCallback, useMemo } from 'react' import { useQuery } from '@tanstack/react-query' import { motion, AnimatePresence } from 'framer-motion' import { RefreshCw, ChevronDown, Flame, Settings2, X, Bell, BellOff, AlertCircle } from 'lucide-react' @@ -218,7 +218,7 @@ function useSealedDegrade(asOf: string, latestDate: string | undefined, sealedRe // ===== 单只股票卡片 ===== -function StockCard({ stock, extFields, direction, sealMode, monitored, monitorRule, onMonitorChange, hasDepth, onClick }: { +const StockCard = React.memo(function StockCard({ stock, extFields, direction, sealMode, monitored, monitorRule, onMonitorChange, hasDepth, onClick }: { stock: LimitLadderStock extFields: ExtFieldConfig direction: Direction @@ -227,7 +227,7 @@ function StockCard({ stock, extFields, direction, sealMode, monitored, monitorRu monitorRule?: MonitorRule onMonitorChange: () => void hasDepth: boolean - onClick: () => void + onClick: (symbol: string, name?: string) => void }) { const [showMonitorMenu, setShowMonitorMenu] = useState(false) const [menuAnchor, setMenuAnchor] = useState(null) @@ -287,7 +287,7 @@ function StockCard({ stock, extFields, direction, sealMode, monitored, monitorRu /> )} ) -} +}) // ===== 封单监控菜单 ===== @@ -1084,7 +1084,7 @@ function TierGroup({ tier, defaultOpen, extFields, filterKeys, bf, onStockClick, monitorRule={ladderRules.get(s.symbol)} onMonitorChange={onMonitorChange} hasDepth={hasDepth} - onClick={() => onStockClick(s.symbol, s.name ?? undefined)} + onClick={onStockClick} /> ))} @@ -1398,7 +1398,7 @@ export function LimitUpLadder() { // 连板梯队封单监控规则 (type=ladder): {symbol → rule} 映射 const { data: monitorRulesData, refetch: refetchMonitorRules } = useQuery({ - queryKey: ['monitor-rules'], + queryKey: QK.monitorRules, queryFn: () => api.monitorRulesList(), staleTime: 30 * 1000, }) @@ -1463,10 +1463,10 @@ export function LimitUpLadder() { storage.limitLadderExtFields.set(f) }, []) - const handleStockClick = (symbol: string, name?: string) => { + const handleStockClick = useCallback((symbol: string, name?: string) => { setPreviewSymbol(symbol) setPreviewName(name ?? '') - } + }, []) const extColumnsParam = useMemo(() => buildExtColumnsParam(extFields), [extFields]) diff --git a/frontend/src/pages/Monitor.tsx b/frontend/src/pages/Monitor.tsx index 82b4ef2..8221719 100644 --- a/frontend/src/pages/Monitor.tsx +++ b/frontend/src/pages/Monitor.tsx @@ -4,6 +4,7 @@ import { motion, AnimatePresence } from 'framer-motion' import { RadioTower, Plus, Trash2, Settings2, Zap, Bell, ListChecks, BellRing, TrendingUp, TrendingDown, Flame } from 'lucide-react' import { PageHeader } from '@/components/PageHeader' import { EmptyState } from '@/components/EmptyState' +import { Skeleton } from '@/components/data/Skeleton' import { api, type MonitorRule, type AlertEvent, type MonitorCondition } from '@/lib/api' import { QK } from '@/lib/queryKeys' import { fmtPrice, fmtPct } from '@/lib/format' @@ -240,7 +241,13 @@ function AlertsList({ alertsQuery, confirmClear, setConfirmClear, total, enterTs return (
- {events.length === 0 ? ( + {alertsQuery.isLoading ? ( +
+ {Array.from({ length: 4 }).map((_, i) => ( + + ))} +
+ ) : events.length === 0 ? ( - {rules.length === 0 ? ( + {rulesQuery.isLoading ? ( +
+ {Array.from({ length: 4 }).map((_, i) => ( + + ))} +
+ ) : rules.length === 0 ? (
} + if (kline.isError) { + return ( + + ) + } + const rows = kline.data?.rows ?? [] if (rows.length === 0) { return diff --git a/frontend/src/pages/Watchlist.tsx b/frontend/src/pages/Watchlist.tsx index 62be399..213a185 100644 --- a/frontend/src/pages/Watchlist.tsx +++ b/frontend/src/pages/Watchlist.tsx @@ -323,7 +323,10 @@ function RealtimeDot({ title = '实时监控中' }: { title?: string }) { // ===== 卡片组件 ===== -function StockCard({ +// 共享的空 K 线数组常量 — 避免每次渲染传入新的 [] 破坏 StockCard 的 memo +const EMPTY_KLINE: KlineRow[] = [] + +const StockCard = React.memo(function StockCard({ r, candleRows, showCandle, @@ -331,7 +334,7 @@ function StockCard({ onConfirmRemove, onCancelRemove, onRequestRemove, - confirmRemove, + isConfirming, extCols, expandedCells, onToggleExpand, @@ -344,7 +347,7 @@ function StockCard({ onConfirmRemove: (symbol: string) => void onCancelRemove: () => void onRequestRemove: (symbol: string) => void - confirmRemove: string | null + isConfirming: boolean extCols: ColumnConfig[] expandedCells: Set onToggleExpand: (key: string) => void @@ -379,7 +382,7 @@ function StockCard({ {/* 删除按钮 / 确认区 */}
- {confirmRemove === r.symbol ? ( + {isConfirming ? (
e.stopPropagation()}>
) -} +}) // ===== 主页面 ===== @@ -656,7 +659,7 @@ export function Watchlist() { // 2. 清除 list 缓存，触发后台 refetch qc.invalidateQueries({ queryKey: QK.watchlist }) qc.invalidateQueries({ queryKey: QK.watchlistEnriched() }) - qc.invalidateQueries({ queryKey: QK.watchlistKlineBatch('') }) + qc.invalidateQueries({ queryKey: ['watchlist-kline-batch'] }) }, }) @@ -680,7 +683,7 @@ export function Watchlist() { qc.setQueryData(['watchlist-enriched', extColumnsParam], { rows: [], as_of: null, elapsed_ms: 0 }) qc.invalidateQueries({ queryKey: QK.watchlist }) qc.invalidateQueries({ queryKey: QK.watchlistEnriched() }) - qc.invalidateQueries({ queryKey: QK.watchlistKlineBatch('') }) + qc.invalidateQueries({ queryKey: ['watchlist-kline-batch'] }) }, }) @@ -688,6 +691,16 @@ export function Watchlist() { const [confirmClear, setConfirmClear] = useState(false) const [confirmRemove, setConfirmRemove] = useState(null) + // 稳定的 per-symbol 回调 (供 memo 化的 StockCard 使用, 避免每次渲染都传新引用) + const handleCardPreview = useCallback((sym: string, name: string) => { + setPreviewSymbol(sym); setPreviewName(name) + }, []) + const handleCardConfirmRemove = useCallback((sym: string) => { + remove.mutate(sym); setConfirmRemove(null) + }, [remove]) + const handleCardCancelRemove = useCallback(() => setConfirmRemove(null), []) + const handleCardRequestRemove = useCallback((sym: string) => setConfirmRemove(sym), []) + const allSymbols = list.data?.symbols?.map(s => s.symbol) ?? [] const rows = enriched.data?.rows ?? [] @@ -1206,13 +1219,13 @@ export function Watchlist() { { setPreviewSymbol(sym); setPreviewName(name) }} - onConfirmRemove={(sym) => { remove.mutate(sym); setConfirmRemove(null) }} - onCancelRemove={() => setConfirmRemove(null)} - onRequestRemove={(sym) => setConfirmRemove(sym)} - confirmRemove={confirmRemove} + onPreview={handleCardPreview} + onConfirmRemove={handleCardConfirmRemove} + onCancelRemove={handleCardCancelRemove} + onRequestRemove={handleCardRequestRemove} + isConfirming={confirmRemove === r.symbol} extCols={visibleExtCols} expandedCells={expandedCells} onToggleExpand={handleToggleExpand} diff --git a/frontend/src/pages/backtest/StrategyBacktest.tsx b/frontend/src/pages/backtest/StrategyBacktest.tsx index e855394..e79bb39 100644 --- a/frontend/src/pages/backtest/StrategyBacktest.tsx +++ b/frontend/src/pages/backtest/StrategyBacktest.tsx @@ -775,7 +775,7 @@ export function StrategyBacktest() { }, [strategies.isLoading, strategyList, selectedStrategy]) const strategyDetail = useQuery({ - queryKey: ['strategy-detail', selectedStrategy], + queryKey: QK.strategyDetail(selectedStrategy ?? ''), queryFn: () => api.strategyGet(selectedStrategy!), enabled: !!selectedStrategy, }) @@ -1497,13 +1497,17 @@ export function StrategyBacktest() {
-
- {backtestTask?.progress - ? `回测中 · 第 ${backtestTask.progress.day}/${backtestTask.progress.total} 天 (${backtestTask.progress.date})` - : '正在重新计算回测…'} +
+ {backtestTask?.reconnecting + ? '连接中断，重试中…' + : backtestTask?.progress + ? `回测中 · 第 ${backtestTask.progress.day}/${backtestTask.progress.total} 天 (${backtestTask.progress.date})` + : '正在重新计算回测…'}
- {result ? '当前展示上次结果，完成后自动替换' : '正在加载回测数据…'} + {backtestTask?.reconnecting + ? '正在尝试恢复连接，若持续失败可停止后重试' + : result ? '当前展示上次结果，完成后自动替换' : '正在加载回测数据…'}
{backtestTask?.progress && ( diff --git a/frontend/src/pages/settings/CustomSignals.tsx b/frontend/src/pages/settings/CustomSignals.tsx index 56b2c9a..2bc7bc7 100644 --- a/frontend/src/pages/settings/CustomSignals.tsx +++ b/frontend/src/pages/settings/CustomSignals.tsx @@ -5,6 +5,7 @@ import { api, type CustomSignal } from '@/lib/api' import { QK } from '@/lib/queryKeys' import { BUILTIN_SIGNAL_DEFINITIONS, type SignalKind } from '@/lib/signals' import { CustomSignalDialog } from '@/components/signals/CustomSignalDialog' +import { Skeleton } from '@/components/data/Skeleton' type SignalSection = 'builtin' | 'custom' @@ -242,7 +243,15 @@ export function SettingsCustomSignalsPanel() {
))} - {signals.length === 0 && ( + {list.isLoading && + Array.from({ length: 2 }).map((_, i) => ( +
+ + + +
+ ))} + {!list.isLoading && signals.length === 0 && (
暂无自定义信号，点击右上角「新建自定义信号」。
diff --git a/frontend/src/pages/settings/ExtPages.tsx b/frontend/src/pages/settings/ExtPages.tsx index be7ee85..6877ad9 100644 --- a/frontend/src/pages/settings/ExtPages.tsx +++ b/frontend/src/pages/settings/ExtPages.tsx @@ -4,6 +4,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import { ExternalLink, Pencil, Plus, Save, Trash2, X } from 'lucide-react' import { api, type AnalysisColumn, type AnalysisMenu, type ExtDataConfig, type ExtDataField } from '@/lib/api' import { QK } from '@/lib/queryKeys' +import { Skeleton } from '@/components/data/Skeleton' function dtypeToColumnType(dtype: string): AnalysisColumn['type'] { return dtype === 'int' || dtype === 'float' ? 'number' : 'string' @@ -282,7 +283,15 @@ export function SettingsExtPagesPanel() { ))} - {menuItems.length === 0 && ( + {menus.isLoading && + Array.from({ length: 3 }).map((_, i) => ( +
+ + + +
+ ))} + {!menus.isLoading && menuItems.length === 0 && (
暂无扩展页面，点击右上角新建。
)} diff --git a/frontend/src/router.tsx b/frontend/src/router.tsx index ef4d6b1..c7df674 100644 --- a/frontend/src/router.tsx +++ b/frontend/src/router.tsx @@ -1,28 +1,33 @@ +import { lazy } from 'react' import { createBrowserRouter, Navigate } from 'react-router-dom' import { Layout } from './components/Layout' -import { Watchlist } from './pages/Watchlist' -import { Screener } from './pages/Screener' -import { Backtest } from './pages/Backtest' -import { Financials } from './pages/Financials' import { Onboarding } from './pages/Onboarding' import { Auth } from './pages/Auth' -import { Data } from './pages/Data' -import { Monitor } from './pages/Monitor' -import { Trading } from './pages/Trading' -import { Dashboard } from './pages/Dashboard' -import { AnalysisDetail } from './pages/AnalysisDetail' -import { ConceptAnalysis } from './pages/ConceptAnalysis' -import { IndustryAnalysis } from './pages/IndustryAnalysis' -import { StockAnalysis } from './pages/StockAnalysis' -import { Review } from './pages/Review' -import { LimitUpLadder } from './pages/LimitUpLadder' -import { Branding } from './pages/Branding' -import { Settings } from './pages/Settings' -import { Indices } from './pages/Indices' -import { Dev } from './pages/Dev' import { useSettings } from './lib/useSharedQueries' import { Logo } from './components/Logo' +// 代码分割: 页面全部 lazy 加载, 避免首屏打包所有页面 (ECharts / lightweight-charts / +// framer-motion 等重库) → 大幅减小首屏 bundle。命名导出用 .then 映射为 default。 +// Layout / Onboarding / Auth 为应用外壳与入口, 保持同步加载。 +const Watchlist = lazy(() => import('./pages/Watchlist').then(m => ({ default: m.Watchlist }))) +const Screener = lazy(() => import('./pages/Screener').then(m => ({ default: m.Screener }))) +const Backtest = lazy(() => import('./pages/Backtest').then(m => ({ default: m.Backtest }))) +const Financials = lazy(() => import('./pages/Financials').then(m => ({ default: m.Financials }))) +const Data = lazy(() => import('./pages/Data').then(m => ({ default: m.Data }))) +const Monitor = lazy(() => import('./pages/Monitor').then(m => ({ default: m.Monitor }))) +const Trading = lazy(() => import('./pages/Trading').then(m => ({ default: m.Trading }))) +const Dashboard = lazy(() => import('./pages/Dashboard').then(m => ({ default: m.Dashboard }))) +const AnalysisDetail = lazy(() => import('./pages/AnalysisDetail').then(m => ({ default: m.AnalysisDetail }))) +const ConceptAnalysis = lazy(() => import('./pages/ConceptAnalysis').then(m => ({ default: m.ConceptAnalysis }))) +const IndustryAnalysis = lazy(() => import('./pages/IndustryAnalysis').then(m => ({ default: m.IndustryAnalysis }))) +const StockAnalysis = lazy(() => import('./pages/StockAnalysis').then(m => ({ default: m.StockAnalysis }))) +const Review = lazy(() => import('./pages/Review').then(m => ({ default: m.Review }))) +const LimitUpLadder = lazy(() => import('./pages/LimitUpLadder').then(m => ({ default: m.LimitUpLadder }))) +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 Dev = lazy(() => import('./pages/Dev').then(m => ({ default: m.Dev }))) + // 首次使用守卫 —— 未完成向导则重定向到 /onboarding // 只挂在根路由上;/onboarding 本身不被守卫,避免循环重定向。 // settings 由 Layout 预取,守卫判定不产生额外请求。 diff --git a/frontend/vite.config.js b/frontend/vite.config.js index 50ac7e1..50f463f 100644 --- a/frontend/vite.config.js +++ b/frontend/vite.config.js @@ -32,5 +32,19 @@ export default defineConfig({ build: { outDir: 'dist', sourcemap: false, + rollupOptions: { + output: { + // 把重型图表库拆到独立 chunk, 避免打进主包 + 让页面按需加载。 + // 用函数形式按 node_modules 路径匹配, 比对象形式更可靠。 + manualChunks(id) { + if (id.includes('node_modules')) { + if (id.includes('echarts')) + return 'echarts'; + if (id.includes('lightweight-charts')) + return 'lightweight-charts'; + } + }, + }, + }, }, }); diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts index 1451aaf..554d93d 100644 --- a/frontend/vite.config.ts +++ b/frontend/vite.config.ts @@ -33,5 +33,17 @@ export default defineConfig({ build: { outDir: 'dist', sourcemap: false, + rollupOptions: { + output: { + // 把重型图表库拆到独立 chunk, 避免打进主包 + 让页面按需加载。 + // 用函数形式按 node_modules 路径匹配, 比对象形式更可靠。 + manualChunks(id) { + if (id.includes('node_modules')) { + if (id.includes('echarts')) return 'echarts' + if (id.includes('lightweight-charts')) return 'lightweight-charts' + } + }, + }, + }, }, })