from __future__ import annotations import asyncio from typing import Annotated import httpx from fastapi import ( APIRouter, Depends, HTTPException, Request, Response, WebSocket, WebSocketDisconnect, ) from app.config import Settings from app.engine.strategies import get_strategy, list_strategies, strategy_spec from app.logs import get_log_handler from app.metrics import render_metrics from app.models import OrderSide, Position, PositionStatus from app.runtime import Runtime from app.schemas import ( BacktestRequest, BacktestResult, ClosePositionRequest, EquityPointOut, EventOut, KillRequest, ManualOrderRequest, MultiStatusOut, OrderOut, PositionOut, StartManyRequest, StartRequest, StatusOut, StrategyConfigIn, StrategyConfigOut, StrategyOut, SweepRequest, SweepResult, ) router = APIRouter(prefix="/api") def get_runtime(request: Request) -> Runtime: return request.app.state.runtime def get_settings(request: Request) -> Settings: return request.app.state.runtime.settings def get_guarded_runtime(request: Request) -> Runtime: """Return the runtime, rejecting requests without the configured API token. The check is a no-op when ``API_TOKEN`` is unset, so local dev and the Basic-auth-only deployment keep working unchanged. """ runtime: Runtime = request.app.state.runtime settings: Settings = runtime.settings if settings.api_token: provided = request.headers.get(settings.api_token_header, "") if provided != settings.api_token: raise HTTPException(status_code=401, detail="invalid or missing API token") return runtime GuardedRuntimeDep = Annotated[Runtime, Depends(get_guarded_runtime)] @router.get("/health") def health() -> dict[str, str]: return {"status": "ok"} @router.get("/metrics") def metrics(runtime: GuardedRuntimeDep) -> Response: """Prometheus text exposition, unauthenticated so a scraper can reach it.""" status = runtime.multi_status() body = render_metrics(status, runtime.settings.mode, runtime.settings.exchange) return Response(content=body, media_type="text/plain; version=0.0.4") @router.get("/logs") def logs(runtime: GuardedRuntimeDep, limit: int = 100) -> dict: handler = get_log_handler() entries = handler.snapshot(limit) if handler is not None else [] return {"entries": entries} @router.get("/version") def version(request: Request) -> dict[str, str]: """Build version of the running backend, used by the client to self-heal from a stale cached bundle after a deploy.""" return {"version": getattr(request.app.state, "build_version", "unknown")} @router.get("/strategies", response_model=list[StrategyOut]) def strategies() -> list[StrategyOut]: return [strategy_spec(cls) for cls in list_strategies()] @router.get("/status", response_model=StatusOut) def status(runtime: GuardedRuntimeDep) -> StatusOut: return runtime.status() @router.post("/bot/start", response_model=StatusOut) async def start_bot(payload: StartRequest, runtime: GuardedRuntimeDep) -> StatusOut: try: return await runtime.start( payload.strategy_key, payload.params, symbol=payload.symbol, timeframe=payload.timeframe, starting_balance=payload.starting_balance, ) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except (ValueError, RuntimeError) as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @router.get("/bot/status", response_model=MultiStatusOut) def multi_status(runtime: GuardedRuntimeDep) -> MultiStatusOut: return runtime.multi_status() @router.get("/live-check") def live_check(runtime: GuardedRuntimeDep) -> dict: """Preflight for live trading: whether the config is safe enough.""" return runtime.live_check() @router.post("/bot/start-many", response_model=MultiStatusOut) async def start_many(payload: StartManyRequest, runtime: GuardedRuntimeDep) -> MultiStatusOut: try: return await runtime.start_many(payload) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except (ValueError, RuntimeError) as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @router.post("/bot/stop", response_model=StatusOut) async def stop_bot(runtime: GuardedRuntimeDep) -> StatusOut: return await runtime.stop() @router.post("/bot/kill", response_model=StatusOut) async def kill_bot(payload: KillRequest, runtime: GuardedRuntimeDep) -> StatusOut: return await runtime.kill(payload.reason) @router.post("/backtest", response_model=BacktestResult) async def run_backtest(payload: BacktestRequest, runtime: GuardedRuntimeDep) -> BacktestResult: try: return await runtime.run_backtest(payload) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except httpx.HTTPError as exc: raise HTTPException(status_code=502, detail=f"market data error: {exc}") from exc @router.get("/presets", response_model=list[StrategyConfigOut]) def list_presets(runtime: GuardedRuntimeDep) -> list[StrategyConfigOut]: return [ StrategyConfigOut.model_validate(config) for config in runtime.store.list_strategy_configs() ] @router.post("/presets", response_model=StrategyConfigOut) def create_preset(payload: StrategyConfigIn, runtime: GuardedRuntimeDep) -> StrategyConfigOut: try: get_strategy(payload.strategy_key) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc config = runtime.store.save_strategy_config( name=payload.name, strategy_key=payload.strategy_key, symbol=payload.symbol, timeframe=payload.timeframe, params=payload.params, ) return StrategyConfigOut.model_validate(config) @router.delete("/presets/{config_id}") def delete_preset(config_id: int, runtime: GuardedRuntimeDep) -> dict[str, bool]: if not runtime.store.delete_strategy_config(config_id): raise HTTPException(status_code=404, detail="preset not found") return {"deleted": True} @router.post("/sweep", response_model=SweepResult) async def run_sweep_endpoint(payload: SweepRequest, runtime: GuardedRuntimeDep) -> SweepResult: try: return await runtime.run_sweep(payload) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except httpx.HTTPError as exc: raise HTTPException(status_code=502, detail=f"market data error: {exc}") from exc def _position_out(position: Position, price: float | None) -> PositionOut: out = PositionOut.model_validate(position) if position.status == PositionStatus.OPEN and price is not None: sign = 1 if position.side.value == "long" else -1 out.unrealized_pnl = sign * (price - position.entry_price) * position.qty return out @router.get("/positions", response_model=list[PositionOut]) def positions(runtime: GuardedRuntimeDep, limit: int = 50) -> list[PositionOut]: price = runtime.engine.last_price if runtime.engine else None return [_position_out(position, price) for position in runtime.store.list_positions(limit)] @router.get("/orders", response_model=list[OrderOut]) def orders(runtime: GuardedRuntimeDep, limit: int = 50) -> list[OrderOut]: return [OrderOut.model_validate(order) for order in runtime.store.list_orders(limit)] @router.get("/events", response_model=list[EventOut]) def events(runtime: GuardedRuntimeDep, limit: int = 50) -> list[EventOut]: return [EventOut.model_validate(event) for event in runtime.store.list_events(limit)] @router.get("/equity", response_model=list[EquityPointOut]) def equity(runtime: GuardedRuntimeDep, limit: int = 500) -> list[EquityPointOut]: return [EquityPointOut.model_validate(point) for point in runtime.store.list_equity(limit)] @router.post("/orders", response_model=StatusOut) async def manual_order(payload: ManualOrderRequest, runtime: GuardedRuntimeDep) -> StatusOut: engine = runtime.engine if engine is None or not engine.running: raise HTTPException(status_code=409, detail="engine is not running") side = OrderSide.BUY if payload.side == "buy" else OrderSide.SELL await engine.manual_order(side, payload.qty, payload.quote_amount, payload.reason) return engine.status() @router.post("/positions/{position_id}/close", response_model=StatusOut) async def close_position( position_id: int, payload: ClosePositionRequest, runtime: GuardedRuntimeDep ) -> StatusOut: engine = runtime.engine if engine is None or not engine.running: raise HTTPException(status_code=409, detail="engine is not running") if engine.position is None or engine.position.id != position_id: raise HTTPException(status_code=404, detail="no open position with that id") await engine.close_current_position(payload.reason) return engine.status() @router.get("/ws") def websocket_info() -> dict[str, str]: """A plain browser hit on the socket URL gets a clear hint, not a 404/426.""" return { "detail": "This endpoint is a WebSocket. Connect with a WebSocket client.", "path": "/api/ws", } @router.websocket("/ws") async def websocket_endpoint(websocket: WebSocket) -> None: runtime: Runtime = websocket.app.state.runtime settings: Settings = runtime.settings if settings.api_token: # The frontend nginx injects the token as a header on the upgrade request # (browsers can't set WebSocket headers themselves); accept a query param # too for direct/manual clients. provided = websocket.headers.get(settings.api_token_header) or websocket.query_params.get( "token", "" ) if provided != settings.api_token: await websocket.close(code=1008) return await websocket.accept() queue = runtime.subscribe() try: await websocket.send_json( {"type": "status", "data": runtime.status().model_dump(mode="json")} ) while True: try: payload = await asyncio.wait_for(queue.get(), timeout=15.0) except TimeoutError: await websocket.send_json({"type": "ping"}) continue await websocket.send_json(payload) except WebSocketDisconnect: pass finally: runtime.unsubscribe(queue)