piker/piker/brokers/ib/venues.py

# 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


# TODO, put this into `._util` and call it from here!
#
# NOTE, this was generated by @guille from a gpt5 prompt
# and was originally thot to be needed before learning about
# `ib_insync.contract.ContractDetails._parseSessions()` and
# it's downstream meths..
#
# This is still likely useful to keep for now to parse the
# `.tradingHours: str` value manually if we ever decide
# to move off `ib_async` and implement our own `trio`/`anyio`
# based version Bp
#
# >attempt to parse the retarted ib "time stampy thing" they
# >do for "venue hours" with this.. written by
# >gpt5-"thinking",
#


def parse_trading_hours(
    spec: str,
    tz: TzInfo|None = None
) -> dict[
    date,
    tuple[datetime, datetime]
]|None:
    '''
    Parse venue hours like:
    'YYYYMMDD:HHMM-YYYYMMDD:HHMM;YYYYMMDD:CLOSED;...'

    Returns `dict[date] = (open_dt, close_dt)` or `None` if
    closed.

    '''
    if (
        not isinstance(spec, str)
        or
        not spec
    ):
        raise ValueError('spec must be a non-empty string')

    out: dict[
        date,
        tuple[datetime, datetime]
    ]|None = {}

    for part in (p.strip() for p in spec.split(';') if p.strip()):
        if part.endswith(':CLOSED'):
            day_s, _ = part.split(':', 1)
            d = datetime.strptime(day_s, '%Y%m%d').date()
            out[d] = None
            continue

        try:
            start_s, end_s = part.split('-', 1)
            start_dt = datetime.strptime(start_s, '%Y%m%d:%H%M')
            end_dt = datetime.strptime(end_s, '%Y%m%d:%H%M')
        except ValueError as exc:
            raise ValueError(f'invalid segment: {part}') from exc

        if tz is not None:
            start_dt = start_dt.replace(tzinfo=tz)
            end_dt = end_dt.replace(tzinfo=tz)

        out[start_dt.date()] = (start_dt, end_dt)

    return out


# ORIG desired usage,
#
# TODO, for non-drunk tomorrow,
# - call above fn and check that `output[today] is not None`
# trading_hrs: dict = parse_trading_hours(
#     details.tradingHours
# )
# liq_hrs: dict = parse_trading_hours(
#     details.liquidHours
    # )