From a72445d782599bc017e038f73c1de49145391bfd Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 29 Jun 2026 17:17:44 +0800 Subject: [PATCH] =?UTF-8?q?fix(pipeline):=20=E4=BF=AE=E5=A4=8D=E7=9B=98?= =?UTF-8?q?=E5=90=8E=E7=AE=A1=E9=81=93=E9=99=A4=E6=9D=83=E5=9B=A0=E5=AD=90?= =?UTF-8?q?=E6=8B=89=E5=8F=96=E8=8C=83=E5=9B=B4=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit daily_pipeline 除权因子改用日K范围回填 (实时增量模式兜底最近数日), 移除 max_date/incremental 依赖;kline_sync 修复 _normalize_adj_factor 重复列重命名崩溃, 新增 fetch_adj_factor_single 单股拉取; kline.py fallback 改用 compute_enriched 应用复权; index_sync 指数/ETF 日K同步增加批次进度回调; 同步指数日K步骤输出详细日志。 --- backend/app/api/kline.py | 13 +++++- backend/app/jobs/daily_pipeline.py | 60 +++++++++++++++----------- backend/app/main.py | 3 +- backend/app/services/financial_sync.py | 18 ++++++-- backend/app/services/index_sync.py | 12 +++++- backend/app/services/kline_sync.py | 25 ++++++++++- backend/uv.lock | 2 +- 7 files changed, 99 insertions(+), 34 deletions(-) diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index 860ece8..8e0261f 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -7,7 +7,7 @@ from typing import Optional from fastapi import APIRouter, HTTPException, Query, Request -from app.indicators.pipeline import compute_enriched_single +from app.indicators.pipeline import compute_enriched, compute_enriched_single from app.services import kline_sync logger = logging.getLogger(__name__) @@ -126,7 +126,16 @@ def get_daily( raise HTTPException(status_code=502, detail=f"TickFlow fetch failed: {e}") from e if raw.is_empty(): return {"symbol": symbol, "name": stock_name, "stock_info": stock_info, "rows": []} - enriched = compute_enriched_single(raw) + # 拉除权因子做前复权 (Starter+ 有权限), 否则空 df → compute_enriched 退回未复权 + factors = pl.DataFrame() + capset = getattr(request.app.state, "capabilities", None) + try: + from app.tickflow.capabilities import Cap + if capset and capset.has(Cap.ADJ_FACTOR): + factors = kline_sync.fetch_adj_factor_single(symbol) + except Exception as e: # noqa: BLE001 + logger.debug("单股除权因子拉取失败 %s: %s", symbol, e) + enriched = compute_enriched(raw, factors=factors) rows = enriched.tail(days).to_dicts() # 即使 live 模式也尝试追加实时蜡烛 rows = _maybe_inject_live_candle(request, symbol, rows) diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index 59a829a..bf02072 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -110,6 +110,9 @@ def run_now( today = _date.today() today_exists = latest_daily and latest_daily >= today new_daily_days = 0 + # 日K范围拉取的起点(分支3补缺口/分支4首次); 实时增量/跳过时为 None。 + # 供 Step 1.5 除权因子回溯范围对齐: 范围拉取→用日K范围, 非范围→最近N天兜底。 + daily_range_start: _date | None = None # A 股日K拉取开关(默认开);关闭时跳过日K同步,保留已有数据 pull_a_share = _prefs.get_pipeline_pull_a_share() @@ -130,6 +133,7 @@ def run_now( # 也覆盖"今天已有数据但无实时行情权限(free/none)"的降级场景: # 此时 start_date = latest_daily = today,batch 刷新当天日K。 start_date = latest_daily + daily_range_start = start_date emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…") logger.info("sync_daily: [%s ~ %s] %s", start_date, today, "refresh today" if today_exists else "gap fill") @@ -150,6 +154,7 @@ def run_now( else: # 首次：无任何数据 → batch 拉 1 年 start_date = today - _td(days=365) + daily_range_start = start_date emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…") logger.info("sync_daily: [%s ~ %s] initial fetch", start_date, today) @@ -167,36 +172,22 @@ def run_now( logger.info("sync_daily: [%s ~ %s] done", start_date, today) _invalidate("daily") - # Step 1.5: 增量同步除权因子 — 从已有数据最新日期的下一天开始获取 + # Step 1.5: 同步除权因子 — 范围与日K拉取方式对齐 + # 日K范围拉取(补缺口/首次) → 除权用日K范围 [daily_range_start, now] + # 首次会覆盖整个日K区间内的历史除权事件; 补缺口天然只增量(起点=latest_daily≈昨天) + # 日K实时增量/跳过(分支2/分支1) → 除权兜底拉最近 30 天, 补可能遗漏的新除权 + # (这两类分支不拉历史日K, 除权不能用日K范围, 只能兜底最近几日) written_adj = 0 affected_symbols: list[str] = [] if capset.has(Cap.ADJ_FACTOR): from datetime import datetime, timedelta adj_end = datetime.now() - # 从已有除权因子数据的最新日期开始获取，避免重复拉取 - adj_factor_path = repo.store.data_dir / "adj_factor" / "all.parquet" - fallback_start = adj_end - timedelta(days=30) - if adj_factor_path.exists(): - try: - from datetime import date as date_cls - max_date = pl.scan_parquet(adj_factor_path).select( - pl.col("trade_date").max() - ).collect().item() - if max_date is not None: - # trade_date 可能是 date / datetime / string 类型 - if isinstance(max_date, str): - td = date_cls.fromisoformat(max_date) - elif isinstance(max_date, datetime): - td = max_date.date() - else: - td = max_date - adj_start = datetime.combine(td, datetime.min.time()) - else: - adj_start = fallback_start - except Exception: - adj_start = fallback_start + if daily_range_start is not None: + adj_start = datetime.combine(daily_range_start, datetime.min.time()) else: - adj_start = fallback_start + # 日K实时增量/跳过时, 除权兜底拉最近 N 天, 覆盖周末/长假/停机期间的新除权事件。 + # 15 天: 覆盖春节/国庆最长约10天长假 + 故障恢复缓冲; sync_adj_factor 内部 merge+unique 幂等, 多拉无副作用。 + adj_start = adj_end - timedelta(days=15) adj_start_str = adj_start.strftime("%Y-%m-%d") adj_end_str = adj_end.strftime("%Y-%m-%d") emit("sync_adj", 50, f"获取除权因子 [{adj_start_str} ~ {adj_end_str}]…") @@ -312,33 +303,46 @@ def run_now( if pull_etf: _types.append("ETF") emit("sync_index", 88, f"同步{'+'.join(_types)}日K…") + # 子阶段进度分配: 88.0(开始) → 89.0(完成), 指数占前半, ETF 占后半 try: if pull_index: + emit("sync_index", 88, "同步指数维表…") index_count = index_sync.sync_index_instruments(repo, pull_index=True, pull_etf=False) + emit("sync_index", 88, f"指数维表完成,{index_count} 只") index_dir = repo.store.data_dir / "kline_index_enriched" index_dates = sorted( 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) + + def _index_chunk(cur: int, tot: int) -> None: + emit("sync_index", 88, f"指数日K批次 {cur}/{tot}", + stage_pct=int(100 * cur / tot) if tot else 100, skip_log=cur < tot) + written_index_daily = index_sync.sync_and_persist_index_daily( repo, capset, start_date=_dt.combine(index_start, _dt.min.time()), end_date=_dt.combine(today, _dt.min.time()), + on_chunk_done=_index_chunk, ) + emit("sync_index", 88, f"指数日K完成,{written_index_daily} 行") _invalidate("index_instruments") _invalidate("index_daily") _invalidate("index_enriched") if pull_etf: + emit("sync_index", 88, "同步 ETF 维表…") etf_count = index_sync.sync_etf_instruments(repo) + emit("sync_index", 88, f"ETF 维表完成,{etf_count} 只") etf_symbols: list[str] = [] etf_inst = repo.get_etf_instruments() if not etf_inst.is_empty() and "symbol" in etf_inst.columns: etf_symbols = sorted(set(etf_inst["symbol"].to_list())) if etf_symbols and capset.has(Cap.ADJ_FACTOR): try: + emit("sync_index", 88, "同步 ETF 除权因子…") from datetime import datetime, timedelta adj_end = datetime.now() adj_path = repo.store.data_dir / "adj_factor_etf" / "all.parquet" @@ -361,6 +365,7 @@ def run_now( end_time=adj_end, ) etf_adj_symbols = len(affected_etfs) + emit("sync_index", 88, f"ETF 除权因子完成,{etf_adj_symbols} 只") except Exception as e: # noqa: BLE001 logger.warning("ETF adj_factor skipped: %s", e) etf_dir = repo.store.data_dir / "kline_etf_enriched" @@ -369,12 +374,19 @@ def run_now( if d.is_dir() and d.name.startswith("date=") ) if etf_dir.exists() else [] etf_start = _date.fromisoformat(etf_dates[-1]) if etf_dates else today - _td(days=365) + + def _etf_chunk(cur: int, tot: int) -> None: + emit("sync_index", 88, f"ETF 日K批次 {cur}/{tot}", + stage_pct=int(100 * cur / tot) if tot else 100, skip_log=cur < tot) + written_etf_daily = index_sync.sync_and_persist_etf_daily( repo, capset, start_date=_dt.combine(etf_start, _dt.min.time()), end_date=_dt.combine(today, _dt.min.time()), + on_chunk_done=_etf_chunk, ) + emit("sync_index", 88, f"ETF 日K完成,{written_etf_daily} 行") _invalidate("etf_instruments") _invalidate("etf_daily") diff --git a/backend/app/main.py b/backend/app/main.py index 4b15d2e..dcc7366 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -98,7 +98,8 @@ async def lifespan(app: FastAPI): except Exception as e: # noqa: BLE001 logger.warning("内置扩展表初始化失败 (不影响启动): %s", e) - # 财务数据独立调度 (需 Expert 套餐) + # 财务数据 (需 Expert 套餐): 仅初始化调度器供 /api/financials/sync/* 手动同步, + # 不启动自动调度——用户在「财务分析」页点「同步」手动拉取。 from app.services.financial_sync import financial_scheduler financial_scheduler.start(store.data_dir, capset) app.state.financial_scheduler = financial_scheduler diff --git a/backend/app/services/financial_sync.py b/backend/app/services/financial_sync.py index 22dee64..983b67f 100644 --- a/backend/app/services/financial_sync.py +++ b/backend/app/services/financial_sync.py @@ -200,13 +200,18 @@ class FinancialScheduler: # 手动同步(run_now)是否正在进行。前端据此显示"同步中"并防重复点击。 self._is_syncing = False - def start(self, data_dir: Path, capset: CapabilitySet) -> None: + def start(self, data_dir: Path, capset: CapabilitySet, *, auto_schedule: bool = False) -> None: + """初始化调度器，并按需启动周期同步后台任务。 + + auto_schedule=False (默认): 仅初始化 (设置数据目录/能力 + 恢复 last_sync), + 供 /api/financials/sync/* 手动同步使用, 不启动自动调度。 + auto_schedule=True: 额外启动每周一次的 metrics 自动同步 (启动后 60s 首跑)。 + """ if not capset.has(Cap.FINANCIAL): logger.info("FinancialScheduler skipped: no FINANCIAL capability") return self._data_dir = data_dir self._capset = capset - self._running = True # 从持久化恢复上次同步时间: 重启后前端仍能显示真实最后同步时间,而非"尚未同步" try: from app.services import preferences @@ -227,8 +232,15 @@ class FinancialScheduler: logger.info("FinancialScheduler restored last_sync: %s", list(self._last_sync.keys())) except Exception as e: # noqa: BLE001 logger.warning("restore financial_sync_times failed: %s", e) + + if not auto_schedule: + # 仅初始化 (手动同步用), 不启动周期任务。 + logger.info("FinancialScheduler initialized (auto-schedule disabled; manual sync only)") + return + + self._running = True self._task = asyncio.create_task(self._run_loop()) - logger.info("FinancialScheduler started") + logger.info("FinancialScheduler started (auto-schedule enabled)") def _record_sync(self, table: str) -> None: """记录一张表的同步完成时间: 更新内存 + 持久化到 preferences.json。 diff --git a/backend/app/services/index_sync.py b/backend/app/services/index_sync.py index 4f689b9..42f3b1b 100644 --- a/backend/app/services/index_sync.py +++ b/backend/app/services/index_sync.py @@ -8,6 +8,7 @@ from __future__ import annotations import logging import gc +from collections.abc import Callable from datetime import datetime, timedelta import polars as pl @@ -194,11 +195,13 @@ def sync_and_persist_index_daily( start_date: datetime | None = None, end_date: datetime | None = None, symbols_override: list[str] | None = None, + on_chunk_done: Callable[[int, int], None] | None = None, ) -> int: """同步指数/ETF 日K到独立 parquet,并计算 enriched。 symbols_override 非空时,只拉这些代码(跳过 instruments 表),用于自定义范围。 否则取 index_instruments 表全量(指数+ETF 合并存储)。 + on_chunk_done(current, total) 每个批次完成后回调。 """ if not capset.has(Cap.KLINE_DAILY_BATCH): return 0 @@ -248,6 +251,8 @@ def sync_and_persist_index_daily( repo.append_index_enriched(enriched) total_rows += raw.height logger.info("index/etf daily synced: %d/%d chunks, +%d rows", i + 1, len(chunks), raw.height) + if on_chunk_done: + on_chunk_done(i + 1, len(chunks)) del raw, enriched gc.collect() repo.refresh_index_views() @@ -292,8 +297,11 @@ def sync_and_persist_etf_daily( start_date: datetime | None = None, end_date: datetime | None = None, symbols_override: list[str] | None = None, + on_chunk_done: Callable[[int, int], None] | None = None, ) -> int: - """同步 ETF 日K到独立 kline_etf_* parquet,并计算 ETF enriched。""" + """同步 ETF 日K到独立 kline_etf_* parquet,并计算 ETF enriched。 + on_chunk_done(current, total) 每个批次完成后回调。 + """ if not capset.has(Cap.KLINE_DAILY_BATCH): return 0 @@ -344,6 +352,8 @@ def sync_and_persist_etf_daily( repo.append_etf_enriched(enriched) total_rows += raw.height logger.info("etf daily synced: %d/%d chunks, +%d rows", i + 1, len(chunks), raw.height) + if on_chunk_done: + on_chunk_done(i + 1, len(chunks)) del raw, enriched gc.collect() repo.refresh_index_views() diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index cff2f70..7a4d5f5 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -241,8 +241,14 @@ def _normalize_adj_factor(raw) -> pl.DataFrame: df = pl.from_pandas(raw.reset_index() if hasattr(raw, "reset_index") else raw) if df.is_empty(): return df - rename_map = {"timestamp": "trade_date", "date": "trade_date", "adj_factor": "ex_factor"} - df = df.rename({k: v for k, v in rename_map.items() if k in df.columns}) + # rename: timestamp/date → trade_date, adj_factor → ex_factor + # 注意: 新版 SDK 可能同时返回 timestamp 和 trade_date (或 adj_factor 和 ex_factor), + # 直接 rename 会产生重复列报错。仅当目标列不存在时才 rename。 + rename_map: dict[str, str] = {} + for src, dst in (("timestamp", "trade_date"), ("date", "trade_date"), ("adj_factor", "ex_factor")): + if src in df.columns and dst not in df.columns: + rename_map[src] = dst + df = df.rename(rename_map) if "trade_date" in df.columns: if df.schema["trade_date"] in {pl.Int64, pl.Int32, pl.UInt64, pl.UInt32, pl.Float64, pl.Float32}: df = df.with_columns( @@ -481,6 +487,21 @@ def fetch_minute_single(symbol: str, trade_date: date) -> pl.DataFrame: return pl.DataFrame() +def fetch_adj_factor_single(symbol: str) -> pl.DataFrame: + """从 TickFlow 实时拉取单股除权因子(不写入本地), 用于单股 K 线即时前复权。 + + 返回结构: symbol, trade_date, ex_factor (空 DataFrame 表示无除权事件或拉取失败)。 + 与 _apply_adj_factor / compute_enriched 的 factors 参数格式一致。 + """ + tf = get_client() + try: + raw = tf.klines.ex_factors([symbol], as_dataframe=True, show_progress=False) + except Exception as e: # noqa: BLE001 + logger.warning("fetch_adj_factor_single(%s) failed: %s", symbol, e) + return pl.DataFrame() + return _normalize_adj_factor(raw) + + def _latest_minute_datetime(repo: KlineRepository) -> datetime | None: """本地分钟 K 数据的最新时间。""" try: diff --git a/backend/uv.lock b/backend/uv.lock index 233de5f..7ec7052 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -2491,7 +2491,7 @@ all = [ [[package]] name = "tickflow-stock-panel-backend" -version = "0.1.63" +version = "0.1.64" source = { editable = "." } dependencies = [ { name = "apscheduler" },