Solidify validation and add executive + freshness dashboard views

Author: Dennis Donaghy

docs/graphql_query_pack.graphql

@@ -60,3 +60,20 @@ query SelectionSetV1Dashboard {
     avg_theta
   }
 }
+
+query SelectionSetV1Executive {
+  analytics_selection_set_v1_executive(order_by: [{block: asc}, {symbol: asc}]) {
+    block
+    symbol
+    source
+    last_ts
+    last_close
+    last_volume
+  }
+  analytics_ingestion_freshness_sla(order_by: [{source: asc}]) {
+    source
+    last_run_at
+    minutes_since_last_run
+    sla_status
+  }
+}

infra/sql/init/010_ops_freshness_and_executive.sql

@@ -0,0 +1,30 @@
+create schema if not exists analytics;
+
+create or replace view analytics.ingestion_freshness_sla as
+select
+  source,
+  max(started_at) as last_run_at,
+  extract(epoch from (now() - max(started_at))) / 60.0 as minutes_since_last_run,
+  case
+    when extract(epoch from (now() - max(started_at))) / 60.0 > 90 then 'warn'
+    else 'ok'
+  end as sla_status
+from ingestion.runs
+group by source;
+
+create or replace view analytics.selection_set_v1_executive as
+select 'rates'::text as block, symbol, source, last_ts, last_close, last_volume
+from analytics.selection_set_v1_latest_bars
+where symbol in ('SOFR','MORTGAGE30US','DGS10','DGS2')
+union all
+select 'volatility'::text, symbol, source, last_ts, last_close, last_volume
+from analytics.selection_set_v1_latest_bars
+where symbol in ('VIXCLS','VXX','UVXY','SVXY')
+union all
+select 'equities_etf'::text, symbol, source, last_ts, last_close, last_volume
+from analytics.selection_set_v1_latest_bars
+where symbol in ('SPY','QQQ','IWM','DIA','VTI','TLT','IEF','XLF','XLK')
+union all
+select 'crypto'::text, symbol, source, last_ts, last_close, last_volume
+from analytics.selection_set_v1_latest_bars
+where symbol in ('BTCUSD','ETHUSD','SOLUSD');

services/ingestion/pull_crypto_coingecko.py

@@ -4,6 +4,7 @@
 from datetime import datetime, timezone
 from urllib import parse, request
 import psycopg
+from validation_models import MarketBarIn
 
 DB_URL = os.getenv("DATABASE_URL", "postgresql://quant:quant_dev_change_me@127.0.0.1:5432/quant")
 COINS = [c.strip().lower() for c in os.getenv("CRYPTO_COINS", "bitcoin,ethereum,solana").split(",") if c.strip()]
@@ -47,14 +48,18 @@ def main():
                     """,
                     (symbol, ts, price, mcap, vol, json.dumps(r)),
                 )
+                try:
+                    v = MarketBarIn(symbol=symbol, ts=ts, open=price, high=price, low=price, close=price, volume=vol or 0, source='coingecko')
+                except Exception:
+                    continue
                 cur.execute(
                     """
                     insert into market.bars(symbol, ts, open, high, low, close, volume, source)
                     values (%s,%s,%s,%s,%s,%s,%s,'coingecko')
                     on conflict (symbol, ts) do update
                     set close=excluded.close, volume=excluded.volume, source=excluded.source
                     """,
-                    (symbol, ts, price, price, price, price, vol or 0),
+                    (v.symbol, v.ts, v.open, v.high, v.low, v.close, v.volume),
                 )
                 written += 1
             cur.execute("update ingestion.runs set status='success', completed_at=now(), rows_written=%s where id=%s", (written, run_id))

services/ingestion/pull_equities_bars_stooq.py

@@ -5,6 +5,7 @@
 from datetime import datetime, timezone
 from urllib import request
 import psycopg
+from validation_models import MarketBarIn
 
 DB_URL = os.getenv("DATABASE_URL", "postgresql://quant:quant_dev_change_me@127.0.0.1:5432/quant")
 SYMBOLS = [s.strip().upper() for s in os.getenv("EQUITY_SYMBOLS", "SPY,QQQ,AAPL,MSFT").split(",") if s.strip()]
@@ -44,14 +45,18 @@ def main():
                 row = fetch_stooq_daily(sym)
                 if not row:
                     continue
+                try:
+                    v = MarketBarIn(**{**row, "source": "stooq"})
+                except Exception:
+                    continue
                 cur.execute(
                     """
                     insert into market.bars(symbol, ts, open, high, low, close, volume, source)
                     values (%s,%s,%s,%s,%s,%s,%s,'stooq')
                     on conflict (symbol, ts) do update
                     set open=excluded.open, high=excluded.high, low=excluded.low, close=excluded.close, volume=excluded.volume, source=excluded.source
                     """,
-                    (row["symbol"], row["ts"], row["open"], row["high"], row["low"], row["close"], row["volume"]),
+                    (v.symbol, v.ts, v.open, v.high, v.low, v.close, v.volume),
                 )
                 written += 1
 

services/ingestion/pull_macro_fred.py

@@ -5,6 +5,7 @@
 from datetime import datetime, timezone
 from urllib import request
 import psycopg
+from validation_models import MarketBarIn
 
 DB_URL = os.getenv("DATABASE_URL", "postgresql://quant:quant_dev_change_me@127.0.0.1:5432/quant")
 FRED_SERIES = [s.strip().upper() for s in os.getenv("FRED_SERIES", "SOFR,MORTGAGE30US,DGS10,DGS2,VIXCLS").split(",") if s.strip()]
@@ -34,14 +35,18 @@ def main():
                 ts, val = fetch_latest(s)
                 if ts is None:
                     continue
+                try:
+                    v = MarketBarIn(symbol=s, ts=ts, open=val, high=val, low=val, close=val, volume=0, source='fred')
+                except Exception:
+                    continue
                 cur.execute(
                     """
                     insert into market.bars(symbol, ts, open, high, low, close, volume, source)
                     values (%s,%s,%s,%s,%s,%s,0,'fred')
                     on conflict (symbol, ts) do update
                     set close=excluded.close, source=excluded.source
                     """,
-                    (s, ts, val, val, val, val),
+                    (v.symbol, v.ts, v.open, v.high, v.low, v.close),
                 )
                 written += 1
             cur.execute("update ingestion.runs set status='success', completed_at=now(), rows_written=%s where id=%s", (written, run_id))

services/ingestion/pull_tradier_greeks.py

@@ -4,6 +4,7 @@
 from datetime import datetime, timezone
 from urllib import parse, request
 import psycopg
+from validation_models import GreeksIn
 
 DB_URL = os.getenv("DATABASE_URL", "postgresql://quant:quant_dev_change_me@127.0.0.1:5432/quant")
 TRADIER_BASE = os.getenv("TRADIER_SANDBOX_BASE_URL") or os.getenv("TRADIER_PAPER_BASE_URL") or "https://sandbox.tradier.com/v1"
@@ -55,6 +56,22 @@ def main():
 
                 for o in options[:150]:
                     greeks = o.get("greeks") or {}
+                    try:
+                        g = GreeksIn(
+                            underlying=sym,
+                            option_symbol=o.get("symbol") or "",
+                            ts=now,
+                            price=o.get("last") or o.get("bid") or o.get("ask") or 0,
+                            iv=greeks.get("mid_iv") or greeks.get("smv_vol"),
+                            delta=greeks.get("delta"),
+                            gamma=greeks.get("gamma"),
+                            vega=greeks.get("vega"),
+                            theta=greeks.get("theta"),
+                            rho=greeks.get("rho"),
+                            model='tradier'
+                        )
+                    except Exception:
+                        continue
                     cur.execute(
                         """
                         insert into options.greeks_snapshot(
@@ -64,9 +81,9 @@ def main():
                         set price=excluded.price, iv=excluded.iv, delta=excluded.delta, gamma=excluded.gamma,
                             vega=excluded.vega, theta=excluded.theta, rho=excluded.rho, model=excluded.model
                         """,
-                        (sym, o.get("symbol"), now, o.get("last") or o.get("bid") or o.get("ask"),
-                         greeks.get("mid_iv") or greeks.get("smv_vol"), greeks.get("delta"), greeks.get("gamma"),
-                         greeks.get("vega"), greeks.get("theta"), greeks.get("rho")),
+                        (g.underlying, g.option_symbol, g.ts, g.price,
+                         g.iv, g.delta, g.gamma,
+                         g.vega, g.theta, g.rho),
                     )
                     written += 1
 

services/ingestion/validation_models.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+from decimal import Decimal
+from pydantic import BaseModel, Field
+
+
+class MarketBarIn(BaseModel):
+    symbol: str = Field(min_length=1, max_length=32)
+    ts: datetime
+    open: Decimal
+    high: Decimal
+    low: Decimal
+    close: Decimal
+    volume: Decimal = Decimal("0")
+    source: str = Field(min_length=1, max_length=32)
+
+
+class GreeksIn(BaseModel):
+    underlying: str = Field(min_length=1, max_length=32)
+    option_symbol: str = Field(min_length=1, max_length=96)
+    ts: datetime
+    price: Decimal
+    iv: Decimal | None = None
+    delta: Decimal | None = None
+    gamma: Decimal | None = None
+    vega: Decimal | None = None
+    theta: Decimal | None = None
+    rho: Decimal | None = None
+    model: str = Field(default="unspecified", max_length=32)