#!/usr/bin/env python3 """ Trading Bot Backend - FastAPI Application Supports Quotex via pyquotex and Tradovate """ import asyncio import json import os from contextlib import asynccontextmanager from datetime import datetime, timedelta from typing import Optional import uvicorn from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Depends, status from fastapi.middleware.cors import CORSMiddleware from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from app.auth import verify_token, create_access_token from app.database import init_db, get_db from app.models import LoginRequest, SignalResponse, BotStatus from app.quotex_client import QuotexClient from app.signal_engine import SignalEngine from app.websocket_manager import WebSocketManager # Global instances quotex_client: Optional[QuotexClient] = None signal_engine: Optional[SignalEngine] = None ws_manager = WebSocketManager() security = HTTPBearer() @asynccontextmanager async def lifespan(app: FastAPI): """Startup and shutdown events""" global quotex_client, signal_engine print("🚀 Trading Bot Backend Starting...") await init_db() signal_engine = SignalEngine(ws_manager) asyncio.create_task(signal_engine.run_signal_loop()) print("✅ Signal engine started") yield print("🛑 Shutting down...") if quotex_client: await quotex_client.disconnect() app = FastAPI( title="Trading Bot API", description="Real-time trading signals for Quotex & Tradovate", version="1.0.0", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): token = credentials.credentials payload = verify_token(token) if not payload: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token", ) return payload # ─── AUTH ROUTES ────────────────────────────────────────────────────────────── @app.post("/api/auth/login") async def login(request: LoginRequest): """Admin/User login endpoint""" from app.auth import authenticate_user user = await authenticate_user(request.username, request.password) if not user: raise HTTPException(status_code=401, detail="Invalid credentials") token = create_access_token({"sub": user["username"], "role": user["role"], "id": user["id"]}) return { "access_token": token, "token_type": "bearer", "user": {"username": user["username"], "role": user["role"]}, } # ─── QUOTEX ROUTES ──────────────────────────────────────────────────────────── @app.post("/api/quotex/connect") async def connect_quotex( email: str, password: str, current_user: dict = Depends(get_current_user), ): global quotex_client quotex_client = QuotexClient(email, password) connected = await quotex_client.connect() if not connected: raise HTTPException(status_code=400, detail="Failed to connect to Quotex") if signal_engine: signal_engine.set_quotex_client(quotex_client) return {"status": "connected", "message": "Successfully connected to Quotex"} @app.get("/api/quotex/assets") async def get_assets(current_user: dict = Depends(get_current_user)): if not quotex_client or not quotex_client.is_connected: raise HTTPException(status_code=400, detail="Not connected to Quotex") assets = await quotex_client.get_available_assets() return {"assets": assets} @app.get("/api/quotex/balance") async def get_balance(current_user: dict = Depends(get_current_user)): if not quotex_client or not quotex_client.is_connected: raise HTTPException(status_code=400, detail="Not connected to Quotex") balance = await quotex_client.get_balance() return {"balance": balance} @app.post("/api/quotex/trade") async def place_trade( asset: str, direction: str, amount: float, duration: int, current_user: dict = Depends(get_current_user), ): if not quotex_client or not quotex_client.is_connected: raise HTTPException(status_code=400, detail="Not connected to Quotex") result = await quotex_client.place_trade(asset, direction, amount, duration) return result # ─── SIGNAL ROUTES ──────────────────────────────────────────────────────────── @app.get("/api/signals") async def get_signals( asset: Optional[str] = None, timeframe: Optional[str] = None, current_user: dict = Depends(get_current_user), ): if not signal_engine: raise HTTPException(status_code=500, detail="Signal engine not initialized") signals = signal_engine.get_latest_signals(asset=asset, timeframe=timeframe) return {"signals": signals} @app.get("/api/signals/history") async def get_signal_history( limit: int = 50, asset: Optional[str] = None, current_user: dict = Depends(get_current_user), ): if not signal_engine: raise HTTPException(status_code=500, detail="Signal engine not initialized") history = signal_engine.get_signal_history(limit=limit, asset=asset) return {"history": history} @app.get("/api/bot/status") async def get_bot_status(current_user: dict = Depends(get_current_user)): return { "quotex_connected": bool(quotex_client and quotex_client.is_connected), "signal_engine_running": bool(signal_engine and signal_engine.running), "active_timeframes": ["5s", "10s", "30s", "1m", "5m"], "timestamp": datetime.utcnow().isoformat(), } # ─── ADMIN ROUTES ───────────────────────────────────────────────────────────── @app.get("/api/admin/users") async def list_users(current_user: dict = Depends(get_current_user)): if current_user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admin access required") from app.database import get_all_users users = await get_all_users() return {"users": users} @app.post("/api/admin/users") async def create_user( username: str, password: str, current_user: dict = Depends(get_current_user), ): if current_user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admin access required") from app.database import create_user_db from app.auth import hash_password user = await create_user_db(username, hash_password(password), role="user") return {"message": "User created", "user": user} @app.delete("/api/admin/users/{user_id}") async def delete_user(user_id: int, current_user: dict = Depends(get_current_user)): if current_user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admin access required") from app.database import delete_user_db await delete_user_db(user_id) return {"message": "User deleted"} # ─── WEBSOCKET ──────────────────────────────────────────────────────────────── @app.websocket("/ws/signals") async def websocket_signals(websocket: WebSocket): """Real-time signal streaming via WebSocket""" token = websocket.query_params.get("token") if not token or not verify_token(token): await websocket.close(code=4001) return await ws_manager.connect(websocket) try: while True: data = await websocket.receive_text() msg = json.loads(data) if msg.get("type") == "subscribe": await ws_manager.subscribe(websocket, msg.get("asset"), msg.get("timeframe")) elif msg.get("type") == "ping": await websocket.send_json({"type": "pong", "timestamp": datetime.utcnow().isoformat()}) except WebSocketDisconnect: ws_manager.disconnect(websocket) if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)