Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions docs/user-guide/sessions.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`,
Expand Down
85 changes: 85 additions & 0 deletions src/ml4t/data/sessions/_cme_crypto.py
Original file line number Diff line number Diff line change
@@ -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,
)
33 changes: 26 additions & 7 deletions src/ml4t/data/sessions/assigner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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, *, padding_days: int = 1) -> 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=padding_days),
end_date=pd.Timestamp(end).normalize() + pd.Timedelta(days=padding_days),
interruptions=True,
)


# Exchange to calendar mapping
EXCHANGE_CALENDARS = {
"CME": "CME Globex Crypto",
Expand Down Expand Up @@ -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__(
Expand Down Expand Up @@ -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}")

Expand Down Expand Up @@ -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 = []
Expand Down
14 changes: 6 additions & 8 deletions src/ml4t/data/sessions/completer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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")
Expand Down
7 changes: 2 additions & 5 deletions src/ml4t/data/utils/gaps.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from ml4t.data.sessions.assigner import (
SessionAssigner,
_daily_session_labels,
_schedule_for_instants,
_session_intervals,
)
from ml4t.specs import (
Expand Down Expand Up @@ -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():
Expand Down
168 changes: 168 additions & 0 deletions tests/test_cme_crypto_cutover.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,168 @@
"""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.data.utils.gaps import GapDetector
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"


@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) == []
Loading