Repository navigation
Expand file tree
/
Copy pathtempCodeRunnerFile.py
More file actions
115 lines (100 loc) · 3.99 KB
/
Copy pathtempCodeRunnerFile.py
File metadata and controls
115 lines (100 loc) · 3.99 KB
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
from fastapi import FastAPI, Depends, HTTPException, Query, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from sqlalchemy.orm import Session
from database import engine, Base, get_db
import models, schemas, auth, backtester
# Create database tables
Base.metadata.create_all(bind=engine)
print("✅ Database tables ready.")
app = FastAPI(title="ASEP Backend")
# Allow frontend to communicate with backend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ==========================================
# AUTHENTICATION ENDPOINTS
# ==========================================
@app.post("/api/register")
def register(request: schemas.RegisterRequest, db: Session = Depends(get_db)):
db_user = db.query(models.User).filter(models.User.email == request.email).first()
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
hashed_pw = auth.get_password_hash(request.password)
new_user = models.User(
email=request.email,
username=request.username,
hashed_password=hashed_pw
)
db.add(new_user)
db.commit()
db.refresh(new_user)
return {"message": "User created successfully"}
@app.post("/api/login")
def login(request: schemas.LoginRequest, db: Session = Depends(get_db)):
db_user = db.query(models.User).filter(models.User.email == request.email).first()
if not db_user or not auth.verify_password(request.password, db_user.hashed_password):
raise HTTPException(status_code=401, detail="Invalid email or password")
return {
"message": "Login successful",
"email": db_user.email,
"username": db_user.username
}
# ==========================================
# BACKTESTING ENDPOINTS
# ==========================================
@app.post("/api/backtest")
def run_backtest(req: schemas.BacktestRequest, db: Session = Depends(get_db)):
try:
# Run the actual backtest logic
results = backtester.run_strategy(
strategy_name=req.strategy,
ticker=req.ticker,
start_date=req.startDate,
end_date=req.endDate,
fast_ma=req.fastMA,
slow_ma=req.slowMA,
mr_k=req.mrK,
initial_capital=req.initialCapital,
risk_percent=req.riskPercent,
atr_period=req.atrPeriod,
stop_loss=req.stopLoss,
slippage=req.slippage,
interval=req.interval
)
# Save results to database
db_backtest = models.BacktestResult(
user_id=1, # Hardcoding user_id 1 for now, in a real app this comes from the logged-in user
strategy=req.strategy,
ticker=req.ticker,
total_return=results.get("total_return", 0),
win_rate=results.get("win_rate", 0),
max_drawdown=results.get("max_drawdown", 0)
)
db.add(db_backtest)
db.commit()
return results
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/history")
def get_history(db: Session = Depends(get_db)):
# Fetch all backtests from the database
history = db.query(models.BacktestResult).order_by(models.BacktestResult.created_at.desc()).all()
return history
# ==========================================
# SERVE FRONTEND FILES
# ==========================================
# Serve the main login page when someone visits the root URL "/"
@app.get("/")
def serve_root():
return FileResponse("login.html")
# Mount the entire current directory to serve HTML, CSS, JS files
app.mount("/", StaticFiles(directory=".", html=True), name="static")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)