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>
65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
"""Hourly market ingestion: fetch every (symbol, group) defined in TOML and
|
|
insert one Quote row per fetch."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import httpx
|
|
|
|
from app.config import get_settings, load_groups
|
|
from app.db import utcnow
|
|
from app.jobs._helpers import job_lifecycle, log
|
|
from app.models import Quote
|
|
from app.services.market import fetch
|
|
|
|
|
|
async def run() -> None:
|
|
async with job_lifecycle("market_job") as (session, run):
|
|
if run.status == "skipped":
|
|
return
|
|
s = get_settings()
|
|
groups = load_groups(s.BASELINE_TOML, s.PORTFOLIO_TOML)
|
|
anchor = s.CASSANDRA_ANCHOR_DATE or None
|
|
|
|
async with httpx.AsyncClient(follow_redirects=True) as client:
|
|
tasks = [
|
|
fetch(client, sym, lab, note, anchor)
|
|
for group, items in groups.items()
|
|
for sym, lab, note in items
|
|
]
|
|
# Run in parallel but bounded — Yahoo can throttle if we hammer.
|
|
sem = asyncio.Semaphore(16)
|
|
async def bounded(t):
|
|
async with sem:
|
|
return await t
|
|
quotes = await asyncio.gather(*(bounded(t) for t in tasks))
|
|
|
|
# Re-index quotes back to their group for persistence.
|
|
items_flat = [
|
|
(group, sym)
|
|
for group, items in groups.items()
|
|
for sym, _, _ in items
|
|
]
|
|
now = utcnow()
|
|
for (group, _sym), q in zip(items_flat, quotes):
|
|
session.add(Quote(
|
|
symbol=q.symbol,
|
|
source=q.source,
|
|
label=q.label,
|
|
group_name=group,
|
|
price=q.price,
|
|
currency=q.currency,
|
|
as_of=q.as_of,
|
|
changes=q.changes or None,
|
|
# Truncate to the column's 255-char limit. Some providers
|
|
# return verbose redirect chains that blow the limit.
|
|
error=(q.error[:250] if q.error else None),
|
|
fetched_at=now,
|
|
))
|
|
await session.commit()
|
|
run.items_written = len(quotes)
|
|
log.info("market_job.done", count=len(quotes))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run())
|