from datetime import date, timedelta import pandas_market_calendars as _mcal # optional dep — fall back gracefully from typing import Optional import logging logger = logging.getLogger(__name__) _NYSE_CAL = None def _get_nyse(): global _NYSE_CAL if _NYSE_CAL is None: try: import pandas_market_calendars as mcal _NYSE_CAL = mcal.get_calendar("NYSE") except ImportError: pass return _NYSE_CAL def is_trading_day(d: date) -> bool: """Return True if d is a NYSE trading day.""" cal = _get_nyse() if cal is None: return d.weekday() < 5 import pandas as pd schedule = cal.schedule(start_date=str(d), end_date=str(d)) return not schedule.empty def next_trading_day(d: date) -> date: """Return the next trading day after d.""" candidate = d + timedelta(days=1) while not is_trading_day(candidate): candidate += timedelta(days=1) return candidate def next_friday(from_date: Optional[date] = None) -> date: """Return the next Friday on or after from_date.""" d = from_date or date.today() days_ahead = 4 - d.weekday() # Friday is weekday 4 if days_ahead < 0: days_ahead += 7 elif days_ahead == 0: pass # today is Friday return d + timedelta(days=days_ahead) def nearest_monthly_expiry(from_date: Optional[date] = None, target_dte: int = 30) -> date: """Return the standard monthly expiry (third Friday) closest to target_dte from from_date.""" d = from_date or date.today() target = d + timedelta(days=target_dte) # Find third Friday of that month year, month = target.year, target.month third_friday = _third_friday(year, month) # If the third Friday of target month is already past relative to today, advance one month if third_friday <= d: if month == 12: year += 1 month = 1 else: month += 1 third_friday = _third_friday(year, month) return third_friday def _third_friday(year: int, month: int) -> date: """Return the third Friday of the given month.""" first = date(year, month, 1) # Find first Friday days_to_friday = (4 - first.weekday()) % 7 first_friday = first + timedelta(days=days_to_friday) return first_friday + timedelta(weeks=2) def market_is_open_now() -> bool: """Best-effort check: is the US market currently open (9:30–16:00 ET)?""" from datetime import datetime import zoneinfo now_et = datetime.now(tz=zoneinfo.ZoneInfo("America/New_York")) if not is_trading_day(now_et.date()): return False open_time = now_et.replace(hour=9, minute=30, second=0, microsecond=0) close_time = now_et.replace(hour=16, minute=0, second=0, microsecond=0) return open_time <= now_et <= close_time def within_dte_window_for_0dte() -> bool: """True if it's a trading day and between 9:30 AM and 2:00 PM ET.""" from datetime import datetime import zoneinfo now_et = datetime.now(tz=zoneinfo.ZoneInfo("America/New_York")) if not is_trading_day(now_et.date()): return False open_time = now_et.replace(hour=9, minute=30, second=0, microsecond=0) cutoff = now_et.replace(hour=14, minute=0, second=0, microsecond=0) return open_time <= now_et <= cutoff