read.markets/app/routers/ops.py

"""HTML-only ops endpoints — /api/markets-bar and /api/health.

These are HTMX partials consumed by the dashboard. They return HTML by
default (not JSON) and are not included in the OpenAPI schema.
"""
from __future__ import annotations

from fastapi import APIRouter, Depends, Query, Request
from fastapi.responses import HTMLResponse, JSONResponse
from sqlalchemy import desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import require_token
from app.db import get_session, utcnow
from app.models import JobRun, Quote
from app.routers.api import _age_seconds, _fmt_age
from app.schemas import HealthOut, JobStatus
from app.templates_env import templates

router = APIRouter(dependencies=[Depends(require_token)])

JOB_NAMES = ("market_job", "news_job", "ai_log_job", "rollup_job",
             "indicator_summary_job", "universe_flush_job",
             "email_digest_job")
JOB_STALE_HOURS = 2.0  # job is "warn" if its last success was >2h ago

# Market → headline index mapping for the sticky bottom bar. Symbols must
# be present in config/default.toml so market_job populates `quotes`.
_MARKET_INDEX = {
    "NYSE":  ("^GSPC",     "S&P 500"),
    "LSE":   ("^FTSE",     "FTSE 100"),
    # XETRA → Euro Stoxx 50 rather than ^GDAXI: Yahoo's DAX ticker is
    # patchy via the chart endpoint, and ^STOXX50E is already tracked in
    # config/default.toml's equity group.
    "XETRA": ("^STOXX50E", "STOXX 50"),
    "JPX":   ("^N225",     "Nikkei 225"),
    "HKEX":  ("^HSI",      "Hang Seng"),
    "SSE":   ("000300.SS", "CSI 300"),
}


def _fmt_price(p: float | None) -> str:
    if p is None:
        return "—"
    if abs(p) >= 1000:
        return f"{p:,.0f}"
    if abs(p) >= 100:
        return f"{p:,.1f}"
    return f"{p:,.2f}"


@router.get("/markets-bar", response_class=HTMLResponse, include_in_schema=False)
async def markets_bar(
    request: Request,
    session: AsyncSession = Depends(get_session),
    as_: str | None = Query(default=None, alias="as"),
):
    """The sticky bottom-bar payload: per-market open/close status with the
    market's headline index price + 1d change. Refreshed by HTMX every 60s.
    """
    from app.services.markets import all_statuses

    statuses = all_statuses()
    # Latest quote per headline-index symbol in one query.
    wanted_syms = [sym for sym, _ in _MARKET_INDEX.values()]
    sub = (
        select(Quote.symbol, func.max(Quote.fetched_at).label("mx"))
        .where(Quote.symbol.in_(wanted_syms))
        .group_by(Quote.symbol)
        .subquery()
    )
    rows = (await session.execute(
        select(Quote).join(
            sub,
            (Quote.symbol == sub.c.symbol) & (Quote.fetched_at == sub.c.mx),
        )
    )).scalars().all()
    by_sym = {q.symbol: q for q in rows}

    markets: list[dict] = []
    for st in statuses:
        sym, label = _MARKET_INDEX.get(st["code"], (None, None))
        q = by_sym.get(sym) if sym else None
        idx = None
        if q is not None and q.price is not None:
            idx = {
                "symbol": q.symbol,
                "label": label,
                "price_fmt": _fmt_price(q.price),
                "change_1d_pct": (q.changes or {}).get("1d"),
            }
        markets.append({
            "code": st["code"],
            "label": st["label"],
            "open": st["open"],
            "until_iso": st["until"].isoformat(),
            "until_hhmm": st["until"].strftime("%H:%M"),
            "index": idx,
        })

    return templates.TemplateResponse(
        request, "partials/markets_bar.html",
        {"markets": markets},
    )


@router.get("/health", response_class=HTMLResponse, include_in_schema=False)
async def health_html(
    request: Request,
    session: AsyncSession = Depends(get_session),
    as_: str | None = Query(default=None, alias="as"),
):
    """Returns an HTML fragment by default (the ops footer); ?as=json returns the
    structured object. The default is HTML because that's how the dashboard
    consumes it; CLI/curl users will pass ?as=json."""
    try:
        await session.execute(select(func.now()))
        db_ok = True
    except Exception:
        db_ok = False

    now = utcnow()
    jobs: list[dict] = []
    structured: list[JobStatus] = []
    for name in JOB_NAMES:
        row = (await session.execute(
            select(JobRun).where(JobRun.name == name)
            .order_by(desc(JobRun.started_at)).limit(1)
        )).scalar_one_or_none()
        if row is None:
            jobs.append({"name": name, "led": "idle", "age": "—",
                         "last_finished": None})
            structured.append(JobStatus(name=name))
            continue
        if row.status == "success":
            secs = _age_seconds(now, row.finished_at or row.started_at) or 0
            led = "ok" if secs < JOB_STALE_HOURS * 3600 else "warn"
        elif row.status == "skipped":
            led = "warn"
        elif row.status == "running":
            led = "warn"
        else:
            led = "err"
        jobs.append({
            "name": name, "led": led,
            "age": _fmt_age(now, row.finished_at or row.started_at),
            "last_finished": row.finished_at,
        })
        structured.append(JobStatus(
            name=name, last_started=row.started_at,
            last_finished=row.finished_at, status=row.status,
            error=row.error, items_written=row.items_written,
        ))

    if as_ == "json":
        return JSONResponse(
            HealthOut(db="ok" if db_ok else "down", jobs=structured).model_dump(mode="json")
        )
    return templates.TemplateResponse(
        request, "partials/ops_footer.html",
        {"db_ok": db_ok, "jobs": jobs},
    )