Harden `.ib.venues` against unknown exchange cals

Deats,
- catch `InvalidCalendarName` in `has_holiday()` so
  venues without an `exchange_calendars` entry (eg.
  `IDEALPRO` for forex, `PAXOS` for crypto) gracefully
  return `False` instead of raising.
- add `log` via `get_logger()` to emit a warning when
  skipping the holiday check for an unmapped venue.
- fix `std_exch` type annot from `dict` -> `str`.
- guard `is_expired()` against empty
  `.realExpirationDate` strings.
- fill in `is_expired()` docstring.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
Gud Boi 2026-03-24 13:24:39 -04:00
parent 6402146116
commit 4b9834a775
1 changed files with 33 additions and 6 deletions

View File

@ -33,6 +33,9 @@ from typing import (
) )
import exchange_calendars as xcals import exchange_calendars as xcals
from exchange_calendars.errors import (
InvalidCalendarName,
)
from pendulum import ( from pendulum import (
parse, parse,
now, now,
@ -41,6 +44,10 @@ from pendulum import (
Time, Time,
) )
from piker.log import get_logger
log = get_logger(__name__)
if TYPE_CHECKING: if TYPE_CHECKING:
from ib_async import ( from ib_async import (
TradingSession, TradingSession,
@ -61,10 +68,15 @@ def is_expired(
con_deats: ContractDetails, con_deats: ContractDetails,
) -> bool: ) -> bool:
''' '''
Predicate Simple predicate whether the provided contract-deats match and
already lifetime-terminated instrument.
''' '''
expiry_dt: datetime = parse(con_deats.realExpirationDate) expiry_str: str = con_deats.realExpirationDate
if not expiry_str:
return False
expiry_dt: datetime = parse(expiry_str)
return expiry_dt.date() >= now().date() return expiry_dt.date() >= now().date()
@ -102,13 +114,28 @@ def has_holiday(
con.exchange con.exchange
) )
# XXX, ad-hoc handle any IB exchange which are non-std # XXX, ad-hoc handle any IB exchange which are
# via lookup table.. # non-std via lookup table..
std_exch: dict = { std_exch: str = {
'ARCA': 'ARCX', 'ARCA': 'ARCX',
}.get(exch, exch) }.get(exch, exch)
cal: ExchangeCalendar = xcals.get_calendar(std_exch) try:
cal: ExchangeCalendar = xcals.get_calendar(
std_exch
)
except InvalidCalendarName:
# venue has no `exchange_calendars` entry
# (eg. IDEALPRO for forex, PAXOS for
# crypto) -> not a holiday by default since
# weekends are already handled by
# `has_weekend()`.
log.warning(
f'No exchange cal for {std_exch!r},'
f' skipping holiday check..\n'
)
return False
end: datetime = period.end end: datetime = period.end
# _start: datetime = period.start # _start: datetime = period.start
# ?TODO, can rm ya? # ?TODO, can rm ya?