diff --git a/piker/brokers/ib/venues.py b/piker/brokers/ib/venues.py new file mode 100644 index 00000000..a374d44d --- /dev/null +++ b/piker/brokers/ib/venues.py @@ -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 . + +''' +(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