gavindiaz/tick-stock-panel
1
0
Fork 0
mirror of https://github.com/shy3130/tick-stock-panel.git synced 2026-11-02 10:26:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #404 from kevin9327/fix/etf-adj-history-recompute

Browse Source 
fix(etf): 增量日K用完整本地历史重算前复权, 除权后不再留下跳空
This commit is contained in:
wshy
2026-09-24 08:54:48 +08:00
committed by GitHub
parent bdc1478e14 e3baba6bfe
commit 9c1abd7fe5
4 changed files with 244 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/jobs/daily_pipeline.py
+3 -11
View File
@@ -585,17 +585,9 @@ def run_now(
from datetime import datetime, timedelta
adj_end = datetime.now()
adj_path = repo.store.data_dir / "adj_factor_etf" / "all.parquet"
fallback_start = adj_end - timedelta(days=30)
adj_start = fallback_start
if adj_path.exists():
max_date = pl.scan_parquet(adj_path).select(pl.col("trade_date").max()).collect().item()
if max_date is not None:
if isinstance(max_date, str):
adj_start = datetime.combine(_date.fromisoformat(max_date), datetime.min.time())
elif isinstance(max_date, datetime):
adj_start = datetime.combine(max_date.date(), datetime.min.time())
else:
adj_start = datetime.combine(max_date, datetime.min.time())
# 首次无文件时与 ETF 日K 默认一年窗口对齐, 不要只拉 30 天。
history_start = adj_end - timedelta(days=365)
adj_start = index_sync.etf_adj_factor_window_start(adj_path, history_start)
_, affected_etfs = index_sync.sync_etf_adj_factor(
etf_symbols,
repo,
backend/app/services/index_sync.py
+49 -3
View File
@@ -265,6 +265,48 @@ def _load_etf_factors(repo: KlineRepository) -> pl.DataFrame:
return pl.DataFrame()
def etf_adj_factor_window_start(adj_path, history_start: datetime) -> datetime:
"""ETF 除权拉取窗口起点。
本地还没有因子文件时与日K历史对齐 (调用方传入 today-365 或最早分区日),
不要只拉最近 30 天: 拆分落在窗口外则全年日K都不复权。已有文件则从最新
事件日续拉。
"""
if not adj_path.exists():
return history_start
try:
max_date = pl.scan_parquet(adj_path).select(pl.col("trade_date").max()).collect().item()
except Exception as e:
logger.warning("读取 ETF 除权因子最新日期失败, 回退日K历史起点: %s", e)
return history_start
if max_date is None:
return history_start
if isinstance(max_date, str):
stored = datetime.combine(datetime.fromisoformat(max_date).date(), datetime.min.time())
elif isinstance(max_date, datetime):
stored = datetime.combine(max_date.date(), datetime.min.time())
else:
stored = datetime.combine(max_date, datetime.min.time())
return stored
def _load_local_etf_daily(repo: KlineRepository, symbols: list[str]) -> pl.DataFrame:
"""读本批 symbol 的完整本地 ETF 日K, 供前复权改写历史价。"""
daily_dir = repo.store.data_dir / "kline_etf_daily"
if not daily_dir.exists() or not any(daily_dir.glob("date=*")):
return pl.DataFrame()
from app.parquet import scan_daily_parquet
try:
return (
scan_daily_parquet((daily_dir / "**" / "*.parquet").as_posix())
.filter(pl.col("symbol").is_in(symbols))
.collect()
)
except Exception as e:
logger.warning("读取本地 ETF 日K失败: %s", e)
return pl.DataFrame()
def sync_etf_adj_factor(
symbols: list[str],
repo: KlineRepository,
@@ -336,14 +378,18 @@ def sync_and_persist_etf_daily(
repo.append_etf_daily(raw)
batch_factors = factors.filter(pl.col("symbol").is_in(chunk)) if not factors.is_empty() else factors
# ETF 使用复权和通用技术指标；不传 instruments，避免套用 A股涨跌停/连板逻辑。
enriched = compute_enriched(raw, factors=batch_factors, instruments=None)
# 前复权 ratio = cum/total, 新除权事件会改写全部历史价。只用本次拉取
# 窗口算 enriched 时, 拆分日前的分区停在未复权价, 日K 留下跳空。
local = _load_local_etf_daily(repo, chunk)
hist = local if not local.is_empty() else raw
# ETF 使用复权和通用技术指标; 不传 instruments, 避免套用 A股涨跌停/连板逻辑。
enriched = compute_enriched(hist, factors=batch_factors, instruments=None)
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
del raw, enriched, local, hist
gc.collect()
repo.refresh_index_views()
return total_rows
backend/app/services/kline_sync.py
+24 -15
View File
@@ -519,21 +519,30 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
on_chunk_done=on_chunk_done,
)
if new_data.is_empty():
return 0, []
affected = new_data["symbol"].unique().to_list()
factor_dir = "adj_factor_etf" if asset_type == "etf" else "adj_factor"
out = repo.store.data_dir / factor_dir / "all.parquet"
out.parent.mkdir(parents=True, exist_ok=True)
if out.exists():
existing = pl.read_parquet(out)
before = existing.height
merged = pl.concat([existing, new_data]).unique(
subset=["symbol", "trade_date"], keep="last",
).sort(["symbol", "trade_date"])
_atomic_write_parquet(merged, out)
return merged.height - before, affected
_atomic_write_parquet(new_data.sort(["symbol", "trade_date"]), out)
return new_data.height, affected
# 扶摇等自定义源对 ETF 直接空返回, 与「该 ETF 无除权」无法区分;
# 有 TickFlow 除权能力时回退, 否则 ETF 日K永远不复权。
if asset_type == "etf":
logger.info(
"custom adj_factor provider %s returned no ETF rows, falling back to TickFlow",
provider_name,
)
else:
return 0, []
else:
affected = new_data["symbol"].unique().to_list()
factor_dir = "adj_factor_etf" if asset_type == "etf" else "adj_factor"
out = repo.store.data_dir / factor_dir / "all.parquet"
out.parent.mkdir(parents=True, exist_ok=True)
if out.exists():
existing = pl.read_parquet(out)
before = existing.height
merged = pl.concat([existing, new_data]).unique(
subset=["symbol", "trade_date"], keep="last",
).sort(["symbol", "trade_date"])
_atomic_write_parquet(merged, out)
return merged.height - before, affected
_atomic_write_parquet(new_data.sort(["symbol", "trade_date"]), out)
return new_data.height, affected
# 自定义源未配置 adj_factor → 回退 TickFlow
if not capset.has(Cap.ADJ_FACTOR):
backend/tests/test_etf_adj_factor.py
+168
View File
@@ -0,0 +1,168 @@
"""ETF 除权因子: 增量日K 必须用完整本地历史重算前复权 (issue #362)。
股票路径在新除权事件到达后会重算受影响标的的全部日期。ETF 路径只把本次
拉取窗口送进 compute_enriched, 历史分区停在未复权价, 日K 在拆分日留下跳空。
"""
from __future__ import annotations
from datetime import date, datetime
from types import SimpleNamespace
import polars as pl
import pytest
from app.services import index_sync, kline_sync
from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet
from app.tickflow.repository import DataStore, KlineRepository
ETF = "588200.SH"
PRE = date(2026, 3, 2)
EX = date(2026, 3, 3)
NEW = date(2026, 3, 4)
EX_FACTOR = 1.25
PRE_RAW = 10.0
EX_RAW = 8.0
NEW_RAW = 8.1
PRE_QFQ = PRE_RAW / EX_FACTOR # 8.0
def _capset() -> CapabilitySet:
return CapabilitySet(
{
Cap.KLINE_DAILY_BATCH: CapabilityLimits(batch=50, rpm=30),
Cap.ADJ_FACTOR: CapabilityLimits(batch=50, rpm=30),
}
)
def _bar(day: date, close: float) -> pl.DataFrame:
return pl.DataFrame(
{
"symbol": [ETF],
"date": [day],
"open": [close],
"high": [close],
"low": [close],
"close": [close],
"volume": [1000.0],
"amount": [close * 1000.0],
}
)
def _write_factor(data_dir, day: date, factor: float) -> None:
out = data_dir / "adj_factor_etf" / "all.parquet"
out.parent.mkdir(parents=True, exist_ok=True)
pl.DataFrame(
{"symbol": [ETF], "trade_date": [day], "ex_factor": [factor]},
schema={"symbol": pl.String, "trade_date": pl.Date, "ex_factor": pl.Float64},
).write_parquet(out)
def _enriched_close(data_dir, day: date) -> float:
part = data_dir / "kline_etf_enriched" / f"date={day.isoformat()}" / "part.parquet"
df = pl.read_parquet(part).filter(pl.col("symbol") == ETF)
assert df.height == 1
return float(df["close"][0])
def test_etf_incremental_daily_reapplies_adj_to_history(tmp_path, monkeypatch):
"""本地已有拆分前未复权 enriched, 增量只拉新一日时, 历史收盘必须改写成前复权价。"""
repo = KlineRepository(DataStore(tmp_path))
repo.append_etf_daily(pl.concat([_bar(PRE, PRE_RAW), _bar(EX, EX_RAW)]))
repo.append_etf_enriched(pl.concat([_bar(PRE, PRE_RAW), _bar(EX, EX_RAW)]))
_write_factor(tmp_path, EX, EX_FACTOR)
monkeypatch.setattr(
index_sync.kline_sync,
"sync_daily_batch",
lambda *a, **k: _bar(NEW, NEW_RAW),
)
monkeypatch.setattr(index_sync.preferences, "get_index_daily_batch_size", lambda: 50)
written = index_sync.sync_and_persist_etf_daily(
repo,
_capset(),
start_date=datetime(NEW.year, NEW.month, NEW.day),
end_date=datetime(NEW.year, NEW.month, NEW.day, 15, 0),
symbols_override=[ETF],
)
assert written == 1
assert abs(_enriched_close(tmp_path, PRE) - PRE_QFQ) < 1e-9
assert abs(_enriched_close(tmp_path, EX) - EX_RAW) < 1e-9
assert abs(_enriched_close(tmp_path, NEW) - NEW_RAW) < 1e-9
pre = pl.read_parquet(
tmp_path / "kline_etf_enriched" / f"date={PRE.isoformat()}" / "part.parquet"
).filter(pl.col("symbol") == ETF)
assert abs(float(pre["raw_close"][0]) - PRE_RAW) < 1e-9
def test_etf_adj_window_without_file_uses_history_start(tmp_path):
start = datetime(2025, 3, 4)
got = index_sync.etf_adj_factor_window_start(tmp_path / "missing.parquet", start)
assert got == start
def test_etf_adj_window_with_file_continues_from_last_event(tmp_path):
path = tmp_path / "all.parquet"
pl.DataFrame(
{
"symbol": [ETF, ETF],
"trade_date": [date(2025, 6, 1), date(2026, 1, 15)],
"ex_factor": [1.05, 1.10],
}
).write_parquet(path)
got = index_sync.etf_adj_factor_window_start(path, datetime(2025, 3, 4))
assert got == datetime(2026, 1, 15, 0, 0)
def test_sync_adj_factor_etf_custom_empty_falls_back_to_tickflow(tmp_path, monkeypatch):
"""扶摇等自定义源对 ETF 空返回时, 有 TickFlow 除权能力则回退, 不能当成无事件。"""
repo = KlineRepository(DataStore(tmp_path))
empty = pl.DataFrame(
schema={
"symbol": pl.String,
"trade_date": pl.Date,
"ex_factor": pl.Float64,
}
)
class _EmptyETF:
def get_adj_factors(
self, symbols, start_time, end_time, asset_type="stock", on_chunk_done=None
):
assert asset_type == "etf"
return empty
monkeypatch.setattr(kline_sync.preferences, "get_adj_factor_provider", lambda: "fuyao")
from app.data_providers import custom as custom_sources
monkeypatch.setattr(
custom_sources,
"provider_has_dataset",
lambda name, dataset: name == "fuyao" and dataset == "adj_factor",
)
monkeypatch.setattr(custom_sources, "get_provider", lambda name: _EmptyETF())
class _Klines:
def ex_factors(self, symbols, **kwargs):
return {ETF: [{"trade_date": EX, "ex_factor": EX_FACTOR}]}
monkeypatch.setattr(
kline_sync,
"get_client",
lambda: SimpleNamespace(klines=_Klines()),
)
written, affected = kline_sync.sync_adj_factor(
[ETF],
repo,
_capset(),
asset_type="etf",
)
assert written == 1
assert affected == [ETF]
stored = pl.read_parquet(tmp_path / "adj_factor_etf" / "all.parquet")
assert stored["ex_factor"][0] == pytest.approx(EX_FACTOR)