64 lines
2 KiB
Python
64 lines
2 KiB
Python
"""Stock market open and close times."""
|
|
|
|
from datetime import timedelta, timezone
|
|
|
|
import dateutil.tz
|
|
import exchange_calendars
|
|
import pandas
|
|
|
|
here = dateutil.tz.tzlocal()
|
|
|
|
|
|
def timedelta_display(delta: timedelta) -> str:
|
|
"""Format timedelta as a human readable string."""
|
|
total_seconds = int(delta.total_seconds())
|
|
days, remainder = divmod(total_seconds, 24 * 60 * 60)
|
|
hours, remainder = divmod(remainder, 60 * 60)
|
|
mins, secs = divmod(remainder, 60)
|
|
|
|
return " ".join(
|
|
f"{v:>3} {label}"
|
|
for v, label in ((days, "days"), (hours, "hrs"), (mins, "mins"))
|
|
if v
|
|
)
|
|
|
|
|
|
def open_and_close() -> list[str]:
|
|
"""Stock markets open and close times."""
|
|
# The trading calendars code is slow, maybe there is a faster way to do this
|
|
# Or we could cache the result
|
|
now = pandas.Timestamp.now(timezone.utc)
|
|
now_local = pandas.Timestamp.now(here)
|
|
markets = [
|
|
("XLON", "London"),
|
|
("XNYS", "US"),
|
|
]
|
|
reply = []
|
|
for code, label in markets:
|
|
cal = exchange_calendars.get_calendar(code)
|
|
|
|
if cal.is_open_on_minute(now_local):
|
|
next_close = cal.next_close(now).tz_convert(here)
|
|
next_close = next_close.replace(minute=round(next_close.minute, -1))
|
|
delta_close = timedelta_display(next_close - now_local)
|
|
|
|
prev_open = cal.previous_open(now).tz_convert(here)
|
|
prev_open = prev_open.replace(minute=round(prev_open.minute, -1))
|
|
delta_open = timedelta_display(now_local - prev_open)
|
|
|
|
msg = (
|
|
f"{label:>6} market opened {delta_open} ago, "
|
|
+ f"closes in {delta_close} ({next_close:%H:%M})"
|
|
)
|
|
else:
|
|
ts = cal.next_open(now)
|
|
ts = ts.replace(minute=round(ts.minute, -1))
|
|
ts = ts.tz_convert(here)
|
|
delta = timedelta_display(ts - now_local)
|
|
msg = f"{label:>6} market opens in {delta}" + (
|
|
f" ({ts:%H:%M})" if (ts - now_local) < timedelta(days=1) else ""
|
|
)
|
|
|
|
reply.append(msg)
|
|
return reply
|