Add `.ib.venues` for mkt-venue-closure checkers

Introduce set of helper-fns for detecting venue open/close status,
session start/end times, and related time-gap detection using
`pendulum`.

Deats,
- add `iter_sessions()` to yield `pendulum.Interval`s from
  a `ContractDetails` instance.
- add `is_venue_open()` to check if active at a given time.
- add `is_venue_closure()` to detect valid closure gaps.
- add `sesh_times()` to extract weekday-agnostic open/close times.
- add `has_weekend()` to check for Sat/Sun in interval.
- move in lowlevel `is_current_time_in_range()` for checking a
  datetime within a `sesh: pendulum.Interval`.

(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
cont_hist_fixes
Gud Boi 2026-02-05 16:36:11 -05:00
parent ef3309adf9
commit 77518f0758
1 changed files with 151 additions and 0 deletions

View File

@ -0,0 +1,151 @@
# piker: trading gear for hackers
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
(Multi-)venue mgmt helpers.
IB generally supports all "legacy" trading venues, those mostly owned
by ICE and friends.
'''
from __future__ import annotations
from datetime import ( # noqa
datetime,
date,
tzinfo as TzInfo,
)
from typing import (
Iterator,
TYPE_CHECKING,
)
from pendulum import (
now,
Duration,
Interval,
Time,
)
if TYPE_CHECKING:
from ib_insync import (
TradingSession,
ContractDetails,
)
def has_weekend(
period: Interval,
) -> bool:
has_weekend: bool = False
for dt in period:
if dt.day_of_week in [0, 6]: # 0=Sunday, 6=Saturday
has_weekend = True
break
return has_weekend
def is_current_time_in_range(
sesh: Interval,
when: datetime|None = None,
) -> bool:
'''
Check if current time is within the datetime range.
Use any/the-same timezone as provided by `start_dt.tzinfo` value
in the range.
'''
when: datetime = when or now()
return when in sesh
def iter_sessions(
con_deats: ContractDetails,
) -> Iterator[Interval]:
'''
Yield `pendulum.Interval`s for all
`ibas.ContractDetails.tradingSessions() -> TradingSession`s.
'''
sesh: TradingSession
for sesh in con_deats.tradingSessions():
yield Interval(*sesh)
def sesh_times(
con_deats: ContractDetails,
) -> tuple[Time, Time]:
'''
Based on the earliest trading session provided by the IB API,
get the (day-agnostic) times for the start/end.
'''
earliest_sesh: Interval = next(iter_sessions(con_deats))
return (
earliest_sesh.start.time(),
earliest_sesh.end.time(),
)
# ^?TODO, use `.diff()` to get point-in-time-agnostic period?
# https://pendulum.eustace.io/docs/#difference
def is_venue_open(
con_deats: ContractDetails,
when: datetime|Duration|None = None,
) -> bool:
'''
Check if market-venue is open during `when`, which defaults to
"now".
'''
sesh: Interval
for sesh in iter_sessions(con_deats):
if is_current_time_in_range(
sesh=sesh,
when=when,
):
return True
return False
def is_venue_closure(
gap: Interval,
con_deats: ContractDetails,
) -> bool:
'''
Check if a provided time-`gap` is just an (expected) trading
venue closure period.
'''
open: Time
close: Time
open, close = sesh_times(con_deats)
# TODO! ensure this works!
# breakpoint()
if (
(
gap.start.time() == close
and
gap.end.time() == open
)
or
has_weekend(gap)
):
return True
return False