diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index 06e3728..fbaf635 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -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: # 向前扩展模式: 从本地最早数据往前补, 叠加已有数据避免缺口。 diff --git a/backend/tests/test_kline_sync_timezone.py b/backend/tests/test_kline_sync_timezone.py index a25f78d..dfee85c 100644 --- a/backend/tests/test_kline_sync_timezone.py +++ b/backend/tests/test_kline_sync_timezone.py @@ -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)