From acd30dd7288c4b9b81eddf007678383347ff900c Mon Sep 17 00:00:00 2001 From: wshy Date: Thu, 25 Jun 2026 10:16:34 +0800 Subject: [PATCH] =?UTF-8?q?fix(pipeline):=20free/none=20=E6=A1=A3=E9=99=8D?= =?UTF-8?q?=E7=BA=A7=E6=97=B6=E8=AF=AF=E8=B0=83=E5=AE=9E=E6=97=B6=E8=A1=8C?= =?UTF-8?q?=E6=83=85=E6=8E=A5=E5=8F=A3=20(#9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从 expert 切换到 free 档后,本地 kline_daily 仍保留今天实时落盘的数据, 导致 run_now() 中 today_exists=True,误入实时行情覆写分支调 quotes.get_by_universes()。但 free/none 档走 free-api 服务器,无实时行情 端点,调用报错或返回空,当天日K数据可能异常。 根因:分支①只判断「今天有无数据」,未判断「当前档位有无实时行情能力」。 sync_daily_by_quotes 内部也无 capability 守卫。 修复:分支①条件增加 capset.has(Cap.QUOTE_POOL)(starter+ 才有)。 free/none 档降级到 elif latest_daily 分支,用 klines.batch(period='1d') 历史日K接口刷新当天,这是 free 档唯一可用且正确的数据源。 --- backend/app/jobs/daily_pipeline.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index d22edfc..0ffa0d2 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -102,8 +102,8 @@ def run_now( emit("resolve_universe", 10, f"标的池规模:{len(universe)} 只") # Step 1: 日 K 同步 - # 今天有数据 → 实时行情接口拉一次覆写（1请求全市场） - # 今天没数据 → batch K-line API 补齐 + # 付费档 + 今天有数据 → 实时行情接口拉一次覆写（1请求全市场） + # 有历史数据 → batch K-line API 补齐缺口 # 无任何数据 → batch K-line API 拉首次 1 年 from datetime import date as _date, timedelta as _td, datetime as _dt latest_daily = repo.latest_daily_date() @@ -111,18 +111,23 @@ def run_now( today_exists = latest_daily and latest_daily >= today new_daily_days = 0 - if today_exists: - # 今天有数据（QuoteService 已落盘）→ 实时行情覆写，确保最新 + if today_exists and capset.has(Cap.QUOTE_POOL): + # 付费档:今天有数据(QuoteService 已落盘)→ 实时行情覆写,确保最新。 + # free/none 档无 quote.pool 能力,即便今天已有数据(如从 expert 降级), + # 也降级到下方 batch 路径刷新,避免调用无权限的实时行情接口。 emit("sync_daily", 12, f"获取日K [{today} ~ {today}] 实时行情…") written_daily = kline_sync.sync_daily_by_quotes(repo) new_daily_days = 1 emit("sync_daily", 45, f"日K 完成,{written_daily} 只标的") logger.info("sync_daily: [%s ~ %s] live quotes, %d symbols", today, today, written_daily) elif latest_daily: - # 有历史但今天没数据 → batch 补齐缺口 + # 有历史 → batch 补齐缺口。 + # 也覆盖"今天已有数据但无实时行情权限(free/none)"的降级场景: + # 此时 start_date = latest_daily = today,batch 刷新当天日K。 start_date = latest_daily emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…") - logger.info("sync_daily: [%s ~ %s] gap fill", start_date, today) + logger.info("sync_daily: [%s ~ %s] %s", start_date, today, + "refresh today" if today_exists else "gap fill") def _daily_chunk_progress(cur: int, tot: int) -> None: emit("sync_daily", 12 + int(33 * cur / tot),