diff --git a/backend/app/plugins/stocksdk/provider.py b/backend/app/plugins/stocksdk/provider.py index c3481a4..c431fba 100644 --- a/backend/app/plugins/stocksdk/provider.py +++ b/backend/app/plugins/stocksdk/provider.py @@ -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() diff --git a/backend/app/services/pipeline_jobs.py b/backend/app/services/pipeline_jobs.py index 365a216..08160e5 100644 --- a/backend/app/services/pipeline_jobs.py +++ b/backend/app/services/pipeline_jobs.py @@ -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: diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index d3d1e5d..1a17521 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -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 用)。""" diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index 81aae5e..f84cb55 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -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。""" diff --git a/backend/tests/test_atomic_write_retry.py b/backend/tests/test_atomic_write_retry.py new file mode 100644 index 0000000..94f22a7 --- /dev/null +++ b/backend/tests/test_atomic_write_retry.py @@ -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 diff --git a/backend/tests/test_job_record_persistence.py b/backend/tests/test_job_record_persistence.py new file mode 100644 index 0000000..9adfdd7 --- /dev/null +++ b/backend/tests/test_job_record_persistence.py @@ -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" diff --git a/backend/tests/test_realtime_mode.py b/backend/tests/test_realtime_mode.py new file mode 100644 index 0000000..12f90f7 --- /dev/null +++ b/backend/tests/test_realtime_mode.py @@ -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" diff --git a/backend/tests/test_stocksdk_provider.py b/backend/tests/test_stocksdk_provider.py index d6ddc57..de2453f 100644 --- a/backend/tests/test_stocksdk_provider.py +++ b/backend/tests/test_stocksdk_provider.py @@ -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" diff --git a/backend/tests/test_watchlist_batch.py b/backend/tests/test_watchlist_batch.py deleted file mode 100644 index 251d240..0000000 --- a/backend/tests/test_watchlist_batch.py +++ /dev/null @@ -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"] diff --git a/backend/tests/test_watchlist_realtime_split.py b/backend/tests/test_watchlist_realtime_split.py deleted file mode 100644 index 7529a96..0000000 --- a/backend/tests/test_watchlist_realtime_split.py +++ /dev/null @@ -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 diff --git a/frontend/src/components/EChartsIntraday.tsx b/frontend/src/components/EChartsIntraday.tsx index ad86406..dae6b00 100644 --- a/frontend/src/components/EChartsIntraday.tsx +++ b/frontend/src/components/EChartsIntraday.tsx @@ -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 && {date}} 开 - {d.open.toFixed(2)} + {d.open != null ? d.open.toFixed(2) : '—'} 高 {d.high.toFixed(2)} 低 diff --git a/frontend/src/components/EChartsMultiDayIntraday.tsx b/frontend/src/components/EChartsMultiDayIntraday.tsx index 63e0cf5..1db4dc3 100644 --- a/frontend/src/components/EChartsMultiDayIntraday.tsx +++ b/frontend/src/components/EChartsMultiDayIntraday.tsx @@ -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 ? ( <> {info.date} {formatMinuteTime(info.row.datetime)} - 开{info.row.open.toFixed(2)} + 开{info.row.open != null ? info.row.open.toFixed(2) : '—'} 高{info.row.high.toFixed(2)} 低{info.row.low.toFixed(2)} 收{info.row.close.toFixed(2)} diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index f9adbdb..bc34732 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -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) // 仅在交易时段立即获取一次行情 diff --git a/frontend/src/components/StockPreviewDialog.tsx b/frontend/src/components/StockPreviewDialog.tsx index 26f422e..606af2b 100644 --- a/frontend/src/components/StockPreviewDialog.tsx +++ b/frontend/src/components/StockPreviewDialog.tsx @@ -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('daily') - const [intradayDays, setIntradayDays] = useState(loadIntradayDays) + const [intradayDays, setIntradayDays] = useState(loadIntradayDays) const [dateRange, setDateRange] = useState(getDefaultRange) const [showMonitorEditor, setShowMonitorEditor] = useState(false) const [priceAlertDraft, setPriceAlertDraft] = useState(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( + () => 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 ) : (
- {INTRADAY_DAY_OPTIONS.map(days => ( + {dayOptions.map(days => (