aeroxw/tick-stock-panel
1
0
Fork 0
mirror of https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git synced 2026-09-12 17:54:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(realtime): 自选实时分流与数据落盘可靠性

Browse Source 
- quote_service 实时通道重构与自选分流, 旧 watchlist 批量测试相应移除
- repository.replace_with_retry 原子写重试; pipeline_jobs 任务记录落盘
- stock-sdk provider 适配增强; 分时图组件与预览入口调整
This commit is contained in:
shy3130
2026-08-30 19:05:27 +08:00
parent d70a960750
commit 3b09fac15e
16 changed files with 653 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/plugins/stocksdk/provider.py
+92 -20
View File
@@ -12,7 +12,8 @@ from __future__ import annotations
import logging
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime
from datetime import datetime, timedelta
from datetime import time as dtime
import polars as pl
@@ -51,6 +52,9 @@ class StockSDKProvider:
name = "stocksdk"
builtin = True
# 分钟历史深度能力(可选声明, 未声明视为深历史): stock-sdk 免费分时接口
# 只保留最近 5 个交易日的 1 分钟数据, 分时档位/默认值据此收窄。
minute_history_days = 5
def __init__(self) -> None:
self.config = _StockSDKConfig()
@@ -145,29 +149,96 @@ class StockSDKProvider:
return pl.DataFrame()
period = "".join(ch for ch in str(freq) if ch.isdigit()) or "1"
logger.info("stock-sdk minute 拉取开始(%d symbols, period=%s)", len(symbols), period)
# 上游区间查询的分钟 open 为日级常量(伪值), 单日查询才给最新交易日真实
# 分钟 open → 末尾 3 个自然日逐日单拉(跳过周末, 覆盖周五收盘后场景),
# 其余历史段仍走单个区间任务控制桥接成本。伪 open 由 _null_degenerate_opens
# 在 _minute_df 内置 null。
windows: list[tuple[datetime | None, datetime | None]] = []
if (
start_time is not None
and end_time is not None
and end_time.date() > start_time.date()
):
tail_start = end_time.date() - timedelta(days=3)
if tail_start > start_time.date():
head_end = datetime.combine(tail_start - timedelta(days=1), dtime.max)
windows.append((start_time, head_end))
else:
tail_start = start_time.date()
day = tail_start
while day <= end_time.date():
if day.weekday() < 5:
windows.append((
datetime.combine(day, dtime.min),
datetime.combine(day, dtime.max),
))
day += timedelta(days=1)
else:
windows.append((start_time, end_time))
frames: list[pl.DataFrame] = []
chunks = chunked(symbols, _BATCH)
for i, chunk in enumerate(chunks):
job = {
"op": "minute",
"symbols": chunk,
"period": period,
"start": _yyyymmdd(start_time),
"end": _yyyymmdd(end_time),
}
try:
result = bridge.run_job(job, timeout=180)
except bridge.StockSDKBridgeError as e:
logger.warning("stock-sdk minute 拉取失败(%d symbols): %s", len(chunk), e)
result = {"rows": {}}
for sym, rows in (result.get("rows") or {}).items():
df = self._minute_df(rows, sym)
if not df.is_empty():
frames.append(df)
if on_chunk_done:
on_chunk_done(i + 1, len(chunks))
total = len(chunks) * len(windows)
step = 0
for win_start, win_end in windows:
for chunk in chunks:
step += 1
job = {
"op": "minute",
"symbols": chunk,
"period": period,
"start": _yyyymmdd(win_start),
"end": _yyyymmdd(win_end),
}
try:
result = bridge.run_job(job, timeout=180)
except bridge.StockSDKBridgeError as e:
logger.warning("stock-sdk minute 拉取失败(%d symbols): %s", len(chunk), e)
result = {"rows": {}}
for sym, rows in (result.get("rows") or {}).items():
df = self._minute_df(rows, sym)
if not df.is_empty():
frames.append(df)
if on_chunk_done:
on_chunk_done(step, total)
# 末窗口(最新一日)的真实 open 与首窗口可能重叠同日(时区/边界), keep="last"
# 由上层 _write_minute_partition 的 unique 处理; 这里仅拼接。
return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame()
@staticmethod
def _null_degenerate_opens(df: pl.DataFrame) -> pl.DataFrame:
"""把"日级常量"的伪分钟 open 置 null。
stock-sdk 上游对历史日的分钟 open 只给全天常量(如涨跌停价/日开),
并非真实分钟开盘价; 只有最新交易日在单日查询下给真实值。伪 open 入库
会让分钟K的 close-vs-open 口径全偏(如分时量恒红), 故按日检测:
同日 rows>10 且 open 唯一值<=3 而 close 唯一值>10 → open 判定非分钟级,
置 null(fail-closed, 不伪造 prev_close 替代)。
"""
if df.is_empty() or "open" not in df.columns or "datetime" not in df.columns:
return df
stats = df.group_by(pl.col("datetime").dt.date()).agg(
pl.len().alias("n"),
pl.col("open").n_unique().alias("uo"),
pl.col("close").n_unique().alias("uc"),
)
fake_dates = stats.filter(
(pl.col("n") > 10) & (pl.col("uo") <= 3) & (pl.col("uc") > 10)
)["datetime"]
if fake_dates.is_empty():
return df
logger.warning(
"stock-sdk minute open 为日级常量, 置 null: %s %s",
df["symbol"][0] if "symbol" in df.columns else "?", fake_dates.to_list(),
)
return df.with_columns(
pl.when(pl.col("datetime").dt.date().is_in(fake_dates))
.then(None)
.otherwise(pl.col("open"))
.alias("open")
)
@staticmethod
def _minute_df(rows: list[dict], symbol: str) -> pl.DataFrame:
if not rows:
@@ -192,6 +263,7 @@ class StockSDKProvider:
for col in ("open", "high", "low", "close", "volume", "amount"):
if col in df.columns:
df = df.with_columns(pl.col(col).cast(pl.Float64, strict=False))
df = StockSDKProvider._null_degenerate_opens(df)
keep = [c for c in _MINUTE_CANONICAL if c in df.columns]
return df.select(keep) if "datetime" in keep else pl.DataFrame()
backend/app/services/pipeline_jobs.py
+52 -5
View File
@@ -2,8 +2,12 @@
设计:
- job_store/ 文件夹,每个 job 一个 {id}.json,最多保留 max_jobs 个文件
- running/pending 状态的 job 仅存内存(高频读写)
- succeeded/failed 后写入独立文件并从内存释放
- create()/start() 即落盘 pending/running 快照(进度只更新内存) ——
进程意外死亡(uvicorn --reload 热重载 / 被 kill)时记录不蒸发
- succeeded/failed 后写入终态并从内存释放
- 实例化时扫描磁盘,把上个进程遗留的 pending/running 孤儿记录补标为
failed(中断);finished_at 取文件 mtime(最后已知存活时刻),不用下次
开机时间虚增时长
- 列表查询 = 内存中的活跃 job + 磁盘文件扫描,按时间排序
- 单个查询 = 内存优先,没有则读磁盘
- 创建新 job 前检查文件数量,>= max_jobs 时删除最老的文件
@@ -15,7 +19,7 @@ import logging
import os
import threading
import uuid
from datetime import datetime
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Literal
@@ -105,11 +109,12 @@ class JobStore:
self._active_id: str | None = None
self._lock = threading.Lock()
self._store_dir.mkdir(parents=True, exist_ok=True)
self._reap_orphans()
# ===== persistence =====
def _write_file(self, job: dict[str, Any]) -> None:
"""将终态 job 写入独立 JSON 文件。"""
"""将 job 快照写入独立 JSON 文件(create/start/终态均落盘)。"""
path = self._store_dir / f"{job['id']}.json"
try:
path.write_text(
@@ -154,6 +159,42 @@ class JobStore:
jobs.sort(key=lambda j: j.get("started_at") or "", reverse=True)
return jobs
def _reap_orphans(self) -> None:
"""启动补录: 把上个进程遗留的 pending/running 记录标为中断。
单例在进程启动时实例化,此时磁盘上的 pending/running 必然来自已死
亡的进程(uvicorn --reload 热重载 / 被 kill)—— 不补录则这些记录
永远停留在「运行中」, 同步历史里既看不到结果也看不到失败, 即
「数据在、记录丢」。finished_at 取文件 mtime(最后一次落盘 = 最后
已知存活时刻), 避免拿下次开机时间虚增 duration。
"""
for f in self._store_dir.glob("*.json"):
try:
j = json.loads(f.read_text("utf-8"))
except Exception:
continue
orig_status = j.get("status")
if not j.get("id") or orig_status not in ("pending", "running"):
continue
j["status"] = "failed"
j["error"] = "后端重启,任务中断(启动时补录)"
try:
mtime = datetime.fromtimestamp(f.stat().st_mtime, tz=UTC)
end = mtime.strftime("%Y-%m-%dT%H:%M:%SZ")
except Exception:
end = datetime.utcnow().isoformat(timespec="seconds") + "Z"
# mtime 早于 started_at(时钟回拨等)时夹住, 避免 duration 为负
if j.get("started_at") and end < j["started_at"]:
end = j["started_at"]
j["finished_at"] = end
j["duration_s"] = _duration_s(j)
logger.warning(
"job_store: 补录中断任务 %s (上个进程遗留 %s 记录)",
j["id"],
orig_status,
)
self._write_file(j)
# ===== lifecycle =====
def create(
@@ -189,7 +230,7 @@ class JobStore:
return self._active_id, False
job_id = uuid.uuid4().hex[:10]
self._active_jobs[job_id] = {
job = {
"id": job_id,
"status": "pending",
"stage": "init",
@@ -204,7 +245,11 @@ class JobStore:
"error": None,
"timeout_s": timeout_s,
}
self._active_jobs[job_id] = job
self._active_id = job_id
# pending 即落盘: 进程在 start() 前死亡时记录也不丢
self._delete_oldest()
self._write_file(job)
_register_cancel_flag(job_id)
return job_id, True
@@ -218,6 +263,8 @@ class JobStore:
# 心跳基准初始化为启动时刻: start() 到首次 progress() 之间的
# 初始化阶段(解析标的池等)同样计入停滞计时。
j["last_progress_at"] = j["started_at"]
# running 快照落盘: 进程死亡后由下次启动的 _reap_orphans 补录
self._write_file(j)
def succeed(self, job_id: str, result: Any) -> None:
with self._lock:
backend/app/services/quote_service.py
+9 -150
View File
@@ -425,15 +425,18 @@ class QuoteService:
@classmethod
def realtime_mode(cls) -> str:
"""当前实时行情模式: none / watchlist / full_market。"""
"""当前实时行情模式: none / full_market。
TickFlow 免费档不再提供"自选前 5 只"降级实时(自定义源 fuyao 的全市场
快照已全面覆盖且免费); TickFlow 免费档 = 无实时, 接入自定义实时源
(如 fuyao)或升级 TickFlow 后恢复全市场模式。
"""
from app.services import preferences
if preferences.get_realtime_data_provider() != "tickflow":
return "full_market"
tier = cls._current_tier()
if tier == "none":
if tier in ("none", "free"):
return "none"
if tier == "free":
return "watchlist"
return "full_market"
@classmethod
@@ -501,7 +504,6 @@ class QuoteService:
def status(self) -> dict:
"""返回行情服务状态。"""
from app.services import preferences
age = (time.perf_counter() - self._fetch_time) * 1000 if self._fetch_time else -1
mode = self.realtime_mode()
phase = self._market_phase()
@@ -514,7 +516,6 @@ class QuoteService:
"paused": self._paused,
"mode": mode,
"realtime_allowed": mode != "none",
"watchlist_symbol_count": len(preferences.get_realtime_watchlist_symbols()),
"interval_s": self._interval,
"symbol_count": self._symbol_count,
"index_symbol_count": self._index_symbol_count,
@@ -568,15 +569,12 @@ class QuoteService:
waited += 0.5
def _fetch_quotes(self, *, final: bool = False) -> bool:
"""按当前档位拉取行情。加锁串行化 (后台轮询 vs 手动 refresh)。返回本轮是否成功更新。"""
"""拉取行情。加锁串行化 (后台轮询 vs 手动 refresh)。返回本轮是否成功更新。"""
with self._fetch_lock:
before = self._fetched_at
if final:
logger.info("最终行情同步开始")
if self.realtime_mode() == "watchlist":
self._fetch_watchlist_quotes()
else:
self._fetch_full_market_quotes()
self._fetch_full_market_quotes()
return self._fetched_at > before
def _fetch_full_market_quotes(self) -> None:
@@ -772,150 +770,11 @@ class QuoteService:
# ---- 策略监控 + 告警评估 ----
self._evaluate_monitors(daily_df, quote_extra)
def _fetch_watchlist_quotes(self) -> None:
"""Free 档自选股实时: 按 capability batch 上限分批拉取。"""
from app.services import preferences
from app.tickflow.client import get_paid_realtime_client
from app.tickflow.capabilities import Cap
from app.tickflow.policy import detect_capabilities
from app.tickflow.rate_limits import chunked, resolve_limit, sleep_between_batches
symbols = preferences.get_realtime_watchlist_symbols()
# 指数监控规则标的并入轮询 (与股票共享 batch 额度)
engine = getattr(self._app_state, "monitor_engine", None) if self._app_state else None
if engine:
for _r in list(engine.rules.values()):
if _r.get("enabled", True) and _r.get("asset_type") == "index" and _r.get("scope") == "symbols":
for _s in _r.get("symbols", []):
if _s and _s not in symbols:
symbols.append(_s)
if not symbols:
logger.info("自选实时未配置标的, 跳过行情拉取")
return
tf = get_paid_realtime_client()
if tf is None:
logger.warning("自选实时拉取失败:未配置付费服务器 API Key")
return
# 按 capability batch 上限分批: 股票+指数共享额度, 超过上限会导致整轮失败
capset = detect_capabilities()
lim = resolve_limit(capset, Cap.QUOTE_BY_SYMBOL, default_batch=5)
batches = chunked(symbols, lim.batch)
t0 = time.perf_counter()
now_ts = time.perf_counter()
resp = []
for i, batch in enumerate(batches):
sleep_between_batches(i, lim.rpm)
try:
resp.extend(tf.quotes.get(symbols=batch) or [])
except Exception as e: # noqa: BLE001
logger.warning("自选实时批次 %d/%d 拉取失败: %s", i + 1, len(batches), e)
if not resp:
logger.warning("自选实时行情数据为空")
return
records = []
for q in resp:
ext = q.get("ext") or {}
last_price = q.get("last_price")
prev_close = q.get("prev_close")
change_amount = ext.get("change_amount")
change_pct = ext.get("change_pct")
if change_amount is None and last_price is not None and prev_close is not None:
change_amount = float(last_price) - float(prev_close)
if change_pct is None and change_amount is not None and prev_close not in (None, 0):
# 小数制, 与 ext.change_pct / enriched 口径一致 (不乘 100)
change_pct = float(change_amount) / float(prev_close)
records.append({
"symbol": q.get("symbol"),
"name": q.get("name") or ext.get("name"),
"last_price": last_price,
"prev_close": prev_close,
"open": q.get("open"),
"high": q.get("high"),
"low": q.get("low"),
"volume": q.get("volume"),
"amount": q.get("amount"),
"change_pct": change_pct,
"change_amount": change_amount,
"amplitude": ext.get("amplitude"),
"turnover_rate": ext.get("turnover_rate"),
"timestamp": q.get("timestamp"),
"session": q.get("session"),
})
index_set = self._repo.get_index_symbol_set() if self._repo else set()
etf_set = self._repo.get_etf_symbol_set() if self._repo else set()
index_records, etf_records, stock_records = self._split_records_by_asset(records, index_set, etf_set)
fetch_ms = (time.perf_counter() - t0) * 1000
fetched_at = time.time() * 1000
with self._lock:
self._fetch_time = now_ts
self._fetch_ms = fetch_ms
self._fetched_at = fetched_at
self._symbol_count = len(stock_records)
self._index_symbol_count = len(index_records)
self._etf_symbol_count = len(etf_records)
self._index_quotes_cache = self._build_index_quotes(index_records) if index_records else None
_persist_last_fetch(fetched_at)
logger.info("自选实时刷新: %d 只股票, %d 只ETF, %d 只指数, 耗时 %.0fms",
len(stock_records), len(etf_records), len(index_records), fetch_ms)
daily_df = self._build_daily(stock_records)
quote_extra = self._build_quote_extra(stock_records)
if not daily_df.is_empty() and self._repo:
try:
self._repo.merge_live_daily_asset("stock", daily_df)
except Exception as e: # noqa: BLE001
logger.warning("自选实时日K写盘失败: %s", e)
self._flush_live_enriched(daily_df, quote_extra, asset_type="stock", merge=True)
# ETF/指数进自选前5时按各自资产落盘, 不污染股票表
etf_daily_df = self._build_daily(etf_records)
if not etf_daily_df.is_empty() and self._repo:
try:
self._repo.merge_live_daily_asset("etf", etf_daily_df)
except Exception as e: # noqa: BLE001
logger.warning("自选实时 ETF 日K写盘失败: %s", e)
self._flush_live_enriched(etf_daily_df, self._build_quote_extra(etf_records), asset_type="etf", merge=True)
index_daily_df = self._build_daily(index_records)
if not index_daily_df.is_empty() and self._repo:
try:
self._repo.merge_live_daily_asset("index", index_daily_df)
except Exception as e: # noqa: BLE001
logger.warning("自选实时指数日K写盘失败: %s", e)
self._flush_live_enriched(index_daily_df, self._build_quote_extra(index_records), asset_type="index", merge=True)
self._broadcast_quote_updated()
self._evaluate_monitors(daily_df, quote_extra)
# ================================================================
# 工具
# ================================================================
@staticmethod
def _split_records_by_asset(
records: list[dict], index_set: set[str], etf_set: set[str],
) -> tuple[list[dict], list[dict], list[dict]]:
"""把行情 records 按资产拆成 (index, etf, stock)。判定顺序与 resolve_asset_type 一致: 先 ETF 后指数。"""
index_records: list[dict] = []
etf_records: list[dict] = []
stock_records: list[dict] = []
for r in records:
sym = r.get("symbol")
if sym in etf_set:
etf_records.append(r)
elif sym in index_set:
index_records.append(r)
else:
stock_records.append(r)
return index_records, etf_records, stock_records
@staticmethod
def _build_daily(records: list[dict]) -> pl.DataFrame:
"""将 API records 转为日K格式 DataFrame (OHLCV + quote_ts, 写 kline_daily 用)。"""
backend/app/tickflow/repository.py
+51 -26
View File
@@ -38,6 +38,33 @@ from app.parquet import scan_enriched_parquet
logger = logging.getLogger(__name__)
def replace_with_retry(src: Path, dst: Path, *, attempts: int = 10, delay_s: float = 0.5) -> None:
"""os.replace 的 Windows 读锁重试版。
分区 parquet 的读端 (polars scan_parquet / DuckDB read_parquet 视图) 在扫描进行
期间持有句柄; Windows 不允许替换"仍被读端打开"的目标文件 (PermissionError,
WinError 5), Linux 的 inode 交换语义则无此限制。读端扫描通常亚秒级完成,
短退避重试即可穿过并发读窗口; attempts 次仍被占用则原样抛出, 由上层记录失败。
"""
last: PermissionError | None = None
for i in range(attempts):
try:
src.replace(dst)
if i:
logger.info("parquet replace succeeded after %d blocked attempt(s): %s", i, dst)
return
except PermissionError as e:
last = e
if i == 0:
logger.warning(
"parquet replace blocked by concurrent reader, retrying (total <= %.1fs): %s",
attempts * delay_s, dst,
)
if i < attempts - 1:
time.sleep(delay_s)
raise last # type: ignore[misc] # attempts >= 1 时 last 必已赋值
def enriched_dirname(asset_type: str) -> str:
"""asset_type → enriched parquet 目录名。ETF 走独立目录, 其余(stock)用日K enriched。"""
return "kline_etf_enriched" if asset_type == "etf" else "kline_daily_enriched"
@@ -1813,13 +1840,15 @@ class KlineRepository:
# ================================================================
def latest_minute_date(self, symbol: str, asset_type: str = "stock") -> date | None:
# 注意: 必须走 execute_one (cursor+close)。直连 self.db.execute(...).fetchone()
# 的未消费结果集会把首个分区 parquet 的句柄钉在共享连接上, Windows 下阻塞
# 同步写入的 os.replace → 个股分时"补齐数据"500。
table = "kline_etf_minute" if asset_type == "etf" else "kline_minute"
try:
with self._lock:
row = self.db.execute(
f"SELECT max(CAST(datetime AS DATE)) FROM {table} WHERE symbol = ?",
[symbol],
).fetchone()
row = self.execute_one(
f"SELECT max(CAST(datetime AS DATE)) FROM {table} WHERE symbol = ?",
[symbol],
)
if row and row[0]:
return row[0] if isinstance(row[0], date) else date.fromisoformat(str(row[0]))
except duckdb.CatalogException:
@@ -1829,10 +1858,9 @@ class KlineRepository:
def latest_minute_date_global(self) -> date | None:
"""全市场最近分钟K日期 (不分 symbol)。用于非交易日回退到上一交易日。"""
try:
with self._lock:
row = self.db.execute(
"SELECT max(CAST(datetime AS DATE)) FROM kline_minute",
).fetchone()
row = self.execute_one(
"SELECT max(CAST(datetime AS DATE)) FROM kline_minute",
)
if row and row[0]:
return row[0] if isinstance(row[0], date) else date.fromisoformat(str(row[0]))
except Exception: # noqa: BLE001
@@ -1841,10 +1869,9 @@ class KlineRepository:
def earliest_daily_date(self) -> date | None:
"""本地日K数据的最早日期。"""
try:
with self._lock:
res = self.db.execute(
"SELECT min(date) FROM kline_daily",
).fetchone()
res = self.execute_one(
"SELECT min(date) FROM kline_daily",
)
if res and res[0]:
d = res[0]
return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -1855,10 +1882,9 @@ class KlineRepository:
def earliest_minute_date(self) -> date | None:
"""本地分钟K数据的最早日期。"""
try:
with self._lock:
res = self.db.execute(
"SELECT min(CAST(datetime AS DATE)) FROM kline_minute",
).fetchone()
res = self.execute_one(
"SELECT min(CAST(datetime AS DATE)) FROM kline_minute",
)
if res and res[0]:
d = res[0]
return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -1892,10 +1918,9 @@ class KlineRepository:
def latest_daily_date(self) -> date | None:
"""本地日K数据的最新日期。"""
try:
with self._lock:
res = self.db.execute(
"SELECT max(date) FROM kline_daily",
).fetchone()
res = self.execute_one(
"SELECT max(date) FROM kline_daily",
)
if res and res[0]:
d = res[0]
return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -1947,10 +1972,9 @@ class KlineRepository:
def _latest_enriched_date_duckdb(self) -> date | None:
try:
with self._lock:
res = self.db.execute(
"SELECT max(date) FROM kline_enriched",
).fetchone()
res = self.execute_one(
"SELECT max(date) FROM kline_enriched",
)
if res and res[0]:
d = res[0]
return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -2118,10 +2142,11 @@ class KlineRepository:
直接 write_parquet(out) 在进程被 kill (dev.sh 清端口用 kill -9)
或断电时会留下半截文件, 之后 scan_parquet glob 整条链路报错。
临时文件后缀 .tmp 不匹配 *.parquet glob, 不会被扫描误读。
Windows 下目标正被并发读取时由 replace_with_retry 短退避穿过。
"""
tmp = out.with_name(out.name + ".tmp")
df.write_parquet(tmp)
tmp.replace(out) # 同目录 rename, POSIX/NTFS 均为原子操作
replace_with_retry(tmp, out)
def _write_daily_partition(self, df: pl.DataFrame, table: str) -> None:
"""按 date 分区写入 parquet，每个日期一个文件，支持 merge-upsert。"""
backend/tests/test_atomic_write_retry.py
+161
View File
@@ -0,0 +1,161 @@
"""Windows 读锁竞态下 parquet 原子替换的重试测试。
根因: polars scan_parquet / DuckDB read_parquet 扫描进行中持有分区句柄,
Windows os.replace 替换"仍被打开"的目标文件抛 PermissionError (WinError 5);
Linux 的 inode 交换语义无此限制。表现为个股分时"补齐数据"500。
修复: replace_with_retry 短退避重试穿过读窗口; 永久占用则原样抛出。
两处 _atomic_write_parquet (repository / kline_sync) 均接入。
另含 DuckDB 句柄泄漏回归: latest_minute_date 等曾用 self.db.execute(...)
.fetchone() 直连共享连接, 未消费结果集把首个分区句柄钉死在连接上,
导致同步 os.replace 永久被拒 (修为 execute_one cursor+close)。
"""
from __future__ import annotations
import os
import sys
from datetime import date, datetime
import polars as pl
import pytest
from app.services import kline_sync
from app.tickflow import repository
try:
import psutil
_PSUTIL = True
except ImportError: # pragma: no cover
_PSUTIL = False
def _minute_frame() -> pl.DataFrame:
return pl.DataFrame({
"symbol": ["600519.SH"],
"datetime": [datetime(2026, 1, 15, 9, 30)],
"open": [10.0], "high": [10.5], "low": [9.5], "close": [10.2],
"volume": [100.0], "amount": [1020.0],
})
def _flaky_replace(monkeypatch, fail_times: int) -> dict:
"""os.replace 前 fail_times 次 raise PermissionError, 之后正常执行。"""
real_replace = os.replace
state = {"calls": 0}
def _flaky(src, dst):
state["calls"] += 1
if state["calls"] <= fail_times:
raise PermissionError(5, "拒绝访问。")
return real_replace(src, dst)
monkeypatch.setattr(os, "replace", _flaky)
return state
# ---------- replace_with_retry 本体 ----------
def test_retry_succeeds_after_transient_blocks(tmp_path, monkeypatch):
out = tmp_path / "part.parquet"
out.write_bytes(b"old")
src = tmp_path / "part.parquet.tmp"
src.write_bytes(b"new")
state = _flaky_replace(monkeypatch, fail_times=2)
repository.replace_with_retry(src, out, attempts=5, delay_s=0)
assert out.read_bytes() == b"new"
assert state["calls"] == 3
assert not src.exists()
def test_retry_exhausted_raises_last_error(tmp_path, monkeypatch):
monkeypatch.setattr(os, "replace", lambda s, d: (_ for _ in ()).throw(PermissionError(5, "拒绝访问。")))
src = tmp_path / "a.tmp"
src.write_bytes(b"x")
with pytest.raises(PermissionError, match="拒绝访问"):
repository.replace_with_retry(src, tmp_path / "a.parquet", attempts=3, delay_s=0)
assert src.exists() # 未被消费, 目标未生成
def test_retry_no_block_single_attempt(tmp_path, monkeypatch):
out = tmp_path / "part.parquet"
out.write_bytes(b"old")
src = tmp_path / "part.parquet.tmp"
src.write_bytes(b"new")
state = _flaky_replace(monkeypatch, fail_times=0)
repository.replace_with_retry(src, out, attempts=5, delay_s=0)
assert state["calls"] == 1
assert out.read_bytes() == b"new"
# ---------- 两处 _atomic_write_parquet 接入 ----------
def test_kline_sync_atomic_write_survives_transient_lock(tmp_path, monkeypatch):
state = _flaky_replace(monkeypatch, fail_times=1)
out = tmp_path / "date=2026-01-15" / "part.parquet"
out.parent.mkdir(parents=True)
kline_sync._atomic_write_parquet(_minute_frame(), out)
assert out.exists()
assert state["calls"] == 2
assert pl.read_parquet(out).height == 1
def test_repository_atomic_write_survives_transient_lock(tmp_path, monkeypatch):
state = _flaky_replace(monkeypatch, fail_times=1)
out = tmp_path / "kline_minute" / "date=2026-01-15" / "part.parquet"
out.parent.mkdir(parents=True)
repository.KlineRepository._atomic_write_parquet(_minute_frame(), out)
assert out.exists()
assert state["calls"] == 2
def test_write_minute_partition_survives_reader_race(tmp_path, monkeypatch):
"""集成: _write_minute_partition 读旧→concat→写新全程有读锁竞态仍完成。"""
state = _flaky_replace(monkeypatch, fail_times=2)
# 预置旧分区 (读改写路径)
old_dir = tmp_path / "date=2026-01-15"
old_dir.mkdir(parents=True)
_minute_frame().write_parquet(old_dir / "part.parquet")
written = kline_sync._write_minute_partition(_minute_frame(), tmp_path)
assert written == 1
assert state["calls"] >= 3 # 至少经历了重试
# ---------- DuckDB 句柄泄漏回归 (Windows 实测语义) ----------
@pytest.mark.skipif(sys.platform != "win32" or not _PSUTIL, reason="Windows 句柄语义 + psutil")
def test_minute_date_queries_do_not_pin_partition_handles(tmp_path):
"""latest_minute_date 等查询后不得残留分区句柄。
旧实现 self.db.execute(...).fetchone() 的未消费结果集经 DuckDB buffer
manager 钉住首个分区句柄, 后续同步 os.replace 永久 PermissionError。
"""
from app.tickflow.repository import DataStore, KlineRepository
minute_dir = tmp_path / "kline_minute"
kline_sync._write_minute_partition(
_minute_frame(), minute_dir) # date=2026-01-15
repo = KlineRepository(DataStore(data_dir=tmp_path))
assert repo.latest_minute_date("600519.SH") == date(2026, 1, 15)
assert repo.latest_minute_date_global() == date(2026, 1, 15)
assert repo.earliest_minute_date() == date(2026, 1, 15)
me = psutil.Process()
held = [f.path for f in me.open_files() if "kline_minute" in f.path]
assert held == []
# 钉住场景的端到端后果: 查询后重写同一分区必须成功 (旧实现在此 PermissionError)
assert kline_sync._write_minute_partition(_minute_frame(), minute_dir) == 1
backend/tests/test_job_record_persistence.py
+125
View File
@@ -0,0 +1,125 @@
"""回归测试: job 记录跨进程死亡持久化(「数据在、记录丢」补丁)。
背景(用户反馈): 全市场同步 12:11~12:42 成功结束后 0.7s, uvicorn --reload
检测到代码变更杀死 worker, 恰好落在管道完成与 job_store.succeed() 落盘之间
—— 数据已写盘但同步历史无任何记录。旧实现 pending/running 仅存内存、终态才
落盘, 存在整段丢失窗口。
修复后契约:
- create()/start() 即落盘 pending/running 快照;
- 下次进程启动(= 新 JobStore 实例, 同目录)把遗留的 pending/running
孤儿记录补标为 failed(中断), finished_at 取文件 mtime;
- 终态记录不受补录影响; 终态写入覆盖 running 快照(同一文件)。
均为纯逻辑, 不触网。
"""
from __future__ import annotations
import json
from app.services.pipeline_jobs import JobStore
def _read_disk(d, jid: str) -> dict:
return json.loads((d / f"{jid}.json").read_text("utf-8"))
# ── 创建/启动即落盘 ──────────────────────────────────────────────────────
def test_create_writes_pending_snapshot_to_disk(tmp_path):
d = tmp_path / "jobs"
store = JobStore(store_dir=d)
jid, _ = store.create(timeout_s=60)
disk = _read_disk(d, jid)
assert disk["status"] == "pending"
assert disk["stage"] == "init"
def test_start_updates_disk_snapshot_to_running(tmp_path):
d = tmp_path / "jobs"
store = JobStore(store_dir=d)
jid, _ = store.create(timeout_s=60)
store.start(jid)
disk = _read_disk(d, jid)
assert disk["status"] == "running"
assert disk["started_at"] is not None
# ── 进程死亡 → 下次启动补录 ──────────────────────────────────────────────
def test_orphan_running_record_is_reaped_on_next_boot(tmp_path):
"""核心场景: 进程死在 running(甚至工作已做完但未终态), 记录必须可见。"""
d = tmp_path / "jobs"
dead = JobStore(store_dir=d)
jid, _ = dead.create(timeout_s=60)
dead.start(jid)
dead.progress(jid, "sync", 50, "halfway") # 进度只更新内存
# 新进程 = 同目录新实例(内存为空, 只有磁盘)
revived = JobStore(store_dir=d)
j = revived.get(jid)
assert j is not None
assert j["status"] == "failed"
assert "中断" in j["error"]
assert j["finished_at"] is not None
# finished_at 基于文件 mtime(≈ start 时刻), 时长不得虚增为负或巨大
assert j["duration_s"] is not None
assert 0 <= j["duration_s"] <= 60
# 同步历史列表可见
assert any(x["id"] == jid for x in revived.list_recent())
def test_orphan_pending_record_is_reaped(tmp_path):
"""进程死在 create() 与 start() 之间: 记录同样可见, 时长为 None。"""
d = tmp_path / "jobs"
dead = JobStore(store_dir=d)
jid, _ = dead.create(timeout_s=60)
# 未 start 即死亡
revived = JobStore(store_dir=d)
j = revived.get(jid)
assert j["status"] == "failed"
assert j["duration_s"] is None
def test_reap_does_not_touch_terminal_records(tmp_path):
d = tmp_path / "jobs"
store = JobStore(store_dir=d)
jid, _ = store.create(timeout_s=60)
store.start(jid)
store.succeed(jid, {"daily_rows": 100})
revived = JobStore(store_dir=d)
j = revived.get(jid)
assert j["status"] == "succeeded"
assert j["result"] == {"daily_rows": 100}
def test_reap_allows_new_job_after_dead_orphan(tmp_path):
"""补录后旧 job 已 failed: 新进程 create() 不被死孤儿阻塞(单飞只看内存)。"""
d = tmp_path / "jobs"
dead = JobStore(store_dir=d)
old_jid, _ = dead.create(timeout_s=60)
dead.start(old_jid)
revived = JobStore(store_dir=d)
new_jid, is_new = revived.create(timeout_s=60)
assert is_new is True
assert new_jid != old_jid
# ── 终态覆盖快照 ─────────────────────────────────────────────────────────
def test_terminal_write_replaces_running_snapshot(tmp_path):
d = tmp_path / "jobs"
store = JobStore(store_dir=d)
jid, _ = store.create(timeout_s=60)
store.start(jid)
store.fail(jid, "boom")
files = list(d.glob("*.json"))
assert len(files) == 1
disk = _read_disk(d, jid)
assert disk["status"] == "failed"
assert disk["error"] == "boom"
backend/tests/test_realtime_mode.py
+31
View File
@@ -0,0 +1,31 @@
"""回归测试: 实时行情模式判定 — 免费档不再提供自选实时降级。
watchlist(自选前 5 只)模式已于 2026-08 移除: 自定义实时源(如 fuyao)的
全市场快照免费且更优, TickFlow 免费档不再保留降级通路。锁定判定结果,
防止该通路被无意恢复。
"""
from app.services.quote_service import QuoteService
def test_custom_realtime_source_is_full_market(monkeypatch):
"""自定义实时源(如 fuyao)无视 TickFlow 档位, 恒为全市场。"""
from app.services import preferences
monkeypatch.setattr(preferences, "get_realtime_data_provider", lambda: "fuyao")
monkeypatch.setattr(QuoteService, "_current_tier", lambda: "free")
assert QuoteService.realtime_mode() == "full_market"
def test_tickflow_free_has_no_realtime(monkeypatch):
"""TickFlow 免费档 = 无实时(不再降级为自选模式)。"""
from app.services import preferences
monkeypatch.setattr(preferences, "get_realtime_data_provider", lambda: "tickflow")
monkeypatch.setattr(QuoteService, "_current_tier", lambda: "free")
assert QuoteService.realtime_mode() == "none"
assert QuoteService.is_realtime_allowed() is False
def test_tickflow_paid_is_full_market(monkeypatch):
from app.services import preferences
monkeypatch.setattr(preferences, "get_realtime_data_provider", lambda: "tickflow")
monkeypatch.setattr(QuoteService, "_current_tier", lambda: "pro")
assert QuoteService.realtime_mode() == "full_market"
backend/tests/test_stocksdk_provider.py
+79
View File
@@ -226,3 +226,82 @@ def test_builtin_not_editable():
raise AssertionError("expected ValueError for builtin")
except ValueError:
pass
# ---------- 分钟 open 数据卫生 (stock-sdk 上游区间模式给日级常量伪 open) ----------
def _sdk_rows(day: str, opens, closes):
"""构造 bridge 返回形状的分钟行 (timestamp 为北京墙钟对应 UTC 毫秒)。"""
from datetime import UTC, datetime, timedelta
rows = []
for i, (o, c) in enumerate(zip(opens, closes, strict=True)):
dt = datetime.fromisoformat(f"{day} 09:30:00") + timedelta(minutes=i)
ts = int(dt.replace(tzinfo=UTC).timestamp() * 1000) - 8 * 3600_000
rows.append({"timestamp": ts, "open": o, "high": max(o, c), "low": min(o, c),
"close": c, "volume": 100 + i, "amount": 1000.0 + i})
return rows
def test_minute_degenerate_open_nulled():
"""历史日 open 为全天常量(uniq=1)而 close 多值 → open 置 null。"""
n = 30
rows = _sdk_rows("2026-08-27", [8.0] * n, [10 + i * 0.01 for i in range(n)])
df = StockSDKProvider._minute_df(rows, "600664.SH")
assert df.height == n
assert df["open"].null_count() == n # 伪 open 全部置 null
assert df["close"].null_count() == 0 # close/high/low 保留
def test_minute_real_open_kept():
"""真实分钟 open (多唯一值) 原样保留。"""
n = 30
rows = _sdk_rows("2026-08-28", [10 + i * 0.01 for i in range(n)], [10.05 + i * 0.01 for i in range(n)])
df = StockSDKProvider._minute_df(rows, "600664.SH")
assert df["open"].null_count() == 0
assert df["open"].n_unique() == n
def test_minute_short_day_open_kept():
"""短交易日 (rows<=10, 如半日/首日少量bar) 不误杀。"""
rows = _sdk_rows("2026-08-26", [8.0] * 6, [8.0 + i * 0.1 for i in range(6)])
df = StockSDKProvider._minute_df(rows, "600664.SH")
assert df["open"].null_count() == 0
def test_get_minute_splits_tail_into_single_day_jobs(monkeypatch):
"""多日区间 → 末尾 3 自然日(跳过周末)逐日单拉 (保住最新交易日真实 open)。"""
from datetime import datetime
jobs = []
def fake_run_job(job, timeout=None):
jobs.append(job)
return {"ok": True, "op": "minute", "rows": {}}
monkeypatch.setattr(sp.bridge, "run_job", fake_run_job)
p = StockSDKProvider()
p.get_minute(["600519.SH"], datetime(2026, 8, 20), datetime(2026, 8, 29, 23, 0))
# 末尾 4 个自然日 (08-26..08-29) 逐日单拉, 周六 08-29 跳过;
# 前段 = [08-20, 08-25] 一个区间任务
spans = [(j["start"], j["end"]) for j in jobs]
for day in ("20260826", "20260827", "20260828"):
assert (day, day) in spans
assert ("20260829", "20260829") not in spans # 周六不单拉
assert ("20260820", "20260825") in spans
def test_get_minute_single_day_no_split(monkeypatch):
"""单日区间不拆分, 保持一个任务。"""
from datetime import datetime
jobs = []
def fake_run_job(job, timeout=None):
jobs.append(job)
return {"ok": True, "op": "minute", "rows": {}}
monkeypatch.setattr(sp.bridge, "run_job", fake_run_job)
StockSDKProvider().get_minute(["600519.SH"], datetime(2026, 8, 28), datetime(2026, 8, 28, 15, 0))
assert len(jobs) == 1
assert jobs[0]["start"] == "20260828"
backend/tests/test_watchlist_batch.py
-123
View File
@@ -1,123 +0,0 @@
"""回归测试: Free 档自选实时 symbols 超过 capability batch 上限时分批请求 (PR #46 问题 4)。"""
from contextlib import ExitStack
from unittest.mock import MagicMock, patch
import polars as pl
from app.services.quote_service import QuoteService
from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet
def _make_svc(engine_rules: dict) -> QuoteService:
"""创建最小可用的 QuoteService 实例 (跳过 __init__)。"""
svc = QuoteService.__new__(QuoteService)
svc._app_state = MagicMock()
svc._repo = MagicMock()
svc._lock = MagicMock()
engine = MagicMock()
engine.rules = engine_rules
svc._app_state.monitor_engine = engine
svc._app_state.repo = svc._repo
svc._repo.get_index_symbol_set.return_value = {"000001.SH"}
svc._repo.get_etf_symbol_set.return_value = set()
return svc
def _run_fetch(svc, tf, watchlist: list[str], capset: CapabilitySet):
"""在完整 patch 环境下执行 _fetch_watchlist_quotes。"""
with ExitStack() as stack:
stack.enter_context(patch(
"app.services.preferences.get_realtime_watchlist_symbols",
return_value=watchlist,
))
stack.enter_context(patch(
"app.tickflow.client.get_paid_realtime_client", return_value=tf,
))
stack.enter_context(patch(
"app.tickflow.policy.detect_capabilities", return_value=capset,
))
stack.enter_context(patch("app.tickflow.rate_limits.sleep_between_batches"))
# patch 分批之后的下游处理
stack.enter_context(patch.object(
QuoteService, "_build_daily", return_value=pl.DataFrame(),
))
stack.enter_context(patch.object(
QuoteService, "_build_quote_extra", return_value=pl.DataFrame(),
))
stack.enter_context(patch.object(
QuoteService, "_build_index_quotes", return_value=pl.DataFrame(),
))
stack.enter_context(patch.object(QuoteService, "_broadcast_quote_updated"))
stack.enter_context(patch.object(QuoteService, "_evaluate_monitors"))
stack.enter_context(patch("app.services.quote_service._persist_last_fetch"))
svc._fetch_watchlist_quotes()
def test_watchlist_batch_respects_capability_limit():
"""6 symbols / batch 5 → 分 2 批请求, 不整轮失败。"""
engine_rules = {
"r_idx": {"enabled": True, "asset_type": "index", "scope": "symbols",
"symbols": ["000001.SH"]},
}
svc = _make_svc(engine_rules)
tf = MagicMock()
tf.quotes.get.return_value = [
{"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}},
]
capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)})
_run_fetch(svc, tf,
["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"],
capset)
# 5 股票 + 1 指数 = 6 symbols, batch 5 → 2 批
assert tf.quotes.get.call_count == 2
first_batch = tf.quotes.get.call_args_list[0][1]["symbols"]
second_batch = tf.quotes.get.call_args_list[1][1]["symbols"]
assert len(first_batch) == 5
assert len(second_batch) == 1
assert "000001.SH" in second_batch
def test_watchlist_batch_partial_failure_keeps_other_batches():
"""某一批拉取失败不影响其他批次 (已有股票实时刷新不丢失)。"""
engine_rules = {
"r_idx": {"enabled": True, "asset_type": "index", "scope": "symbols",
"symbols": ["000001.SH"]},
}
svc = _make_svc(engine_rules)
tf = MagicMock()
# 第一批 (股票) 成功, 第二批 (指数) 失败
tf.quotes.get.side_effect = [
[{"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}}],
ConnectionError("timeout"),
]
capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)})
_run_fetch(svc, tf,
["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"],
capset)
# 两批都被尝试 (第二批失败不阻断)
assert tf.quotes.get.call_count == 2
def test_watchlist_no_index_rules_no_extra_symbols():
"""无指数监控规则时, symbols 不追加指数标的。"""
svc = _make_svc({}) # 无规则
tf = MagicMock()
tf.quotes.get.return_value = [
{"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}},
]
capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)})
_run_fetch(svc, tf, ["600000.SH", "600001.SH"], capset)
# 2 symbols / batch 5 → 1 批
assert tf.quotes.get.call_count == 1
assert tf.quotes.get.call_args_list[0][1]["symbols"] == ["600000.SH", "600001.SH"]
backend/tests/test_watchlist_realtime_split.py
-19
View File
@@ -1,19 +0,0 @@
"""Free 档自选实时资产分流测试。"""
from app.services.quote_service import QuoteService
def test_split_records_by_asset():
records = [
{"symbol": "600000.SH"}, {"symbol": "510300.SH"}, {"symbol": "000001.SH"},
]
index, etf, stock = QuoteService._split_records_by_asset(
records, {"000001.SH"}, {"510300.SH"},
)
assert [r["symbol"] for r in index] == ["000001.SH"]
assert [r["symbol"] for r in etf] == ["510300.SH"]
assert [r["symbol"] for r in stock] == ["600000.SH"]
# etf 优先于 index (与 resolve_asset_type 判定顺序一致)
index2, etf2, stock2 = QuoteService._split_records_by_asset(
[{"symbol": "X"}], {"X"}, {"X"},
)
assert etf2 and not index2 and not stock2
frontend/src/components/EChartsIntraday.tsx
+12 -2
View File
@@ -72,6 +72,9 @@ function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgP
const volumes = new Array(FULL_DAY_TIMES.length).fill(null) as (any | null)[]
const volNeutral = 'rgba(161,161,170,0.5)'
// 量柱着色基准: 前一分钟 close; 第一根用昨收。
// 不用 row.open — stock-sdk 历史日无真实分钟 open(为 null), close-vs-open 会全偏。
let prevRef: number | null = prevClose ?? null
for (let i = 0; i < data.length; i++) {
const timeKey = formatMinuteTime(data[i].datetime)
const idx = timeIndexMap.get(timeKey)
@@ -83,9 +86,16 @@ function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgP
volumes[idx] = {
value: data[i].volume,
itemStyle: {
color: data[i].close > data[i].open ? THEME.volUp : data[i].close < data[i].open ? THEME.volDown : volNeutral,
color: prevRef == null
? volNeutral
: data[i].close > prevRef
? THEME.volUp
: data[i].close < prevRef
? THEME.volDown
: volNeutral,
},
}
prevRef = data[i].close
}
}
@@ -570,7 +580,7 @@ export function EChartsIntraday({
<>
{date && <span className="text-muted">{date}</span>}
<span className="text-muted">开</span>
<span style={{ color: priceClr }}>{d.open.toFixed(2)}</span>
<span style={{ color: priceClr }}>{d.open != null ? d.open.toFixed(2) : '—'}</span>
<span className="text-muted">高</span>
<span style={{ color: priceClr }}>{d.high.toFixed(2)}</span>
<span className="text-muted">低</span>
frontend/src/components/EChartsMultiDayIntraday.tsx
+12 -6
View File
@@ -74,6 +74,9 @@ function buildModel(sessions: MinuteKlineSession[]) {
const dayValues: (number | null)[] = []
const dayAverages: (number | null)[] = []
// 量柱着色基准: 前一分钟 close; 当日第一根用 session 昨收。
// 不用 row.open — stock-sdk 历史日无真实分钟 open(为 null), close-vs-open 会全偏。
let prevRef: number | null = session.prev_close
for (const time of FULL_DAY_TIMES) {
const point = rowsByTime.get(time)
const index = categories.length
@@ -91,13 +94,16 @@ function buildModel(sessions: MinuteKlineSession[]) {
volumeData.push({
value: row.volume,
itemStyle: {
color: row.close > row.open
? COLORS.volumeUp
: row.close < row.open
? COLORS.volumeDown
: COLORS.volumeFlat,
color: prevRef == null
? COLORS.volumeFlat
: row.close > prevRef
? COLORS.volumeUp
: row.close < prevRef
? COLORS.volumeDown
: COLORS.volumeFlat,
},
})
prevRef = row.close
priceValues.push(row.low, row.high, average)
pointByIndex.set(index, {
date: session.date,
@@ -408,7 +414,7 @@ export function EChartsMultiDayIntraday({
{info ? (
<>
<span className="text-muted">{info.date} {formatMinuteTime(info.row.datetime)}</span>
<span className="text-muted">开</span><span style={{ color: infoColor }}>{info.row.open.toFixed(2)}</span>
<span className="text-muted">开</span><span style={{ color: infoColor }}>{info.row.open != null ? info.row.open.toFixed(2) : '—'}</span>
<span className="text-muted">高</span><span style={{ color: infoColor }}>{info.row.high.toFixed(2)}</span>
<span className="text-muted">低</span><span style={{ color: infoColor }}>{info.row.low.toFixed(2)}</span>
<span className="text-muted">收</span><span className="font-semibold" style={{ color: infoColor }}>{info.row.close.toFixed(2)}</span>
frontend/src/components/Layout.tsx
-4
View File
@@ -575,10 +575,6 @@ export function Layout() {
toast('当前数据源无实时行情能力, 请先配置数据源', 'error')
return
}
if (fresh.mode === 'watchlist' && (prefs?.realtime_watchlist_symbols?.length ?? 0) === 0) {
navigate('/watchlist')
return
}
}
await toggleQuote.mutateAsync(enabled)
// 仅在交易时段立即获取一次行情
frontend/src/components/StockPreviewDialog.tsx
+22 -7
View File
@@ -50,11 +50,11 @@ interface PriceAlertDraft {
}
const INTRADAY_DAY_OPTIONS = [1, 5, 10, 20] as const
function loadIntradayDays(): number {
function loadIntradayDays(): number | null {
const saved = storage.stockPreviewIntradayDays.get(10)
return INTRADAY_DAY_OPTIONS.includes(saved as typeof INTRADAY_DAY_OPTIONS[number])
? saved
: 10
: null
}
function boardTag(symbol: string): { label: string; color: string } | null {
@@ -81,7 +81,7 @@ function fmtAbnormalCalcTime(asofSec: number): string {
export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props) {
const [view, setView] = useState<PreviewView>('daily')
const [intradayDays, setIntradayDays] = useState(loadIntradayDays)
const [intradayDays, setIntradayDays] = useState<number | null>(loadIntradayDays)
const [dateRange, setDateRange] = useState(getDefaultRange)
const [showMonitorEditor, setShowMonitorEditor] = useState(false)
const [priceAlertDraft, setPriceAlertDraft] = useState<PriceAlertDraft | null>(null)
@@ -167,6 +167,21 @@ export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props
const { data: prefs } = usePreferences()
const intradayRefetchMs = (prefs?.minute_intraday_refresh_interval ?? 6) * 1000
// 分时档位按分钟源历史深度收窄: 浅源(如 stock-sdk=5日)只显示可行档位、默认 5日;
// 深源(tickflow/未声明)全档位、默认 20日。用户已保存的可行选择优先保留。
const minuteHistoryDays = prefs?.minute_history_days ?? null
const dayOptions = useMemo<number[]>(
() => INTRADAY_DAY_OPTIONS.filter(d => minuteHistoryDays == null || d <= minuteHistoryDays),
[minuteHistoryDays],
)
const defaultIntradayDays = minuteHistoryDays != null && minuteHistoryDays < 20 ? 5 : 20
const effectiveIntradayDays = intradayDays ?? defaultIntradayDays
useEffect(() => {
if (!dayOptions.includes(effectiveIntradayDays)) {
setIntradayDays(defaultIntradayDays)
}
}, [dayOptions, effectiveIntradayDays, defaultIntradayDays])
const handleRefresh = () => {
if (!symbol) return
if (view === 'daily') {
@@ -270,14 +285,14 @@ export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props
) : (
<div className="flex items-center gap-1">
<div className="inline-flex shrink-0 items-center rounded border border-border bg-elevated p-0.5" aria-label="分时周期">
{INTRADAY_DAY_OPTIONS.map(days => (
{dayOptions.map(days => (
<button
key={days}
type="button"
aria-pressed={intradayDays === days}
aria-pressed={effectiveIntradayDays === days}
onClick={() => selectIntradayDays(days)}
className={`h-5 rounded px-1.5 font-mono text-[10px] transition-colors ${
intradayDays === days
effectiveIntradayDays === days
? 'bg-accent/20 text-accent'
: 'text-muted hover:text-secondary'
}`}
@@ -486,7 +501,7 @@ export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props
/>
<StockMultiDayIntradayChart
symbol={symbol}
days={intradayDays}
days={effectiveIntradayDays}
height={480}
refetchIntervalMs={intradayRefetchMs}
priceLines={monitorPriceLines}
frontend/src/components/stock-table/MiniIntraday.tsx
+5 -5
View File
@@ -36,23 +36,23 @@ export function MiniIntraday({ rows, prevClose, changePct, width = 100, height =
const n = rows.length
// 涨跌着色: 优先用 changePct (后端 enriched 字段, 最可靠);
// 其次用 prevClose vs lastClose; 最后回退到第一根 open
// 其次用 prevClose vs lastClose; 最后回退到首根 open(可能为 null, 再退首根 close)
const lastClose = rows[n - 1].close
const firstOpen = rows[0].open
const firstRef = rows[0].open ?? rows[0].close
const isUp = changePct != null
? changePct >= 0
: prevClose != null && prevClose > 0
? lastClose >= prevClose
: lastClose >= firstOpen
: lastClose >= firstRef
const color = isUp ? BULL : BEAR
// 昨收基准线: 优先用 prevClose; 其次用 changePct 反算 (close/(1+changePct));
// 最后回退到第一根 open
// 最后回退到首根 open(为 null 时退首根 close)
const baseline = (prevClose != null && prevClose > 0)
? prevClose
: (changePct != null && changePct !== 0)
? lastClose / (1 + changePct)
: firstOpen
: firstRef
// 价格区间: close + 昨收 + 均线 全部纳入, 确保都在可视范围
let hi = -Infinity, lo = Infinity
frontend/src/lib/intraday-chart.ts
+2 -2
View File
@@ -1,10 +1,10 @@
import type { MinuteKlineRow } from '@/lib/api'
/** 从 datetime 串取 HH:MM。契约: 分钟K datetime 已在后端入口统一为北京墙钟, 前端不做时区换算。 */
export function formatMinuteTime(datetime: string): string {
const match = datetime.match(/(\d{2}):(\d{2})/)
if (!match) return datetime.slice(11, 16)
const hour = (parseInt(match[1]) + 8) % 24
return `${String(hour).padStart(2, '0')}:${match[2]}`
return `${match[1]}:${match[2]}`
}
export function computeIntradayAverage(data: MinuteKlineRow[]): number[] {