From d99c60f37506c00df802fec6ff27570a0492da0e Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 15:46:43 -0400 Subject: [PATCH 1/2] fix: honor CME crypto cutover and business trade dates --- docs/user-guide/sessions.md | 21 ++++ src/ml4t/data/sessions/_cme_crypto.py | 85 ++++++++++++++++ src/ml4t/data/sessions/assigner.py | 33 +++++-- src/ml4t/data/sessions/completer.py | 14 ++- tests/test_cme_crypto_cutover.py | 137 ++++++++++++++++++++++++++ 5 files changed, 275 insertions(+), 15 deletions(-) create mode 100644 src/ml4t/data/sessions/_cme_crypto.py create mode 100644 tests/test_cme_crypto_cutover.py diff --git a/docs/user-guide/sessions.md b/docs/user-guide/sessions.md index a312d45..528f39f 100644 --- a/docs/user-guide/sessions.md +++ b/docs/user-guide/sessions.md @@ -114,6 +114,27 @@ assigner = SessionAssigner("CME Globex Crypto") Calendar names and schedules can change upstream. Check the installed calendar before assigning a large dataset. +### CME Crypto and Spot Crypto + +`CME` selects CME Globex cryptocurrency futures and options, not all CME products and not +spot-quoted futures. Data preserves the upstream historical schedule through May 29, 2026. After +that day's 16:00-16:30 Chicago maintenance, trading includes weekends and holidays. Regular Globex +maintenance is Monday-Friday 16:00-16:02 Chicago time and Saturday 02:00-04:00. Pre-open periods +are not trading intervals. See the [CME crypto FAQ](https://www.cmegroup.com/articles/faqs/frequently-asked-questions-cryptocurrency-futures.html) +and [launch notice](https://cmegroupclientsite.atlassian.net/wiki/spaces/EPICSANDBOX/pages/1283194884/Cryptocurrency+Futures+and+Options+Migration+to+24-7+Trading). + +CME observations retain the following business trade date across weekends and holidays, rather +than acquiring a separate Saturday or Sunday label. Data uses the upstream `CME_TradeDate` +business calendar, qualified against CME's +[2026 reporting calendar](https://www.cmegroup.com/content/dam/cmegroup/notices/market-regulation/2026/01/msn01-09-2026revisedholidayprocessinglargetrader2026.pdf). +Schedule queries include the next business date when needed; a Saturday observation can therefore +map to Monday, or Tuesday after Labor Day. Assignment and completion share these intervals and +exclude maintenance. The adapter does not change the upstream calendar registry or other products. + +Ordinary spot crypto uses `calendar="24/7"`, with civil-day labels and no CME maintenance. Explicit +`calendar=None` disables session handling instead. Select the schedule from the feed's actual +market, not from its symbol or asset-class name. + ## Complete Minute Sessions `SessionCompleter` expands minute bars to the exchange schedule. It requires `timestamp`, `open`, diff --git a/src/ml4t/data/sessions/_cme_crypto.py b/src/ml4t/data/sessions/_cme_crypto.py new file mode 100644 index 0000000..ad76c1f --- /dev/null +++ b/src/ml4t/data/sessions/_cme_crypto.py @@ -0,0 +1,85 @@ +"""CME crypto's May 2026 transition, without changing the upstream registry.""" + +from __future__ import annotations + +from typing import Any + +import pandas as pd + + +class CMEGlobexCryptoSchedule: + def __init__(self, legacy: Any, trade_dates: Any) -> None: + self._legacy = legacy + self._trade_dates = trade_dates + self._business_day = trade_dates.holidays() + self.tz = legacy.tz + self.name = legacy.name + + def schedule( + self, start_date: Any, end_date: Any, tz: Any = "UTC", *, interruptions: bool = False + ) -> pd.DataFrame: + start = pd.Timestamp(start_date).tz_localize(None).normalize() + end = pd.Timestamp(end_date).tz_localize(None).normalize() + if start > end: + raise ValueError("start_date must not be later than end_date") + cutover = pd.Timestamp("2026-05-29") + frames = [] + if start <= cutover: + frames.append( + self._legacy.schedule(start, min(end, cutover), tz=tz, interruptions=interruptions) + ) + if end > cutover: + labels = self._trade_dates.valid_days( + max(start, cutover + pd.Timedelta(days=1)), end, tz=None + ) + rows = [] + for label in labels: + previous = label - self._business_day + opening = previous + pd.Timedelta(hours=16, minutes=2) + if previous == cutover: + opening = previous + pd.Timedelta(hours=16, minutes=30) + closing = label + pd.Timedelta(hours=16) + row = { + "market_open": opening.tz_localize(self.tz).tz_convert(tz), + "market_close": closing.tz_localize(self.tz).tz_convert(tz), + } + if interruptions: + closures = [] + for day in pd.date_range(previous, label): + if day.dayofweek < 5: + halt = day + pd.Timedelta(hours=16) + reopen = halt + pd.Timedelta(minutes=2) + elif day.dayofweek == 5: + halt = day + pd.Timedelta(hours=2) + reopen = halt + pd.Timedelta(hours=2) + else: + continue + if opening < reopen and halt < closing: + closures.append((halt, reopen)) + for index, (halt, reopen) in enumerate(closures, start=1): + row[f"interruption_start_{index}"] = halt.tz_localize(self.tz).tz_convert( + tz + ) + row[f"interruption_end_{index}"] = reopen.tz_localize(self.tz).tz_convert( + tz + ) + rows.append(row) + frames.append( + pd.DataFrame( + rows, index=labels, columns=None if rows else ["market_open", "market_close"] + ) + ) + return pd.concat(frames).sort_index() + + def schedule_for_instants(self, start: Any, end: Any) -> pd.DataFrame: + start_timestamp, end_timestamp = pd.Timestamp(start), pd.Timestamp(end) + start_local = ( + start_timestamp.tz_convert(self.tz) if start_timestamp.tzinfo else start_timestamp + ) + end_local = end_timestamp.tz_convert(self.tz) if end_timestamp.tzinfo else end_timestamp + next_trade_date = pd.Timestamp(end_local.date()) + self._business_day + return self.schedule( + pd.Timestamp(start_local.date()) - pd.Timedelta(days=1), + next_trade_date, + interruptions=True, + ) diff --git a/src/ml4t/data/sessions/assigner.py b/src/ml4t/data/sessions/assigner.py index 67935ff..da2e238 100644 --- a/src/ml4t/data/sessions/assigner.py +++ b/src/ml4t/data/sessions/assigner.py @@ -52,6 +52,20 @@ def _session_intervals(row: Any) -> list[tuple[datetime, datetime]]: return intervals +def _schedule_for_instants(calendar: Any, start: Any, end: Any) -> Any: + import pandas as pd + + from ml4t.data.sessions._cme_crypto import CMEGlobexCryptoSchedule + + if isinstance(calendar, CMEGlobexCryptoSchedule): + return calendar.schedule_for_instants(start, end) + return calendar.schedule( + start_date=pd.Timestamp(start).normalize() - pd.Timedelta(days=1), + end_date=pd.Timestamp(end).normalize() + pd.Timedelta(days=1), + interruptions=True, + ) + + # Exchange to calendar mapping EXCHANGE_CALENDARS = { "CME": "CME Globex Crypto", @@ -111,8 +125,7 @@ class SessionAssigner: Uses pandas_market_calendars to determine trading sessions and assigns a session_date column to each timestamp. - For CME futures, sessions start at 5pm CT Sunday and end at 4pm CT Friday. - The session_date is the date when the session ENDS (4pm date). + CME crypto uses business trade dates across overnight and weekend trading. """ def __init__( @@ -168,6 +181,16 @@ def __init__( try: self.calendar = mcal.get_calendar(self.calendar_name) + from pandas_market_calendars.calendars.cme_globex_crypto import ( + CMEGlobexCryptoExchangeCalendar, + ) + + if type(self.calendar) is CMEGlobexCryptoExchangeCalendar: + from ml4t.data.sessions._cme_crypto import CMEGlobexCryptoSchedule + + self.calendar = CMEGlobexCryptoSchedule( + self.calendar, mcal.get_calendar("CME_TradeDate") + ) except Exception as e: raise ValueError(f"Unknown calendar '{self.calendar_name}': {e}") @@ -265,11 +288,7 @@ def assign_sessions( end_timestamp, pd.Timestamp ): raise ValueError("start_date and end_date must not be NaT") - start_pd = start_timestamp.normalize() - pd.Timedelta(days=1) - end_pd = end_timestamp.normalize() + pd.Timedelta(days=1) - schedule = self.calendar.schedule( - start_date=start_pd, end_date=end_pd, interruptions=True - ) + schedule = _schedule_for_instants(self.calendar, start_timestamp, end_timestamp) logger.debug(f"Got {len(schedule)} sessions from calendar") session_map = [] diff --git a/src/ml4t/data/sessions/completer.py b/src/ml4t/data/sessions/completer.py index 4efc3da..36838df 100644 --- a/src/ml4t/data/sessions/completer.py +++ b/src/ml4t/data/sessions/completer.py @@ -11,7 +11,11 @@ import polars as pl import structlog -from ml4t.data.sessions.assigner import SessionAssigner, _session_intervals +from ml4t.data.sessions.assigner import ( + SessionAssigner, + _schedule_for_instants, + _session_intervals, +) from ml4t.specs import UNSET_CALENDAR, BarLabel, CalendarUnset, TimestampSemantics logger = structlog.get_logger() @@ -166,13 +170,7 @@ def complete_sessions( ): raise ValueError("start_date and end_date must be minute-aligned") - start_pd = request_start.normalize() - pd.Timedelta(days=1) - end_pd = request_end.normalize() + pd.Timedelta(days=1) - - # Get trading schedule - schedule = self.calendar.schedule( - start_date=start_pd, end_date=end_pd, interruptions=True - ) + schedule = _schedule_for_instants(self.calendar, request_start, request_end) if len(schedule) == 0: logger.warning("No trading sessions found in date range") diff --git a/tests/test_cme_crypto_cutover.py b/tests/test_cme_crypto_cutover.py new file mode 100644 index 0000000..374dc22 --- /dev/null +++ b/tests/test_cme_crypto_cutover.py @@ -0,0 +1,137 @@ +"""Real provider rows must survive the CME crypto schedule transition.""" + +from datetime import date, datetime + +import polars as pl +import pytest +from polars.testing import assert_frame_equal + +from ml4t.data import DataManager +from ml4t.data.providers import SyntheticProvider +from ml4t.data.sessions import SessionAssigner, SessionCompleter +from ml4t.specs import FeedSpec + + +def provider_rows(day: str, clocks: list[str]) -> pl.DataFrame: + with SyntheticProvider(calendar_mode="continuous") as provider: + frame = provider.fetch_ohlcv("BTCUSD", day, day, "minute") + selected = frame.filter(pl.col("timestamp").dt.strftime("%H:%M").is_in(clocks)) + assert selected.height == len(clocks) + return selected + + +@pytest.mark.parametrize( + "day,clock,expected", + [ + ("2026-05-23", "12:00", None), + ("2026-05-30", "12:00", date(2026, 6, 1)), + ("2026-10-17", "12:00", date(2026, 10, 19)), + ("2026-10-16", "21:03", date(2026, 10, 19)), + ("2026-10-16", "21:00", None), + ("2026-10-16", "21:01", None), + ("2026-10-17", "07:59", None), + ("2026-10-17", "09:00", date(2026, 10, 19)), + ("2026-12-05", "08:00", None), + ("2026-12-05", "10:00", date(2026, 12, 7)), + ("2026-05-29", "21:15", None), + ("2026-05-29", "21:30", date(2026, 6, 1)), + ("2026-09-05", "12:00", date(2026, 9, 8)), + ("2026-07-02", "21:03", date(2026, 7, 6)), + ("2026-06-18", "21:03", date(2026, 6, 22)), + ("2026-11-25", "22:03", date(2026, 11, 27)), + ("2026-12-24", "22:03", date(2026, 12, 28)), + ("2026-10-12", "12:00", date(2026, 10, 12)), + ("2026-11-11", "12:00", date(2026, 11, 11)), + ("2026-05-29", "20:59", date(2026, 5, 29)), + ("2026-05-28", "21:30", None), + ("2026-05-28", "22:00", date(2026, 5, 29)), + ("2026-11-01", "07:30", date(2026, 11, 2)), + ], +) +def test_cme_crypto_uses_cutover_maintenance_and_business_dates(day, clock, expected): + frame = provider_rows(day, [clock]) + assigned = SessionAssigner.from_exchange("CME").assign_sessions(frame, bar_frequency="intraday") + assert assigned["session_date"].to_list() == [expected] + assert_frame_equal(assigned.select(frame.columns), frame) + + +def test_cme_completion_preserves_weekend_observations_and_fills_only_open_gap(): + frame = provider_rows("2026-10-17", ["12:00", "12:02"]) + completed = SessionCompleter("CME Globex Crypto").complete_sessions(frame) + assert completed.height == 3 + assert completed["is_imputed"].to_list() == [False, True, False] + assert completed["session_date"].to_list() == [date(2026, 10, 19)] * 3 + observed = completed.filter(~pl.col("is_imputed")).select(frame.columns).cast(frame.schema) + assert_frame_equal(observed, frame) + + +def test_spot_continuous_calendar_keeps_civil_dates_not_cme_trade_dates(): + frame = provider_rows("2026-10-17", ["12:00", "12:02"]) + assigned = SessionAssigner("24/7").assign_sessions(frame, bar_frequency="intraday") + assert assigned["session_date"].to_list() == [date(2026, 10, 17)] * 2 + completed = SessionCompleter("24/7").complete_sessions(frame) + assert completed.height == 3 + assert completed["session_date"].to_list() == [date(2026, 10, 17)] * 3 + + +def test_calendar_none_never_assigns_or_manufactures_cme_or_spot_rows(): + frame = provider_rows("2026-10-17", ["12:00", "12:02"]) + assert_frame_equal(SessionAssigner(None).assign_sessions(frame), frame) + assert_frame_equal(SessionCompleter(None).complete_sessions(frame), frame) + + +@pytest.mark.parametrize( + "day,clocks,expected", + [ + ("2026-10-17", ["06:59", "09:00"], [date(2026, 10, 19)] * 2), + ("2026-09-07", ["20:59", "21:02"], [date(2026, 9, 8)] * 2), + ("2026-05-29", ["20:59", "21:30"], [date(2026, 5, 29), date(2026, 6, 1)]), + ], +) +def test_completion_does_not_fill_weekly_holiday_or_launch_maintenance(day, clocks, expected): + frame = provider_rows(day, clocks) + completed = SessionCompleter("CME Globex Crypto").complete_sessions(frame) + assert completed.height == 2 + assert completed["is_imputed"].to_list() == [False, False] + assert completed["session_date"].to_list() == expected + assert_frame_equal(completed.select(frame.columns).cast(frame.schema), frame) + + +@pytest.mark.parametrize("alias", ["CME Globex Crypto", "CME Globex Cryptocurrencies"]) +def test_cme_aliases_and_facade_honor_shared_override_and_disable(alias): + frame = provider_rows("2026-10-17", ["12:00"]) + feed = FeedSpec(calendar=alias, timestamp_semantics="event_time") + with DataManager() as manager: + assert manager.assign_sessions(frame, feed_spec=feed)["session_date"].to_list() == [ + date(2026, 10, 19) + ] + assert manager.assign_sessions(frame, calendar="24/7", feed_spec=feed)[ + "session_date" + ].to_list() == [date(2026, 10, 17)] + assert_frame_equal(manager.assign_sessions(frame, calendar=None, feed_spec=feed), frame) + + +def test_selected_crypto_schedule_does_not_modify_upstream_or_other_product_calendars(): + import pandas_market_calendars as mcal + from pandas_market_calendars.calendars.cme_globex_crypto import CMEGlobexCryptoExchangeCalendar + + SessionAssigner("CME Globex Crypto") + legacy = mcal.get_calendar("CME Globex Crypto") + assert type(legacy) is CMEGlobexCryptoExchangeCalendar + assert legacy.schedule("2026-10-17", "2026-10-17").empty + other = SessionAssigner("CME_Equity") + frame = provider_rows("2026-10-17", ["12:00"]) + assert other.assign_sessions(frame, bar_frequency="intraday")["session_date"].to_list() == [ + None + ] + + +def test_cme_daily_labels_are_business_dates_not_each_open_weekend_civil_date(): + with SyntheticProvider(calendar_mode="continuous") as provider: + daily = provider.fetch_ohlcv("BTCUSD", "2026-10-17", "2026-10-19", "daily") + feed = FeedSpec(calendar="CME Globex Crypto", timestamp_semantics="session_label") + assigned = SessionAssigner(feed_spec=feed).assign_sessions(daily) + assert assigned["session_date"].to_list() == [None, None, date(2026, 10, 19)] + info = SessionCompleter("CME Globex Crypto").get_session_info(datetime(2026, 10, 19)) + assert info["market_open"].isoformat() == "2026-10-16T21:02:00+00:00" + assert info["market_close"].isoformat() == "2026-10-19T21:00:00+00:00" From b75d05f085cb44e0dd0e0eda169fa76600646fbf Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 16:06:05 -0400 Subject: [PATCH 2/2] fix: extend intraday grid queries through CME business trade dates --- src/ml4t/data/sessions/assigner.py | 6 +++--- src/ml4t/data/utils/gaps.py | 7 ++----- tests/test_cme_crypto_cutover.py | 31 ++++++++++++++++++++++++++++++ 3 files changed, 36 insertions(+), 8 deletions(-) diff --git a/src/ml4t/data/sessions/assigner.py b/src/ml4t/data/sessions/assigner.py index da2e238..9193a38 100644 --- a/src/ml4t/data/sessions/assigner.py +++ b/src/ml4t/data/sessions/assigner.py @@ -52,7 +52,7 @@ def _session_intervals(row: Any) -> list[tuple[datetime, datetime]]: return intervals -def _schedule_for_instants(calendar: Any, start: Any, end: Any) -> Any: +def _schedule_for_instants(calendar: Any, start: Any, end: Any, *, padding_days: int = 1) -> Any: import pandas as pd from ml4t.data.sessions._cme_crypto import CMEGlobexCryptoSchedule @@ -60,8 +60,8 @@ def _schedule_for_instants(calendar: Any, start: Any, end: Any) -> Any: if isinstance(calendar, CMEGlobexCryptoSchedule): return calendar.schedule_for_instants(start, end) return calendar.schedule( - start_date=pd.Timestamp(start).normalize() - pd.Timedelta(days=1), - end_date=pd.Timestamp(end).normalize() + pd.Timedelta(days=1), + start_date=pd.Timestamp(start).normalize() - pd.Timedelta(days=padding_days), + end_date=pd.Timestamp(end).normalize() + pd.Timedelta(days=padding_days), interruptions=True, ) diff --git a/src/ml4t/data/utils/gaps.py b/src/ml4t/data/utils/gaps.py index 4d4e81d..9aa863e 100644 --- a/src/ml4t/data/utils/gaps.py +++ b/src/ml4t/data/utils/gaps.py @@ -14,6 +14,7 @@ from ml4t.data.sessions.assigner import ( SessionAssigner, _daily_session_labels, + _schedule_for_instants, _session_intervals, ) from ml4t.specs import ( @@ -247,11 +248,7 @@ def _detect_intraday_gaps( 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, - ) + schedule = _schedule_for_instants(assigner.calendar, start, end, padding_days=2) grid_intervals: list[tuple[datetime, datetime, int, int]] = [] total_bars = 0 for _, row in schedule.iterrows(): diff --git a/tests/test_cme_crypto_cutover.py b/tests/test_cme_crypto_cutover.py index 374dc22..f2697e0 100644 --- a/tests/test_cme_crypto_cutover.py +++ b/tests/test_cme_crypto_cutover.py @@ -9,6 +9,7 @@ from ml4t.data import DataManager from ml4t.data.providers import SyntheticProvider from ml4t.data.sessions import SessionAssigner, SessionCompleter +from ml4t.data.utils.gaps import GapDetector from ml4t.specs import FeedSpec @@ -135,3 +136,33 @@ def test_cme_daily_labels_are_business_dates_not_each_open_weekend_civil_date(): info = SessionCompleter("CME Globex Crypto").get_session_info(datetime(2026, 10, 19)) assert info["market_open"].isoformat() == "2026-10-16T21:02:00+00:00" assert info["market_close"].isoformat() == "2026-10-19T21:00:00+00:00" + + +@pytest.mark.parametrize( + "day,clocks,missing", + [ + ("2026-09-05", ["12:00", "12:02"], 1), + ("2026-10-17", ["06:59", "09:00"], 0), + ("2026-09-07", ["20:59", "21:02"], 0), + ("2026-05-29", ["20:59", "21:30"], 0), + ], +) +def test_intraday_grid_uses_business_date_coverage_and_excludes_cme_maintenance( + day, clocks, missing +): + frame = provider_rows(day, clocks) + feed = FeedSpec( + time_contract_version=2, + calendar="CME Globex Crypto", + data_frequency="minute", + timestamp_semantics="event_time", + bar_grid_anchor="session_open", + bar_label="left", + partial_bar_policy="exclude", + ) + gaps = GapDetector().detect_gaps(frame, feed_spec=feed) + assert sum(gap.missing_periods for gap in gaps) == missing + if day == "2026-10-17": + spot = GapDetector().detect_gaps(frame, calendar="24/7", feed_spec=feed) + assert sum(gap.missing_periods for gap in spot) == 120 + assert GapDetector().detect_gaps(frame, calendar=None, feed_spec=feed) == []