chore: api routers

Author: ayushkrtiwari

src/quant_research_starter/api/routers/__init__.py

@@ -0,0 +1,9 @@
+"""API routers package."""
+
+from fastapi import APIRouter
+
+router = APIRouter()
+
+from . import auth as auth_router  # noqa: E402,F401
+from . import backtest as backtest_router  # noqa: E402,F401
+from . import assets as assets_router  # noqa: E402,F401

src/quant_research_starter/api/routers/assets.py

@@ -0,0 +1,15 @@
+"""Assets router to expose available symbols / sample data."""
+from fastapi import APIRouter
+from quant_research_starter.data.sample_loader import SampleDataLoader
+
+router = APIRouter(prefix="/api/assets", tags=["assets"])
+
+
+@router.get("/")
+async def list_assets():
+    loader = SampleDataLoader()
+    df = loader.load_sample_prices()
+    symbols = []
+    for sym in df.columns:
+        symbols.append({"symbol": sym, "price": float(df[sym].iloc[-1])})
+    return symbols

src/quant_research_starter/api/routers/auth.py

@@ -0,0 +1,35 @@
+"""Authentication routes: register and token endpoints."""
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+from fastapi.security import OAuth2PasswordRequestForm
+
+from .. import schemas, db, models, auth
+
+router = APIRouter(prefix="/api/auth", tags=["auth"])
+
+
+@router.post("/register", response_model=schemas.UserRead)
+async def register_user(user_in: schemas.UserCreate, session: AsyncSession = Depends(db.get_session)):
+    q = await session.execute(models.User.__table__.select().where(models.User.username == user_in.username))
+    if q.first():
+        raise HTTPException(status_code=400, detail="Username already registered")
+    hashed = auth.get_password_hash(user_in.password)
+    user = models.User(username=user_in.username, hashed_password=hashed)
+    session.add(user)
+    await session.commit()
+    await session.refresh(user)
+    return schemas.UserRead(id=user.id, username=user.username, is_active=user.is_active, role=user.role)
+
+
+@router.post("/token", response_model=schemas.Token)
+async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), session: AsyncSession = Depends(db.get_session)):
+    q = await session.execute(models.User.__table__.select().where(models.User.username == form_data.username))
+    row = q.first()
+    if not row:
+        raise HTTPException(status_code=400, detail="Incorrect username or password")
+    user = row[0]
+    if not auth.verify_password(form_data.password, user.hashed_password):
+        raise HTTPException(status_code=400, detail="Incorrect username or password")
+
+    access_token = auth.create_access_token(data={"sub": user.username})
+    return {"access_token": access_token, "token_type": "bearer"}

src/quant_research_starter/api/routers/backtest.py

@@ -0,0 +1,67 @@
+"""Backtest endpoints: enqueue backtest jobs and fetch results."""
+from __future__ import annotations
+import uuid
+import os
+from fastapi import APIRouter, Depends, BackgroundTasks, UploadFile, File, HTTPException, WebSocket
+from sqlalchemy.ext.asyncio import AsyncSession
+from typing import Optional
+
+from .. import schemas, db, models, auth
+from ..tasks.celery_app import celery_app
+from ..utils.ws_manager import manager
+
+router = APIRouter(prefix="/api/backtest", tags=["backtest"])
+
+
+@router.post("/", response_model=schemas.BacktestStatus)
+async def submit_backtest(req: schemas.BacktestRequest, current_user=Depends(auth.require_active_user), session: AsyncSession = Depends(db.get_session)):
+    # Create job
+    job_id = uuid.uuid4().hex
+    job = models.BacktestJob(id=job_id, user_id=current_user.id, status="queued", params=req.dict())
+    session.add(job)
+    await session.commit()
+
+    # Enqueue celery task
+    celery_app.send_task("quant_research_starter.api.tasks.tasks.run_backtest", args=[job_id, req.dict()])
+
+    return {"job_id": job_id, "status": "queued"}
+
+
+@router.get("/{job_id}/results")
+async def get_results(job_id: str, current_user=Depends(auth.require_active_user), session: AsyncSession = Depends(db.get_session)):
+    q = await session.execute(models.BacktestJob.__table__.select().where(models.BacktestJob.id == job_id))
+    row = q.first()
+    if not row:
+        raise HTTPException(status_code=404, detail="Job not found")
+    job = row[0]
+    if job.user_id != current_user.id and current_user.role != "admin":
+        raise HTTPException(status_code=403, detail="Not authorized to view this job")
+
+    if job.result_path and os.path.exists(job.result_path):
+        import json
+
+        with open(job.result_path, "r") as f:
+            return json.load(f)
+    return {"status": job.status}
+
+
+@router.websocket("/ws/{job_id}")
+async def websocket_backtest(websocket: WebSocket, job_id: str):
+    """WebSocket endpoint that registers the client and relays messages from Redis pub/sub.
+
+    The Redis listener broadcasts messages to the ConnectionManager which then sends
+    them to connected WebSocket clients.
+    """
+    await manager.connect(job_id, websocket)
+    try:
+        while True:
+            # keep the connection alive; client may send ping messages
+            msg = await websocket.receive_text()
+            # ignore incoming messages; server pushes updates
+            await websocket.send_text("ok")
+    except Exception:
+        manager.disconnect(job_id, websocket)
+        try:
+            await websocket.close()
+        except Exception:
+            pass