Spaces:

deveshc23
/

drift-quant

Running on Zero

drift-quant / src /drift_quant /api /app.py
deveshc23
feat: Add Free Gradio Space support, Git LFS tracking for data artifacts, and Quant Allocator
3e33e2e 20 days ago
30 kB
"""FastAPI application serving persisted research artifacts and quantitative analytics."""

from __future__ import annotations

import io
import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any

from fastapi import FastAPI, HTTPException, Query, Response
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from drift_quant.api.repository import SECTOR_MAP, ArtifactNotFoundError, ArtifactRepository
from drift_quant.backtesting import (
    AtrTrendStrategy,
    BacktestConfig,
    BuyAndHoldStrategy,
    EmaMomentumStrategy,
    SectorCrossMomentumStrategy,
    SectorPairsReversionStrategy,
    VwapProxyReversionStrategy,
    run_backtest,
)
from drift_quant.portfolio import PortfolioAllocator

# Global in-memory job registry for research simulation jobs
_JOBS: dict[str, dict[str, Any]] = {}


class HealthResponse(BaseModel):
    status: str
    artifacts_directory: str
    completed_experiments: int
    dataset_audit_available: bool
    sqlite_market_data_available: bool
    cached_sqlite_bars: int
    currency: str


class DatasetSummaryResponse(BaseModel):
    archive_bytes: int
    archive_sha256: str
    total_rows: int
    total_symbols: int
    first_timestamp: str
    last_timestamp: str
    zero_volume_rows: int
    invalid_ohlc_rows: int
    intraday_unexpected_gaps: int
    duplicate_timestamps: int
    timestamp_regressions: int
    frequency: str
    currency: str


class SymbolCatalogItem(BaseModel):
    symbol: str
    sector: str
    rows: int
    first_timestamp: str
    last_timestamp: str
    zero_volume_rows: int
    invalid_ohlc_rows: int
    intraday_unexpected_gaps: int
    is_cached: bool


class MarketBarItem(BaseModel):
    timestamp: str
    open: float
    high: float
    low: float
    close: float
    volume: float
    bars_aggregated: int | None = None


class CoverageTimelineResponse(BaseModel):
    yearly: list[dict[str, Any]]
    weekdays: dict[str, int]
    total_rows: int


class StrategyCatalogItem(BaseModel):
    id: str
    name: str
    version: str
    title: str
    category: str
    description: str
    required_indicators: list[str]
    parameters_schema: dict[str, Any]
    supported_instruments: str
    warmup_bars: int
    known_limitations: str


class BacktestRunRequest(BaseModel):
    strategy_name: str
    universe: list[str] = Field(min_length=1, max_length=50)
    start_date: str | None = None
    end_date: str | None = None
    initial_capital: float = Field(default=1_000_000.0, gt=0)
    max_gross_exposure: float = Field(default=0.95, gt=0, le=1.0)
    transaction_cost_bps: float = Field(default=10.0, ge=0)
    slippage_bps: float = Field(default=5.0, ge=0)
    strategy_params: dict[str, Any] = Field(default_factory=dict)


class JobStatusResponse(BaseModel):
    job_id: str
    status: str  # queued, running, completed, failed, cancelled
    progress: float
    message: str
    experiment_id: str | None = None
    error: str | None = None


class ExperimentSummary(BaseModel):
    experiment_id: str
    strategy_name: str
    strategy_version: str
    dataset_id: str
    period_start: str
    period_end_exclusive: str
    metrics: dict[str, float | int | None]
    currency: str = "INR"


class ExperimentListResponse(BaseModel):
    total: int
    offset: int
    limit: int
    items: list[ExperimentSummary]


class SeriesPoint(BaseModel):
    timestamp: str
    equity: float
    gross_exposure: float


class DrawdownPoint(BaseModel):
    timestamp: str
    equity: float
    peak: float
    drawdown: float


class TradeResponse(BaseModel):
    symbol: str
    entry_time: str
    exit_time: str
    quantity: int
    entry_price: float
    exit_price: float
    gross_pnl: float
    costs: float
    net_pnl: float
    exit_reason: str


class TradePageResponse(BaseModel):
    total: int
    offset: int
    limit: int
    items: list[TradeResponse]


class AttributionItem(BaseModel):
    symbol: str
    sector: str
    trade_count: int
    win_rate: float
    gross_pnl: float
    costs: float
    net_pnl: float
    return_contribution: float
    profit_factor: float | None = None


class MonthlyReturnsResponse(BaseModel):
    years: list[str]
    matrix: dict[str, dict[str, float | None]]


class ExperimentDetailResponse(BaseModel):
    experiment_id: str
    strategy_name: str
    strategy_version: str
    dataset_id: str
    period_start: str | None = None
    period_end_exclusive: str | None = None
    config: dict[str, Any]
    metrics: dict[str, float | int | None]
    currency: str = "INR"


def _repository() -> ArtifactRepository:
    configured = os.environ.get("DRIFT_ARTIFACTS_DIR", "artifacts")
    return ArtifactRepository(Path(configured))


def _experiment_or_404(repository: ArtifactRepository, experiment_id: str) -> dict[str, Any]:
    try:
        return repository.experiment(experiment_id)
    except ArtifactNotFoundError as exc:
        raise HTTPException(status_code=404, detail="Experiment not found.") from exc


def _downsample(points: list[dict[str, Any]], maximum: int) -> list[dict[str, Any]]:
    if len(points) <= maximum:
        return points
    step = (len(points) - 1) / (maximum - 1)
    indices = {round(index * step) for index in range(maximum)}
    return [point for index, point in enumerate(points) if index in indices]


app = FastAPI(
    title="Drift Quant NSE Research API",
    version="0.2.0",
    description=(
        "Quantitative research and simulation API for NSE 15-minute data with"
        " real-time backtest execution."
    ),
)


@app.get("/health", response_model=HealthResponse, tags=["system"])
def health() -> dict[str, object]:
    return _repository().health()


@app.get("/api/v1/dataset/audit", tags=["dataset"])
def dataset_audit() -> dict[str, Any]:
    try:
        return _repository().dataset_audit()
    except ArtifactNotFoundError as exc:
        raise HTTPException(status_code=404, detail="Dataset audit not found.") from exc


@app.get("/api/v1/dataset/summary", response_model=DatasetSummaryResponse, tags=["dataset"])
def dataset_summary() -> DatasetSummaryResponse:
    repo = _repository()
    return DatasetSummaryResponse.model_validate(repo.dataset_summary())


@app.get("/api/v1/dataset/symbols", response_model=list[SymbolCatalogItem], tags=["dataset"])
def dataset_symbols(
    search: str | None = Query(default=None),
    sector: str | None = Query(default=None),
) -> list[SymbolCatalogItem]:
    repo = _repository()
    items = repo.symbols_catalog(search=search, sector=sector)
    return [SymbolCatalogItem.model_validate(item) for item in items]


@app.get(
    "/api/v1/dataset/coverage-timeline", response_model=CoverageTimelineResponse, tags=["dataset"]
)
def coverage_timeline() -> CoverageTimelineResponse:
    repo = _repository()
    return CoverageTimelineResponse.model_validate(repo.coverage_timeline())


@app.get("/api/v1/market/bars", response_model=list[MarketBarItem], tags=["market"])
def market_bars(
    symbol: str = Query(..., min_length=1, max_length=30),
    start: str | None = Query(default=None),
    end: str | None = Query(default=None),
    resolution: str = Query(default="15m", pattern="^(15m|1h|1d)$"),
    limit: int = Query(default=1500, ge=10, le=5000),
) -> list[MarketBarItem]:
    repo = _repository()
    try:
        bars = repo.market_bars(symbol, start=start, end=end, resolution=resolution, limit=limit)
        return [MarketBarItem.model_validate(b) for b in bars]
    except ArtifactNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc


@app.get("/api/v1/strategies", response_model=list[StrategyCatalogItem], tags=["strategies"])
def list_strategies() -> list[StrategyCatalogItem]:
    catalog = [
        StrategyCatalogItem(
            id="buy_and_hold_baseline",
            name="buy_and_hold_baseline",
            version="1.0.0",
            title="Buy & Hold Baseline",
            category="Benchmark",
            description=(
                "Enters long on each configured instrument after its first completed bar and"
                " maintains the position until end of simulation."
            ),
            required_indicators=[],
            parameters_schema={},
            supported_instruments="All NSE equity equities with valid OHLCV",
            warmup_bars=1,
            known_limitations="Exposed to full market drawdowns; no risk management.",
        ),
        StrategyCatalogItem(
            id="sector_pairs_reversion",
            name="sector_pairs_reversion",
            version="1.0.0",
            title="Sector Statistical Arbitrage / Pairs Trading",
            category="Statistical Arbitrage",
            description=(
                "Monitors cointegrated intra-sector equity pairs (e.g. Banking: ICICI/HDFC,"
                " SBI/Axis; IT: TCS/Infosys). Computes rolling log-price spread Z-scores and"
                " allocates long to underpriced peer. Uses multi-day holding to cut fee drag."
            ),
            required_indicators=["Spread Z-Score", "Rolling Mean", "Standard Deviation"],
            parameters_schema={
                "lookback_period": {"type": "integer", "default": 160, "min": 20, "max": 500},
                "entry_z": {"type": "number", "default": 2.0, "min": 1.0, "max": 4.0},
                "exit_z": {"type": "number", "default": 0.4, "min": 0.1, "max": 1.0},
                "stop_z": {"type": "number", "default": 3.6, "min": 2.0, "max": 5.0},
                "min_holding_bars": {"type": "integer", "default": 40, "min": 4, "max": 200},
            },
            supported_instruments="Paired equities within Banking, IT, Energy, FMCG, Auto",
            warmup_bars=160,
            known_limitations=(
                "Pair correlation may break down during market regime changes or corporate actions."
            ),
        ),
        StrategyCatalogItem(
            id="sector_cross_momentum",
            name="sector_cross_momentum",
            version="1.0.0",
            title="Sector Cross-Sectional Momentum",
            category="Cross-Sectional Momentum",
            description=(
                "Evaluates multi-day price momentum across sector leaders. Allocates to stocks"
                " exhibiting strong relative momentum confirmed by an overarching trend regime."
            ),
            required_indicators=["Multi-day Return Momentum", "Trend Filter SMA"],
            parameters_schema={
                "momentum_period": {"type": "integer", "default": 120, "min": 20, "max": 300},
                "trend_sma_period": {"type": "integer", "default": 200, "min": 50, "max": 500},
                "min_holding_bars": {"type": "integer", "default": 40, "min": 8, "max": 200},
            },
            supported_instruments="Liquid large-cap sector representatives",
            warmup_bars=200,
            known_limitations=(
                "Requires sustained trending markets; subject to whipsaws in sideways regimes."
            ),
        ),
        StrategyCatalogItem(
            id="intraday_ema_momentum",
            name="intraday_ema_momentum",
            version="1.0.0",
            title="Intraday Fast/Slow EMA Crossover",
            category="Trend Following",
            description=(
                "Classic exponential moving average crossover (fast 8, slow 26 bars)."
                " Decided on bar close and filled at next bar open."
            ),
            required_indicators=["Fast EMA", "Slow EMA"],
            parameters_schema={
                "fast_period": {"type": "integer", "default": 8, "min": 2, "max": 50},
                "slow_period": {"type": "integer", "default": 26, "min": 10, "max": 200},
            },
            supported_instruments="All liquid equities",
            warmup_bars=26,
            known_limitations="High turnover leads to severe fee drag on 15m bars.",
        ),
        StrategyCatalogItem(
            id="vwap_proxy_reversion",
            name="vwap_proxy_reversion",
            version="1.0.0",
            title="Intraday VWAP Proxy Reversion",
            category="Mean Reversion",
            description=(
                "Intraday session typical-price VWAP proxy. Enters when close dips below VWAP"
                " band and exits on convergence."
            ),
            required_indicators=["Session VWAP Proxy"],
            parameters_schema={
                "entry_deviation": {
                    "type": "number",
                    "default": 0.003,
                    "min": 0.001,
                    "max": 0.05,
                },
            },
            supported_instruments="High volume equities",
            warmup_bars=5,
            known_limitations="Overtrades counter-trend in persistent strong directional trends.",
        ),
        StrategyCatalogItem(
            id="atr_trend",
            name="atr_trend",
            version="1.0.0",
            title="ATR Breakout & Trailing Stop",
            category="Volatility Breakout",
            description=(
                "Breakout beyond recent highs confirmed by close, with an ATR-multiple stop."
            ),
            required_indicators=["ATR", "Donchian High", "Trailing Stop"],
            parameters_schema={
                "atr_period": {"type": "integer", "default": 14, "min": 5, "max": 50},
                "breakout_period": {"type": "integer", "default": 20, "min": 5, "max": 100},
                "stop_atr_multiple": {"type": "number", "default": 2.5, "min": 1.0, "max": 6.0},
            },
            supported_instruments="All equities",
            warmup_bars=20,
            known_limitations="False breakouts incur frequent stop-outs in range-bound markets.",
        ),
    ]
    return catalog


def _resolve_strategy(strat_name: str, params: dict[str, Any]) -> Any:
    strat_clean = strat_name.strip().lower()
    if strat_clean == "buy_and_hold_baseline":
        return BuyAndHoldStrategy()
    elif strat_clean == "sector_pairs_reversion":
        return SectorPairsReversionStrategy(
            lookback_period=int(params.get("lookback_period", 160)),
            entry_z=float(params.get("entry_z", 2.0)),
            exit_z=float(params.get("exit_z", 0.4)),
            stop_z=float(params.get("stop_z", 3.6)),
            min_holding_bars=int(params.get("min_holding_bars", 40)),
        )
    elif strat_clean == "sector_cross_momentum":
        return SectorCrossMomentumStrategy(
            momentum_period=int(params.get("momentum_period", 120)),
            trend_sma_period=int(params.get("trend_sma_period", 200)),
            min_holding_bars=int(params.get("min_holding_bars", 40)),
        )
    elif strat_clean == "intraday_ema_momentum":
        return EmaMomentumStrategy(
            fast_period=int(params.get("fast_period", 8)),
            slow_period=int(params.get("slow_period", 26)),
        )
    elif strat_clean == "vwap_proxy_reversion":
        return VwapProxyReversionStrategy(
            entry_deviation=float(params.get("entry_deviation", 0.003))
        )
    elif strat_clean == "atr_trend":
        return AtrTrendStrategy(
            atr_period=int(params.get("atr_period", 14)),
            breakout_period=int(params.get("breakout_period", 20)),
            stop_atr_multiple=float(params.get("stop_atr_multiple", 2.5)),
        )
    else:
        raise HTTPException(
            status_code=400,
            detail=f"Unknown strategy '{strat_name}'. Must be one of registered strategies.",
        )


@app.post("/api/v1/backtest/run", response_model=ExperimentDetailResponse, tags=["backtesting"])
def run_backtest_endpoint(req: BacktestRunRequest) -> ExperimentDetailResponse:
    repo = _repository()
    strat_inst = _resolve_strategy(req.strategy_name, req.strategy_params or {})

    cleaned_symbols = tuple(sorted({s.strip().upper() for s in req.universe if s.strip()}))
    if not cleaned_symbols:
        raise HTTPException(
            status_code=400, detail="Universe must contain at least one valid symbol."
        )

    try:
        dataset = repo.get_backtest_dataset(cleaned_symbols, req.start_date, req.end_date)
    except Exception as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc

    config = BacktestConfig(
        initial_capital=req.initial_capital,
        max_gross_exposure=req.max_gross_exposure,
        transaction_cost_bps=req.transaction_cost_bps,
        slippage_bps=req.slippage_bps,
        allow_overnight=True,
        force_close_at_end=True,
    )

    result = run_backtest(dataset, strat_inst, config)
    result_dict = result.to_dict()
    first_ts = min(b[0].timestamp for b in dataset.bars_by_symbol.values() if b)
    last_ts = max(b[-1].timestamp for b in dataset.bars_by_symbol.values() if b)
    result_dict["period_start"] = first_ts.isoformat(sep=" ")
    result_dict["period_end_exclusive"] = last_ts.isoformat(sep=" ")
    result_dict["currency"] = "INR"

    # Persist experiment
    repo.save_experiment(result_dict)

    return ExperimentDetailResponse.model_validate(result_dict)


@app.post("/api/v1/backtest/stream", tags=["backtesting"])
def stream_backtest_endpoint(req: BacktestRunRequest) -> StreamingResponse:
    repo = _repository()
    strat_inst = _resolve_strategy(req.strategy_name, req.strategy_params or {})

    cleaned_symbols = tuple(sorted({s.strip().upper() for s in req.universe if s.strip()}))
    if not cleaned_symbols:
        raise HTTPException(
            status_code=400, detail="Universe must contain at least one valid symbol."
        )

    try:
        dataset = repo.get_backtest_dataset(cleaned_symbols, req.start_date, req.end_date)
    except Exception as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc

    config = BacktestConfig(
        initial_capital=req.initial_capital,
        max_gross_exposure=req.max_gross_exposure,
        transaction_cost_bps=req.transaction_cost_bps,
        slippage_bps=req.slippage_bps,
        allow_overnight=True,
        force_close_at_end=True,
    )

    def event_stream():
        events: list[dict[str, Any]] = []

        def on_progress(
            pct: float,
            current_ts: datetime,
            equity: float,
            trades_count: int,
            last_trade: Any | None,
        ):
            payload: dict[str, Any] = {
                "type": "progress",
                "progress_pct": pct,
                "current_date": current_ts.strftime("%Y-%m-%d %H:%M"),
                "current_equity": round(equity, 2),
                "trades_count": trades_count,
                "point": {
                    "timestamp": current_ts.strftime("%Y-%m-%d %H:%M"),
                    "equity": round(equity, 2),
                },
            }
            if last_trade:
                payload["last_trade"] = {
                    "symbol": last_trade.symbol,
                    "net_pnl": round(last_trade.net_pnl, 2),
                    "exit_time": last_trade.exit_time.strftime("%Y-%m-%d %H:%M"),
                    "exit_reason": last_trade.exit_reason,
                }
            events.append(payload)

        result = run_backtest(dataset, strat_inst, config, progress_callback=on_progress)

        for ev in events:
            yield f"data: {json.dumps(ev)}\n\n"

        result_dict = result.to_dict()
        first_ts = min(b[0].timestamp for b in dataset.bars_by_symbol.values() if b)
        last_ts = max(b[-1].timestamp for b in dataset.bars_by_symbol.values() if b)
        result_dict["period_start"] = first_ts.isoformat(sep=" ")
        result_dict["period_end_exclusive"] = last_ts.isoformat(sep=" ")
        result_dict["currency"] = "INR"

        repo.save_experiment(result_dict)

        complete_payload = {
            "type": "complete",
            "experiment_id": result_dict["experiment_id"],
            "strategy_name": result_dict["strategy_name"],
            "strategy_version": result_dict["strategy_version"],
            "dataset_id": result_dict["dataset_id"],
            "period_start": result_dict["period_start"],
            "period_end_exclusive": result_dict["period_end_exclusive"],
            "config": result_dict["config"],
            "metrics": result_dict["metrics"],
            "equity_curve": result_dict["equity_curve"],
            "trades": result_dict.get("trades", [])[:100],
        }
        yield f"data: {json.dumps(complete_payload)}\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )


@app.get("/api/v1/jobs/{job_id}", response_model=JobStatusResponse, tags=["backtesting"])
def job_status(job_id: str) -> JobStatusResponse:
    job = _JOBS.get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found.")
    return JobStatusResponse.model_validate(job)


@app.get("/api/v1/experiments", response_model=ExperimentListResponse, tags=["experiments"])
def list_experiments(
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=50, ge=1, le=100),
) -> ExperimentListResponse:
    experiments = _repository().list_experiments()
    summaries = [
        ExperimentSummary.model_validate(experiment)
        for experiment in experiments[offset : offset + limit]
    ]
    return ExperimentListResponse(
        total=len(experiments), offset=offset, limit=limit, items=summaries
    )


@app.get(
    "/api/v1/experiments/{experiment_id}",
    response_model=ExperimentDetailResponse,
    tags=["experiments"],
)
def experiment_detail(experiment_id: str) -> ExperimentDetailResponse:
    experiment = _experiment_or_404(_repository(), experiment_id)
    return ExperimentDetailResponse.model_validate(experiment)


@app.get(
    "/api/v1/experiments/{experiment_id}/equity",
    response_model=list[SeriesPoint],
    tags=["experiments"],
)
def equity_curve(
    experiment_id: str, maximum_points: int = Query(default=1_500, ge=10, le=5_000)
) -> list[SeriesPoint]:
    experiment = _experiment_or_404(_repository(), experiment_id)
    points = _downsample(experiment["equity_curve"], maximum_points)
    return [SeriesPoint.model_validate(point) for point in points]


@app.get(
    "/api/v1/experiments/{experiment_id}/drawdown",
    response_model=list[DrawdownPoint],
    tags=["experiments"],
)
def drawdown_curve(
    experiment_id: str, maximum_points: int = Query(default=1_500, ge=10, le=5_000)
) -> list[DrawdownPoint]:
    repo = _repository()
    _experiment_or_404(repo, experiment_id)
    points = repo.drawdown_series(experiment_id, maximum_points=maximum_points)
    return [DrawdownPoint.model_validate(pt) for pt in points]


@app.get(
    "/api/v1/experiments/{experiment_id}/trades",
    response_model=TradePageResponse,
    tags=["experiments"],
)
def trades(
    experiment_id: str,
    symbol: str | None = Query(default=None),
    outcome: str | None = Query(default=None, pattern="^(win|loss)$"),
    exit_reason: str | None = Query(default=None),
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=100, ge=1, le=500),
) -> TradePageResponse:
    experiment = _experiment_or_404(_repository(), experiment_id)
    rows = experiment["trades"]

    if symbol:
        sym_clean = symbol.strip().upper()
        rows = [t for t in rows if t["symbol"] == sym_clean]
    if outcome == "win":
        rows = [t for t in rows if t["net_pnl"] > 0]
    elif outcome == "loss":
        rows = [t for t in rows if t["net_pnl"] <= 0]
    if exit_reason:
        rows = [t for t in rows if t["exit_reason"] == exit_reason]

    return TradePageResponse(
        total=len(rows),
        offset=offset,
        limit=limit,
        items=[TradeResponse.model_validate(row) for row in rows[offset : offset + limit]],
    )


@app.get(
    "/api/v1/experiments/{experiment_id}/attribution",
    response_model=list[AttributionItem],
    tags=["experiments"],
)
def experiment_attribution(experiment_id: str) -> list[AttributionItem]:
    repo = _repository()
    _experiment_or_404(repo, experiment_id)
    items = repo.attribution(experiment_id)
    return [AttributionItem.model_validate(it) for it in items]


@app.get(
    "/api/v1/experiments/{experiment_id}/monthly",
    response_model=MonthlyReturnsResponse,
    tags=["experiments"],
)
def experiment_monthly(experiment_id: str) -> MonthlyReturnsResponse:
    repo = _repository()
    _experiment_or_404(repo, experiment_id)
    res = repo.monthly_returns(experiment_id)
    return MonthlyReturnsResponse.model_validate(res)


@app.get(
    "/api/v1/experiments/compare",
    response_model=list[ExperimentSummary],
    tags=["experiments"],
)
def compare_experiments(
    ids: list[str] = Query(min_length=2, max_length=8),  # noqa: B008
) -> list[ExperimentSummary]:
    repository = _repository()
    index = {item["experiment_id"]: item for item in repository.list_experiments()}
    missing = [experiment_id for experiment_id in ids if experiment_id not in index]
    if missing:
        raise HTTPException(status_code=404, detail="One or more experiments were not found.")
    return [ExperimentSummary.model_validate(index[experiment_id]) for experiment_id in ids]


@app.get(
    "/api/v1/reports/{experiment_id}/export-csv",
    tags=["reports"],
)
def export_csv(
    experiment_id: str,
    kind: str = Query(default="trades", pattern="^(trades|equity)$"),
) -> Response:
    experiment = _experiment_or_404(_repository(), experiment_id)
    output = io.StringIO()

    if kind == "trades":
        trades_list = experiment.get("trades", [])
        output.write(
            "symbol,entry_time,exit_time,quantity,entry_price,exit_price,gross_pnl,costs,net_pnl,exit_reason\n"
        )
        for t in trades_list:
            output.write(
                f"{t['symbol']},{t['entry_time']},{t['exit_time']},{t['quantity']},{t['entry_price']},{t['exit_price']},{t['gross_pnl']},{t['costs']},{t['net_pnl']},{t['exit_reason']}\n"
            )
        filename = f"trades_{experiment_id[:8]}.csv"
    else:
        equity_list = experiment.get("equity_curve", [])
        output.write("timestamp,equity,gross_exposure\n")
        for pt in equity_list:
            output.write(f"{pt['timestamp']},{pt['equity']},{pt['gross_exposure']}\n")
        filename = f"equity_{experiment_id[:8]}.csv"

    return Response(
        content=output.getvalue(),
        media_type="text/csv",
        headers={"Content-Disposition": f"attachment; filename={filename}"},
    )


class PortfolioAllocateRequest(BaseModel):
    universe: list[str] = Field(min_length=2, max_length=50)
    model: str = Field(
        default="equal_weight",
        pattern="^(equal_weight|inverse_volatility|risk_parity|multi_factor_momentum|minimum_variance)$",
    )
    start_date: str | None = None
    end_date: str | None = None
    max_stock_cap: float = Field(default=0.15, gt=0, le=1.0)
    max_sector_cap: float = Field(default=0.35, gt=0, le=1.0)


@app.post("/api/v1/portfolio/allocate", tags=["portfolio"])
def allocate_portfolio(payload: PortfolioAllocateRequest) -> dict[str, Any]:
    """Execute quantitative portfolio construction and risk-budgeting allocation."""
    repo = _repository()
    bars_by_symbol: dict[str, list[dict[str, Any]]] = {}
    for sym in payload.universe:
        bars = repo.market_bars(
            sym, start=payload.start_date, end=payload.end_date, resolution="1d", limit=1000
        )
        if bars:
            bars_by_symbol[sym] = bars

    if len(bars_by_symbol) < 2:
        raise HTTPException(
            status_code=400,
            detail="At least 2 symbols with active historical data in the date range are required.",
        )

    allocator = PortfolioAllocator(SECTOR_MAP)
    res = allocator.allocate(
        payload.model,
        bars_by_symbol,
        max_stock_cap=payload.max_stock_cap,
        max_sector_cap=payload.max_sector_cap,
    )
    return res.to_dict()


# ==============================================================================
# STATIC FRONTEND SERVING & SPA FALLBACK (FOR HUGGING FACE & PRODUCTION)
# ==============================================================================
_env_dist = os.environ.get("FRONTEND_DIST_DIR")
_dist_candidates = [
    Path(_env_dist) if _env_dist else None,
    Path(__file__).resolve().parents[3] / "frontend" / "dist",
    Path.cwd() / "frontend" / "dist",
    Path.cwd() / "dist",
]
_dist_dir = next((d for d in _dist_candidates if d is not None and d.is_dir()), None)

if _dist_dir is not None:
    from fastapi.responses import FileResponse
    from fastapi.staticfiles import StaticFiles

    _assets_dir = _dist_dir / "assets"
    if _assets_dir.is_dir():
        app.mount("/assets", StaticFiles(directory=str(_assets_dir)), name="assets")

    @app.get("/", include_in_schema=False)
    def serve_root():
        index_file = _dist_dir / "index.html"
        if index_file.is_file():
            return FileResponse(index_file)
        raise HTTPException(status_code=404, detail="Frontend index.html not found.")

    @app.get("/{full_path:path}", include_in_schema=False)
    def serve_spa(full_path: str):
        if (
            full_path.startswith("api/")
            or full_path == "health"
            or full_path == "docs"
            or full_path == "openapi.json"
        ):
            raise HTTPException(status_code=404, detail="Not Found")
        if full_path:
            target_file = _dist_dir / full_path
            if target_file.is_file():
                return FileResponse(target_file)
        index_file = _dist_dir / "index.html"
        if index_file.is_file():
            return FileResponse(index_file)
        raise HTTPException(status_code=404, detail="Frontend build index.html not found.")