diff --git a/.gitignore b/.gitignore index 7fba4b6..fe9f019 100644 --- a/.gitignore +++ b/.gitignore @@ -65,6 +65,7 @@ backend/user_prompts/ # ===== AI IDE ===== .trae/ +.zcode/ # ===== TypeScript build cache ===== frontend/tsconfig*.tsbuildinfo diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index 91b3de2..be3d12f 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -717,6 +717,90 @@ async def extend_history(request: Request): raise HTTPException(status_code=500, detail=str(e)) from e +@router.post("/repair_daily") +async def repair_daily(request: Request): + """修正 / 补全日K数据 — 从指定起始日期重拉到今天。 + + 典型场景: 昨天没看盘 / 服务挂了,本地日K缺了若干天。 + 用户选起始日期,复用盘后管道全流程重拉 [start_date ~ 今天]。 + + body: { "start_date": "YYYY-MM-DD" } + 返回 job_id,可轮询 /api/pipeline/jobs 查看进度。 + """ + import asyncio + import traceback as _tb + from datetime import date as _date + try: + body = await request.json() + raw = body.get("start_date") + if not raw: + raise HTTPException(status_code=400, detail="start_date 必填 (YYYY-MM-DD)") + try: + start_date = _date.fromisoformat(str(raw)) + except ValueError: + raise HTTPException(status_code=400, detail="start_date 格式错误 (应为 YYYY-MM-DD)") + + if start_date > _date.today(): + raise HTTPException(status_code=400, detail="起始日期不能晚于今天") + + repo = request.app.state.repo + capset = request.app.state.capabilities + + from app.tickflow.capabilities import Cap + if not capset.has(Cap.KLINE_DAILY_BATCH): + raise HTTPException(status_code=403, detail="需要 Pro+ 权限 (batch K-line)") + + from app.services.repair_daily import run_repair_daily + from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot + from app.api.data import invalidate_storage_cache + + job_id, is_new = job_store.create() + if not is_new: + return {"status": "reused", "job_id": job_id} + + async def task() -> None: + if not try_acquire_run_slot(): + job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试") + return + loop = asyncio.get_event_loop() + qs = getattr(request.app.state, "quote_service", None) + + def progress(stage: str, pct: int, msg: str, + stage_pct: int | None = None, skip_log: bool = False) -> None: + job_store.progress(job_id, stage, pct, msg, + stage_pct=stage_pct, skip_log=skip_log) + + def _run() -> dict: + # 修正运行期间暂停实时行情, 防止覆写同一批 parquet 竞态 + if qs: + with qs.paused(): + return run_repair_daily(repo, capset, start_date, on_progress=progress) + return run_repair_daily(repo, capset, start_date, on_progress=progress) + + try: + job_store.start(job_id) + result = await loop.run_in_executor(_long_task_executor, _run) + if "error" in result: + job_store.fail(job_id, result["error"]) + else: + job_store.succeed(job_id, result) + invalidate_storage_cache() + except Exception as e: + logger.exception("repair_daily failed: job_id=%s", job_id) + job_store.fail(job_id, str(e)) + invalidate_storage_cache() + finally: + release_run_slot() + + asyncio.create_task(task()) + return {"status": "started", "job_id": job_id} + except HTTPException: + raise + except Exception as e: + logger.error("repair_daily error: %s\n%s", e, _tb.format_exc()) + raise HTTPException(status_code=500, detail=str(e)) from e + + @router.post("/rebuild_enriched") async def rebuild_enriched(request: Request): """全量重算 enriched 表 — 不获取任何数据,仅基于已有 kline_daily + adj_factor 重算复权+指标。 diff --git a/backend/app/api/pipeline.py b/backend/app/api/pipeline.py index fcd86a6..89968f1 100644 --- a/backend/app/api/pipeline.py +++ b/backend/app/api/pipeline.py @@ -44,6 +44,8 @@ async def run_now(request: Request) -> dict: if not try_acquire_run_slot(): job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试") return + # 管道运行期间暂停实时行情取数, 防止覆写同一批 parquet 竞态 + qs = getattr(request.app.state, "quote_service", None) try: job_store.start(job_id) loop = asyncio.get_event_loop() @@ -52,10 +54,13 @@ async def run_now(request: Request) -> dict: skip_log: bool = False) -> None: job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log) - result = await loop.run_in_executor( - _long_task_executor, - lambda: daily_pipeline.run_now(repo, capset, on_progress=progress), - ) + def _run() -> dict: + if qs: + with qs.paused(): + return daily_pipeline.run_now(repo, capset, on_progress=progress) + return daily_pipeline.run_now(repo, capset, on_progress=progress) + + result = await loop.run_in_executor(_long_task_executor, _run) job_store.succeed(job_id, result) invalidate_storage_cache() repo.refresh_cache() # 刷新 Polars 缓存 diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 38bc288..b72dc49 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -662,6 +662,9 @@ def update_realtime_quotes(req: RealtimeQuotesPrefs, request: Request) -> dict: if qs: qs.disable() return {"realtime_quotes_enabled": False, "realtime_allowed": False} + if req.realtime_quotes_enabled and qs and qs.is_paused(): + # 管道/数据修正运行期间禁止开启实时行情 — 防止写盘竞态 + raise HTTPException(status_code=409, detail="数据同步运行中,实时行情已临时暂停,请稍后再开启") if req.realtime_quotes_enabled and qs and qs.realtime_mode() == "watchlist" and not preferences.get_realtime_watchlist_symbols(): preferences.save({"realtime_quotes_enabled": False}) return {"realtime_quotes_enabled": False, "realtime_allowed": True, "mode": "watchlist", "error": "watchlist_empty"} diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index af992d0..21cc113 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -102,11 +102,15 @@ def run_now( repo: KlineRepository, capset: CapabilitySet, on_progress: ProgressCb | None = None, + override_start_date: _date | None = None, ) -> dict: """立即执行一次盘后管道,支持进度回调。 跳过的 stage **不 emit**,避免前端把"无 capability"的卡片错误标记为 active/done。 result 里带 skipped_stages 列表供前端展示。 + + override_start_date: 传入时强制走 batch 拉取分支,用该日期作为日K/除权/指数的 + 拉取起点(到今天),用于「数据修正/补数据」场景。None 时走原有自动判定逻辑。 """ emit = on_progress or _noop skipped: list[str] = [] @@ -127,6 +131,7 @@ def run_now( emit("resolve_universe", 10, f"标的池规模:{len(universe)} 只") # Step 1: 日 K 同步 + # override_start_date 传入 → 强制 batch 拉取 [override_start_date ~ today] (数据修正) # 付费档 + 今天有数据 → 实时行情接口拉一次覆写(1请求全市场) # 有历史数据 → batch K-line API 补齐缺口 # 无任何数据 → batch K-line API 拉首次 1 年 @@ -135,15 +140,36 @@ def run_now( today = _date.today() today_exists = latest_daily and latest_daily >= today new_daily_days = 0 - # 日K范围拉取的起点(分支3补缺口/分支4首次); 实时增量/跳过时为 None。 + # 日K范围拉取的起点(分支3补缺口/分支4首次/数据修正); 实时增量/跳过时为 None。 # 供 Step 1.5 除权因子回溯范围对齐: 范围拉取→用日K范围, 非范围→最近N天兜底。 daily_range_start: _date | None = None - # A 股日K拉取开关(默认开);关闭时跳过日K同步,保留已有数据 + # A 股日K拉取开关(默认开);关闭时跳过日K同步,保留已有数据。 + # 数据修正(override_start_date)时即使关闭开关也强制拉取 — 修正就是来补数据的。 pull_a_share = _prefs.get_pipeline_pull_a_share() - if not pull_a_share: + if not pull_a_share and not override_start_date: emit("sync_daily", 45, "已跳过 A 股日K同步(拉取内容未勾选)") logger.info("sync_daily: skipped (pipeline_pull_a_share=False)") + elif override_start_date: + # 数据修正: 强制用传入日期作起点 batch 拉取, 忽略实时行情覆写分支。 + start_date = override_start_date + daily_range_start = start_date + emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…") + logger.info("sync_daily: [%s ~ %s] repair/override", start_date, today) + + def _daily_chunk_progress(cur: int, tot: int) -> None: + emit("sync_daily", 12 + int(33 * cur / tot), + f"日K 批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True) + written_daily = kline_sync.sync_and_persist_daily_batch( + universe, repo, capset, + start_date=_dt.combine(start_date, _dt.min.time()), + end_date=_dt.combine(today, _dt.min.time()), + on_chunk_done=_daily_chunk_progress, + ) + gap_days = (today - start_date).days + new_daily_days = gap_days + emit("sync_daily", 45, f"日K 完成,覆盖 {gap_days} 天") + logger.info("sync_daily: [%s ~ %s] done, %d days", start_date, today, gap_days) elif today_exists and capset.has(Cap.QUOTE_POOL) and _prefs.get_daily_data_provider() == "tickflow": # 付费档:今天有数据(QuoteService 已落盘)→ 实时行情覆写,确保最新。 # free/none 档无 quote.pool 能力,即便今天已有数据(如从 expert 降级), @@ -358,7 +384,11 @@ def run_now( d.name[5:] for d in index_dir.glob("date=*") if d.is_dir() and d.name.startswith("date=") ) if index_dir.exists() else [] - index_start = _date.fromisoformat(index_dates[-1]) if index_dates else today - _td(days=365) + # 数据修正模式下用传入起点; 否则用本地指数最新日期补到今天 + if override_start_date: + index_start = override_start_date + else: + index_start = _date.fromisoformat(index_dates[-1]) if index_dates else today - _td(days=365) def _index_chunk(cur: int, tot: int) -> None: emit("sync_index", 88, f"指数日K批次 {cur}/{tot}", @@ -835,8 +865,14 @@ def start_scheduler(repo: KlineRepository, capset: CapabilitySet) -> AsyncIOSche # 旧 capset —— 否则 Key 中途过期/续费后, 调度管道仍按旧档位打端点。 app_state = _get_app_state() capset_live = getattr(app_state, "capabilities", None) or capset + # 管道运行期间暂停实时行情取数, 防止覆写同一批 parquet 竞态 + qs = getattr(app_state, "quote_service", None) try: - result = run_now(repo, capset_live, on_progress=on_progress) + if qs: + with qs.paused(): + result = run_now(repo, capset_live, on_progress=on_progress) + else: + result = run_now(repo, capset_live, on_progress=on_progress) finally: # 即便有阶段软失败(run_now 末尾抛 PipelineStageError), 已落盘的日K/enriched # 仍需刷进内存缓存, 否则 live_agg 基准列停留在旧交易日。放 finally 保证部分 diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 22252fd..f3580ce 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -27,6 +27,7 @@ import logging import threading import time from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager from datetime import date, time as dt_time import polars as pl @@ -136,6 +137,10 @@ class QuoteService: self._fetch_lock = threading.Lock() self._running = False self._enabled = False # 全局开关 (持久化到 preferences) + # 暂停态: 盘后管道/数据修正运行期间临时暂停取数, 防止与管道写同一批 parquet 竞态。 + # 与 _enabled 不同 — pause 不改 preferences、不 stop 线程, 仅让轮询循环跳过取数; + # 进程重启后 _paused 归零, 从 preferences 恢复真实开关态, 无"假关闭"副作用。 + self._paused = False self._interval = self.DEFAULT_INTERVAL self._thread: threading.Thread | None = None self._repo = None # 延迟注入, 避免循环导入 @@ -210,6 +215,44 @@ class QuoteService: self.stop() logger.info("行情服务已关闭") + # ================================================================ + # 临时暂停 (盘后管道/数据修正期间, 防止写盘竞态) + # ================================================================ + + def pause(self) -> None: + """临时暂停行情轮询取数 (不关闭线程、不改 preferences)。 + + 用于盘后管道/数据修正运行期间, 防止实时行情覆写管道正在写的 parquet。 + 与 stop() 的区别: 线程继续存活但跳过 _fetch_quotes; preferences 开关态不变, + 管道结束调用 resume() 即恢复。线程级检查, 即时生效, 无 join 等待。 + """ + self._paused = True + logger.info("行情轮询已临时暂停 (管道/修正运行中)") + + def resume(self) -> None: + """恢复暂停的行情轮询取数 (对应 pause)。""" + self._paused = False + logger.info("行情轮询已恢复") + + def is_paused(self) -> bool: + """是否处于临时暂停态 (管道运行期间)。""" + return self._paused + + @contextmanager + def paused(self): + """上下文管理器: 进入时暂停轮询取数, 退出时(含异常)自动恢复。 + + 供盘后管道/数据修正复用: + with quote_service.paused(): + run_pipeline(...) + 无论正常结束还是异常/crash, finally 都会 resume (除非进程直接被 kill)。 + """ + self.pause() + try: + yield + finally: + self.resume() + def boot_check(self) -> None: """启动时检查 preferences,若 enabled 则自动启动。 @@ -391,6 +434,7 @@ class QuoteService: return { "enabled": self._enabled, "running": self._running, + "paused": self._paused, "mode": mode, "realtime_allowed": mode != "none", "watchlist_symbol_count": len(preferences.get_realtime_watchlist_symbols()), @@ -420,21 +464,24 @@ class QuoteService: def _poll_loop(self) -> None: while self._running and self._enabled: try: - phase = self._market_phase() - if self._should_fetch_for_phase(phase): - is_final = phase in {"morning_final", "close_final"} - ok = self._fetch_quotes(final=is_final) - if is_final: - key = self._final_sync_key(phase) - if key and ok: - self._final_sync_done.add(key) - self._final_sync_failed.pop(key, None) - logger.info("%s 最终行情同步完成, 进入休盘态", "午休" if phase == "morning_final" else "收盘") - elif key: - self._final_sync_failed[key] = "fetch_failed" - logger.warning("%s 最终行情同步失败, 将继续重试", "午休" if phase == "morning_final" else "收盘") - else: - logger.debug("非轮询阶段(%s), 跳过行情轮询", phase) + # 管道/数据修正运行期间临时暂停取数, 防止与管道写同一批 parquet 竞态。 + # 线程继续存活 + 分片 sleep, resume() 后即时恢复, 无需重启线程。 + if not self._paused: + phase = self._market_phase() + if self._should_fetch_for_phase(phase): + is_final = phase in {"morning_final", "close_final"} + ok = self._fetch_quotes(final=is_final) + if is_final: + key = self._final_sync_key(phase) + if key and ok: + self._final_sync_done.add(key) + self._final_sync_failed.pop(key, None) + logger.info("%s 最终行情同步完成, 进入休盘态", "午休" if phase == "morning_final" else "收盘") + elif key: + self._final_sync_failed[key] = "fetch_failed" + logger.warning("%s 最终行情同步失败, 将继续重试", "午休" if phase == "morning_final" else "收盘") + else: + logger.debug("非轮询阶段(%s), 跳过行情轮询", phase) except Exception as e: # noqa: BLE001 logger.warning("行情轮询异常: %s", e) diff --git a/backend/app/services/repair_daily.py b/backend/app/services/repair_daily.py new file mode 100644 index 0000000..5d70f02 --- /dev/null +++ b/backend/app/services/repair_daily.py @@ -0,0 +1,54 @@ +"""修正 / 补全日K数据 — 完全复用盘后管道,只是日期范围由用户传入。 + +典型场景: 昨天没看盘 / 服务挂了一天,本地日K缺了若干天。 + +设计原则: 不重复盘后管道的任何逻辑。直接调用 daily_pipeline.run_now(), +通过 override_start_date 参数把"自动算日期"换成"用户指定起点", +其余 (维表 / A股日K / 除权因子 / enriched / 指数 / ETF / 错误兜底) 全部原样复用。 +这样修正功能与盘后管道永远保持一致,不会出现遗漏。 + +落盘是 merge-upsert (按 symbol+date 去重 keep="last"), 新数据天然覆盖旧值, +不会产生重复,也不需要先删分区。 +""" +from __future__ import annotations + +import logging +from collections.abc import Callable +from datetime import date + +from app.tickflow.capabilities import CapabilitySet +from app.tickflow.repository import KlineRepository + +logger = logging.getLogger(__name__) + + +def run_repair_daily( + repo: KlineRepository, + capset: CapabilitySet, + start_date: date, + end_date: date | None = None, + on_progress: Callable | None = None, +) -> dict: + """修正 / 补全数据 — 复用盘后管道,日期范围由用户指定。 + + 通过 run_now(override_start_date=start_date) 把日K/除权/指数的拉取起点 + 统一设为 start_date (到今天),其余流程与盘后管道完全一致。 + + Args: + repo: 数据仓库 + capset: 权限集 + start_date: 用户选定的起始日期 + end_date: 保留参数(目前盘后管道固定拉到今天, 此值未使用, 为接口兼容保留) + on_progress: 进度回调 + + Returns: + run_now() 的完整结果 dict。 + """ + today = date.today() + if start_date > today: + return {"error": "起始日期不能晚于今天"} + + logger.info("repair_daily: run pipeline with override_start_date=%s", start_date) + + from app.jobs.daily_pipeline import run_now + return run_now(repo, capset, on_progress=on_progress, override_start_date=start_date) diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index 25bef41..db1dea9 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -344,6 +344,8 @@ export function Layout() { const toggleQuote = useToggleRealtimeQuotes() const isRunning = quoteStatus?.running ?? false const isTrading = quoteStatus?.is_trading_hours ?? false + // 管道/数据修正运行期间实时行情被临时暂停 — 此时禁止开启 + const isPaused = quoteStatus?.paused ?? false const tier = tierRank(caps?.label ?? '') const isNoneTier = tier < 0 const isWatchlistMode = tier === 0 @@ -589,12 +591,13 @@ export function Layout() { + + {!hasBatchCap && ( + + 需 Pro+ 权限 + + )} + + ) +} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index ac6ef72..01ee7d0 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -957,6 +957,7 @@ export const api = { request<{ enabled: boolean running: boolean + paused?: boolean mode?: 'none' | 'watchlist' | 'full_market' realtime_allowed?: boolean interval_s: number @@ -1205,6 +1206,11 @@ export const api = { method: 'POST', body: JSON.stringify({ value, unit }), }), + repairDaily: (startDate: string) => + request<{ status: string; job_id: string }>('/api/kline/repair_daily', { + method: 'POST', + body: JSON.stringify({ start_date: startDate }), + }), extendMinuteHistory: (value: number, unit: 'day' | 'month') => request<{ status: string; job_id: string }>('/api/kline/extend_minute_history', { method: 'POST', diff --git a/frontend/src/pages/Data.tsx b/frontend/src/pages/Data.tsx index b84b92d..f13e18a 100644 --- a/frontend/src/pages/Data.tsx +++ b/frontend/src/pages/Data.tsx @@ -15,6 +15,7 @@ import { SlidersHorizontal, AlertTriangle, Info, + WandSparkles, } from 'lucide-react' import { Link } from 'react-router-dom' import { EndpointTestDialog } from '@/components/EndpointTestDialog' @@ -39,6 +40,7 @@ import { SectionTitle, HistoryRow } from '@/components/data/SectionTitle' import { SettingsModal } from '@/components/data/SettingsModal' import { ScheduleEditor } from '@/components/data/ScheduleEditor' import { ExtendHistoryPanel } from '@/components/data/ExtendHistoryPanel' +import { RepairDailyPanel } from '@/components/data/RepairDailyPanel' import { EnrichedRebuildPanel } from '@/components/data/EnrichedRebuildPanel' import { MinuteSyncConfig } from '@/components/data/MinuteSyncConfig' import { PipelineScopeConfig } from '@/components/data/PipelineScopeConfig' @@ -131,6 +133,7 @@ export function Data() { const [schemaTable, setSchemaTable] = useState(null) const [showEndpointTest, setShowEndpointTest] = useState(false) const [showCreateExt, setShowCreateExt] = useState(false) + const [showRepair, setShowRepair] = useState(false) const [editingExt, setEditingExt] = useState(null) const [indexBatchInput, setIndexBatchInput] = useState('100') @@ -557,6 +560,14 @@ export function Data() { 数据范围 +