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 16:44:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(pipeline): 修复盘后管道除权因子拉取范围逻辑

Browse Source 
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步骤输出详细日志。
This commit is contained in:
shy3130
2026-06-29 17:17:44 +08:00
parent 37934091ac
commit a72445d782
7 changed files with 99 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/kline.py
+11 -2
View File
@@ -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)
backend/app/jobs/daily_pipeline.py
+36 -24
View File
@@ -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")
backend/app/main.py
+2 -1
View File
@@ -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
backend/app/services/financial_sync.py
+15 -3
View File
@@ -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。
backend/app/services/index_sync.py
+11 -1
View File
@@ -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()
backend/app/services/kline_sync.py
+23 -2
View File
@@ -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:
backend/uv.lock
Generated
+1 -1
View File
@@ -2491,7 +2491,7 @@ all = [
[[package]]
name = "tickflow-stock-panel-backend"
version = "0.1.63"
version = "0.1.64"
source = { editable = "." }
dependencies = [
{ name = "apscheduler" },