Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import yfinance as yf
5import numpy as np
6from datetime import datetime, timedelta
7import hashlib
8import random
9import string
10 
11app = FastAPI()
12 
13users = {}
14tokens = {}
15orders = {}
16 
17def generate_token():
18    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
19 
20def hash_password(password: str):
21    return hashlib.sha256(password.encode()).hexdigest()
22 
23def get_current_user(authorization: Optional[str] = Header(None)):
24    if not authorization:
25        raise HTTPException(status_code=401, detail="Missing auth header")
26    token = authorization.replace("Bearer ", "")
27    if token not in tokens:
28        raise HTTPException(status_code=401, detail="Invalid token")
29    return tokens[token]
30 
31class SignupRequest(BaseModel):
32    username: str
33    password: str
34 
35class LoginRequest(BaseModel):
36    username: str
37    password: str
38 
39class OrderCreate(BaseModel):
40    ticker: str
41    quantity: int
42    price: float
43 
44class PortfolioRequest(BaseModel):
45    tickers: List[str]
46    weights: List[float]
47    start_date: str
48    end_date: str
49 
50@app.post("/signup")
51def signup(req: SignupRequest):
52    if req.username in users:
53        raise HTTPException(status_code=400, detail="User exists")
54    users[req.username] = hash_password(req.password)
55    return {"message": "User created"}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    if req.username not in users or users[req.username] != hash_password(req.password):
60        raise HTTPException(status_code=401, detail="Invalid credentials")
61    token = generate_token()
62    tokens[token] = req.username
63    return {"token": token}
64 
65@app.post("/orders")
66def create_order(order: OrderCreate, authorization: Optional[str] = Header(None)):
67    user = get_current_user(authorization)
68    order_id = len(orders) + 1
69    orders[order_id] = {"id": order_id, "user": user, **order.dict()}
70    return orders[order_id]
71 
72@app.get("/orders/{order_id}")
73def get_order(order_id: int, authorization: Optional[str] = Header(None)):
74    user = get_current_user(authorization)
75    if order_id not in orders:
76        raise HTTPException(status_code=404, detail="Order not found")
77    return orders[order_id]
78 
79@app.get("/returns")
80def portfolio_returns(tickers: str, weights: str, start_date: str, end_date: str, authorization: Optional[str] = Header(None)):
81    user = get_current_user(authorization)
82    ticker_list = [t.strip() for t in tickers.split(",")]
83    weight_list = [float(w.strip()) for w in weights.split(",")]
84    
85    if len(ticker_list) != len(weight_list):
86        raise HTTPException(status_code=400, detail="Tickers and weights length mismatch")
87    
88    if abs(sum(weight_list) - 1.0) > 0.01:
89        raise HTTPException(status_code=400, detail="Weights must sum to 1")
90    
91    data = yf.download(ticker_list, start=start_date, end=end_date)["Adj Close"]
92    if data.empty:
93        raise HTTPException(status_code=400, detail="No data for given range")
94    
95    returns = data.pct_change().dropna()
96    portfolio_returns = (returns * weight_list).sum(axis=1)
97    
98    # VaR (95% confidence)
99    var_95 = np.percentile(portfolio_returns, 5)
100    
101    # Sharpe ratio (assuming 0% risk-free rate)
102    sharpe = portfolio_returns.mean() / portfolio_returns.std() * np.sqrt(252)
103    
104    # Max drawdown
105    cumulative = (1 + portfolio_returns).cumprod()
106    running_max = cumulative.cummax()
107    drawdown = (cumulative - running_max) / running_max
108    max_drawdown = drawdown.min()
109    
110    return {
111        "var_95": round(var_95, 4),
112        "sharpe_ratio": round(sharpe, 4),
113        "max_drawdown": round(max_drawdown, 4)
114    }

1fastapi
2uvicorn
3yfinance
4numpy
5pydantic