From f484726a8cf748d417fe3b622ef4caaa1fcf055b Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 5 Feb 2026 16:51:26 -0500 Subject: [PATCH] Mv `parse_trading_hours()` from `._util` to `.venues` It was an AI-model draft that we can prolly toss but figured might as well org it appropriately. (this commit msg was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- piker/brokers/ib/_util.py | 97 -------------------------------------- piker/brokers/ib/venues.py | 80 +++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 97 deletions(-) diff --git a/piker/brokers/ib/_util.py b/piker/brokers/ib/_util.py index 5ecd4e55..58114233 100644 --- a/piker/brokers/ib/_util.py +++ b/piker/brokers/ib/_util.py @@ -326,7 +326,6 @@ def i3ipc_fin_wins_titled( ) - def i3ipc_xdotool_manual_click_hack() -> None: ''' Do the data reset hack but expecting a local X-window using `xdotool`. @@ -388,99 +387,3 @@ def i3ipc_xdotool_manual_click_hack() -> None: ]) except subprocess.TimeoutExpired: log.exception('xdotool timed out?') - - - -def is_current_time_in_range( - start_dt: datetime, - end_dt: datetime, -) -> 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. - - ''' - now: datetime = datetime.now(start_dt.tzinfo) - return start_dt <= now <= end_dt - - -# 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 - # ) diff --git a/piker/brokers/ib/venues.py b/piker/brokers/ib/venues.py index a374d44d..3d9bf4a1 100644 --- a/piker/brokers/ib/venues.py +++ b/piker/brokers/ib/venues.py @@ -149,3 +149,83 @@ def is_venue_closure( 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 + # )