diff --git a/docs/user-guide/incremental-updates.md b/docs/user-guide/incremental-updates.md index e7c9a54..99c3318 100644 --- a/docs/user-guide/incremental-updates.md +++ b/docs/user-guide/incremental-updates.md @@ -52,9 +52,11 @@ Daily detection counts missing sessions from the selected exchange schedule, not The Python updater resolves its explicit `calendar`, then supplied feed or stored metadata, then NYSE. Asset class and descriptive exchange metadata do not select a schedule. Continuous feeds must declare `calendar="24/7"`; `calendar=None` disables schedule-derived detection and filling. -Intraday detection through the updater requires a supported declared-grid consumer, which is not -implemented yet. Use `fill_gaps=False` for intraday updates and inspect -[trading-session completion](sessions.md) for the supported one-minute operation. +Intraday detection requires the complete version-2 grid declaration described below, supplied +through `feed_spec` or retained in stored metadata. Undeclared intraday datasets must migrate their +metadata or use `--no-fill-gaps` / `fill_gaps=False` for updates. +[Trading-session completion](sessions.md) is a separate one-minute operation. The acquisition-range +planner's intraday limitations are described separately below. Detection does not itself fill or modify observations. The updater's separately enabled forward fill uses a regular elapsed-time grid, not the selected exchange schedule, and can insert closed @@ -92,10 +94,49 @@ mapping, even with daily frequency or midnight timestamps. Right-labeled continu 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 -imputation. +Unknown calendars and malformed daily timestamps fail without modifying the input. `fill_gaps()` +remains a separate elapsed-grid operation, not calendar-aware imputation. + +### Inspect Intraday Gaps Against a Declared Grid + +```python +feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + data_frequency="minute", + timestamp_semantics="event_time", + bar_grid_anchor="session_open", + bar_label="left", + partial_bar_policy="include", +) +gaps = GapDetector().detect_gaps(df, frequency="minute", feed_spec=feed) +``` + +Intraday classification requires the complete version-2 declaration above. `event_time` alone +does not choose a bar side. `bar_close` requires `bar_label="right"`. Naive timestamps also require +`timezone`; ambiguous/nonexistent local clocks fail rather than selecting a DST interpretation. +Supported fixed intervals use positive minute/hour counts, such as `5minute`, `15min` or `hourly`. +Omitted `frequency` inherits the feed declaration, otherwise defaults to daily. An explicit +frequency must agree with the declaration. Event-count bars and session-label timestamps +cannot define an intraday grid. + +The grid starts at each authoritative tradable interval's open and restarts after lunch breaks +and declared calendar interruptions. Zero-length breaks on shortened sessions remove no bars. +`partial_bar_policy="include"` retains a shorter final bar; `"exclude"` omits it. Right labels name +the bar's end, including the actual shortened close. Holidays, early closes, DST, overnight sessions +and registered custom calendars use their selected schedule, not the observed rows. Off-grid +observations fail clearly instead of changing the definition to fit the input. + +Run classification separately per asset. Missing bars strictly between the first and last observed +timestamps are counted, including a completely missing session, even if every asset has that outage. +Returned intraday bounds are UTC instants; `missing_periods` counts expected bars, while `duration` +is elapsed time. Detection does not infer missing boundaries outside the observed range or fill rows. +Explicit `calendar=None` disables classification, while `"24/7"` still expects continuous bars. + +This replaces the old fixed-US-clock approximation. Unqualified legacy intraday calls now require +a complete feed declaration or explicit `calendar=None`; the detector does not invent an anchor, +label side or partial-bar policy. Daily classification is unchanged. The separate intraday/weekly +backfill range planner below still has its documented limitations. ### Plan Daily Backfill Acquisition diff --git a/src/ml4t/data/utils/gaps.py b/src/ml4t/data/utils/gaps.py index cae0b17..4d4e81d 100644 --- a/src/ml4t/data/utils/gaps.py +++ b/src/ml4t/data/utils/gaps.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from bisect import bisect_left, bisect_right from dataclasses import dataclass from datetime import UTC, date, datetime, timedelta @@ -10,8 +11,19 @@ import polars as pl import structlog -from ml4t.data.sessions.assigner import SessionAssigner, _daily_session_labels -from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec +from ml4t.data.sessions.assigner import ( + SessionAssigner, + _daily_session_labels, + _session_intervals, +) +from ml4t.specs import ( + UNSET_CALENDAR, + BarLabel, + CalendarUnset, + FeedSpec, + PartialBarPolicy, + TimestampSemantics, +) logger = structlog.get_logger() @@ -70,7 +82,7 @@ def __init__(self, tolerance: float = DEFAULT_TOLERANCE) -> None: def detect_gaps( self, df: pl.DataFrame, - frequency: str = "daily", + frequency: str | None = None, timestamp_col: str = "timestamp", is_crypto: bool = False, *, @@ -82,16 +94,18 @@ def detect_gaps( Args: df: DataFrame with time series data - frequency: Data frequency (minute, hourly, daily, etc.) + frequency: Data frequency; inherit the feed declaration, otherwise daily. timestamp_col: Name of timestamp column is_crypto: If True, expect 24/7 data; if False, market hours only - calendar: Daily schedule override; None disables schedule-based detection. + calendar: Schedule override; None disables schedule-based detection. feed_spec: Shared feed declaration, resolved before the NYSE default. Returns: List of detected gaps """ feed = FeedSpec.from_any(feed_spec) + if frequency is None: + frequency = str(feed.data_frequency) if feed.data_frequency is not None else "daily" # Preserve the old crypto hint only when the feed has no declaration. if ( frequency.lower() in {"daily", "1day"} @@ -106,8 +120,20 @@ def detect_gaps( selected_calendar = feed.resolve_calendar(calendar=calendar) if selected_calendar is None: return [] + declared_interval = ( + self._intraday_interval(str(feed.data_frequency)) + if feed.data_frequency is not None + else None + ) + if ( + declared_interval is not None + and self._intraday_interval(frequency) != declared_interval + ): + raise ValueError("declared grid data_frequency must match the requested frequency") if frequency.lower() in {"daily", "1day"}: return self._detect_daily_gaps(df, timestamp_col, frequency, selected_calendar, feed) + if self._intraday_interval(frequency) is not None: + return self._detect_intraday_gaps(df, timestamp_col, frequency, selected_calendar, feed) if calendar is not UNSET_CALENDAR or feed_spec is not None: raise NotImplementedError( "calendar/feed declarations currently support daily gaps only" @@ -139,21 +165,6 @@ def detect_gaps( actual_seconds = actual_interval.total_seconds() expected_seconds = expected_interval.total_seconds() - # Check if this is a gap (considering market hours for non-crypto) - if ( - not is_crypto - and frequency - in [ - "minute", - "5minute", - "15minute", - "30minute", - "hourly", - ] - and self._spans_non_trading_hours(current, next_ts, frequency) - ): - continue - # Detect gap if interval exceeds expected + tolerance if actual_seconds > expected_seconds + tolerance_seconds: missing_periods = int(actual_seconds / expected_seconds) - 1 @@ -177,6 +188,124 @@ def detect_gaps( return gaps + @staticmethod + def _intraday_interval(frequency: str) -> timedelta | None: + value = frequency.lower().strip().replace("hourly", "hour") + match = re.fullmatch(r"(\d+)?(minutes?|min|m|hours?|h)", value) + if match is None: + return None + count = int(match[1] or 1) + if count <= 0: + raise ValueError("bar frequency must be positive") + return timedelta(hours=count) if match[2].startswith("h") else timedelta(minutes=count) + + def _detect_intraday_gaps( + self, + df: pl.DataFrame, + timestamp_col: str, + frequency: str, + calendar: str, + feed: FeedSpec, + ) -> list[DataGap]: + interval = self._intraday_interval(frequency) + if interval is None: + raise ValueError("unsupported intraday bar frequency") + required = ( + "data_frequency", + "timestamp_semantics", + "bar_grid_anchor", + "bar_label", + "partial_bar_policy", + ) + if feed.time_contract_version != 2 or any(getattr(feed, name) is None for name in required): + raise ValueError("intraday detection requires a complete declared version-2 bar grid") + if feed.timestamp_semantics == TimestampSemantics.SESSION_LABEL: + raise ValueError("intraday detection requires a declared event/close bar grid") + if feed.bar_type not in (None, "time"): + raise ValueError("intraday grid detection requires fixed time bars") + if interval != self._intraday_interval(str(feed.data_frequency)): + raise ValueError("declared grid data_frequency must match the requested frequency") + assigner = SessionAssigner(calendar, feed_spec=feed) + if timestamp_col not in df.columns: + raise ValueError(f"DataFrame must have '{timestamp_col}' column") + dtype = df.schema[timestamp_col] + if not isinstance(dtype, pl.Datetime): + raise TypeError(f"'{timestamp_col}' must contain Datetime values for an intraday grid") + if df[timestamp_col].null_count(): + raise ValueError(f"'{timestamp_col}' must not contain null timestamps") + if ( + dtype.time_unit == "ns" + and df.select((pl.col(timestamp_col).dt.nanosecond() % 1000 != 0).any()).item() + ): + raise ValueError("intraday observations contain off-grid sub-microsecond timestamps") + if dtype.time_zone is None and feed.timezone is None: + raise ValueError("naive intraday timestamps require a declared timezone") + if df.is_empty(): + return [] + expr = pl.col(timestamp_col) + if dtype.time_zone is None: + expr = expr.dt.replace_time_zone(feed.timezone) + observed = df.select(expr.dt.convert_time_zone("UTC")).to_series().unique().sort().to_list() + start, end = observed[0], observed[-1] + schedule = assigner.calendar.schedule( + start_date=(start - timedelta(days=2)).date(), + end_date=(end + timedelta(days=2)).date(), + interruptions=True, + ) + grid_intervals: list[tuple[datetime, datetime, int, int]] = [] + total_bars = 0 + for _, row in schedule.iterrows(): + for opening, closing in _session_intervals(row): + opening, closing = opening.astimezone(UTC), closing.astimezone(UTC) + count, remainder = divmod(closing - opening, interval) + if remainder and feed.partial_bar_policy == PartialBarPolicy.INCLUDE: + count += 1 + if count <= 0: + continue + if grid_intervals and opening < grid_intervals[-1][1]: + raise ValueError("calendar returned overlapping tradable intervals") + grid_intervals.append((opening, closing, count, total_bars)) + total_bars += count + right_labeled = feed.bar_label == BarLabel.RIGHT + boundaries = [row[1] if right_labeled else row[0] for row in grid_intervals] + + def bar_index(timestamp: datetime) -> int: + index = ( + bisect_left(boundaries, timestamp) + if right_labeled + else bisect_right(boundaries, timestamp) - 1 + ) + if index < 0 or index >= len(grid_intervals): + raise ValueError("intraday observations contain off-grid timestamps") + opening, closing, count, offset = grid_intervals[index] + if right_labeled and timestamp == closing: + if (closing - opening) % interval == timedelta(0) or ( + feed.partial_bar_policy == PartialBarPolicy.INCLUDE + ): + return offset + count - 1 + raise ValueError("intraday observations contain off-grid partial-close timestamps") + position, remainder = divmod(timestamp - opening, interval) + if right_labeled: + position -= 1 + if ( + remainder + or timestamp < opening + or timestamp >= closing + or not 0 <= position < count + ): + raise ValueError("intraday observations contain off-grid timestamps") + return offset + position + + positions = [bar_index(timestamp) for timestamp in observed] + gaps = [] + for current, following, left_index, right_index in zip( + observed, observed[1:], positions, positions[1:] + ): + missing = right_index - left_index - 1 + if missing > 0: + gaps.append(DataGap(current, following, missing, following - current, frequency)) + return gaps + def _detect_daily_gaps( self, df: pl.DataFrame, @@ -224,22 +353,11 @@ def _get_expected_interval(self, frequency: str, is_crypto: bool = False) -> tim Expected timedelta between consecutive points """ freq_lower = frequency.lower() + interval = self._intraday_interval(frequency) + if interval is not None: + return interval - # Handle minute-based frequencies - if "minute" in freq_lower: - if freq_lower == "minute" or freq_lower == "1minute": - return timedelta(minutes=1) - if freq_lower == "5minute": - return timedelta(minutes=5) - if freq_lower == "15minute": - return timedelta(minutes=15) - if freq_lower == "30minute": - return timedelta(minutes=30) - - # Handle other frequencies - elif freq_lower == "hourly" or freq_lower == "1hour": - return timedelta(hours=1) - elif freq_lower == "daily" or freq_lower == "1day": + if freq_lower == "daily" or freq_lower == "1day": if is_crypto: return timedelta(days=1) # For stocks, next trading day could be 1-3 days away @@ -252,41 +370,6 @@ def _get_expected_interval(self, frequency: str, is_crypto: bool = False) -> tim # Default to daily return timedelta(days=1) - def _spans_non_trading_hours( - self, - start: datetime, - end: datetime, - frequency: str, # noqa: ARG002 - ) -> bool: - """ - Check if gap spans non-trading hours (nights/weekends). - - Args: - start: Start of gap - end: End of gap - frequency: Data frequency - - Returns: - True if gap is expected (non-trading hours) - """ - # Convert to UTC if needed - if start.tzinfo is None: - start = start.replace(tzinfo=UTC) - if end.tzinfo is None: - end = end.replace(tzinfo=UTC) - - # Check for weekend - if start.weekday() == 4 and end.weekday() == 0: # Friday to Monday - return True - - # Check for overnight gap (4pm to 9:30am ET) - # Convert to ET (UTC-5 or UTC-4 for DST) - market_close_hour = 21 # 4pm ET in UTC (during standard time) - market_open_hour = 14 # 9:30am ET in UTC (during standard time) - - # Simplified check: if gap starts after market close and ends at/after market open - return start.hour >= market_close_hour or end.hour < market_open_hour - def summarize_gaps(self, gaps: list[DataGap]) -> dict[str, Any]: """ Summarize detected gaps. diff --git a/tests/test_gaps.py b/tests/test_gaps.py index b499608..61641d5 100644 --- a/tests/test_gaps.py +++ b/tests/test_gaps.py @@ -6,6 +6,7 @@ import pytest from ml4t.data.utils.gaps import DataGap, GapDetector +from ml4t.specs import FeedSpec class TestGapDetector: @@ -87,7 +88,7 @@ def test_detect_multiple_gaps(self) -> None: def test_minute_data_gaps(self) -> None: """Test gap detection in minute-level data.""" - base = datetime(2024, 1, 1, 14, 30, tzinfo=UTC) # 9:30 AM ET + base = datetime(2024, 1, 2, 14, 30, tzinfo=UTC) # 9:30 AM ET, a NYSE session timestamps = [ base, base + timedelta(minutes=1), @@ -105,7 +106,18 @@ def test_minute_data_gaps(self) -> None: ) detector = GapDetector() - gaps = detector.detect_gaps(df, frequency="minute") + gaps = detector.detect_gaps( + df, + frequency="minute", + feed_spec=FeedSpec( + time_contract_version=2, + data_frequency="minute", + timestamp_semantics="event_time", + bar_grid_anchor="session_open", + bar_label="left", + partial_bar_policy="include", + ), + ) assert len(gaps) == 1 assert gaps[0].missing_periods == 3 diff --git a/tests/test_intraday_schedule_gaps.py b/tests/test_intraday_schedule_gaps.py new file mode 100644 index 0000000..aae6b2b --- /dev/null +++ b/tests/test_intraday_schedule_gaps.py @@ -0,0 +1,381 @@ +"""Real provider outages must not be hidden by fixed US clocks or observed-session inference.""" + +from datetime import UTC, date, datetime, time +from zoneinfo import ZoneInfo + +import polars as pl +import pytest +from polars.testing import assert_frame_equal + +from ml4t.data import DataManager +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.utils.gaps import GapDetector +from ml4t.specs import FeedSpec + + +def provider_frame(start="2025-07-07", end="2025-07-07", frequency="minute"): + with SyntheticProvider(calendar_mode="continuous") as provider: + return provider.fetch_ohlcv("SPY", start, end, frequency) + + +def declaration(calendar="NYSE", frequency="minute", **overrides): + return FeedSpec( + time_contract_version=2, + calendar=calendar, + data_frequency=frequency, + timestamp_semantics=overrides.pop("timestamp_semantics", "event_time"), + bar_grid_anchor="session_open", + bar_label=overrides.pop("bar_label", "left"), + partial_bar_policy=overrides.pop("partial_bar_policy", "include"), + **overrides, + ) + + +def gaps(frame, feed, **kwargs): + return GapDetector().detect_gaps(frame, frequency=feed.data_frequency, feed_spec=feed, **kwargs) + + +@pytest.mark.parametrize("backend", [FlatStorage, HiveStorage]) +@pytest.mark.parametrize("disabled", [False, True]) +def test_stored_intraday_grid_reaches_real_provider_updates( + tmp_path, monkeypatch, backend, disabled +): + class Clock(datetime): + @classmethod + def now(cls, tz=UTC): + return cls(2025, 7, 7, 23, tzinfo=tz) + + monkeypatch.setattr("ml4t.data.managers.storage_manager.datetime", Clock) + with SyntheticProvider(seed=42, calendar_mode="continuous") as provider: + complete = provider.fetch_ohlcv("SPY", "2025-07-07", "2025-07-07", "hourly") + storage = backend(StorageConfig(base_path=tmp_path)) + with DataManager( + storage=storage, + providers={"synthetic": {"extra": {"seed": 42, "calendar_mode": "continuous"}}}, + ) as manager: + key = manager.import_data( + complete.head(3), + "SPY", + "synthetic", + frequency="hourly", + feed_spec=declaration("24/7", "hourly"), + ) + overrides = {"calendar": None} if disabled else {} + assert manager.update("SPY", frequency="hourly", lookback_days=0, **overrides) == key + assert_frame_equal(storage.read(key).collect(), complete) + feed = FeedSpec.from_any(manager.get_metadata_for_key(key)) + assert feed.resolve_calendar() == (None if disabled else "24/7") + assert GapDetector().detect_gaps(storage.read(key).collect(), feed_spec=feed) == [] + + +def test_unqualified_summer_outage_cannot_silently_return_no_gap(): + frame = provider_frame().filter( + (pl.col("timestamp").dt.hour() == 13) & pl.col("timestamp").dt.minute().is_in([30, 32]) + ) + with pytest.raises(ValueError, match="declared.*grid"): + GapDetector().detect_gaps(frame, frequency="minute") + + +def test_declared_intraday_frequency_cannot_silently_fall_back_to_daily_labels(): + frame = provider_frame().filter( + (pl.col("timestamp").dt.hour() == 13) & pl.col("timestamp").dt.minute().is_in([30, 32]) + ) + found = GapDetector().detect_gaps(frame, feed_spec=declaration()) + assert sum(g.missing_periods for g in found) == 1 + with pytest.raises(ValueError, match="frequency"): + GapDetector().detect_gaps(frame, frequency="daily", feed_spec=declaration()) + + +@pytest.mark.parametrize("day,hour,minute", [("2025-07-07", 13, 30), ("2025-01-07", 14, 30)]) +def test_declared_nyse_dst_and_winter_outage_each_count_one_bar(day, hour, minute): + frame = provider_frame(day, day).filter( + (pl.col("timestamp").dt.hour() == hour) + & pl.col("timestamp").dt.minute().is_in([minute, minute + 2]) + ) + original = frame.clone() + assert sum(g.missing_periods for g in gaps(frame, declaration())) == 1 + assert_frame_equal(frame, original) + + +def test_london_morning_outage_is_not_discarded_by_the_us_clock(): + frame = provider_frame(frequency="hourly").filter(pl.col("timestamp").dt.hour().is_in([8, 10])) + found = gaps(frame, declaration("LSE", "hourly")) + assert len(found) == 1 + assert found[0].missing_periods == 1 + + +def test_holiday_and_early_close_are_not_missing_expected_bars(): + frame = provider_frame("2025-07-03", "2025-07-07").filter( + ( + (pl.col("timestamp").dt.date() == date(2025, 7, 3)) + & (pl.col("timestamp").dt.hour() == 16) + & (pl.col("timestamp").dt.minute() == 59) + ) + | ( + (pl.col("timestamp").dt.date() == date(2025, 7, 7)) + & (pl.col("timestamp").dt.hour() == 13) + & (pl.col("timestamp").dt.minute() == 30) + ) + ) + assert gaps(frame, declaration()) == [] + + +def test_missing_shortened_session_counts_its_actual_210_minutes(): + frame = provider_frame("2025-07-02", "2025-07-07").filter( + ( + (pl.col("timestamp").dt.date() == date(2025, 7, 2)) + & (pl.col("timestamp").dt.hour() == 19) + & (pl.col("timestamp").dt.minute() == 59) + ) + | ( + (pl.col("timestamp").dt.date() == date(2025, 7, 7)) + & (pl.col("timestamp").dt.hour() == 13) + & (pl.col("timestamp").dt.minute() == 30) + ) + ) + assert sum(g.missing_periods for g in gaps(frame, declaration())) == 210 + + +def test_all_asset_missing_session_keeps_390_expected_minutes(): + frame = provider_frame("2025-07-15", "2025-07-17").filter( + ( + (pl.col("timestamp").dt.date() == date(2025, 7, 15)) + & (pl.col("timestamp").dt.hour() == 19) + & (pl.col("timestamp").dt.minute() == 59) + ) + | ( + (pl.col("timestamp").dt.date() == date(2025, 7, 17)) + & (pl.col("timestamp").dt.hour() == 13) + & (pl.col("timestamp").dt.minute() == 30) + ) + ) + panel = pl.concat([frame.with_columns(pl.lit(s).alias("symbol")) for s in ("SPY", "QQQ")]) + for asset in panel.partition_by("symbol"): + assert sum(g.missing_periods for g in gaps(asset, declaration())) == 390 + + +def test_hong_kong_break_does_not_count_as_missing_minutes(): + frame = provider_frame().filter( + ((pl.col("timestamp").dt.hour() == 3) & (pl.col("timestamp").dt.minute() == 59)) + | ((pl.col("timestamp").dt.hour() == 5) & (pl.col("timestamp").dt.minute() == 0)) + ) + assert gaps(frame, declaration("HKEX")) == [] + + +def test_cme_overnight_calendar_keeps_a_missing_midnight_minute(): + frame = provider_frame("2025-07-06", "2025-07-07").filter( + ( + (pl.col("timestamp").dt.date() == date(2025, 7, 6)) + & (pl.col("timestamp").dt.hour() == 23) + & (pl.col("timestamp").dt.minute() == 59) + ) + | ( + (pl.col("timestamp").dt.date() == date(2025, 7, 7)) + & (pl.col("timestamp").dt.hour() == 0) + & (pl.col("timestamp").dt.minute() == 1) + ) + ) + assert sum(g.missing_periods for g in gaps(frame, declaration("CME_Equity"))) == 1 + + +def test_continuous_weekend_is_a_real_grid_but_explicit_none_is_not(): + frame = provider_frame("2025-07-11", "2025-07-14", frequency="hourly").filter( + pl.col("timestamp").dt.hour().eq(0) + & pl.col("timestamp").dt.date().is_in([date(2025, 7, 11), date(2025, 7, 14)]) + ) + feed = declaration("24/7", "hourly") + assert sum(g.missing_periods for g in gaps(frame, feed)) == 71 + assert gaps(frame, feed, calendar=None) == [] + assert gaps(frame, declaration(None, "hourly")) == [] + + +def test_right_labeled_close_bar_is_valid_and_missing_predecessor_is_counted(): + frame = provider_frame().filter( + ((pl.col("timestamp").dt.hour() == 19) & (pl.col("timestamp").dt.minute() == 58)) + | ((pl.col("timestamp").dt.hour() == 20) & (pl.col("timestamp").dt.minute() == 0)) + ) + feed = declaration(timestamp_semantics="bar_close", bar_label="right") + assert sum(g.missing_periods for g in gaps(frame, feed)) == 1 + + +def test_naive_local_timestamps_require_and_use_the_declared_timezone(): + frame = ( + provider_frame() + .filter( + (pl.col("timestamp").dt.hour() == 13) & pl.col("timestamp").dt.minute().is_in([30, 32]) + ) + .with_columns( + pl.col("timestamp").dt.convert_time_zone("America/New_York").dt.replace_time_zone(None) + ) + ) + assert ( + sum(g.missing_periods for g in gaps(frame, declaration(timezone="America/New_York"))) == 1 + ) + with pytest.raises(ValueError, match="timezone"): + gaps(frame, declaration()) + + +@pytest.mark.parametrize( + "missing", + ["bar_grid_anchor", "bar_label", "partial_bar_policy", "timestamp_semantics", "data_frequency"], +) +def test_incomplete_grid_metadata_cannot_silently_choose_a_bar_definition(missing): + frame = provider_frame().head(2) + values = declaration("24/7").to_dict() + values.pop(missing) + with pytest.raises(ValueError, match="declared.*grid"): + GapDetector().detect_gaps(frame, frequency="minute", feed_spec=values) + + +def test_off_grid_provider_rows_fail_without_mutating_input(): + frame = provider_frame(frequency="hourly").filter(pl.col("timestamp").dt.hour().is_in([14, 15])) + original = frame.clone() + with pytest.raises(ValueError, match="off.grid"): + gaps(frame, declaration("NYSE", "hourly")) + assert_frame_equal(frame, original) + + +@pytest.mark.parametrize("policy,expected", [("include", 1), ("exclude", 0)]) +def test_registered_custom_break_grid_restarts_and_honors_partial_bar_policy(policy, expected): + from pandas_market_calendars.market_calendar import MarketCalendar + + class DataGridRegressionCalendar(MarketCalendar): + name = "DataGridRegression" + aliases = ["DataGridRegression"] + tz = ZoneInfo("UTC") + regular_market_times = { + "market_open": ((None, time(9)),), + "break_start": ((None, time(11)),), + "break_end": ((None, time(12)),), + "market_close": ((None, time(14, 30)),), + } + + frame = provider_frame("2025-07-07", "2025-07-08", frequency="hourly").filter( + ( + (pl.col("timestamp").dt.date() == date(2025, 7, 7)) + & (pl.col("timestamp").dt.hour() == 13) + ) + | ( + (pl.col("timestamp").dt.date() == date(2025, 7, 8)) + & (pl.col("timestamp").dt.hour() == 9) + ) + ) + feed = declaration("DataGridRegression", "hourly", partial_bar_policy=policy) + assert sum(g.missing_periods for g in gaps(frame, feed)) == expected + + +def test_naive_ambiguous_local_clock_is_rejected_without_a_guessed_dst_fold(): + frame = ( + provider_frame("2025-11-02", "2025-11-02") + .filter( + (pl.col("timestamp").dt.hour() == 5) & pl.col("timestamp").dt.minute().is_in([30, 32]) + ) + .with_columns( + pl.col("timestamp").dt.convert_time_zone("America/New_York").dt.replace_time_zone(None) + ) + ) + with pytest.raises(pl.exceptions.ComputeError, match="ambiguous"): + gaps(frame, declaration("24/7", timezone="America/New_York")) + + +def test_a_frequency_conflict_and_event_count_feed_cannot_define_a_time_grid(): + frame = provider_frame().head(2) + with pytest.raises(ValueError, match="frequency"): + GapDetector().detect_gaps(frame, frequency="hourly", feed_spec=declaration("24/7")) + with pytest.raises(ValueError, match="fixed time bars"): + gaps(frame, declaration("24/7", bar_type="volume")) + + +def test_right_labeled_break_start_is_the_last_valid_morning_bar(): + frame = provider_frame().filter( + ((pl.col("timestamp").dt.hour() == 3) & (pl.col("timestamp").dt.minute() == 59)) + | ((pl.col("timestamp").dt.hour() == 4) & (pl.col("timestamp").dt.minute() == 0)) + ) + assert ( + gaps(frame, declaration("HKEX", timestamp_semantics="bar_close", bar_label="right")) == [] + ) + + +def test_sub_microsecond_offset_cannot_be_truncated_into_a_valid_grid_label(): + frame = ( + provider_frame() + .head(2) + .with_columns( + pl.col("timestamp").cast(pl.Datetime("ns", "UTC")) + pl.duration(nanoseconds=1) + ) + ) + with pytest.raises(ValueError, match="off.grid"): + gaps(frame, declaration("24/7")) + + +@pytest.mark.parametrize("policy,expected", [("include", 1), ("exclude", 0)]) +def test_right_labels_count_the_partial_shortened_close_at_its_actual_end(policy, expected): + frame = provider_frame("2025-07-03", "2025-07-07").filter( + ( + (pl.col("timestamp").dt.date() == date(2025, 7, 3)) + & (pl.col("timestamp").dt.hour() == 16) + & (pl.col("timestamp").dt.minute() == 30) + ) + | ( + (pl.col("timestamp").dt.date() == date(2025, 7, 7)) + & (pl.col("timestamp").dt.hour() == 14) + & (pl.col("timestamp").dt.minute() == 30) + ) + ) + feed = declaration( + frequency="hourly", + timestamp_semantics="bar_close", + bar_label="right", + partial_bar_policy=policy, + ) + assert sum(g.missing_periods for g in gaps(frame, feed)) == expected + + +def test_excluded_partial_close_is_not_accepted_as_the_last_full_bar_label(): + frame = provider_frame("2025-07-03", "2025-07-03").filter( + ((pl.col("timestamp").dt.hour() == 16) & (pl.col("timestamp").dt.minute() == 30)) + | ((pl.col("timestamp").dt.hour() == 17) & (pl.col("timestamp").dt.minute() == 0)) + ) + assert ( + gaps( + frame, + declaration(frequency="hourly", timestamp_semantics="bar_close", bar_label="right"), + ) + == [] + ) + with pytest.raises(ValueError, match="off.grid"): + gaps( + frame, + declaration( + frequency="hourly", + timestamp_semantics="bar_close", + bar_label="right", + partial_bar_policy="exclude", + ), + ) + + +def test_registered_calendar_halt_is_not_missing_observations(): + from pandas_market_calendars.market_calendar import MarketCalendar + + class DataGridHaltCalendar(MarketCalendar): + name = "DataGridHalt" + aliases = ["DataGridHalt"] + tz = ZoneInfo("UTC") + regular_market_times = { + "market_open": ((None, time(9)),), + "market_close": ((None, time(14)),), + } + + @property + def interruptions(self): + return [("2025-07-07", time(10), time(10, 30))] + + frame = provider_frame().filter( + ((pl.col("timestamp").dt.hour() == 9) & (pl.col("timestamp").dt.minute() == 59)) + | ((pl.col("timestamp").dt.hour() == 10) & (pl.col("timestamp").dt.minute() == 30)) + ) + assert gaps(frame, declaration("DataGridHalt")) == [] diff --git a/tests/test_utils_gaps.py b/tests/test_utils_gaps.py index 327b817..458c675 100644 --- a/tests/test_utils_gaps.py +++ b/tests/test_utils_gaps.py @@ -3,11 +3,35 @@ from datetime import datetime, timedelta import polars as pl +import pytest +from ml4t.data.providers import SyntheticProvider from ml4t.data.utils.gaps import ( DataGap, GapDetector, ) +from ml4t.specs import FeedSpec + + +def intraday_feed(frequency, calendar="24/7"): + return FeedSpec( + time_contract_version=2, + calendar=calendar, + timezone="UTC", + data_frequency=frequency, + timestamp_semantics="event_time", + bar_grid_anchor="session_open", + bar_label="left", + partial_bar_policy="include", + ) + + +def intraday_rows(timestamps): + with SyntheticProvider(calendar_mode="continuous") as provider: + frame = provider.fetch_ohlcv( + "SYNTH", timestamps[0].date().isoformat(), timestamps[-1].date().isoformat(), "minute" + ) + return frame.filter(pl.col("timestamp").dt.replace_time_zone(None).is_in(timestamps)) class TestDataGap: @@ -161,48 +185,20 @@ def test_detect_gaps_custom_timestamp_column(self): assert len(gaps) >= 1 def test_detect_gaps_hourly_frequency(self): - """Test gap detection with hourly frequency.""" + """A three-hour continuous interval must retain two missing hourly bars.""" detector = GapDetector() - df = pl.DataFrame( - { - "timestamp": [ - datetime(2024, 1, 1, 10, 0), - datetime(2024, 1, 1, 13, 0), # Missing 3 hours - ], - "value": [10.0, 40.0], - } - ) - - gaps = detector.detect_gaps(df, frequency="hourly", is_crypto=True) - - # Should detect a gap (crypto = 24/7) - assert len(gaps) >= 1 + df = intraday_rows([datetime(2024, 1, 1, 10), datetime(2024, 1, 1, 13)]) + gaps = detector.detect_gaps(df, feed_spec=intraday_feed("hourly")) + assert [gap.missing_periods for gap in gaps] == [2] def test_detect_gaps_crypto_vs_traditional(self): - """Test difference between crypto (24/7) and traditional market hours.""" + """An exchange weekend must not erase a continuous-feed outage.""" detector = GapDetector() - - # Weekend data (Saturday to Monday) - df = pl.DataFrame( - { - "timestamp": [ - datetime(2024, 1, 6, 15, 0), # Saturday - datetime(2024, 1, 8, 9, 0), # Monday - ], - "value": [10.0, 20.0], - } - ) - - # For crypto (24/7), this should be a gap - crypto_gaps = detector.detect_gaps(df, frequency="hourly", is_crypto=True) - - # For traditional markets, weekend gap might be expected - traditional_gaps = detector.detect_gaps(df, frequency="hourly", is_crypto=False) - - # Results might differ (crypto should detect more gaps over weekends) - # This tests the logic even if implementation varies - assert isinstance(crypto_gaps, list) - assert isinstance(traditional_gaps, list) + df = intraday_rows([datetime(2024, 1, 5, 20, 59), datetime(2024, 1, 8, 14, 30)]) + crypto_gaps = detector.detect_gaps(df, feed_spec=intraday_feed("minute")) + traditional_gaps = detector.detect_gaps(df, feed_spec=intraday_feed("minute", "NYSE")) + assert [gap.missing_periods for gap in crypto_gaps] == [3930] + assert traditional_gaps == [] def test_detect_gaps_unsorted_data(self): """Test gap detection handles unsorted data.""" @@ -224,22 +220,11 @@ def test_detect_gaps_unsorted_data(self): assert isinstance(gaps, list) def test_detect_gaps_minute_frequency(self): - """Test gap detection with minute frequency.""" + """A five-minute continuous interval must retain four missing minute bars.""" detector = GapDetector() - df = pl.DataFrame( - { - "timestamp": [ - datetime(2024, 1, 1, 9, 30, 0), - datetime(2024, 1, 1, 9, 35, 0), # 5 minute gap - ], - "value": [100.0, 105.0], - } - ) - - gaps = detector.detect_gaps(df, frequency="minute", is_crypto=True) - - # Should detect multiple 1-minute gaps - assert len(gaps) >= 1 + df = intraday_rows([datetime(2024, 1, 1, 9, 30), datetime(2024, 1, 1, 9, 35)]) + gaps = detector.detect_gaps(df, feed_spec=intraday_feed("minute")) + assert [gap.missing_periods for gap in gaps] == [4] def test_periods_per_day_constants(self): """Test PERIODS_PER_DAY constants are reasonable.""" @@ -313,107 +298,57 @@ def test_monthly_frequency(self): # Should detect gaps in monthly data assert isinstance(gaps, list) - def test_tolerance_effect(self): - """Test that tolerance affects gap detection.""" - # High tolerance should detect fewer gaps - lenient_detector = GapDetector(tolerance=0.5) # 50% tolerance - strict_detector = GapDetector(tolerance=0.01) # 1% tolerance - - df = pl.DataFrame( - { - "timestamp": [ - datetime(2024, 1, 1, 0, 0), - datetime(2024, 1, 1, 1, 30), # 1.5 hours (slight gap) - ], - "value": [10.0, 20.0], - } - ) - - lenient_gaps = lenient_detector.detect_gaps(df, frequency="hourly", is_crypto=True) - strict_gaps = strict_detector.detect_gaps(df, frequency="hourly", is_crypto=True) - - # Strict detector should potentially detect more gaps than lenient - assert isinstance(lenient_gaps, list) - assert isinstance(strict_gaps, list) + @pytest.mark.parametrize("tolerance", [0.5, 0.01]) + def test_tolerance_cannot_reclassify_off_grid_timestamps(self, tolerance): + """Elapsed-time tolerance must not admit half-hour labels to an hourly grid.""" + df = intraday_rows([datetime(2024, 1, 1), datetime(2024, 1, 1, 1, 30)]) + with pytest.raises(ValueError, match="off-grid"): + GapDetector(tolerance=tolerance).detect_gaps(df, feed_spec=intraday_feed("hourly")) class TestGapDetectorIntegration: """Integration tests for GapDetector.""" def test_realistic_stock_data_gaps(self): - """Test gap detection on realistic stock market data.""" + """An hourly NYSE grid excludes the weekend but retains Monday's missing bar.""" detector = GapDetector() - - # Simulate stock data with weekend gap - df = pl.DataFrame( - { - "timestamp": [ - datetime(2024, 1, 5, 15, 0), # Friday market close - datetime(2024, 1, 8, 9, 30), # Monday market open - datetime(2024, 1, 8, 10, 30), # Monday 1 hour later - datetime(2024, 1, 8, 12, 0), # Monday with small gap - ], - "value": [100.0, 102.0, 103.0, 104.0], - } + df = intraday_rows( + [ + datetime(2024, 1, 5, 20, 30), + datetime(2024, 1, 8, 14, 30), + datetime(2024, 1, 8, 15, 30), + datetime(2024, 1, 8, 17, 30), + ] ) - - gaps = detector.detect_gaps(df, frequency="hourly", is_crypto=False) - - # Should handle weekend gaps appropriately for traditional markets - assert isinstance(gaps, list) + gaps = detector.detect_gaps(df, feed_spec=intraday_feed("hourly", "NYSE")) + assert [gap.missing_periods for gap in gaps] == [1] + assert gaps[0].start.hour == 15 + assert gaps[0].end.hour == 17 def test_crypto_24_7_data_gaps(self): - """Test gap detection on 24/7 crypto data.""" + """Separate outages on a continuous hourly grid must keep their exact sizes.""" detector = GapDetector() - - # Simulate crypto data with regular hourly gaps base_time = datetime(2024, 1, 1, 0, 0) - timestamps = [] - values = [] - - # Add hourly data with some gaps - for hour in [0, 1, 2, 5, 6, 9]: # Missing hours 3,4 and 7,8 - timestamps.append(base_time + timedelta(hours=hour)) - values.append(50000.0 + hour * 100) - - df = pl.DataFrame( - { - "timestamp": timestamps, - "value": values, - } - ) - - gaps = detector.detect_gaps(df, frequency="hourly", is_crypto=True) - - # Should detect the missing hours in 24/7 crypto data - assert len(gaps) >= 1 + df = intraday_rows([base_time + timedelta(hours=hour) for hour in [0, 1, 2, 5, 6, 9]]) + gaps = detector.detect_gaps(df, feed_spec=intraday_feed("hourly")) + assert [gap.missing_periods for gap in gaps] == [2, 2] def test_mixed_frequency_analysis(self): - """Test gap detection across different frequencies on same data.""" + """Observations valid on a minute grid need not belong to an hourly grid.""" detector = GapDetector() - - # Create data with gaps at different scales - df = pl.DataFrame( - { - "timestamp": [ - datetime(2024, 1, 1, 9, 0), - datetime(2024, 1, 1, 9, 5), # 5 minutes later - datetime(2024, 1, 1, 11, 0), # 2 hour gap - datetime(2024, 1, 2, 9, 0), # Next day - ], - "value": [100.0, 101.0, 103.0, 110.0], - } + df = intraday_rows( + [ + datetime(2024, 1, 1, 9, 0), + datetime(2024, 1, 1, 9, 5), + datetime(2024, 1, 1, 11, 0), + datetime(2024, 1, 2, 9, 0), + ] ) - - # Test different frequencies - minute_gaps = detector.detect_gaps(df, frequency="minute", is_crypto=True) - hourly_gaps = detector.detect_gaps(df, frequency="hourly", is_crypto=True) - daily_gaps = detector.detect_gaps(df, frequency="daily", is_crypto=True) - - # Different frequencies should detect different gap patterns - assert isinstance(minute_gaps, list) - assert isinstance(hourly_gaps, list) - assert isinstance(daily_gaps, list) + minute_gaps = detector.detect_gaps(df, feed_spec=intraday_feed("minute")) + assert [gap.missing_periods for gap in minute_gaps] == [4, 114, 1319] + with pytest.raises(ValueError, match="off-grid"): + detector.detect_gaps(df, feed_spec=intraday_feed("hourly")) + assert detector.detect_gaps(df, frequency="daily", calendar="24/7") == [] def test_large_dataset_performance(self): """Test gap detection performance on larger dataset."""