diff --git a/docs/user-guide/incremental-updates.md b/docs/user-guide/incremental-updates.md index 64a0aec..e7c9a54 100644 --- a/docs/user-guide/incremental-updates.md +++ b/docs/user-guide/incremental-updates.md @@ -77,15 +77,21 @@ applies, then NYSE. `calendar="24/7"` expects a daily observation on weekends an feed declaration selects a calendar. Version-1 feed metadata with `calendar=None` still means unspecified; version-2 metadata preserves an explicit null as disabled after serialization. -Daily inputs are session labels (Polars Date, or the date component of Datetime values in their -existing timezone). Non-midnight labels remain supported. Exchange holidays and weekends do not +Undeclared daily timestamps and explicit `session_label` inputs are labels (Polars Date, or the +date component of Datetime values). Aware labels restore a declared `timezone` before extracting +dates, including after storage's canonical UTC read. Naive labels remain civil identifiers rather +than localized event clocks. Non-midnight labels remain supported. Exchange holidays and weekends do not count as missing NYSE sessions; a shortened session still requires one daily observation. Gap bounds remain the two observed timestamps, while `missing_periods` counts only scheduled dates strictly between them. Inspect each asset separately; observed dates from other assets must not define the schedule. This operation does not validate off-session observations or discover gaps outside the observed range. -Explicit event-time/bar-close metadata is rejected rather than reinterpreted as session labels. +Explicit `event_time`, `bar_close` or bar-side metadata instead uses native authoritative session +mapping, even with daily frequency or midnight timestamps. Right-labeled continuous midnight belongs +to the preceding session; event time belongs to the new session. Out-of-session instants fail clearly. +Gap classification does not add a new off-session rejection policy for legacy daily labels. The +requested daily frequency cannot silently override a declared intraday frequency. Named calendar/feed declarations for intraday detection are not supported yet; legacy intraday calls retain their existing behavior. Unknown calendars and malformed daily timestamps fail without modifying the input. `fill_gaps()` remains a separate elapsed-grid operation, not calendar-aware @@ -111,10 +117,11 @@ or feed declaration is present. `exclude_weekends=True` now follows NYSE holiday Omission follows NYSE, so callers requiring the old elapsed-day counts should pass `calendar="24/7"`. Intraday/weekly range planning retains its legacy elapsed-time behavior; named calendar/feed declarations for those frequencies are not supported by this planner yet. `calendar=None` disables -classification. Session-label dates retain their existing timezone's civil date. Explicit event-time -or bar-close declarations are rejected for daily planning rather than silently relabeled. Declare -the feed's `timezone` when a non-UTC label must survive storage's canonical UTC reads; the planner -restores that declared timezone before interpreting dates. +classification. Session-label dates use their declared timezone, or their existing timezone when +undeclared. Explicit event-time/bar-close/bar-side declarations map to authoritative session labels +before planning, including for provider-row selection during backfill. Out-of-session instants fail +clearly; original provider timestamps are retained in storage. Declare the feed's `timezone` when a +non-UTC label must survive storage's canonical UTC reads. If a provider's history limit cannot reach the stored dataset, the updater records `provider_history_limited` and skips gap filling across the unavailable interval. diff --git a/src/ml4t/data/sessions/assigner.py b/src/ml4t/data/sessions/assigner.py index d583b7b..67935ff 100644 --- a/src/ml4t/data/sessions/assigner.py +++ b/src/ml4t/data/sessions/assigner.py @@ -66,6 +66,45 @@ def _session_intervals(row: Any) -> list[tuple[datetime, datetime]]: } +def _daily_session_labels( + df: pl.DataFrame, assigner: SessionAssigner, timestamp_col: str = "timestamp" +) -> pl.Series: + if timestamp_col not in df.columns: + raise ValueError(f"DataFrame must have '{timestamp_col}' column") + feed = assigner.feed_spec + if feed.data_frequency is not None and str(feed.data_frequency).lower() not in { + "daily", + "1day", + "1d", + }: + raise ValueError("daily session projection requires a declared daily frequency") + if df.is_empty(): + return pl.Series(timestamp_col, [], dtype=pl.Date) + dtype = df.schema[timestamp_col] + if dtype != pl.Date and not isinstance(dtype, pl.Datetime): + raise TypeError(f"'{timestamp_col}' must contain Date or Datetime values") + if df[timestamp_col].null_count(): + raise ValueError(f"'{timestamp_col}' must not contain null timestamps") + if feed.timestamp_semantics in ( + TimestampSemantics.EVENT_TIME, + TimestampSemantics.BAR_CLOSE, + ) or ( + feed.timestamp_semantics != TimestampSemantics.SESSION_LABEL and feed.bar_label is not None + ): + assigned = assigner.assign_sessions( + df.select(pl.col(timestamp_col).alias("timestamp")), + outside_session="raise", + bar_frequency="daily", + ) + return assigned["session_date"].rename(timestamp_col) + labels = df[timestamp_col] + if isinstance(dtype, pl.Datetime): + if dtype.time_zone is not None and feed.timezone is not None: + labels = labels.dt.convert_time_zone(feed.timezone) + labels = labels.dt.date() + return labels + + class SessionAssigner: """Assigns session dates to trading data based on exchange calendars. diff --git a/src/ml4t/data/update_manager.py b/src/ml4t/data/update_manager.py index 3b8ab0a..b1a7d53 100644 --- a/src/ml4t/data/update_manager.py +++ b/src/ml4t/data/update_manager.py @@ -13,10 +13,10 @@ from ml4t.data.core.schemas import align_frames_for_concat from ml4t.data.providers.base import BaseProvider -from ml4t.data.sessions.assigner import SessionAssigner +from ml4t.data.sessions.assigner import SessionAssigner, _daily_session_labels from ml4t.data.storage.backend import StorageBackend, normalize_storage_metadata from ml4t.data.storage.metadata_tracker import MetadataTracker, UpdateRecord -from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec, TimestampSemantics +from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec logger = structlog.get_logger() @@ -136,34 +136,20 @@ def _feed_calendar( return feed, feed.resolve_calendar(calendar=calendar) @staticmethod - def _label_frame(df: pl.DataFrame, feed: FeedSpec) -> pl.DataFrame: - dtype = df.schema["timestamp"] - if dtype != pl.Date and not isinstance(dtype, pl.Datetime): - raise TypeError("'timestamp' must contain Date or Datetime values") - if df["timestamp"].null_count(): - raise ValueError("'timestamp' must not contain null timestamps") - if isinstance(dtype, pl.Datetime) and feed.timezone is not None: - timestamp = pl.col("timestamp") - if dtype.time_zone is None: - timestamp = timestamp.dt.replace_time_zone(feed.timezone) - else: - timestamp = timestamp.dt.convert_time_zone(feed.timezone) - return df.with_columns(timestamp) - return df + def _label_frame(df: pl.DataFrame, feed: FeedSpec, calendar: str | None) -> pl.DataFrame: + labels = _daily_session_labels(df, SessionAssigner(calendar, feed_spec=feed)) + return df.with_columns(labels.cast(pl.Datetime("us")).dt.replace_time_zone("UTC")) @staticmethod def _daily_ranges( df: pl.DataFrame, start: datetime, end: datetime, - feed: FeedSpec, calendar: str, tolerance_days: int = 0, ) -> list[dict[str, Any]]: if start.date() > end.date(): raise ValueError("start_date must not be after end_date") - if feed.timestamp_semantics not in (None, TimestampSemantics.SESSION_LABEL): - raise NotImplementedError("daily range planning requires session_label timestamps") assigner = SessionAssigner(calendar) observed = { value.date() if isinstance(value, datetime) else value @@ -222,12 +208,11 @@ def detect_gaps( if df.is_empty() or "timestamp" not in df.columns: return [] if frequency.lower() in {"daily", "1day"}: - df = self._label_frame(df, feed) + df = self._label_frame(df, feed, selected) return self._daily_ranges( df, _session_label(df["timestamp"].min()), _session_label(df["timestamp"].max()), - feed, selected, tolerance_days, ) @@ -312,7 +297,7 @@ def detect_gaps_in_storage( except KeyError: df = pl.DataFrame(schema={"timestamp": pl.Datetime("us", "UTC")}) return self._daily_ranges( - self._label_frame(df, feed), start_date, end_date, feed, selected + self._label_frame(df, feed, selected), start_date, end_date, selected ) start_date = _ensure_datetime(start_date) end_date = _ensure_datetime(end_date) @@ -677,7 +662,9 @@ def apply_strategy( start_date=gap["start"].date(), end_date=gap["end"].date() ) gap_rows = new_data.filter( - self.gap_detector._label_frame(new_data, feed)["timestamp"] + self.gap_detector._label_frame(new_data, feed, selected_calendar)[ + "timestamp" + ] .dt.date() .is_in([label.date() for label in expected.index]) ) diff --git a/src/ml4t/data/utils/gaps.py b/src/ml4t/data/utils/gaps.py index 2baf493..cae0b17 100644 --- a/src/ml4t/data/utils/gaps.py +++ b/src/ml4t/data/utils/gaps.py @@ -10,8 +10,8 @@ import polars as pl import structlog -from ml4t.data.sessions.assigner import SessionAssigner -from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec, TimestampSemantics +from ml4t.data.sessions.assigner import SessionAssigner, _daily_session_labels +from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec logger = structlog.get_logger() @@ -185,21 +185,12 @@ def _detect_daily_gaps( calendar: str, feed: FeedSpec, ) -> list[DataGap]: - assigner = SessionAssigner(calendar) - if timestamp_col not in df.columns: - raise ValueError(f"DataFrame must have '{timestamp_col}' column") + assigner = SessionAssigner(calendar, feed_spec=feed) + ordered = df.sort(timestamp_col) if timestamp_col in df.columns else df + labels = _daily_session_labels(ordered, assigner, timestamp_col).to_list() if df.is_empty(): return [] - dtype = df.schema[timestamp_col] - if dtype != pl.Date and not isinstance(dtype, pl.Datetime): - raise TypeError(f"'{timestamp_col}' must contain Date or Datetime values") - if df[timestamp_col].null_count(): - raise ValueError(f"'{timestamp_col}' must not contain null timestamps") - if feed.timestamp_semantics not in (None, TimestampSemantics.SESSION_LABEL): - raise NotImplementedError("daily gap detection requires session_label timestamps") - - timestamps = df.sort(timestamp_col)[timestamp_col].to_list() - labels = [value.date() if isinstance(value, datetime) else value for value in timestamps] + timestamps = ordered[timestamp_col].to_list() sessions = [ label.date() for label in assigner.calendar.schedule(start_date=labels[0], end_date=labels[-1]).index diff --git a/tests/test_daily_session_gaps.py b/tests/test_daily_session_gaps.py index 2199093..ee5751b 100644 --- a/tests/test_daily_session_gaps.py +++ b/tests/test_daily_session_gaps.py @@ -142,10 +142,9 @@ def test_date_session_labels_have_the_same_schedule_as_datetime_labels() -> None @pytest.mark.parametrize("semantics", ["event_time", "bar_close"]) -def test_event_semantics_are_not_silently_reinterpreted_as_daily_labels(semantics) -> None: +def test_explicit_daily_instants_are_mapped_without_creating_a_weekend_gap(semantics) -> None: frame = provider_frame("2025-07-11", "2025-07-14") - with pytest.raises(NotImplementedError, match="session_label"): - GapDetector().detect_gaps(frame, feed_spec={"timestamp_semantics": semantics}) + assert GapDetector().detect_gaps(frame, feed_spec={"timestamp_semantics": semantics}) == [] def test_null_timestamps_fail_without_input_mutation() -> None: diff --git a/tests/test_daily_timestamp_declarations.py b/tests/test_daily_timestamp_declarations.py new file mode 100644 index 0000000..66095f7 --- /dev/null +++ b/tests/test_daily_timestamp_declarations.py @@ -0,0 +1,180 @@ +"""Daily consumers must preserve declared label dates and map explicit instants to sessions.""" + +from datetime import date + +import polars as pl +import pytest + +from ml4t.data import DataManager +from ml4t.data.core.models import Metadata +from ml4t.data.providers.synthetic import SyntheticProvider +from ml4t.data.storage.config import StorageConfig +from ml4t.data.storage.flat import FlatStorage +from ml4t.data.storage.hive import HiveStorage +from ml4t.data.update_manager import GapDetector as RangeDetector +from ml4t.data.update_manager import IncrementalUpdater, UpdateStrategy +from ml4t.data.utils.gaps import GapDetector +from ml4t.specs import FeedSpec + + +def provider_frame(start="2025-07-15", end="2025-07-17"): + with SyntheticProvider() as provider: + return provider.fetch_ohlcv("SPY", start, end, "daily") + + +@pytest.fixture(params=[FlatStorage, HiveStorage]) +def stored_labels(request, tmp_path): + storage = request.param(StorageConfig(base_path=tmp_path)) + frame = provider_frame("2025-07-11", "2025-07-14").with_columns( + pl.col("timestamp").dt.date().cast(pl.Datetime("us")).dt.replace_time_zone("Asia/Tokyo") + ) + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + bar_params={"frequency": "daily"}, + calendar="JPX", + time_contract_version=2, + timezone="Asia/Tokyo", + timestamp_semantics="session_label", + ) + key = "equities/daily/SPY" + storage.write(frame, key, metadata.model_dump()) + with DataManager(storage=storage) as manager: + yield storage.read(key).collect(), manager.get_metadata_for_key(key) + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +def test_real_utc_storage_roundtrip_does_not_turn_tokyo_labels_into_a_false_friday_outage( + stored_labels, consumer +): + frame, metadata = stored_labels + assert frame["timestamp"].dt.date().to_list() == [date(2025, 7, 10), date(2025, 7, 13)] + assert consumer().detect_gaps(frame, feed_spec=metadata) == [] + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +@pytest.mark.parametrize("semantics", ["event_time", "bar_close"]) +def test_explicit_daily_instants_use_their_session_and_keep_a_real_missing_session( + consumer, semantics +): + frame = provider_frame().filter(pl.col("timestamp").dt.date() != date(2025, 7, 16)) + feed = FeedSpec(time_contract_version=2, calendar="NYSE", timestamp_semantics=semantics) + found = consumer().detect_gaps(frame, feed_spec=feed) + count = sum(g.missing_periods if consumer is GapDetector else g["size_days"] for g in found) + assert count == 1 + + +@pytest.mark.parametrize("semantics,expected", [("event_time", [12, 13]), ("bar_close", [11, 12])]) +def test_continuous_midnight_right_labels_change_the_missing_session_dates(semantics, expected): + frame = provider_frame("2025-07-11", "2025-07-14").with_columns( + pl.col("timestamp").dt.date().cast(pl.Datetime("us")).dt.replace_time_zone("UTC") + ) + found = RangeDetector().detect_gaps( + frame, + feed_spec=FeedSpec( + time_contract_version=2, + calendar="24/7", + timestamp_semantics=semantics, + ), + ) + assert len(found) == 1 + assert [found[0]["start"].day, found[0]["end"].day] == expected + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +def test_disabled_daily_event_declaration_does_not_restore_a_schedule(consumer): + frame = provider_frame().gather_every(2) + feed = FeedSpec(time_contract_version=2, calendar=None, timestamp_semantics="event_time") + assert consumer().detect_gaps(frame, feed_spec=feed) == [] + + +def test_declared_outside_session_instant_is_not_reinterpreted_as_a_daily_label(): + with SyntheticProvider(calendar_mode="continuous") as provider: + frame = provider.fetch_ohlcv("SPY", "2025-07-07", "2025-07-07", "hourly").head(2) + feed = FeedSpec(time_contract_version=2, calendar="NYSE", timestamp_semantics="event_time") + with pytest.raises(ValueError, match="outside a trading session"): + GapDetector().detect_gaps(frame, feed_spec=feed) + + +def test_declared_instants_support_the_public_custom_timestamp_column(): + frame = ( + provider_frame() + .filter(pl.col("timestamp").dt.date() != date(2025, 7, 16)) + .rename({"timestamp": "stamp"}) + ) + feed = FeedSpec(time_contract_version=2, calendar="NYSE", timestamp_semantics="event_time") + found = GapDetector().detect_gaps(frame, timestamp_col="stamp", feed_spec=feed) + assert sum(g.missing_periods for g in found) == 1 + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +def test_declared_hourly_bars_cannot_silently_be_reinterpreted_as_daily_observations(consumer): + frame = provider_frame() + feed = FeedSpec( + time_contract_version=2, data_frequency="hourly", timestamp_semantics="event_time" + ) + with pytest.raises(ValueError, match="frequency"): + consumer().detect_gaps(frame, frequency="daily", feed_spec=feed) + + +@pytest.mark.parametrize("backend", [FlatStorage, HiveStorage]) +@pytest.mark.parametrize("semantics", ["event_time", "bar_close"]) +def test_actual_backfill_selects_provider_instants_by_their_missing_session( + tmp_path, backend, semantics +): + storage = backend(StorageConfig(base_path=tmp_path)) + complete = provider_frame() + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + bar_params={"frequency": "daily"}, + calendar="NYSE", + time_contract_version=2, + timestamp_semantics=semantics, + ) + key = "equities/daily/SPY" + storage.write( + complete.filter(pl.col("timestamp").dt.date() != date(2025, 7, 16)), + key, + metadata.model_dump(), + ) + result = IncrementalUpdater().apply_strategy(storage, key, complete, UpdateStrategy.BACKFILL) + assert result.success + assert result.rows_added == 1 + assert storage.read(key).collect().equals(complete) + + +@pytest.mark.parametrize("backend", [FlatStorage, HiveStorage]) +def test_actual_backfill_restores_tokyo_labels_without_retimestamping_provider_rows( + tmp_path, backend +): + storage = backend(StorageConfig(base_path=tmp_path)) + complete = provider_frame("2025-07-11", "2025-07-15").with_columns( + pl.col("timestamp").dt.date().cast(pl.Datetime("us")).dt.replace_time_zone("Asia/Tokyo") + ) + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + bar_params={"frequency": "daily"}, + calendar="JPX", + time_contract_version=2, + timezone="Asia/Tokyo", + timestamp_semantics="session_label", + ) + key = "equities/daily/SPY" + storage.write( + complete.filter(pl.col("timestamp").dt.date() != date(2025, 7, 14)), + key, + metadata.model_dump(), + ) + result = IncrementalUpdater().apply_strategy(storage, key, complete, UpdateStrategy.BACKFILL) + assert result.success + assert result.rows_added == 1 + assert ( + storage.read(key) + .collect() + .equals(complete.with_columns(pl.col("timestamp").dt.convert_time_zone("UTC"))) + )