Giorgio Gilestro
1edf9cad41
Three new data sources hooked into the existing SOURCES registry. All
open APIs, no keys:
- EUROSTAT: prefix EUROSTAT:dataset?dim=val&... — current EU bond
yields (Bund/OAT/BTP/EZ) and Eurozone economic indicators that
FRED's OECD-mirror series stopped updating in 2022-2023.
- ONS: prefix ONS:topic/cdid/dataset — current UK CPI, unemployment,
GDP, industrial production. Replaces the 5+ month-stale FRED
LRHUTTTTGBM156S mirror.
New indicator groups in default.toml feed the strategic/fundamental
lens we converged on: valuation (CAPE/Buffett anchors), bubble_watch
(SKEW/VVIX/RSP vs SPY/HYG vs TLT/IPO/crypto), economy (multi-region,
ALL current-or-stale-flagged), bonds (UK/EU/US/JPN sovereign yields).
Indicator panel now opens with an AI "read" interpretation per group
(generated hourly at :07 UTC alongside an aggregate cross-group read
shown in the dashboard header). The aggregate is grounded by a markets
strip — NYSE/LSE/Frankfurt/Tokyo/HK/Shanghai with open/closed LEDs and
next-open countdown, computed locally from each exchange's tz.
Other UX bits: indicator-row tooltips populated from TOML notes;
rows whose last observation is >90 days old get a 'stale' chip;
ghost symbols (in DB but no longer in TOML) filtered out of the
panel; Eurostat/ONS symbols display as short codes rather than the
full API path.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""Market-open/close status for the dashboard header. Pure computation —
|
|
no API needed; the schedules are known constants. Holidays are NOT modelled
|
|
(would require a region-specific calendar); a closed Monday will still show
|
|
"open" if the time-of-day fits. Good enough for the strategic dashboard.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, time, timedelta, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Market:
|
|
code: str
|
|
name: str
|
|
tz: str # IANA zone (handles DST automatically)
|
|
open: time # local time
|
|
close: time # local time
|
|
|
|
|
|
# Mon=0 .. Sun=6. Markets observe Mon-Fri unless overridden.
|
|
_WORKWEEK = {0, 1, 2, 3, 4}
|
|
|
|
|
|
MARKETS: list[Market] = [
|
|
Market("NYSE", "NYSE", "America/New_York", time(9, 30), time(16, 0)),
|
|
Market("LSE", "LSE", "Europe/London", time(8, 0), time(16, 30)),
|
|
Market("XETRA", "Frankfurt","Europe/Berlin", time(9, 0), time(17, 30)),
|
|
Market("JPX", "Tokyo", "Asia/Tokyo", time(9, 0), time(15, 0)),
|
|
Market("HKEX", "Hong Kong","Asia/Hong_Kong", time(9, 30), time(16, 0)),
|
|
Market("SSE", "Shanghai", "Asia/Shanghai", time(9, 30), time(15, 0)),
|
|
]
|
|
|
|
|
|
def _next_open_at(m: Market, now_utc: datetime) -> datetime:
|
|
"""Earliest future open datetime (UTC) for this market, scanning ahead
|
|
up to 7 days for the next weekday."""
|
|
tz = ZoneInfo(m.tz)
|
|
local = now_utc.astimezone(tz)
|
|
candidate_date = local.date()
|
|
for _ in range(8): # today + 7 days
|
|
weekday = candidate_date.weekday()
|
|
if weekday in _WORKWEEK:
|
|
local_open = datetime.combine(candidate_date, m.open, tzinfo=tz)
|
|
if local_open > local:
|
|
return local_open.astimezone(timezone.utc)
|
|
candidate_date = candidate_date + timedelta(days=1)
|
|
return now_utc + timedelta(days=7) # fallback (shouldn't happen)
|
|
|
|
|
|
def _close_at(m: Market, now_utc: datetime) -> datetime:
|
|
"""Today's close in UTC (assumes we've already established it's open)."""
|
|
tz = ZoneInfo(m.tz)
|
|
local = now_utc.astimezone(tz)
|
|
return datetime.combine(local.date(), m.close, tzinfo=tz).astimezone(timezone.utc)
|
|
|
|
|
|
def status_for(m: Market, now_utc: datetime) -> dict:
|
|
tz = ZoneInfo(m.tz)
|
|
local = now_utc.astimezone(tz)
|
|
is_workday = local.weekday() in _WORKWEEK
|
|
in_session = is_workday and m.open <= local.time() < m.close
|
|
if in_session:
|
|
return {
|
|
"code": m.code,
|
|
"name": m.name,
|
|
"open": True,
|
|
"until": _close_at(m, now_utc),
|
|
"label": "open",
|
|
}
|
|
return {
|
|
"code": m.code,
|
|
"name": m.name,
|
|
"open": False,
|
|
"until": _next_open_at(m, now_utc),
|
|
"label": "closed",
|
|
}
|
|
|
|
|
|
def all_statuses(now_utc: datetime | None = None) -> list[dict]:
|
|
if now_utc is None:
|
|
now_utc = datetime.now(timezone.utc)
|
|
return [status_for(m, now_utc) for m in MARKETS]
|