"""FastAPI visualization API for the React dashboard.""" from __future__ import annotations import math import os from datetime import date, datetime from pathlib import Path from typing import Any import duckdb import polars as pl from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from etl.chat_engine import chat DB_PATH = os.getenv("DB_PATH", "./data/equity.duckdb") TS_EXACT_DUPLICATE_LIMIT = int(os.getenv("TS_EXACT_DUPLICATE_LIMIT", "1000000")) app = FastAPI(title="Equity Workbench API", version="0.1.0") app.add_middleware( CORSMiddleware, allow_origins=os.getenv("API_CORS_ORIGINS", "http://localhost:5173,http://127.0.0.1:5173").split(","), allow_methods=["GET", "POST"], allow_headers=["*"], ) def _connect() -> duckdb.DuckDBPyConnection: return duckdb.connect(DB_PATH, read_only=True) def _frame(sql: str, params: tuple[Any, ...] = ()) -> pl.DataFrame: with _connect() as conn: rows = conn.execute(sql, params).fetchall() columns = [col[0] for col in conn.description or []] return pl.DataFrame(rows, schema=columns, orient="row") if columns else pl.DataFrame() def _scalar(sql: str, params: tuple[Any, ...] = (), default: Any = None) -> Any: try: with _connect() as conn: row = conn.execute(sql, params).fetchone() return row[0] if row else default except duckdb.Error: return default def _table_exists(name: str) -> bool: return bool( _scalar( """ SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'main' AND table_name = ? """, (name,), 0, ) ) def _column_exists(table: str, column: str) -> bool: return bool( _scalar( """ SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = 'main' AND table_name = ? AND column_name = ? """, (table, column), 0, ) ) def _table_columns(table: str) -> set[str]: if not _table_exists(table): return set() df = _frame( """ SELECT column_name FROM information_schema.columns WHERE table_schema = 'main' AND table_name = ? """, (table,), ) return set(df.get_column("column_name").to_list()) if not df.is_empty() else set() def _json_value(value: Any) -> Any: if isinstance(value, (datetime, date)): return value.isoformat() if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): return None return value def _records(df: pl.DataFrame) -> list[dict[str, Any]]: return [{key: _json_value(value) for key, value in row.items()} for row in df.to_dicts()] def _period_limit(period: str) -> int: return { "1m": 23, "3m": 66, "6m": 132, "1y": 252, "2y": 504, "5y": 1260, }.get(period, 252) TIME_SERIES_TABLES = ( {"name": "polygon_bars", "stage": "bronze", "symbol": "ticker", "time": "ts"}, {"name": "polygon_option_bars", "stage": "bronze", "symbol": "underlying", "time": "ts"}, {"name": "silver_stock_features", "stage": "silver", "symbol": "ticker", "time": "trade_date"}, {"name": "silver_option_greeks", "stage": "silver", "symbol": "underlying", "time": "trade_date"}, {"name": "silver_option_positioning", "stage": "silver", "symbol": "underlying", "time": "trade_date"}, {"name": "silver_cot_features", "stage": "silver", "symbol": "ticker", "time": "report_date"}, {"name": "gold_portfolio", "stage": "gold", "symbol": None, "time": "trade_date"}, {"name": "gold_trades", "stage": "gold", "symbol": "ticker", "time": "trade_date"}, ) def _series_profile( spec: dict[str, Any], include_coverage: bool = False, conn: duckdb.DuckDBPyConnection | None = None, ) -> dict[str, Any]: if conn is None: with _connect() as opened: return _series_profile(spec, include_coverage=include_coverage, conn=opened) def scalar(sql: str, params: tuple[Any, ...] = (), default: Any = None) -> Any: row = conn.execute(sql, params).fetchone() return row[0] if row else default table = spec["name"] columns = { row[0] for row in conn.execute( """ SELECT column_name FROM information_schema.columns WHERE table_schema = 'main' AND table_name = ? """, (table,), ).fetchall() } time_col = spec["time"] if spec["time"] in columns else None symbol_col = spec["symbol"] if spec.get("symbol") in columns else None if not columns: return {**spec, "available": False, "status": "missing", "rows": 0, "coverage": []} rows = int(scalar(f"SELECT COUNT(*) FROM {table}", default=0) or 0) detailed = rows <= TS_EXACT_DUPLICATE_LIMIT profile: dict[str, Any] = { **spec, "available": True, "status": "empty" if rows == 0 else "ready", "rows": rows, "profile_scope": "exact" if detailed else "inventory_only", "symbols": int(scalar(f"SELECT COUNT(DISTINCT {symbol_col}) FROM {table}", default=0) or 0) if symbol_col and detailed else None, "first_timestamp": _json_value(scalar(f"SELECT MIN({time_col}) FROM {table}")) if time_col and detailed else None, "last_timestamp": _json_value(scalar(f"SELECT MAX({time_col}) FROM {table}")) if time_col and detailed else None, "null_timestamps": int(scalar(f"SELECT COUNT(*) FROM {table} WHERE {time_col} IS NULL", default=0) or 0) if time_col and detailed else None, "duplicate_points": None, "duplicate_check": "not_applicable", "coverage_check": "not_applicable", "coverage": [], } key_columns = [column for column in (symbol_col, time_col) if column] if time_col and detailed: keys = ", ".join(key_columns) profile["duplicate_points"] = int( scalar( f"SELECT COALESCE(SUM(n - 1), 0) FROM (SELECT COUNT(*) AS n FROM {table} GROUP BY {keys} HAVING COUNT(*) > 1)", default=0, ) or 0 ) profile["duplicate_check"] = "exact" elif time_col: profile["duplicate_check"] = "skipped_large_table" if include_coverage and time_col and rows and detailed: coverage_rows = conn.execute( f""" SELECT CAST(DATE_TRUNC('month', TRY_CAST({time_col} AS TIMESTAMP)) AS DATE) AS period, COUNT(*) AS rows FROM {table} WHERE TRY_CAST({time_col} AS TIMESTAMP) IS NOT NULL GROUP BY period ORDER BY period DESC LIMIT 120 """ ).fetchall() coverage = pl.DataFrame(coverage_rows, schema=["period", "rows"], orient="row") if coverage_rows else pl.DataFrame() profile["coverage"] = _records(coverage.sort("period")) if not coverage.is_empty() else [] profile["coverage_check"] = "exact" elif include_coverage and time_col and rows: profile["coverage_check"] = "skipped_large_table" return profile def _series_profiles(include_coverage: bool = False) -> list[dict[str, Any]]: with _connect() as conn: return [_series_profile(spec, include_coverage=include_coverage, conn=conn) for spec in TIME_SERIES_TABLES] @app.get("/api/health") def health() -> dict[str, Any]: db_exists = os.path.exists(DB_PATH) return { "ok": db_exists, "db_path": DB_PATH, "tables": _scalar("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'main'", default=0) if db_exists else 0, } @app.get("/api/visualizations/overview") def overview() -> dict[str, Any]: tables = [ "polygon_bars", "silver_stock_features", "silver_option_greeks", "silver_option_positioning", "cot_reports", "edgar_facts", "ticker_embeddings", ] counts = [ {"table": table, "rows": int(_scalar(f"SELECT COUNT(*) FROM {table}", default=0) or 0)} for table in tables if _table_exists(table) ] ret_z = "zscore_ret_20" if _column_exists("silver_stock_features", "zscore_ret_20") else "NULL::DOUBLE" movers = _frame( f""" SELECT ticker, trade_date, close, daily_return, {ret_z} AS zscore_ret_20, zscore_20 FROM silver_stock_features QUALIFY ROW_NUMBER() OVER (PARTITION BY ticker ORDER BY trade_date DESC) = 1 ORDER BY ABS(COALESCE({ret_z}, zscore_20, 0)) DESC LIMIT 12 """ ) return { "counts": counts, "movers": _records(movers), "freshness": { "stocks": _json_value(_scalar("SELECT MAX(trade_date) FROM silver_stock_features")), "options": _json_value(_scalar("SELECT MAX(trade_date) FROM silver_option_positioning")), "polygon": _json_value(_scalar("SELECT MAX(ts) FROM polygon_bars WHERE timespan = 'day'")), }, } @app.get("/api/visualizations/tickers") def tickers() -> dict[str, Any]: df = _frame( """ SELECT ticker, COUNT(*) AS rows, MIN(trade_date) AS first_date, MAX(trade_date) AS last_date FROM silver_stock_features GROUP BY ticker ORDER BY ticker """ ) return {"tickers": _records(df)} @app.get("/api/visualizations/stock-history") def stock_history( ticker: str = Query("NVDA", min_length=1, max_length=12), period: str = Query("1y", pattern="^(1m|3m|6m|1y|2y|5y|all)$"), ) -> dict[str, Any]: symbol = ticker.upper() limit = 5000 if period == "all" else _period_limit(period) df = _frame( f""" WITH bars AS ( SELECT ticker, ts, open, high, low, close, volume, vwap FROM polygon_bars WHERE ticker = ? AND timespan = 'day' ORDER BY ts DESC LIMIT ? ) SELECT b.ticker, TRY_CAST(b.ts AS TIMESTAMPTZ)::DATE AS trade_date, b.open, b.high, b.low, b.close, b.volume, b.vwap, s.daily_return, s.zscore_20, {"s.zscore_ret_20" if _column_exists("silver_stock_features", "zscore_ret_20") else "NULL::DOUBLE"} AS zscore_ret_20, s.ma_20, s.ma_50, s.vwap_20 FROM bars b LEFT JOIN silver_stock_features s ON s.ticker = b.ticker AND s.trade_date = TRY_CAST(b.ts AS TIMESTAMPTZ)::DATE ORDER BY trade_date """, (symbol, limit), ) if df.is_empty(): raise HTTPException(status_code=404, detail=f"No daily bars found for {symbol}.") latest = df.tail(1).to_dicts()[0] first_close = df.select(pl.col("close").drop_nulls().first()).item() last_close = latest.get("close") period_return = ((last_close / first_close) - 1) if first_close and last_close else None return { "ticker": symbol, "period": period, "latest": {key: _json_value(value) for key, value in latest.items()}, "metrics": { "period_return": period_return, "high": df.select(pl.col("high").max()).item(), "low": df.select(pl.col("low").min()).item(), "avg_volume": df.select(pl.col("volume").mean()).item(), }, "series": _records(df), } @app.get("/api/visualizations/zscore-alerts") def zscore_alerts(limit: int = Query(25, ge=1, le=100)) -> dict[str, Any]: pct_change = ( "pct_change" if _column_exists("silver_stock_features", "pct_change") else "daily_return * 100" if _column_exists("silver_stock_features", "daily_return") else "NULL::DOUBLE" ) zret20 = "zscore_ret_20" if _column_exists("silver_stock_features", "zscore_ret_20") else "NULL::DOUBLE" zret50 = "zscore_ret_50" if _column_exists("silver_stock_features", "zscore_ret_50") else "NULL::DOUBLE" zret100 = "zscore_ret_100" if _column_exists("silver_stock_features", "zscore_ret_100") else "NULL::DOUBLE" sql = f""" WITH latest AS ( SELECT *, ROW_NUMBER() OVER (PARTITION BY ticker ORDER BY trade_date DESC) AS rn FROM silver_stock_features ) SELECT ticker, trade_date, close, {pct_change} AS pct_change, zscore_20, zscore_50, zscore_100, {zret20} AS zscore_ret_20, {zret50} AS zscore_ret_50, {zret100} AS zscore_ret_100, GREATEST( ABS(COALESCE(zscore_20, 0)), ABS(COALESCE(zscore_50, 0)), ABS(COALESCE(zscore_100, 0)), ABS(COALESCE({zret20}, 0)), ABS(COALESCE({zret50}, 0)), ABS(COALESCE({zret100}, 0)) ) AS max_abs_zscore, false AS any_breach FROM latest WHERE rn = 1 ORDER BY max_abs_zscore DESC LIMIT ? """ df = _frame(sql, (limit,)) return {"alerts": _records(df)} @app.get("/api/visualizations/options-positioning") def options_positioning( underlying: str = Query("NVDA", min_length=1, max_length=12), period: str = Query("1y", pattern="^(1m|3m|6m|1y|2y|all)$"), ) -> dict[str, Any]: symbol = underlying.upper() limit = 5000 if period == "all" else _period_limit(period) df = _frame( """ SELECT underlying, trade_date, total_volume, call_volume, put_volume, put_call_ratio, atm_iv, call_iv_25d, put_iv_25d, iv_skew_25d, n_contracts FROM silver_option_positioning WHERE underlying = ? ORDER BY trade_date DESC LIMIT ? """, (symbol, limit), ) if df.is_empty(): return {"underlying": symbol, "period": period, "series": []} df = df.sort("trade_date") return {"underlying": symbol, "period": period, "series": _records(df)} @app.get("/api/visualizations/backtests") def backtests() -> dict[str, Any]: if not _table_exists("gold_backtest_runs"): return {"runs": [], "portfolio": [], "metrics": []} runs = _frame( """ SELECT run_id, start_date, end_date, created_at, fold_id FROM gold_backtest_runs ORDER BY created_at DESC LIMIT 20 """ ) latest_run = runs["run_id"][0] if not runs.is_empty() else None portfolio = ( _frame( """ SELECT trade_date, nav, cash, drawdown_pct, n_positions FROM gold_portfolio WHERE run_id = ? ORDER BY trade_date """, (latest_run,), ) if latest_run and _table_exists("gold_portfolio") else pl.DataFrame() ) metrics = ( _frame("SELECT * FROM gold_metrics WHERE run_id = ?", (latest_run,)) if latest_run and _table_exists("gold_metrics") else pl.DataFrame() ) return {"runs": _records(runs), "portfolio": _records(portfolio), "metrics": _records(metrics)} @app.get("/api/research/methodology") def research_methodology() -> dict[str, Any]: return { "title": "Point-in-time market research methodology", "principles": [ { "id": "clock", "title": "Event time before ingestion time", "body": "Bars, features, signals, and fills are aligned on the market event timestamp. Ingestion and computation timestamps are retained for freshness and replay audits, but never substitute for event time in a backtest.", "controls": ["Normalize timestamps to UTC", "Declare exchange session and calendar", "Reject future-dated observations"], }, { "id": "alignment", "title": "Point-in-time alignment", "body": "A decision at time t may only use values published and observable at or before t. Lower-frequency releases are joined backward from their release timestamp, never from the period they describe.", "controls": ["Backward as-of joins", "Publication-lag fields", "No forward fill across unknown releases"], }, { "id": "features", "title": "Windows, resampling, and warmup", "body": "OHLCV resampling uses explicit session boundaries. Rolling features require a complete minimum window and remain null during warmup; signals are shifted to the next executable bar.", "controls": ["Closed-bar inputs", "Minimum observations", "Signal-to-fill lag"], }, { "id": "adjustments", "title": "Corporate actions and universe history", "body": "Price-return studies must declare split and dividend treatment. Delisted names and historical membership are retained so the tested universe is the universe known at that date.", "controls": ["Adjustment policy per run", "Point-in-time membership", "Delisting outcome handling"], }, { "id": "validation", "title": "Walk-forward validation", "body": "Model selection occurs only inside each training window. Validation and out-of-sample windows are chronological, non-overlapping, and separated by an embargo at least as long as the maximum label horizon.", "controls": ["Expanding or rolling splits", "Embargo recorded per fold", "Untouched final holdout"], }, { "id": "costs", "title": "Execution and slippage", "body": "Reported performance is net of spread, commission, and configured market impact. Costs are calculated per fill from information available at execution and are preserved beside gross and net P&L.", "controls": ["Half-spread per leg", "IBKR commission schedule", "Square-root participation impact", "Cost toggles stored per run"], }, ], "slippage_model": { "spread": "round-trip half-spread x quantity x multiplier x 2", "commission": "asset-specific IBKR schedule with minimums and caps", "market_impact": "volatility x sqrt(quantity / ADV) x notional x 2", "cost_bps": "total execution cost / entry notional x 10,000", "net_pnl": "gross_pnl - slippage_cost - commission_cost", }, "required_run_metadata": [ "data snapshot and event-time range", "universe and corporate-action policy", "feature windows and signal lag", "walk-forward folds and embargo", "slippage toggles and parameters", "code version and creation timestamp", ], } @app.get("/api/research/analytics/summary") def research_analytics_summary() -> dict[str, Any]: profiles = _series_profiles(include_coverage=True) return { "tables": profiles, "totals": { "rows": sum(item["rows"] for item in profiles), "available_tables": sum(1 for item in profiles if item["available"]), "tables_with_duplicates": sum(1 for item in profiles if (item.get("duplicate_points") or 0) > 0), "tables_with_null_time": sum(1 for item in profiles if (item.get("null_timestamps") or 0) > 0), }, } def _quality_checks() -> list[dict[str, Any]]: checks: list[dict[str, Any]] = [] for profile in _series_profiles(): spec = profile if not profile["available"]: checks.append({"table": spec["name"], "check": "table_available", "status": "not_applicable", "value": None}) continue for field, check in (("null_timestamps", "event_time_not_null"), ("duplicate_points", "unique_series_key")): value = profile.get(field) checks.append( { "table": spec["name"], "check": check, "status": "pass" if value == 0 else "fail" if value is not None else "not_applicable", "value": value, } ) return checks @app.get("/api/research/audit") def research_audit(limit: int = Query(100, ge=1, le=500)) -> dict[str, Any]: runs = ( _frame( """ SELECT id, run_type, status, rows_written, started_at, finished_at, message FROM etl_runs ORDER BY TRY_CAST(started_at AS TIMESTAMP) DESC NULLS LAST, id DESC LIMIT ? """, (limit,), ) if _table_exists("etl_runs") else pl.DataFrame() ) return {"runs": _records(runs), "checks": _quality_checks()} @app.get("/api/research/audit/summary") def research_audit_summary() -> dict[str, Any]: checks = _quality_checks() status_rows = ( _frame("SELECT status, COUNT(*) AS runs FROM etl_runs GROUP BY status ORDER BY runs DESC") if _table_exists("etl_runs") else pl.DataFrame() ) return { "etl_status": _records(status_rows), "checks": { "passed": sum(1 for item in checks if item["status"] == "pass"), "failed": sum(1 for item in checks if item["status"] == "fail"), "not_applicable": sum(1 for item in checks if item["status"] == "not_applicable"), }, } @app.get("/api/research/system") def research_system() -> dict[str, Any]: profiles = _series_profiles() stages = [] for stage in ("bronze", "silver", "gold"): members = [item for item in profiles if item["stage"] == stage] available = sum(1 for item in members if item["available"] and item["rows"] > 0) stages.append( { "stage": stage, "status": "ready" if available == len(members) else "partial" if available else "missing", "available": available, "expected": len(members), "latest_timestamp": max( (str(item["last_timestamp"]) for item in members if item.get("last_timestamp")), default=None, ), } ) return {"database": health(), "stages": stages, "tables": profiles} def _slippage_available() -> bool: required = {"run_id", "ticker", "gross_pnl", "slippage_cost", "commission_cost", "net_pnl"} return required.issubset(_table_columns("gold_trades")) @app.get("/api/research/slippage/summary") def slippage_summary() -> dict[str, Any]: if not _slippage_available(): return {"available": False, "reason": "gold_trades with cost fields is not available", "summary": None} df = _frame( """ SELECT COUNT(*) AS trades, COUNT(DISTINCT run_id) AS runs, SUM(COALESCE(gross_pnl, 0)) AS gross_pnl, SUM(COALESCE(slippage_cost, 0)) AS slippage_cost, SUM(COALESCE(commission_cost, 0)) AS commission_cost, SUM(COALESCE(slippage_cost, 0) + COALESCE(commission_cost, 0)) AS total_cost, SUM(COALESCE(net_pnl, 0)) AS net_pnl, CASE WHEN ABS(SUM(COALESCE(gross_pnl, 0))) > 0 THEN SUM(COALESCE(slippage_cost, 0) + COALESCE(commission_cost, 0)) / ABS(SUM(COALESCE(gross_pnl, 0))) END AS cost_drag FROM gold_trades """ ) return {"available": True, "summary": _records(df)[0] if not df.is_empty() else None} @app.get("/api/research/slippage/by-ticker") def slippage_by_ticker(run_id: str | None = None) -> dict[str, Any]: if not _slippage_available(): return {"available": False, "reason": "gold_trades with cost fields is not available", "tickers": []} where = "WHERE run_id = ?" if run_id else "" params: tuple[Any, ...] = (run_id,) if run_id else () df = _frame( f""" SELECT ticker, COUNT(*) AS trades, SUM(COALESCE(gross_pnl, 0)) AS gross_pnl, SUM(COALESCE(slippage_cost, 0)) AS slippage_cost, SUM(COALESCE(commission_cost, 0)) AS commission_cost, SUM(COALESCE(slippage_cost, 0) + COALESCE(commission_cost, 0)) AS total_cost, SUM(COALESCE(net_pnl, 0)) AS net_pnl FROM gold_trades {where} GROUP BY ticker ORDER BY total_cost DESC, ticker """, params, ) return {"available": True, "tickers": _records(df)} @app.get("/api/backtests/{run_id}/costs") def backtest_costs(run_id: str) -> dict[str, Any]: if not _slippage_available(): return {"available": False, "run_id": run_id, "reason": "gold_trades with cost fields is not available", "trades": []} columns = _table_columns("gold_trades") order_col = "trade_date" if "trade_date" in columns else "ticker" selected = ["ticker", "gross_pnl", "slippage_cost", "commission_cost", "net_pnl"] for optional in ("trade_date", "asset_type", "quantity", "price", "cost_bps"): if optional in columns: selected.append(optional) df = _frame( f"SELECT {', '.join(selected)} FROM gold_trades WHERE run_id = ? ORDER BY {order_col}", (run_id,), ) totals = slippage_by_ticker(run_id=run_id) return {"available": True, "run_id": run_id, "by_ticker": totals["tickers"], "trades": _records(df)} @app.post("/api/chat") def chat_endpoint(payload: dict[str, Any]) -> dict[str, Any]: message = str(payload.get("message", "")).strip() if not message: raise HTTPException(status_code=400, detail="message is required") result = chat(message, history=payload.get("history") or [], max_rows=int(payload.get("max_rows", 100))) data = result.get("data") if hasattr(data, "to_dict"): data = data.to_dict(orient="records") return { "type": result.get("type"), "response": result.get("answer"), "sql": result.get("sql"), "data": data, } # The Docker Space builds the Vite app into frontend/dist. Keep this route last # so every /api endpoint wins before the single-page-app fallback. FRONTEND_DIST = Path(__file__).resolve().parent.parent / "frontend" / "dist" if FRONTEND_DIST.is_dir(): @app.get("/{full_path:path}", include_in_schema=False) def serve_frontend(full_path: str) -> FileResponse: requested = (FRONTEND_DIST / full_path).resolve() if requested.is_file() and FRONTEND_DIST.resolve() in requested.parents: return FileResponse(requested) return FileResponse(FRONTEND_DIST / "index.html")