Spaces:
Paused
Paused
File size: 4,319 Bytes
5545e4d 461ed71 5545e4d 461ed71 5545e4d 461ed71 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """HTTP API over the trading floor, for a separate frontend to consume.
The Gradio dashboard in demo/ reads accounts.db in-process. This serves the same
data as JSON so a decoupled web frontend can render it. Everything here is
read-only; the trading floor writes the database out of band.
Run it from the 6_mcp directory so it shares the engine's accounts.db:
uv run uvicorn backend.api:app --port 8000
"""
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.staticfiles import StaticFiles
from backend import market
from backend.accounts import Account
from backend.database import read_log
from backend.trading_floor import names, lastnames, short_model_names
# Mirrors the log colours in demo/ so the frontend reproduces the same panel.
LOG_COLORS = {
"trace": "#87CEEB",
"agent": "#00dddd",
"function": "#00dd00",
"generation": "#dddd00",
"response": "#aa00dd",
"account": "#dd0000",
}
DEFAULT_LOG_COLOR = "#87CEEB"
roster = [
{"name": name, "lastname": lastname, "model_name": model_name}
for name, lastname, model_name in zip(names, lastnames, short_model_names)
]
roster_by_name = {trader["name"].lower(): trader for trader in roster}
app = FastAPI(title="Trading Floor")
def average_cost(account: Account, symbol: str) -> float:
"""Average price paid across this symbol's buys, for per-holding profit."""
spend = sum(t.price * t.quantity for t in account.transactions if t.symbol == symbol and t.quantity > 0)
bought = sum(t.quantity for t in account.transactions if t.symbol == symbol and t.quantity > 0)
return spend / bought if bought else 0.0
def holdings_detail(account: Account) -> list[dict]:
"""Current holdings enriched with price, market value and unrealised profit."""
details = []
for symbol, quantity in account.holdings.items():
price = market.get_share_price(symbol)
cost = average_cost(account, symbol)
details.append(
{
"symbol": symbol,
"quantity": quantity,
"price": price,
"avg_cost": cost,
"market_value": price * quantity,
"unrealized_pnl": (price - cost) * quantity,
}
)
return details
def require_trader(name: str) -> dict:
trader = roster_by_name.get(name.lower())
if not trader:
raise HTTPException(status_code=404, detail=f"Unknown trader {name}")
return trader
@app.get("/api/traders")
def get_traders() -> list[dict]:
"""The four traders on the floor."""
return roster
@app.get("/api/market")
def get_market() -> dict:
"""Which price source is live, and whether the market is open."""
source = "massive" if market.massive_api_key else "simulator"
return {"source": source, "is_market_open": market.is_market_open()}
@app.get("/api/traders/{name}")
def get_trader(name: str) -> dict:
"""A trader's full state: value, profit, holdings, transactions and history."""
trader = require_trader(name)
account = Account.get(name)
holdings = holdings_detail(account)
portfolio_value = account.balance + sum(h["market_value"] for h in holdings)
return {
"name": trader["name"],
"lastname": trader["lastname"],
"model_name": trader["model_name"],
"balance": account.balance,
"strategy": account.strategy,
"portfolio_value": portfolio_value,
"pnl": account.calculate_profit_loss(portfolio_value),
"holdings": holdings,
"transactions": account.list_transactions(),
"time_series": [{"datetime": ts, "value": value} for ts, value in account.portfolio_value_time_series],
}
@app.get("/api/traders/{name}/logs")
def get_trader_logs(name: str, last_n: int = 13) -> list[dict]:
"""Recent trace and account log lines, oldest first, with their panel colour."""
require_trader(name)
rows = list(read_log(name, last_n))
return [
{"datetime": ts, "type": kind, "message": message, "color": LOG_COLORS.get(kind, DEFAULT_LOG_COLOR)}
for ts, kind, message in rows
]
static_dir = Path(__file__).resolve().parent.parent / "frontend" / "dist"
if static_dir.is_dir():
app.mount("/", StaticFiles(directory=str(static_dir), html=True), name="frontend")
|