mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 14:24:15 +08:00
fix(kline): 分钟K增量同步窗口统一按北京时区
sync_and_persist_minute 的起点取自本地分钟K的北京墙钟, 终点却用服务器 本地墙钟 datetime.now(); _datetime_to_ms 按服务器时区解释两者, UTC 容器上 起点反而晚于终点, time_segments 为空, 增量补拉一个请求都发不出去。 _latest_minute_datetime / _earliest_minute_datetime 返回值带上 CN_TZ, now 改用 cn_now(), 与 fetch_minute_single 已有的时区纪律保持一致。
This commit is contained in:
@@ -1277,29 +1277,38 @@ def fetch_adj_factor_single(symbol: str) -> pl.DataFrame:
|
||||
return _normalize_adj_factor(raw)
|
||||
|
||||
|
||||
def _as_beijing(d: datetime) -> datetime:
|
||||
"""落盘的分钟 datetime 是北京墙钟 naive, 带上北京时区再交给取数窗口。
|
||||
|
||||
naive 值经 _datetime_to_ms 会被 .timestamp() 按服务器本地时区解释, 与同
|
||||
窗口另一端的服务器本地时间混用后整体错位 (UTC 容器上错 8 小时)。
|
||||
"""
|
||||
return d if d.tzinfo is not None else d.replace(tzinfo=CN_TZ)
|
||||
|
||||
|
||||
def _latest_minute_datetime(repo: KlineRepository) -> datetime | None:
|
||||
"""本地分钟 K 数据的最新时间。"""
|
||||
"""本地分钟 K 数据的最新时间 (北京时区)。"""
|
||||
try:
|
||||
res = repo.execute_one("SELECT max(datetime) FROM kline_minute")
|
||||
if res and res[0]:
|
||||
d = res[0]
|
||||
if isinstance(d, datetime):
|
||||
return d
|
||||
return datetime.fromisoformat(str(d))
|
||||
return _as_beijing(d)
|
||||
return _as_beijing(datetime.fromisoformat(str(d)))
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _earliest_minute_datetime(repo: KlineRepository) -> datetime | None:
|
||||
"""本地分钟 K 数据的最早时间 (用于向前扩展的起点)。"""
|
||||
"""本地分钟 K 数据的最早时间 (北京时区, 用于向前扩展的起点)。"""
|
||||
try:
|
||||
res = repo.execute_one("SELECT min(datetime) FROM kline_minute")
|
||||
if res and res[0]:
|
||||
d = res[0]
|
||||
if isinstance(d, datetime):
|
||||
return d
|
||||
return datetime.fromisoformat(str(d))
|
||||
return _as_beijing(d)
|
||||
return _as_beijing(datetime.fromisoformat(str(d)))
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
return None
|
||||
@@ -1421,7 +1430,9 @@ def sync_and_persist_minute(
|
||||
# 迁移:旧版按 symbol= 分区转为 date= 分区
|
||||
_migrate_symbol_to_date_partition(repo)
|
||||
|
||||
now = datetime.now()
|
||||
# 窗口两端统一为北京时区: 起止点会与本地分钟 K 的北京墙钟混用, 用服务器
|
||||
# 本地时间会让窗口整体错位 (UTC 容器上起点晚于终点, 增量补拉一个请求都发不出)。
|
||||
now = cn_now()
|
||||
|
||||
if extend_backward:
|
||||
# 向前扩展模式: 从本地最早数据往前补, 叠加已有数据避免缺口。
|
||||
|
||||
@@ -5,7 +5,11 @@ fetch_minute_single 构造的 naive datetime 会被 _datetime_to_ms 的 .timesta
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime
|
||||
from datetime import date, datetime, timedelta
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import polars as pl
|
||||
|
||||
from app.market_time import CN_TZ
|
||||
from app.services import kline_sync
|
||||
@@ -41,3 +45,70 @@ def test_fetch_minute_single_window_is_beijing_wall_clock(monkeypatch):
|
||||
end = datetime.fromtimestamp(captured["end_ms"] / 1000, tz=CN_TZ)
|
||||
assert (start.date(), start.hour, start.minute) == (date(2026, 8, 14), 9, 25)
|
||||
assert (end.date(), end.hour, end.minute) == (date(2026, 8, 14), 15, 5)
|
||||
|
||||
|
||||
def _capture_minute_window(monkeypatch, tmp_path, local_dt: datetime, **kwargs) -> dict:
|
||||
"""桩掉 sync_and_persist_minute 的外部依赖, 只截获传给取数层的时间窗口。
|
||||
|
||||
local_dt 模拟 kline_minute 里的极值 (落盘口径为北京墙钟 naive)。
|
||||
"""
|
||||
captured: dict = {}
|
||||
|
||||
def _fake_sync_minute_batch(symbols, **call_kwargs):
|
||||
captured.update(call_kwargs)
|
||||
return pl.DataFrame()
|
||||
|
||||
monkeypatch.setattr(kline_sync.preferences, "get_minute_data_provider", lambda: "tickflow")
|
||||
monkeypatch.setattr(kline_sync.preferences, "get_minute_sync_segment_days", lambda: 20)
|
||||
monkeypatch.setattr(kline_sync, "_cleanup_null_datetime_minute", lambda repo: None)
|
||||
monkeypatch.setattr(kline_sync, "_migrate_symbol_to_date_partition", lambda repo: None)
|
||||
monkeypatch.setattr(kline_sync, "resolve_limit", lambda *a, **kw: SimpleNamespace(batch=100, rpm=30))
|
||||
monkeypatch.setattr(kline_sync, "sync_minute_batch", _fake_sync_minute_batch)
|
||||
|
||||
repo = MagicMock()
|
||||
repo.store.data_dir = tmp_path
|
||||
repo.execute_one = lambda sql: (local_dt,)
|
||||
kline_sync.sync_and_persist_minute(
|
||||
["600000.SH"],
|
||||
repo,
|
||||
CapabilitySet({Cap.KLINE_MINUTE_BATCH: CapabilityLimits()}),
|
||||
**kwargs,
|
||||
)
|
||||
return captured
|
||||
|
||||
|
||||
def test_sync_and_persist_minute_incremental_window_is_beijing_wall_clock(monkeypatch, tmp_path):
|
||||
"""增量分钟同步: 起点取自本地分钟 K 的北京墙钟, 终点必须同口径。
|
||||
|
||||
终点用服务器本地墙钟时, _datetime_to_ms 会把两端按同一本地时区换算 →
|
||||
UTC 容器上起点(北京 14:30)反而晚于终点, 一个请求都发不出去, 分钟 K 永远
|
||||
停在首次拉取的位置。
|
||||
"""
|
||||
local_latest = datetime(2026, 8, 14, 14, 30) # 本地最新分钟 K: 北京墙钟 naive
|
||||
|
||||
captured = _capture_minute_window(monkeypatch, tmp_path, local_latest)
|
||||
start, end = captured["start_time"], captured["end_time"]
|
||||
|
||||
assert start.utcoffset() == timedelta(hours=8)
|
||||
assert end.utcoffset() == timedelta(hours=8)
|
||||
assert start < end
|
||||
assert datetime.fromtimestamp(
|
||||
kline_sync._datetime_to_ms(start) / 1000, tz=CN_TZ
|
||||
) == local_latest.replace(tzinfo=CN_TZ)
|
||||
|
||||
|
||||
def test_sync_and_persist_minute_extend_backward_window_is_beijing_wall_clock(monkeypatch, tmp_path):
|
||||
"""向前扩展: 终点取自本地最早分钟 K 的北京墙钟, 同样不能被服务器时区重解释。"""
|
||||
local_earliest = datetime(2026, 8, 14, 9, 30)
|
||||
|
||||
captured = _capture_minute_window(
|
||||
monkeypatch, tmp_path, local_earliest, days=5, extend_backward=True,
|
||||
)
|
||||
start, end = captured["start_time"], captured["end_time"]
|
||||
|
||||
assert start.utcoffset() == timedelta(hours=8)
|
||||
assert end.utcoffset() == timedelta(hours=8)
|
||||
assert end - start == timedelta(days=7) # 5 交易日 x 7/5
|
||||
assert datetime.fromtimestamp(
|
||||
kline_sync._datetime_to_ms(end) / 1000, tz=CN_TZ
|
||||
) == local_earliest.replace(tzinfo=CN_TZ)
|
||||
|
||||
Reference in New Issue
Block a user