|
|
"""UTC/Eastern conversion and NYSE calendar helpers."""
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import datetime as dt
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
import exchange_calendars as xcals
|
|
|
|
|
|
_EASTERN = ZoneInfo("America/New_York")
|
|
|
_UTC = ZoneInfo("UTC")
|
|
|
_XNYS = None
|
|
|
|
|
|
|
|
|
def _get_xnys() -> xcals.ExchangeCalendar:
|
|
|
global _XNYS
|
|
|
if _XNYS is None:
|
|
|
_XNYS = xcals.get_calendar("XNYS")
|
|
|
return _XNYS
|
|
|
|
|
|
|
|
|
def utc_now() -> dt.datetime:
|
|
|
return dt.datetime.now(tz=_UTC)
|
|
|
|
|
|
|
|
|
def to_eastern(d: dt.datetime) -> dt.datetime:
|
|
|
if d.tzinfo is None:
|
|
|
raise ValueError("Naive datetime rejected; must be timezone-aware.")
|
|
|
return d.astimezone(_EASTERN)
|
|
|
|
|
|
|
|
|
def to_utc(d: dt.datetime) -> dt.datetime:
|
|
|
if d.tzinfo is None:
|
|
|
raise ValueError("Naive datetime rejected; must be timezone-aware.")
|
|
|
return d.astimezone(_UTC)
|
|
|
|
|
|
|
|
|
def is_trading_day(date: dt.date) -> bool:
|
|
|
cal = _get_xnys()
|
|
|
return cal.is_session(date.isoformat())
|
|
|
|
|
|
|
|
|
def previous_trading_day(date: dt.date) -> dt.date:
|
|
|
cal = _get_xnys()
|
|
|
idx = cal.sessions.get_loc(date.isoformat()) if date.isoformat() in cal.sessions else None
|
|
|
if idx is None:
|
|
|
# Find previous session
|
|
|
prev = cal.previous_session(date.isoformat())
|
|
|
return prev.date()
|
|
|
if idx > 0:
|
|
|
return cal.sessions[idx - 1].date()
|
|
|
raise ValueError(f"No previous trading day before {date}")
|
|
|
|
|
|
|
|
|
def next_trading_day(date: dt.date) -> dt.date:
|
|
|
cal = _get_xnys()
|
|
|
return cal.next_session(date.isoformat()).date()
|
|
|
|
|
|
|
|
|
def filing_time_bucket(filed_at: dt.datetime) -> str:
|
|
|
"""Classify filing time as pre_market, regular_hours, post_market, or unknown."""
|
|
|
if filed_at.tzinfo is None:
|
|
|
return "unknown"
|
|
|
eastern = to_eastern(filed_at)
|
|
|
hour = eastern.hour
|
|
|
minute = eastern.minute
|
|
|
total_minutes = hour * 60 + minute
|
|
|
# Pre-market: before 9:30 ET
|
|
|
if total_minutes < 9 * 60 + 30:
|
|
|
return "pre_market"
|
|
|
# Regular hours: 9:30–16:00 ET
|
|
|
if total_minutes <= 16 * 60:
|
|
|
return "regular_hours"
|
|
|
# Post-market: after 16:00 ET
|
|
|
return "post_market"
|
|
|
|
|
|
|
|
|
def trading_days_between(start: dt.date, end: dt.date) -> list[dt.date]:
|
|
|
cal = _get_xnys()
|
|
|
sessions = cal.sessions_in_range(start.isoformat(), end.isoformat())
|
|
|
return [s.date() for s in sessions]
|