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-codecont_hist_fixes
parent
77518f0758
commit
f484726a8c
|
|
@ -326,7 +326,6 @@ def i3ipc_fin_wins_titled(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def i3ipc_xdotool_manual_click_hack() -> None:
|
def i3ipc_xdotool_manual_click_hack() -> None:
|
||||||
'''
|
'''
|
||||||
Do the data reset hack but expecting a local X-window using `xdotool`.
|
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:
|
except subprocess.TimeoutExpired:
|
||||||
log.exception('xdotool timed out?')
|
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
|
|
||||||
# )
|
|
||||||
|
|
|
||||||
|
|
@ -149,3 +149,83 @@ def is_venue_closure(
|
||||||
return True
|
return True
|
||||||
|
|
||||||
return False
|
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
|
||||||
|
# )
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue