gcloud-lab/trade-dashboard/app/main.py

"""Trade Dashboard — FastAPI service for tracking PnL and open positions."""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from decimal import Decimal
from pathlib import Path
from uuid import UUID

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import FileResponse, HTMLResponse
from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from models import positions
from schemas import (
    Direction,
    PnLSnapshot,
    PositionCreate,
    PositionOut,
    PositionUpdate,
    WebhookTrade,
)

app = FastAPI(title="Trade Dashboard", version="0.1.0")

STATIC_DIR = Path(__file__).parent / "static"


# ── Helpers ─────────────────────────────────────────────────────────────

def position_to_out(row: dict) -> PositionOut:
    return PositionOut(
        id=row["id"],
        symbol=row["symbol"],
        direction=row["direction"],
        entry_price=row["entry_price"],
        exit_price=row["exit_price"],
        quantity=row["quantity"],
        exchange=row["exchange"],
        opened_at=row["opened_at"],
        closed_at=row["closed_at"],
        pnl=row["pnl"],
        metadata=row["metadata"],
    )


# ── Health ─────────────────────────────────────────────────────────────

@app.get("/api/health")
async def health():
    async with async_session() as session:
        result = await session.execute(select(func.now()))
        db_time = result.scalar()
    return {"status": "ok", "db_time": db_time.isoformat()}


# ── Positions ────────────────────────────────────────────────────────────

@app.get("/api/positions", response_model=list[PositionOut])
async def list_positions(
    open_only: bool = Query(True, description="Only show open positions"),
):
    async with async_session() as session:
        if open_only:
            stmt = select(positions).where(positions.c.closed_at.is_(None)).order_by(positions.c.opened_at.desc())
        else:
            stmt = select(positions).order_by(positions.c.opened_at.desc())
        rows = (await session.execute(stmt)).mappings().all()
    return [position_to_out(r) for r in rows]


@app.post("/api/positions", status_code=201)
async def create_position(payload: PositionCreate):
    async with async_session() as session:
        values = payload.model_dump()
        result = await session.execute(positions.insert().values(**values))
        session.commit()
        pk = result.inserted_primary_key[0]
    return {"id": str(pk)}


@app.patch("/api/positions/{position_id}")
async def update_position(position_id: UUID, payload: PositionUpdate):
    async with async_session() as session:
        row = await session.execute(
            select(positions).where(positions.c.id == position_id)
        )
        row = row.mappings().one_or_none()
        if not row:
            raise HTTPException(404, "Position not found")

        updates = payload.model_dump(exclude_unset=True)

        # Auto-compute PnL if closing
        if "exit_price" in updates:
            entry = row["entry_price"]
            qty = row["quantity"]
            exit_p = updates["exit_price"]
            direction = row["direction"]
            if direction == "long":
                updates["pnl"] = float((exit_p - entry) * qty)
            else:
                updates["pnl"] = float((entry - exit_p) * qty)
            updates["closed_at"] = datetime.now(timezone.utc)

        await session.execute(
            positions.update().where(positions.c.id == position_id).values(**updates)
        )
        session.commit()

    return {"ok": True}


@app.delete("/api/positions/{position_id}")
async def close_position(position_id: UUID, exit_price: Decimal = Query(None)):
    async with async_session() as session:
        row = await session.execute(
            select(positions).where(positions.c.id == position_id)
        )
        row = row.mappings().one_or_none()
        if not row:
            raise HTTPException(404, "Position not found")

        if row["closed_at"]:
            raise HTTPException(400, "Position already closed")

        exit_p = exit_price or row["entry_price"]  # breakeven default
        entry = row["entry_price"]
        qty = row["quantity"]
        direction = row["direction"]

        if direction == "long":
            pnl = float((exit_p - entry) * qty)
        else:
            pnl = float((entry - exit_p) * qty)

        await session.execute(
            positions.update()
            .where(positions.c.id == position_id)
            .values(exit_price=exit_p, closed_at=datetime.now(timezone.utc), pnl=pnl)
        )
        session.commit()

    return {"ok": True, "pnl": pnl, "exit_price": float(exit_p)}


# ── PnL ─────────────────────────────────────────────────────────────────

@app.get("/api/pnl", response_model=PnLSnapshot)
async def get_pnl():
    async with async_session() as session:
        now = datetime.now(timezone.utc)
        today = now.replace(hour=0, minute=0, second=0, microsecond=0)
        week_start = today - timedelta(days=now.weekday())
        month_start = today.replace(day=1)

        # Summary for closed trades
        closed = select(
            func.coalesce(func.sum(positions.c.pnl), 0).label("total"),
            func.count(positions.c.id).label("count"),
        ).where(positions.c.closed_at.isnot(None))

        result = (await session.execute(closed)).mappings().one()
        all_time_pnl = float(result["total"])
        total_trades = result["count"]

        # PnL by period
        def period_query(start):
            return select(
                func.coalesce(func.sum(positions.c.pnl), 0)
            ).where(
                and_(
                    positions.c.closed_at.isnot(None),
                    positions.c.closed_at >= start,
                )
            )

        today_pnl = float((await session.execute(period_query(today))).scalar())
        week_pnl = float((await session.execute(period_query(week_start))).scalar())
        month_pnl = float((await session.execute(period_query(month_start))).scalar())

        # Open count
        open_count = (await session.execute(
            select(func.count()).where(positions.c.closed_at.is_(None))
        )).scalar()

    return PnLSnapshot(
        today_pnl=Decimal(str(today_pnl)),
        week_pnl=Decimal(str(week_pnl)),
        month_pnl=Decimal(str(month_pnl)),
        all_time_pnl=Decimal(str(all_time_pnl)),
        total_trades=total_trades,
        open_positions=open_count,
    )


@app.get("/api/pnl/history", response_model=list[PositionOut])
async def pnl_history(
    limit: int = Query(50, ge=1, le=500),
):
    async with async_session() as session:
        stmt = (
            select(positions)
            .where(positions.c.closed_at.isnot(None))
            .order_by(positions.c.closed_at.desc())
            .limit(limit)
        )
        rows = (await session.execute(stmt)).mappings().all()
    return [position_to_out(r) for r in rows]


# ── Webhook (for scanner scripts) ───────────────────────────────────────

@app.post("/webhook/trade", status_code=201)
async def webhook_trade(payload: WebhookTrade):
    meta = {"strategy": payload.strategy} if payload.strategy else {}
    async with async_session() as session:
        result = await session.execute(positions.insert().values(**{
            "symbol": payload.symbol,
            "direction": payload.direction,
            "entry_price": payload.entry_price,
            "quantity": payload.quantity,
            "exchange": payload.exchange,
            "metadata": meta,
        }))
        session.commit()
        pk = result.inserted_primary_key[0]
    return {"id": str(pk)}


# ── Frontend ────────────────────────────────────────────────────────────

@app.get("/", response_class=HTMLResponse)
async def index():
    return FileResponse(str(STATIC_DIR / "index.html"))


# ── Startup: run Alembic migrations ──────────────────────────────────────

@app.on_event("startup")
async def startup():
    import alembic.config
    cfg = alembic.config.AlembicConfig(
        str(Path(__file__).parent.parent / "alembic.ini")
    )
    alembic.config.main.main(command="upgrade", args=["head"], config=cfg)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)