Browse the data

1import random
2import time
3import threading
4from fastapi import FastAPI, HTTPException, Header
5from pydantic import BaseModel
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11next_user_id = 1
12 
13portfolios = {}
14next_portfolio_id = 1
15 
16holdings = {}
17next_holding_id = 1
18 
19alerts = []
20next_alert_id = 1
21 
22base_prices = {
23    "AAPL": 150.0,
24    "GOOGL": 2800.0,
25    "MSFT": 300.0,
26    "AMZN": 3400.0,
27    "TSLA": 700.0
28}
29 
30current_prices = {k: v for k, v in base_prices.items()}
31price_history = {k: [] for k in base_prices}
32 
33def price_simulator():
34    while True:
35        for sym in current_prices:
36            change = random.uniform(-0.05, 0.05)
37            current_prices[sym] = round(current_prices[sym] * (1 + change), 2)
38            price_history[sym].append(current_prices[sym])
39            if len(price_history[sym]) > 20:
40                price_history[sym].pop(0)
41            if abs(change) > 0.03:
42                alert_id = next_alert_id
43                next_alert_id += 1
44                alerts.append({
45                    "id": alert_id,
46                    "symbol": sym,
47                    "price": current_prices[sym],
48                    "change_pct": round(change * 100, 2),
49                    "timestamp": time.time()
50                })
51        time.sleep(2)
52 
53threading.Thread(target=price_simulator, daemon=True).start()
54 
55def require_auth(authorization: str = Header(None)):
56    if not authorization:
57        raise HTTPException(status_code=401, detail="No auth header")
58    token = authorization.replace("Bearer ", "")
59    for uid, t in tokens.items():
60        if t == token:
61            return uid
62    raise HTTPException(status_code=401, detail="Invalid token")
63 
64class SignupRequest(BaseModel):
65    username: str
66    password: str
67 
68class LoginRequest(BaseModel):
69    username: str
70    password: str
71 
72class PortfolioCreate(BaseModel):
73    name: str
74 
75class HoldingCreate(BaseModel):
76    symbol: str
77    quantity: int
78    portfolio_id: int
79 
80@app.post("/signup")
81def signup(req: SignupRequest):
82    global next_user_id
83    for u in users.values():
84        if u["username"] == req.username:
85            raise HTTPException(status_code=400, detail="Username taken")
86    uid = next_user_id
87    next_user_id += 1
88    users[uid] = {"id": uid, "username": req.username, "password": req.password}
89    return {"id": uid, "username": req.username}
90 
91@app.post("/login")
92def login(req: LoginRequest):
93    for uid, u in users.items():
94        if u["username"] == req.username and u["password"] == req.password:
95            token = f"tok_{uid}_{random.randint(10000,99999)}"
96            tokens[uid] = token
97            return {"token": token}
98    raise HTTPException(status_code=401, detail="Invalid credentials")
99 
100@app.get("/portfolio/{portfolio_id}")
101def get_portfolio(portfolio_id: int, authorization: str = Header(None)):
102    uid = require_auth(authorization)
103    if portfolio_id not in portfolios:
104        raise HTTPException(status_code=404, detail="Portfolio not found")
105    portfolio = portfolios[portfolio_id]
106    portfolio_holdings = [h for h in holdings.values() if h["portfolio_id"] == portfolio_id]
107    total_pnl = 0
108    total_risk = 0
109    enriched = []
110    for h in portfolio_holdings:
111        sym = h["symbol"]
112        current_price = current_prices.get(sym, 0)
113        cost_basis = h["avg_cost"]
114        pnl = round((current_price - cost_basis) * h["quantity"], 2)
115        total_pnl += pnl
116        hist = price_history.get(sym, [])
117        if len(hist) > 1:
118            returns = [(hist[i] - hist[i-1]) / hist[i-1] for i in range(1, len(hist))]
119            vol = round((sum(r**2 for r in returns) / len(returns))**0.5 * 100, 2) if returns else 0
120        else:
121            vol = 0
122        total_risk += vol * h["quantity"]
123        enriched.append({
124            "id": h["id"],
125            "symbol": sym,
126            "quantity": h["quantity"],
127            "avg_cost": cost_basis,
128            "current_price": current_price,
129            "pnl": pnl,
130            "risk_score": vol
131        })
132    avg_risk = round(total_risk / len(enriched), 2) if enriched else 0
133    return {
134        "portfolio": portfolio,
135        "holdings": enriched,
136        "total_pnl": total_pnl,
137        "risk_score": avg_risk
138    }
139 
140@app.get("/alerts")
141def get_alerts(authorization: str = Header(None)):
142    uid = require_auth(authorization)
143    return {"alerts": alerts[-20:]}
144 
145@app.post("/portfolio")
146def create_portfolio(req: PortfolioCreate, authorization: str = Header(None)):
147    uid = require_auth(authorization)
148    global next_portfolio_id
149    pid = next_portfolio_id
150    next_portfolio_id += 1
151    portfolios[pid] = {"id": pid, "name": req.name, "user_id": uid}
152    return portfolios[pid]
153 
154@app.post("/holding")
155def create_holding(req: HoldingCreate, authorization: str = Header(None)):
156    uid = require_auth(authorization)
157    if req.portfolio_id not in portfolios:
158        raise HTTPException(status_code=404, detail="Portfolio not found")
159    if portfolios[req.portfolio_id]["user_id"] != uid:
160        raise HTTPException(status_code=403, detail="Not your portfolio")
161    global next_holding_id
162    hid = next_holding_id
163    next_holding_id += 1
164    holdings[hid] = {
165        "id": hid,
166        "symbol": req.symbol,
167        "quantity": req.quantity,
168        "avg_cost": current_prices.get(req.symbol, 100.0),
169        "portfolio_id": req.portfolio_id
170    }
171    return holdings[hid]
172 
173@app.get("/portfolio")
174def list_portfolios(authorization: str = Header(None)):
175    uid = require_auth(authorization)
176    user_portfolios = [p for p in portfolios.values() if p["user_id"] == uid]
177    return {"portfolios": user_portfolios}

1fastapi
2uvicorn