diff --git a/alembic/versions/021_add_system_events.py b/alembic/versions/021_add_system_events.py new file mode 100644 index 0000000..4e4c04a --- /dev/null +++ b/alembic/versions/021_add_system_events.py @@ -0,0 +1,55 @@ +"""add system_events table for operational warnings/errors + +Revision ID: 021 +Revises: 020 +Create Date: 2026-07-14 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "021" +down_revision: Union[str, None] = "020" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "system_events", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("severity", sa.String(length=16), nullable=False), + sa.Column("source", sa.String(length=64), nullable=False), + sa.Column("code", sa.String(length=64), nullable=False), + sa.Column("message", sa.Text(), nullable=False), + sa.Column("symbol", sa.String(length=20), nullable=True), + sa.Column("dedup_key", sa.String(length=200), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("acknowledged_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "ix_system_events_created_at", + "system_events", + ["created_at"], + ) + op.create_index( + "ix_system_events_ack_created", + "system_events", + ["acknowledged_at", "created_at"], + ) + op.create_index( + "ix_system_events_dedup_created", + "system_events", + ["dedup_key", "created_at"], + ) + + +def downgrade() -> None: + op.drop_index("ix_system_events_dedup_created", table_name="system_events") + op.drop_index("ix_system_events_ack_created", table_name="system_events") + op.drop_index("ix_system_events_created_at", table_name="system_events") + op.drop_table("system_events") diff --git a/app/models/__init__.py b/app/models/__init__.py index cad6683..b59a7c8 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -13,6 +13,7 @@ from app.models.paper_trade import PaperTrade from app.models.regime_snapshot import RegimeSnapshot from app.models.benchmark_price import BenchmarkPrice from app.models.signal_context_snapshot import SignalContextSnapshot +from app.models.system_event import SystemEvent __all__ = [ "Ticker", @@ -32,4 +33,5 @@ __all__ = [ "RegimeSnapshot", "BenchmarkPrice", "SignalContextSnapshot", + "SystemEvent", ] diff --git a/app/models/system_event.py b/app/models/system_event.py new file mode 100644 index 0000000..50477c0 --- /dev/null +++ b/app/models/system_event.py @@ -0,0 +1,37 @@ +"""Operational system events (warnings/errors) for the admin UI and top-nav badge.""" + +from datetime import datetime + +from sqlalchemy import DateTime, Index, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class SystemEvent(Base): + """Durable warning/error record (jobs, ingestion, pipelines). + + ``acknowledged_at`` is set when a user dismisses the nav badge / clears + events — history still shows in Admin → Jobs for the retention window. + """ + + __tablename__ = "system_events" + __table_args__ = ( + Index("ix_system_events_created_at", "created_at"), + Index("ix_system_events_ack_created", "acknowledged_at", "created_at"), + Index("ix_system_events_dedup_created", "dedup_key", "created_at"), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + severity: Mapped[str] = mapped_column(String(16), nullable=False) # warning | error + source: Mapped[str] = mapped_column(String(64), nullable=False) + code: Mapped[str] = mapped_column(String(64), nullable=False) + message: Mapped[str] = mapped_column(Text, nullable=False) + symbol: Mapped[str | None] = mapped_column(String(20), nullable=True) + dedup_key: Mapped[str | None] = mapped_column(String(200), nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=datetime.utcnow, nullable=False + ) + acknowledged_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) diff --git a/app/routers/admin.py b/app/routers/admin.py index 5bdd4e2..6323adc 100644 --- a/app/routers/admin.py +++ b/app/routers/admin.py @@ -6,7 +6,7 @@ All endpoints require admin role. from fastapi import APIRouter, Depends, Query from sqlalchemy.ext.asyncio import AsyncSession -from app.dependencies import get_db, require_admin +from app.dependencies import get_db, require_access, require_admin from app.models.user import User from app.schemas.admin import ( ActivationConfigUpdate, @@ -29,6 +29,7 @@ from app.schemas.common import APIEnvelope from app.services import admin_service from app.services import alert_service from app.services import sentiment_provider_service +from app.services import system_event_service from app.services import ticker_universe_service router = APIRouter(tags=["admin"]) @@ -403,3 +404,50 @@ async def toggle_job( status="success", data={"key": setting.key, "value": setting.value}, ) + + +# --------------------------------------------------------------------------- +# System events (operational warnings / errors) +# --------------------------------------------------------------------------- + +@router.get("/admin/system-events", response_model=APIEnvelope) +async def list_system_events( + days: int = Query(7, ge=1, le=30), + severity: str | None = Query(None, description="warning | error"), + unacknowledged_only: bool = Query(False), + _user: User = Depends(require_access), + db: AsyncSession = Depends(get_db), +): + """List recent system events (default last 7 days).""" + rows = await system_event_service.list_events( + db, + days=days, + severity=severity, + unacknowledged_only=unacknowledged_only, + ) + return APIEnvelope( + status="success", + data=[system_event_service.event_to_dict(r) for r in rows], + ) + + +@router.get("/admin/system-events/summary", response_model=APIEnvelope) +async def system_events_summary( + days: int = Query(7, ge=1, le=30), + _user: User = Depends(require_access), + db: AsyncSession = Depends(get_db), +): + """Badge counts for the top nav.""" + data = await system_event_service.summary(db, days=days) + return APIEnvelope(status="success", data=data) + + +@router.post("/admin/system-events/acknowledge", response_model=APIEnvelope) +async def acknowledge_system_events( + days: int = Query(7, ge=1, le=30), + _user: User = Depends(require_access), + db: AsyncSession = Depends(get_db), +): + """Dismiss unacknowledged events in the lookback window (clears the badge).""" + count = await system_event_service.acknowledge_all(db, days=days) + return APIEnvelope(status="success", data={"acknowledged": count}) diff --git a/app/routers/ingestion.py b/app/routers/ingestion.py index 1b797d2..c2c391d 100644 --- a/app/routers/ingestion.py +++ b/app/routers/ingestion.py @@ -117,12 +117,32 @@ async def fetch_symbol( result = await ingestion_service.fetch_and_ingest( db, provider, symbol_upper, start_date, end_date ) - status_map = {"complete": "ok", "partial": "ok", "no_data": "warning"} + # "stale" = provider returned nothing but our last bar is old + # (rename/delist/halt) — must not look like a successful refresh. + status_map = { + "complete": "ok", + "partial": "ok", + "no_data": "warning", + "stale": "warning", + } sources_out["ohlcv"] = { "status": status_map.get(result.status, "error"), "records": result.records_ingested, "message": result.message, + "last_date": result.last_date.isoformat() if result.last_date else None, } + if result.status in ("stale", "no_data", "error"): + from app.services.system_event_service import log_event + + await log_event( + db, + severity="warning" if result.status != "error" else "error", + source="ingestion", + code=f"ohlcv_{result.status}", + message=result.message or f"OHLCV fetch {result.status} for {symbol_upper}", + symbol=symbol_upper, + dedup_key=f"ohlcv_{result.status}:{symbol_upper}", + ) except Exception as exc: logger.error("OHLCV fetch failed for %s: %s", symbol_upper, exc) sources_out["ohlcv"] = {"status": "error", "records": 0, "message": str(exc)} diff --git a/app/scheduler.py b/app/scheduler.py index cd7621e..5259af1 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -152,6 +152,28 @@ def _log_job_error(job_name: str, ticker: str, error: Exception) -> None: ) +async def _record_system_event( + *, + severity: str, + source: str, + code: str, + message: str, + symbol: str | None = None, + dedup_key: str | None = None, +) -> None: + """Best-effort durable event for Admin → Jobs and the top-nav badge.""" + from app.services.system_event_service import log_event_standalone + + await log_event_standalone( + severity=severity, + source=source, + code=code, + message=message, + symbol=symbol, + dedup_key=dedup_key, + ) + + def _runtime_start(job_name: str, total: int | None = None, message: str | None = None) -> None: _job_runtime[job_name] = { **_idle_runtime(), @@ -206,6 +228,22 @@ def _runtime_finish( "message": message, }) _job_runtime[job_name] = runtime + # Durable event for error / rate-limit finishes (badge + Admin → Jobs panel). + if status in ("error", "rate_limited"): + severity = "error" if status == "error" else "warning" + try: + loop = asyncio.get_running_loop() + loop.create_task( + _record_system_event( + severity=severity, + source=job_name, + code=f"job_{status}", + message=message or f"Job {job_name} finished with status {status}", + dedup_key=f"job:{job_name}:{status}:{(message or '')[:80]}"[:200], + ) + ) + except RuntimeError: + pass def get_job_runtime_snapshot(job_name: str | None = None) -> dict[str, dict[str, object]] | dict[str, object]: @@ -489,6 +527,15 @@ async def collect_ohlcv(full_backfill: bool = False, job_name: str = "data_colle processed += 1 _runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol) _log_event(logging.INFO, "ticker_collected", job=job_name, ticker=symbol, status=result.status, records=result.records_ingested) + if result.status == "stale": + await _record_system_event( + severity="warning", + source=job_name, + code="ohlcv_stale", + message=result.message or f"No new OHLCV bars for {symbol}", + symbol=symbol, + dedup_key=f"ohlcv_stale:{symbol}", + ) if result.status == "partial": # Rate limited — stop and resume next run _log_event(logging.WARNING, "rate_limited", job=job_name, ticker=symbol, processed=processed) @@ -496,6 +543,14 @@ async def collect_ohlcv(full_backfill: bool = False, job_name: str = "data_colle return except Exception as exc: _log_job_error(job_name, symbol, exc) + await _record_system_event( + severity="error", + source=job_name, + code="job_ticker_error", + message=f"{type(exc).__name__}: {exc}", + symbol=symbol, + dedup_key=f"job_ticker_error:{job_name}:{symbol}:{type(exc).__name__}", + ) # Reset resume pointer on full completion _last_successful[job_name] = None diff --git a/app/services/ingestion_service.py b/app/services/ingestion_service.py index 94bd4b2..b8a5e6e 100644 --- a/app/services/ingestion_service.py +++ b/app/services/ingestion_service.py @@ -59,6 +59,19 @@ async def _get_ohlcv_bar_count(db: AsyncSession, ticker_id: int) -> int: return int(result.scalar() or 0) +async def _get_latest_ohlcv_date(db: AsyncSession, ticker_id: int) -> date | None: + result = await db.execute( + select(func.max(OHLCVRecord.date)).where(OHLCVRecord.ticker_id == ticker_id) + ) + return result.scalar_one_or_none() + + +# If the provider returns no bars but our last stored session is older than this, +# treat the run as stale (not "success / up to date"). Common causes: ticker +# rename, delisting, or multi-day halt — SATS→ECHO is the canonical example. +_STALE_OHLCV_GAP_DAYS = 5 + + async def _update_progress( db: AsyncSession, ticker_id: int, last_date: date ) -> None: @@ -145,8 +158,9 @@ async def fetch_and_ingest( # Provider returned nothing. With no history at all this almost always means # the provider doesn't cover this symbol (Alpaca = US listings only) — surface - # that instead of a misleading "success". With existing bars it just means - # there were no new bars in the requested window. + # that instead of a misleading "success". With recent history, an empty window + # usually means weekends/holidays. With a multi-day gap, the symbol is likely + # halted, delisted, or *renamed* (e.g. SATS → ECHO) and we must not claim success. if not records: existing = await _get_ohlcv_bar_count(db, ticker.id) if existing == 0: @@ -160,10 +174,24 @@ async def fetch_and_ingest( "(Alpaca serves US-listed securities only)." ), ) + latest = await _get_latest_ohlcv_date(db, ticker.id) + gap_days = (end_date - latest).days if latest is not None else None + if gap_days is not None and gap_days > _STALE_OHLCV_GAP_DAYS: + return IngestionResult( + symbol=ticker.symbol, + records_ingested=0, + last_date=latest, + status="stale", + message=( + f"No new bars since {latest.isoformat()} ({gap_days}d gap). " + "The symbol may be halted, delisted, or renamed under a new ticker — " + "check the listing and add/fetch the current symbol if it changed." + ), + ) return IngestionResult( symbol=ticker.symbol, records_ingested=0, - last_date=None, + last_date=latest, status="complete", message="Already up to date — no new bars.", ) diff --git a/app/services/system_event_service.py b/app/services/system_event_service.py new file mode 100644 index 0000000..8822aa8 --- /dev/null +++ b/app/services/system_event_service.py @@ -0,0 +1,191 @@ +"""Persist and query operational system events (warnings / errors).""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.system_event import SystemEvent + +logger = logging.getLogger(__name__) + +DEFAULT_LOOKBACK_DAYS = 7 +DEFAULT_DEDUP_HOURS = 24 +SEVERITIES = frozenset({"warning", "error"}) + + +async def log_event( + db: AsyncSession, + *, + severity: str, + source: str, + code: str, + message: str, + symbol: str | None = None, + dedup_key: str | None = None, + dedup_hours: int = DEFAULT_DEDUP_HOURS, +) -> SystemEvent | None: + """Insert a system event, optionally de-duplicating recent identical keys. + + Returns the new row, or None when a recent dedup_key already exists. + Commits the session. + """ + severity = (severity or "").strip().lower() + if severity not in SEVERITIES: + severity = "warning" + source = (source or "system")[:64] + code = (code or "unknown")[:64] + message = (message or "").strip() or code + symbol = symbol.strip().upper()[:20] if symbol else None + dedup_key = dedup_key[:200] if dedup_key else None + + if dedup_key: + cutoff = datetime.now(timezone.utc) - timedelta(hours=max(1, dedup_hours)) + existing = await db.execute( + select(SystemEvent.id) + .where( + SystemEvent.dedup_key == dedup_key, + SystemEvent.created_at >= cutoff, + ) + .limit(1) + ) + if existing.scalar_one_or_none() is not None: + return None + + row = SystemEvent( + severity=severity, + source=source, + code=code, + message=message[:4000], + symbol=symbol, + dedup_key=dedup_key, + created_at=datetime.now(timezone.utc), + ) + db.add(row) + await db.commit() + await db.refresh(row) + return row + + +async def log_event_standalone( + *, + severity: str, + source: str, + code: str, + message: str, + symbol: str | None = None, + dedup_key: str | None = None, +) -> None: + """Open a short-lived session and log an event (for scheduler / fire-and-forget).""" + try: + from app.database import async_session_factory + + async with async_session_factory() as db: + await log_event( + db, + severity=severity, + source=source, + code=code, + message=message, + symbol=symbol, + dedup_key=dedup_key, + ) + except Exception: + logger.exception("Failed to persist system event %s/%s", source, code) + + +async def list_events( + db: AsyncSession, + *, + days: int = DEFAULT_LOOKBACK_DAYS, + severity: str | None = None, + unacknowledged_only: bool = False, + limit: int = 200, +) -> list[SystemEvent]: + days = max(1, min(int(days), 30)) + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + stmt = ( + select(SystemEvent) + .where(SystemEvent.created_at >= cutoff) + .order_by(SystemEvent.created_at.desc()) + .limit(max(1, min(limit, 500))) + ) + if severity in SEVERITIES: + stmt = stmt.where(SystemEvent.severity == severity) + if unacknowledged_only: + stmt = stmt.where(SystemEvent.acknowledged_at.is_(None)) + result = await db.execute(stmt) + return list(result.scalars().all()) + + +async def summary( + db: AsyncSession, + *, + days: int = DEFAULT_LOOKBACK_DAYS, +) -> dict: + """Counts for badge + admin header.""" + days = max(1, min(int(days), 30)) + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + rows = ( + await db.execute( + select(SystemEvent.severity, SystemEvent.acknowledged_at).where( + SystemEvent.created_at >= cutoff + ) + ) + ).all() + total = len(rows) + unacked = 0 + errors = 0 + warnings = 0 + for severity, acknowledged_at in rows: + if acknowledged_at is not None: + continue + unacked += 1 + if severity == "error": + errors += 1 + elif severity == "warning": + warnings += 1 + return { + "days": days, + "total": total, + "unacknowledged": unacked, + "unacknowledged_errors": errors, + "unacknowledged_warnings": warnings, + } + + +async def acknowledge_all( + db: AsyncSession, + *, + days: int = DEFAULT_LOOKBACK_DAYS, +) -> int: + """Mark unacknowledged events in the lookback window as dismissed. Returns count.""" + days = max(1, min(int(days), 30)) + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + now = datetime.now(timezone.utc) + result = await db.execute( + update(SystemEvent) + .where( + SystemEvent.created_at >= cutoff, + SystemEvent.acknowledged_at.is_(None), + ) + .values(acknowledged_at=now) + ) + await db.commit() + return int(result.rowcount or 0) + + +def event_to_dict(row: SystemEvent) -> dict: + return { + "id": row.id, + "severity": row.severity, + "source": row.source, + "code": row.code, + "message": row.message, + "symbol": row.symbol, + "created_at": row.created_at.isoformat() if row.created_at else None, + "acknowledged_at": row.acknowledged_at.isoformat() if row.acknowledged_at else None, + } diff --git a/frontend/src/api/admin.ts b/frontend/src/api/admin.ts index 1780228..eb1ad72 100644 --- a/frontend/src/api/admin.ts +++ b/frontend/src/api/admin.ts @@ -225,6 +225,50 @@ export function triggerJob(jobName: string, options?: { target_model?: BacktestT .then((r) => r.data); } +// System events (operational warnings / errors) +export interface SystemEvent { + id: number; + severity: 'warning' | 'error' | string; + source: string; + code: string; + message: string; + symbol: string | null; + created_at: string | null; + acknowledged_at: string | null; +} + +export interface SystemEventSummary { + days: number; + total: number; + unacknowledged: number; + unacknowledged_errors: number; + unacknowledged_warnings: number; +} + +export function listSystemEvents(params?: { + days?: number; + severity?: string; + unacknowledged_only?: boolean; +}) { + return apiClient + .get('admin/system-events', { params }) + .then((r) => r.data); +} + +export function getSystemEventSummary(days = 7) { + return apiClient + .get('admin/system-events/summary', { params: { days } }) + .then((r) => r.data); +} + +export function acknowledgeSystemEvents(days = 7) { + return apiClient + .post<{ acknowledged: number }>('admin/system-events/acknowledge', null, { + params: { days }, + }) + .then((r) => r.data); +} + // Data cleanup export function cleanupData(olderThanDays: number) { return apiClient diff --git a/frontend/src/components/admin/SystemEventsPanel.tsx b/frontend/src/components/admin/SystemEventsPanel.tsx new file mode 100644 index 0000000..39c85f2 --- /dev/null +++ b/frontend/src/components/admin/SystemEventsPanel.tsx @@ -0,0 +1,96 @@ +import { useSystemEvents, useAcknowledgeSystemEvents } from '../../hooks/useAdmin'; +import { SkeletonTable } from '../ui/Skeleton'; +import { Button } from '../ui/Button'; + +function formatWhen(iso: string | null): string { + if (!iso) return '—'; + const d = new Date(iso); + const mins = Math.floor((Date.now() - d.getTime()) / 60_000); + if (mins < 1) return 'just now'; + if (mins < 60) return `${mins}m ago`; + const hrs = Math.floor(mins / 60); + if (hrs < 24) return `${hrs}h ago`; + const days = Math.floor(hrs / 24); + if (days < 7) return `${days}d ago`; + return d.toLocaleString(); +} + +function severityClass(severity: string): string { + if (severity === 'error') return 'text-red-300 border-red-400/30 bg-red-500/10'; + return 'text-amber-300 border-amber-400/30 bg-amber-500/10'; +} + +export function SystemEventsPanel() { + const { data: events, isLoading, isError } = useSystemEvents(7); + const ack = useAcknowledgeSystemEvents(); + const unacked = (events ?? []).filter((e) => !e.acknowledged_at).length; + + if (isLoading) return ; + if (isError) { + return ( +
+ Failed to load system events. +
+ ); + } + + const rows = events ?? []; + + return ( +
+
+
+

+ System alerts +

+

+ Operational warnings and errors — jobs, ingestion, renames/stale data, and other sources · last 7 days + {unacked > 0 ? ` · ${unacked} open` : ''} +

+
+ +
+ + {rows.length === 0 ? ( +

No warnings or errors in the last 7 days.

+ ) : ( +
    + {rows.map((e) => ( +
  • +
    +
    + + {e.severity} + + {e.source} + {e.symbol && ( + {e.symbol} + )} + {e.acknowledged_at && ( + dismissed + )} +
    +

    {e.message}

    +

    {e.code}

    +
    + {formatWhen(e.created_at)} +
    • + ))} +
+ )} +
+ ); +} diff --git a/frontend/src/components/layout/TopBar.tsx b/frontend/src/components/layout/TopBar.tsx index 49b63f6..dc77cac 100644 --- a/frontend/src/components/layout/TopBar.tsx +++ b/frontend/src/components/layout/TopBar.tsx @@ -1,9 +1,11 @@ +import { useState } from 'react'; import { NavLink } from 'react-router-dom'; import { useQuery } from '@tanstack/react-query'; import { useAuthStore } from '../../stores/authStore'; import { check as healthCheck } from '../../api/health'; import { getRunningJobs } from '../../api/jobs'; import { useMarketRegime } from '../../hooks/useMarketRegime'; +import { useSystemEventSummary, useAcknowledgeSystemEvents } from '../../hooks/useAdmin'; import { regimeDot, regimeHeadline } from '../../lib/regime'; import TickerSearch from './TickerSearch'; @@ -24,6 +26,7 @@ const linkClasses = (isActive: boolean) => /** Desktop command bar — the mockup's top navigation. MobileNav covers )} + + {/* System alerts badge — open issues from jobs / ingestion */} +
+ + {alertsOpen && ( + <> + +
+ + + )} + + adminApi.listSystemEvents({ days }), + refetchInterval: 30_000, + }); +} + +export function useSystemEventSummary(days = 7) { + return useQuery({ + queryKey: ['admin', 'system-events-summary', days], + queryFn: () => adminApi.getSystemEventSummary(days), + refetchInterval: 30_000, + }); +} + +export function useAcknowledgeSystemEvents() { + const qc = useQueryClient(); + const { addToast } = useToast(); + + return useMutation({ + mutationFn: (days = 7) => adminApi.acknowledgeSystemEvents(days), + onSuccess: (data) => { + qc.invalidateQueries({ queryKey: ['admin', 'system-events'] }); + qc.invalidateQueries({ queryKey: ['admin', 'system-events-summary'] }); + addToast('success', `Dismissed ${data.acknowledged} alert${data.acknowledged === 1 ? '' : 's'}`); + }, + onError: (error: Error) => { + addToast('error', error.message || 'Failed to dismiss alerts'); + }, + }); +} + export function useTriggerJob() { const qc = useQueryClient(); const { addToast } = useToast(); diff --git a/frontend/src/pages/AdminPage.tsx b/frontend/src/pages/AdminPage.tsx index ca45c7e..d49e42b 100644 --- a/frontend/src/pages/AdminPage.tsx +++ b/frontend/src/pages/AdminPage.tsx @@ -6,6 +6,7 @@ import { SentimentProviderSettings } from '../components/admin/SentimentProvider import { DataCleanup } from '../components/admin/DataCleanup'; import { JobControls } from '../components/admin/JobControls'; import { PipelineReadinessPanel } from '../components/admin/PipelineReadinessPanel'; +import { SystemEventsPanel } from '../components/admin/SystemEventsPanel'; import { RecommendationSettings } from '../components/admin/RecommendationSettings'; import { ScheduleSettings } from '../components/admin/ScheduleSettings'; import { SettingsForm } from '../components/admin/SettingsForm'; @@ -15,7 +16,7 @@ import { UserTable } from '../components/admin/UserTable'; import { PageHeader } from '../components/ui/PageHeader'; import { Tabs } from '../components/ui/Tabs'; -const tabs = ['Users', 'Tickers', 'Settings', 'Jobs', 'Cleanup'] as const; +const tabs = ['Users', 'Tickers', 'Settings', 'Jobs', 'Alerts', 'Cleanup'] as const; type Tab = (typeof tabs)[number]; export default function AdminPage() { @@ -49,6 +50,11 @@ export default function AdminPage() { )} + {activeTab === 'Alerts' && ( +
+ +
+ )} {activeTab === 'Cleanup' && } diff --git a/frontend/src/pages/TickerDetailPage.tsx b/frontend/src/pages/TickerDetailPage.tsx index 8a0f078..1bea549 100644 --- a/frontend/src/pages/TickerDetailPage.tsx +++ b/frontend/src/pages/TickerDetailPage.tsx @@ -171,8 +171,10 @@ export default function TickerDetailPage() { const dataStatus: DataStatusItem[] = useMemo(() => [ { label: 'OHLCV', + // Market age of the latest bar (session date), not DB insert time — + // created_at stays frozen when the provider returns no new sessions. available: !!ohlcv.data && ohlcv.data.length > 0, - timestamp: ohlcv.data?.[ohlcv.data.length - 1]?.created_at, + timestamp: ohlcv.data?.[ohlcv.data.length - 1]?.date, selector: ['ohlcv'] as FetchSelector, paid: true, }, diff --git a/tests/unit/test_ingestion_service.py b/tests/unit/test_ingestion_service.py index caec764..1efa1d6 100644 --- a/tests/unit/test_ingestion_service.py +++ b/tests/unit/test_ingestion_service.py @@ -60,3 +60,30 @@ async def test_empty_fetch_with_existing_history_is_up_to_date(session): assert result.status == "complete" assert result.records_ingested == 0 + + +async def test_empty_fetch_with_stale_history_reports_stale(session): + # Provider returns nothing but last stored bar is weeks old — classic + # rename/delist case (e.g. SATS after it became ECHO). Must not look like success. + await _add_ticker(session, "SATS") + today = date.today() + old = [ + OHLCVData( + ticker="SATS", + date=today - timedelta(days=20 + i), + open=100.0, + high=101.0, + low=99.0, + close=100.0, + volume=1000, + ) + for i in range(3) + ] + await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=old), "SATS") + + result = await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=[]), "SATS") + + assert result.status == "stale" + assert result.records_ingested == 0 + assert result.last_date is not None + assert "renamed" in (result.message or "").lower() or "halted" in (result.message or "").lower() diff --git a/tests/unit/test_system_event_service.py b/tests/unit/test_system_event_service.py new file mode 100644 index 0000000..909c426 --- /dev/null +++ b/tests/unit/test_system_event_service.py @@ -0,0 +1,85 @@ +"""System event log: persist, dedup, summary, acknowledge.""" + +from __future__ import annotations + +import pytest + +from app.services import system_event_service as svc +from tests.conftest import _test_session_factory # type: ignore + + +@pytest.fixture +async def session(): + async with _test_session_factory() as s: + yield s + + +async def test_log_and_list(session): + row = await svc.log_event( + session, + severity="warning", + source="ingestion", + code="ohlcv_stale", + message="No new bars since 2026-06-23", + symbol="SATS", + dedup_key="ohlcv_stale:SATS", + ) + assert row is not None + assert row.severity == "warning" + + events = await svc.list_events(session, days=7) + assert len(events) == 1 + assert events[0].symbol == "SATS" + + +async def test_dedup_suppresses_repeat(session): + first = await svc.log_event( + session, + severity="error", + source="data_collector", + code="job_error", + message="boom", + dedup_key="job:data_collector:error", + ) + second = await svc.log_event( + session, + severity="error", + source="data_collector", + code="job_error", + message="boom again", + dedup_key="job:data_collector:error", + ) + assert first is not None + assert second is None + events = await svc.list_events(session, days=7) + assert len(events) == 1 + + +async def test_summary_and_acknowledge(session): + await svc.log_event( + session, + severity="error", + source="job", + code="job_error", + message="fail", + dedup_key="a", + ) + await svc.log_event( + session, + severity="warning", + source="job", + code="job_rate_limited", + message="slow", + dedup_key="b", + ) + s = await svc.summary(session, days=7) + assert s["total"] == 2 + assert s["unacknowledged"] == 2 + assert s["unacknowledged_errors"] == 1 + assert s["unacknowledged_warnings"] == 1 + + n = await svc.acknowledge_all(session, days=7) + assert n == 2 + s2 = await svc.summary(session, days=7) + assert s2["unacknowledged"] == 0 + assert s2["total"] == 2