313 lines
7.4 KiB
Python
313 lines
7.4 KiB
Python
# 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,
|
|
)
|
|
|
|
import exchange_calendars as xcals
|
|
from pendulum import (
|
|
now,
|
|
Duration,
|
|
Interval,
|
|
Time,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from ib_insync import (
|
|
TradingSession,
|
|
ContractDetails,
|
|
)
|
|
from exchange_calendars.exchange_calendars import (
|
|
ExchangeCalendar,
|
|
)
|
|
from pandas import (
|
|
# DatetimeIndex,
|
|
TimeDelta,
|
|
Timestamp,
|
|
)
|
|
|
|
|
|
def has_weekend(
|
|
period: Interval,
|
|
) -> bool:
|
|
'''
|
|
Predicate to for a period being within
|
|
days 6->0 (sat->sun).
|
|
|
|
'''
|
|
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 has_holiday(
|
|
con_deats: ContractDetails,
|
|
period: Interval,
|
|
) -> bool:
|
|
'''
|
|
Using the `exchange_calendars` lib detect if a time-gap `period`
|
|
is contained in a known "cash hours" closure.
|
|
|
|
'''
|
|
tz: str = con_deats.timeZoneId
|
|
exch: str = con_deats.contract.primaryExchange
|
|
cal: ExchangeCalendar = xcals.get_calendar(exch)
|
|
end: datetime = period.end
|
|
# _start: datetime = period.start
|
|
# ?TODO, can rm ya?
|
|
# => not that useful?
|
|
# dti: DatetimeIndex = cal.sessions_in_range(
|
|
# _start.date(),
|
|
# end.date(),
|
|
# )
|
|
prev_close: Timestamp = cal.previous_close(
|
|
end.date()
|
|
).tz_convert(tz)
|
|
prev_open: Timestamp = cal.previous_open(
|
|
end.date()
|
|
).tz_convert(tz)
|
|
# now do relative from prev_ values ^
|
|
# to get the next open which should match
|
|
# "contain" the end of the gap.
|
|
next_open: Timestamp = cal.next_open(
|
|
prev_open,
|
|
).tz_convert(tz)
|
|
next_open: Timestamp = cal.next_open(
|
|
prev_open,
|
|
).tz_convert(tz)
|
|
_next_close: Timestamp = cal.next_close(
|
|
prev_close
|
|
).tz_convert(tz)
|
|
cash_gap: TimeDelta = next_open - prev_close
|
|
is_holiday_gap = (
|
|
cash_gap
|
|
>
|
|
period
|
|
)
|
|
# XXX, debug
|
|
# breakpoint()
|
|
return is_holiday_gap
|
|
|
|
|
|
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,
|
|
time_step_s: int,
|
|
) -> 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)
|
|
|
|
# ensure times are in mkt-native timezone
|
|
tz: str = con_deats.timeZoneId
|
|
start = gap.start.in_tz(tz)
|
|
start_t = start.time()
|
|
end = gap.end.in_tz(tz)
|
|
end_t = end.time()
|
|
if (
|
|
(
|
|
start_t in (
|
|
close,
|
|
close.subtract(seconds=time_step_s)
|
|
)
|
|
and
|
|
end_t in (
|
|
open,
|
|
open.add(seconds=time_step_s),
|
|
)
|
|
)
|
|
or
|
|
has_weekend(gap)
|
|
or
|
|
has_holiday(
|
|
con_deats=con_deats,
|
|
period=gap,
|
|
)
|
|
):
|
|
return True
|
|
|
|
# breakpoint()
|
|
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
|
|
# )
|