api/app.py

"""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")