改进: 并发韧性 + 数据性能 + 死代码清理 + 前端 UX + ST/Sharpe 修复 (#78)

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jinfeng Sun
2026-07-08 18:10:19 +08:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 5732155ec3
commit 9aa96edbd7
46 changed files with 1352 additions and 624 deletions
+17 -5
View File
@@ -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}
+3 -24
View File
@@ -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)
+17 -8
View File
@@ -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
+40 -14
View File
@@ -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")
+19 -10
View File
@@ -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:
+60 -42
View File
@@ -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
+215 -109
View File
@@ -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)
+37 -29
View File
@@ -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(
+9 -60
View File
@@ -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()
+16 -3
View File
@@ -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")
+21 -4
View File
@@ -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)
-52
View File
@@ -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()
+126
View File
@@ -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")
+10 -56
View File
@@ -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)
+48 -1
View File
@@ -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
+7 -54
View File
@@ -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)
+31 -3
View File
@@ -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)
@@ -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)
)
+18 -5
View File
@@ -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:
+37 -5
View File
@@ -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/rpmnow 早于该 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:
+35
View File
@@ -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。
+7 -5
View File
@@ -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)
+88
View File
@@ -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
+17 -3
View File
@@ -102,7 +102,12 @@ export function AlertToastContainer() {
if (!items.length) return null
return (
<div className="fixed bottom-4 right-4 z-[9999] flex flex-col gap-2 w-[320px] pointer-events-none">
<div
role="status"
aria-live="polite"
aria-atomic="false"
className="fixed bottom-4 right-4 z-[9999] flex flex-col gap-2 w-[320px] pointer-events-none"
>
<AnimatePresence>
{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"
>
{/* 左侧色条 */}
<div className={cn('absolute left-0 top-0 h-full w-0.5', sev)} />
@@ -143,7 +157,7 @@ export function AlertToastContainer() {
{fmtPct(pct)}
</span>
)}
<button onClick={(e) => { e.stopPropagation(); dismiss(item.id) }} className="shrink-0 p-0.5 rounded text-muted/50 hover:text-foreground hover:bg-elevated transition-colors cursor-pointer">
<button aria-label="关闭通知" onClick={(e) => { e.stopPropagation(); dismiss(item.id) }} className="shrink-0 p-0.5 rounded text-muted/50 hover:text-foreground hover:bg-elevated transition-colors cursor-pointer">
<X className="h-3 w-3" />
</button>
</div>
+24 -3
View File
@@ -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"
>
<Outlet />
{streamStatus === 'reconnecting' && (
<div
role="status"
aria-live="polite"
className="fixed top-3 right-4 z-[9998] flex items-center gap-1.5 rounded-full border border-warning/30 bg-warning/10 px-2.5 py-1 text-[11px] font-medium text-warning shadow-lg backdrop-blur-md"
>
<WifiOff className="h-3 w-3 shrink-0 animate-pulse" />
·
</div>
)}
<Suspense
fallback={
<div className="flex items-center justify-center py-24">
<Loader2 className="h-5 w-5 animate-spin text-muted" />
</div>
}
>
<Outlet />
</Suspense>
</motion.main>
<ToastContainer />
<AlertToastContainer />
+137
View File
@@ -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<HTMLElement>
/** 点击遮罩是否关闭 (默认 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<HTMLDivElement>(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<HTMLElement>(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<HTMLElement>(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 (
<div
className={overlayClassName}
onClick={closeOnBackdrop ? (e) => { if (e.target === e.currentTarget) onClose() } : undefined}
>
<div
ref={panelRef}
role="dialog"
aria-modal="true"
aria-labelledby={labelledBy}
aria-label={labelledBy ? undefined : ariaLabel}
tabIndex={-1}
className={`outline-none ${panelClassName}`}
onClick={(e) => e.stopPropagation()}
>
{children}
</div>
</div>
)
}
+10 -26
View File
@@ -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 (
<AnimatePresence>
<motion.div
initial={{ opacity: 0 }}
animate={{ opacity: 1 }}
exit={{ opacity: 0 }}
className="fixed inset-0 z-50 flex items-center justify-center bg-black/50"
onClick={e => { if (e.target === e.currentTarget) onClose() }}
>
<motion.div
initial={{ opacity: 0, scale: 0.95, y: 10 }}
animate={{ opacity: 1, scale: 1, y: 0 }}
exit={{ opacity: 0, scale: 0.95, y: 10 }}
transition={{ duration: 0.15, ease: [0.16, 1, 0.3, 1] }}
className="w-[92vw] max-w-[1100px] h-[88vh] bg-surface border border-border rounded-card shadow-xl flex flex-col"
>
<Modal
onClose={onClose}
labelledBy="rps-rotation-title"
overlayClassName="fixed inset-0 z-50 flex items-center justify-center bg-black/50"
panelClassName="w-[92vw] max-w-[1100px] h-[88vh] bg-surface border border-border rounded-card shadow-xl flex flex-col"
>
{/* 标题栏 */}
<div className="flex items-center justify-between px-4 py-2.5 border-b border-border shrink-0">
<div className="flex items-center gap-2">
<Repeat className="h-4 w-4 text-accent" />
<span className="text-sm font-medium text-foreground"></span>
<span id="rps-rotation-title" className="text-sm font-medium text-foreground"></span>
<span className="text-[11px] text-muted">
{conceptCount > 0 ? `${dates.length} 天 · ${conceptCount} 个概念` : '暂无数据'}
</span>
</div>
<button onClick={onClose} className="p-1 rounded hover:bg-elevated transition-colors cursor-pointer">
<button aria-label="关闭" onClick={onClose} className="p-1 rounded hover:bg-elevated transition-colors cursor-pointer">
<X className="h-4 w-4 text-muted" />
</button>
</div>
@@ -438,8 +424,6 @@ export function RpsRotationDialog({ onClose }: Props) {
·
</span>
</div>
</motion.div>
</motion.div>
</AnimatePresence>
</Modal>
)
}
+6 -1
View File
@@ -31,7 +31,12 @@ export function ToastContainer() {
if (!items.length) return null
return (
<div className="fixed bottom-4 right-4 z-[9999] flex flex-col gap-2 pointer-events-none">
<div
role="status"
aria-live="polite"
aria-atomic="false"
className="fixed bottom-4 right-4 z-[9999] flex flex-col gap-2 pointer-events-none"
>
{items.map(t => (
<div
key={t.id}
@@ -1,5 +1,5 @@
import { useState, useEffect, useCallback } from 'react'
import { motion, AnimatePresence } from 'framer-motion'
import { Modal } from '@/components/Modal'
import { X, Sparkles, Save, Loader2, ChevronLeft, ChevronRight, AlertTriangle, Settings2, FileText, Copy, Check, Terminal } from 'lucide-react'
import { api } from '@/lib/api'
import { storage } from '@/lib/storage'
@@ -253,12 +253,12 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create
const hasParams = params.length > 0 || entrySignals.length > 0 || exitSignals.length > 0 || Object.keys(scoring).length > 0
return (
<AnimatePresence>
<motion.div initial={{ opacity: 0 }} animate={{ opacity: 1 }} exit={{ opacity: 0 }}
className="fixed inset-0 z-50 flex items-center justify-center bg-black/40 backdrop-blur-sm"
onClick={e => { if (e.target === e.currentTarget) handleClose() }}>
<motion.div initial={{ opacity: 0, scale: 0.95, y: 10 }} animate={{ opacity: 1, scale: 1, y: 0 }} exit={{ opacity: 0, scale: 0.95, y: 10 }}
className="w-[820px] max-h-[88vh] bg-surface/95 backdrop-blur-xl border border-border/50 rounded-2xl shadow-2xl flex flex-col overflow-hidden">
<Modal
onClose={handleClose}
labelledBy="strategy-builder-title"
overlayClassName="fixed inset-0 z-50 flex items-center justify-center bg-black/40 backdrop-blur-sm"
panelClassName="w-[820px] max-h-[88vh] bg-surface/95 backdrop-blur-xl border border-border/50 rounded-2xl shadow-2xl flex flex-col overflow-hidden"
>
{/* 标题 */}
<div className="grid grid-cols-[1fr_auto_1fr] items-center px-5 py-2.5 border-b border-border/50">
@@ -272,7 +272,7 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create
</button>
</div>
{/* 中间:标题 */}
<span className="text-sm font-semibold text-foreground">
<span id="strategy-builder-title" className="text-sm font-semibold text-foreground">
{strategyId ? '修改策略' : '创建策略'}
</span>
{/* 右侧:步骤 + 关闭 */}
@@ -284,7 +284,7 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create
<span className={'w-5 h-5 rounded-full flex items-center justify-center text-[10px] font-bold ' + (step === 2 ? 'bg-amber-400/20 text-amber-400' : 'bg-border/50 text-muted')}>2</span>
</div>
)}
<button onClick={handleClose} className="p-1.5 rounded-lg hover:bg-elevated"><X className="h-4 w-4 text-muted" /></button>
<button aria-label="关闭" onClick={handleClose} className="p-1.5 rounded-lg hover:bg-elevated"><X className="h-4 w-4 text-muted" /></button>
</div>
</div>
@@ -506,8 +506,6 @@ export function StrategyBuilderDialog({ open, onClose, onSavedId, mode = 'create
</div>
</div>
)}
</motion.div>
</motion.div>
</AnimatePresence>
</Modal>
)
}
@@ -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<string, string> = {}
@@ -320,30 +321,21 @@ export function StrategySettingsDialog({ strategyId, onClose, onSaved, onAiModif
return (
<>
<AnimatePresence>
<motion.div
initial={{ opacity: 0 }}
animate={{ opacity: 1 }}
exit={{ opacity: 0 }}
className="fixed inset-0 z-50 flex items-center justify-center bg-black/40 backdrop-blur-sm"
onClick={e => { if (e.target === e.currentTarget) onClose() }}
>
<motion.div
initial={{ opacity: 0, scale: 0.95, y: 10 }}
animate={{ opacity: 1, scale: 1, y: 0 }}
exit={{ opacity: 0, scale: 0.95, y: 10 }}
transition={{ duration: 0.2, ease: [0.16, 1, 0.3, 1] }}
className="w-[980px] max-h-[88vh] bg-surface/95 backdrop-blur-xl border border-border/50 rounded-2xl shadow-2xl flex flex-col overflow-hidden"
>
<Modal
onClose={onClose}
labelledBy="strategy-settings-title"
overlayClassName="fixed inset-0 z-50 flex items-center justify-center bg-black/40 backdrop-blur-sm"
panelClassName="w-[980px] max-h-[88vh] bg-surface/95 backdrop-blur-xl border border-border/50 rounded-2xl shadow-2xl flex flex-col overflow-hidden"
>
{/* 标题 */}
<div className="flex items-center justify-between px-5 py-3 border-b border-border/50">
<div className="flex items-center gap-2.5">
<Settings2 className="h-4 w-4 text-accent" />
<span className="text-sm font-semibold text-foreground">{detail?.name ?? strategyId}</span>
<span id="strategy-settings-title" className="text-sm font-semibold text-foreground">{detail?.name ?? strategyId}</span>
{detail && <span className="text-[10px] px-1.5 py-0.5 rounded bg-elevated text-muted">{{ builtin: '内置', custom: '自定义', ai: 'AI' }[detail.source] ?? detail.source}</span>}
<span className="text-[10px] text-muted/40 font-mono">{strategyId}</span>
</div>
<button onClick={onClose} className="p-1.5 rounded-lg hover:bg-elevated transition-colors cursor-pointer"><X className="h-4 w-4 text-muted" /></button>
<button aria-label="关闭" onClick={onClose} className="p-1.5 rounded-lg hover:bg-elevated transition-colors cursor-pointer"><X className="h-4 w-4 text-muted" /></button>
</div>
{/* 内容 */}
@@ -568,11 +560,10 @@ export function StrategySettingsDialog({ strategyId, onClose, onSaved, onAiModif
</button>
</div>
</div>
</motion.div>
</motion.div>
</AnimatePresence>
</Modal>
{/* 删除确认弹窗 */}
{/* Modal : Modal backdrop-blur ( fixed )
+ overflow-hidden, / */}
{showDeleteConfirm && (
<AnimatePresence>
<motion.div
@@ -615,5 +606,6 @@ export function StrategySettingsDialog({ strategyId, onClose, onSaved, onAiModif
</AnimatePresence>
)}
</>
)
}
+50 -9
View File
@@ -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<void> {
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
+1
View File
@@ -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,
+63 -2
View File
@@ -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])
}
+10 -1
View File
@@ -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() {
</Link>
</div>
))}
{menuItems.length === 0 && (
{menus.isLoading &&
Array.from({ length: 3 }).map((_, i) => (
<div key={`sk-${i}`} className="rounded-card border border-border bg-surface p-4 space-y-3">
<Skeleton w="w-1/2" h="h-4" />
<Skeleton w="w-1/3" h="h-3" />
<Skeleton h="h-8" rounded="rounded-btn" />
</div>
))}
{!menus.isLoading && menuItems.length === 0 && (
<div className="rounded-card border border-border bg-surface px-5 py-10 text-center text-sm text-muted md:col-span-2 xl:col-span-3"></div>
)}
</section>
+2
View File
@@ -96,6 +96,8 @@ export function Data() {
const clearData = useMutation({
mutationFn: api.dataClear,
onSuccess: () => {
// 清库删除全部 parquet + alerts, 各页 (看板/自选/选股/指数/财务/个股/连板/监控) 缓存
// 数据均已失效, 故广域失效所有 query 令其重取 —— 数据已清空, 全量刷新是正确行为而非误伤。
qc.invalidateQueries()
setShowClearConfirm(false)
},
+2 -1
View File
@@ -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: () => {
+9 -9
View File
@@ -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<DOMRect | null>(null)
@@ -287,7 +287,7 @@ function StockCard({ stock, extFields, direction, sealMode, monitored, monitorRu
/>
)}
<button
onClick={onClick}
onClick={() => onClick(stock.symbol, stock.name ?? undefined)}
className={`w-full flex flex-col items-start gap-1 px-2.5 py-2 rounded-md transition-all duration-200 cursor-pointer hover:opacity-100 ${style.bg} ${style.bar} ${monitored ? 'ring-1 ring-amber-400/50 ring-inset' : ''}`}
style={style.cardStyle ? { ...style.cardStyle } : undefined}
onMouseEnter={e => {
@@ -357,7 +357,7 @@ function StockCard({ stock, extFields, direction, sealMode, monitored, monitorRu
</button>
</div>
)
}
})
// ===== 封单监控菜单 =====
@@ -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}
/>
))}
</div>
@@ -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])
+15 -2
View File
@@ -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 (
<div className="space-y-3">
{events.length === 0 ? (
{alertsQuery.isLoading ? (
<div className="space-y-2">
{Array.from({ length: 4 }).map((_, i) => (
<Skeleton key={i} h="h-14" rounded="rounded-card" />
))}
</div>
) : events.length === 0 ? (
<EmptyState
icon={Bell}
title="暂无触发记录"
@@ -514,7 +521,13 @@ function RulesList({ rulesQuery, onEdit }: {
return (
<div className="space-y-2.5">
{rules.length === 0 ? (
{rulesQuery.isLoading ? (
<div className="space-y-2">
{Array.from({ length: 4 }).map((_, i) => (
<Skeleton key={i} h="h-16" rounded="rounded-card" />
))}
</div>
) : rules.length === 0 ? (
<EmptyState
icon={RadioTower}
title="暂无监控规则"
+11 -1
View File
@@ -1,6 +1,6 @@
import { useState, useEffect } from 'react'
import { useQuery } from '@tanstack/react-query'
import { Sparkles, LineChart, History as HistoryIcon, Loader2, ExternalLink, Bell } from 'lucide-react'
import { Sparkles, LineChart, History as HistoryIcon, Loader2, ExternalLink, Bell, AlertTriangle } from 'lucide-react'
import { PageHeader } from '@/components/PageHeader'
import { EmptyState } from '@/components/EmptyState'
import { StockFinancialSearch } from '@/components/financials/StockFinancialSearch'
@@ -189,6 +189,16 @@ function StockAnalysisBoard({ symbol }: { symbol: string }) {
return <div className="flex items-center justify-center py-20"><Loader2 className="h-5 w-5 animate-spin text-muted" /></div>
}
if (kline.isError) {
return (
<EmptyState
icon={AlertTriangle}
title="日 K 数据加载失败"
hint="请检查网络或数据源配置后重试。"
/>
)
}
const rows = kline.data?.rows ?? []
if (rows.length === 0) {
return <EmptyState icon={LineChart} title="暂无日 K 数据" hint="该标的尚未同步日 K,请先在数据页或自选页同步。" />
+26 -13
View File
@@ -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<string>
onToggleExpand: (key: string) => void
@@ -379,7 +382,7 @@ function StockCard({
{/* 删除按钮 / 确认区 */}
<div className="absolute top-1.5 right-1.5 z-10">
{confirmRemove === r.symbol ? (
{isConfirming ? (
<div className="flex items-center gap-1" onClick={e => e.stopPropagation()}>
<button
onClick={() => onConfirmRemove(r.symbol)}
@@ -488,7 +491,7 @@ function StockCard({
)}
</div>
)
}
})
// ===== 主页面 =====
@@ -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<string | null>(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() {
<StockCard
key={r.symbol}
r={r}
candleRows={klineData[r.symbol] ?? []}
candleRows={klineData[r.symbol] ?? EMPTY_KLINE}
showCandle={dailyKVisible}
onPreview={(sym, name) => { 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}
@@ -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() {
<Loader2 className="relative h-4 w-4 animate-spin text-accent" />
</span>
<div className="min-w-0">
<div className="text-xs font-medium text-accent">
{backtestTask?.progress
? `回测中 · 第 ${backtestTask.progress.day}/${backtestTask.progress.total} 天 (${backtestTask.progress.date})`
: '正在重新计算回测…'}
<div className={backtestTask?.reconnecting ? 'text-xs font-medium text-warning' : 'text-xs font-medium text-accent'}>
{backtestTask?.reconnecting
? '连接中断,重试中…'
: backtestTask?.progress
? `回测中 · 第 ${backtestTask.progress.day}/${backtestTask.progress.total} 天 (${backtestTask.progress.date})`
: '正在重新计算回测…'}
</div>
<div className="mt-0.5 text-[11px] text-secondary">
{result ? '当前展示上次结果,完成后自动替换' : '正在加载回测数据…'}
{backtestTask?.reconnecting
? '正在尝试恢复连接,若持续失败可停止后重试'
: result ? '当前展示上次结果,完成后自动替换' : '正在加载回测数据…'}
</div>
</div>
{backtestTask?.progress && (
+10 -1
View File
@@ -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() {
</div>
</div>
))}
{signals.length === 0 && (
{list.isLoading &&
Array.from({ length: 2 }).map((_, i) => (
<div key={`sk-${i}`} className="rounded-card border border-border bg-base p-4 space-y-3">
<Skeleton w="w-1/2" h="h-4" />
<Skeleton w="w-1/3" h="h-3" />
<Skeleton h="h-4" />
</div>
))}
{!list.isLoading && signals.length === 0 && (
<div className="rounded-card border border-border bg-base px-5 py-10 text-center text-sm text-muted md:col-span-2">
</div>
+10 -1
View File
@@ -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() {
</Link>
</div>
))}
{menuItems.length === 0 && (
{menus.isLoading &&
Array.from({ length: 3 }).map((_, i) => (
<div key={`sk-${i}`} className="rounded-card border border-border bg-surface p-4 space-y-3">
<Skeleton w="w-1/2" h="h-4" />
<Skeleton w="w-1/3" h="h-3" />
<Skeleton h="h-8" rounded="rounded-btn" />
</div>
))}
{!menus.isLoading && menuItems.length === 0 && (
<div className="rounded-card border border-border bg-surface px-5 py-10 text-center text-sm text-muted md:col-span-2 xl:col-span-3"></div>
)}
</section>
+23 -18
View File
@@ -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 预取,守卫判定不产生额外请求。
+14
View File
@@ -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';
}
},
},
},
},
});
+12
View File
@@ -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'
}
},
},
},
},
})