Spaces:
Running
Running
| import os | |
| import json | |
| import asyncio | |
| import logging | |
| from typing import Dict, Any, List, Optional | |
| from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.responses import FileResponse | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from pydantic import BaseModel | |
| from dotenv import load_dotenv | |
| from backend.alpaca_helper import AlpacaHelper | |
| from backend.risk_engine import RiskEngine | |
| from backend.agent import SentryThetaAgent | |
| load_dotenv() | |
| # Setup Logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s [%(levelname)s] %(name)s: %(message)s" | |
| ) | |
| logger = logging.getLogger("SentryTheta.Server") | |
| # FastAPI App Setup | |
| app = FastAPI() | |
| # Enable CORS for local development | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # Initialize components | |
| alpaca = AlpacaHelper() | |
| risk_engine = RiskEngine() | |
| agent = SentryThetaAgent(alpaca, risk_engine) | |
| # Active WebSockets connection pool | |
| class ConnectionManager: | |
| def __init__(self): | |
| self.active_connections: List[WebSocket] = [] | |
| async def connect(self, websocket: WebSocket): | |
| await websocket.accept() | |
| self.active_connections.append(websocket) | |
| logger.info(f"New client connected. Active connections: {len(self.active_connections)}") | |
| def disconnect(self, websocket: WebSocket): | |
| if websocket in self.active_connections: | |
| self.active_connections.remove(websocket) | |
| logger.info(f"Client disconnected. Active connections: {len(self.active_connections)}") | |
| async def broadcast(self, message: Dict[str, Any]): | |
| for connection in self.active_connections: | |
| try: | |
| await connection.send_text(json.dumps(message)) | |
| except Exception as e: | |
| # Connection might be dead | |
| pass | |
| manager = ConnectionManager() | |
| db = alpaca.db | |
| # Load settings from DB with fallback to env/defaults | |
| db_settings = db.get_all_settings() | |
| trading_mode = db_settings.get("trading_mode", os.getenv("TRADING_MODE", "copilot")).lower() | |
| scan_interval = int(db_settings.get("scan_interval", os.getenv("SCAN_INTERVAL_SECONDS", "60"))) | |
| tickers_str = db_settings.get("tickers", os.getenv("TARGET_TICKERS", "AAPL,MSFT,NVDA,SPY,QQQ")) | |
| tickers = [t.strip() for t in tickers_str.split(",")] | |
| # Load risk parameters into RiskEngine | |
| risk_engine.max_position_size_pct = float(db_settings.get("max_position_size_pct", os.getenv("MAX_POSITION_SIZE_PERCENT", "5.0"))) | |
| risk_engine.max_daily_drawdown_pct = float(db_settings.get("max_daily_drawdown_pct", os.getenv("MAX_DAILY_DRAWDOWN_PERCENT", "3.0"))) | |
| risk_engine.stop_loss_pct = float(db_settings.get("stop_loss_pct", os.getenv("STOP_LOSS_PERCENT", "15.0"))) | |
| risk_engine.take_profit_pct = float(db_settings.get("take_profit_pct", os.getenv("TAKE_PROFIT_PERCENT", "30.0"))) | |
| # Load historical logs from DB | |
| terminal_logs = db.get_logs(100) | |
| if not terminal_logs: | |
| terminal_logs = ["SentryTheta Server initialized. Waiting for scanner cycle..."] | |
| db.add_log(terminal_logs[0]) | |
| # Global State | |
| state = { | |
| "trading_mode": trading_mode, | |
| "scan_interval": scan_interval, | |
| "tickers": tickers, | |
| "pending_proposal": None, | |
| "terminal_logs": terminal_logs, | |
| "optionable_assets": ["AAPL", "MSFT", "NVDA", "SPY", "QQQ"] # default fallback | |
| } | |
| # Settings Schema | |
| class SettingsRequest(BaseModel): | |
| trading_mode: str | |
| scan_interval: int | |
| tickers: List[str] | |
| max_position_size_pct: float | |
| max_daily_drawdown_pct: float | |
| stop_loss_pct: float | |
| take_profit_pct: float | |
| # Append terminal log and broadcast | |
| async def log_and_broadcast(message: str): | |
| logger.info(message) | |
| state["terminal_logs"].append(message) | |
| if len(state["terminal_logs"]) > 100: | |
| state["terminal_logs"].pop(0) | |
| db.add_log(message) # Persist log to SQLite | |
| await manager.broadcast({"type": "log", "data": message}) | |
| # REST API Endpoints | |
| async def get_status(): | |
| acc_info = alpaca.get_account_info() | |
| positions = alpaca.get_active_positions() | |
| return { | |
| "account": acc_info, | |
| "positions": positions, | |
| "settings": { | |
| "trading_mode": state["trading_mode"], | |
| "scan_interval": state["scan_interval"], | |
| "tickers": state["tickers"], | |
| "max_position_size_pct": risk_engine.max_position_size_pct, | |
| "max_daily_drawdown_pct": risk_engine.max_daily_drawdown_pct, | |
| "stop_loss_pct": risk_engine.stop_loss_pct, | |
| "take_profit_pct": risk_engine.take_profit_pct, | |
| }, | |
| "pending_proposal": state["pending_proposal"], | |
| "logs": state["terminal_logs"] | |
| } | |
| async def get_assets_endpoint(): | |
| return state.get("optionable_assets", ["AAPL", "MSFT", "NVDA", "SPY", "QQQ"]) | |
| async def update_settings(req: SettingsRequest): | |
| if req.trading_mode not in ["copilot", "autopilot"]: | |
| raise HTTPException(status_code=400, detail="Invalid trading mode.") | |
| state["trading_mode"] = req.trading_mode | |
| state["scan_interval"] = max(10, req.scan_interval) # minimum 10 seconds | |
| state["tickers"] = [t.upper().strip() for t in req.tickers] | |
| risk_engine.max_position_size_pct = req.max_position_size_pct | |
| risk_engine.max_daily_drawdown_pct = req.max_daily_drawdown_pct | |
| risk_engine.stop_loss_pct = req.stop_loss_pct | |
| risk_engine.take_profit_pct = req.take_profit_pct | |
| # Save settings to SQLite | |
| db.set_setting("trading_mode", state["trading_mode"]) | |
| db.set_setting("scan_interval", state["scan_interval"]) | |
| db.set_setting("tickers", ",".join(state["tickers"])) | |
| db.set_setting("max_position_size_pct", risk_engine.max_position_size_pct) | |
| db.set_setting("max_daily_drawdown_pct", risk_engine.max_daily_drawdown_pct) | |
| db.set_setting("stop_loss_pct", risk_engine.stop_loss_pct) | |
| db.set_setting("take_profit_pct", risk_engine.take_profit_pct) | |
| await log_and_broadcast( | |
| f"⚙️ System settings updated. Mode={state['trading_mode'].upper()}, " | |
| f"Tickers={state['tickers']}" | |
| ) | |
| # Broadcast updated status to all clients | |
| await broadcast_state() | |
| return {"status": "success"} | |
| async def approve_trade(): | |
| proposal = state["pending_proposal"] | |
| if not proposal: | |
| raise HTTPException(status_code=400, detail="No pending proposal to approve.") | |
| await log_and_broadcast(f"👍 User APPROVED proposal for {proposal['symbol']}") | |
| # Execute order | |
| res = alpaca.execute_options_trade( | |
| symbol=proposal["symbol"], | |
| qty=proposal["qty"], | |
| side=proposal["side"], | |
| strategy_name=proposal["strategy"] | |
| ) | |
| if res["success"]: | |
| await log_and_broadcast( | |
| f"✅ [Executor] Order filled: {res['side'].upper()} {res['qty']}x {res['symbol']} " | |
| f"at ${res.get('price', proposal['premium']):.2f}. Order ID: {res['order_id']}" | |
| ) | |
| state["pending_proposal"] = None | |
| await broadcast_state() | |
| return {"status": "success", "detail": res} | |
| else: | |
| await log_and_broadcast(f"❌ [Executor] Order failed: {res.get('error')}") | |
| state["pending_proposal"] = None | |
| await broadcast_state() | |
| raise HTTPException(status_code=500, detail=res.get("error")) | |
| async def reject_trade(): | |
| proposal = state["pending_proposal"] | |
| if not proposal: | |
| raise HTTPException(status_code=400, detail="No pending proposal to reject.") | |
| await log_and_broadcast(f"👎 User REJECTED proposal for {proposal['symbol']}. Clearing proposal.") | |
| state["pending_proposal"] = None | |
| await broadcast_state() | |
| return {"status": "success"} | |
| # WebSocket Handler | |
| async def websocket_endpoint(websocket: WebSocket): | |
| await manager.connect(websocket) | |
| try: | |
| # Send initial status | |
| acc_info = alpaca.get_account_info() | |
| positions = alpaca.get_active_positions() | |
| await websocket.send_text(json.dumps({ | |
| "type": "init", | |
| "data": { | |
| "account": acc_info, | |
| "positions": positions, | |
| "settings": { | |
| "trading_mode": state["trading_mode"], | |
| "scan_interval": state["scan_interval"], | |
| "tickers": state["tickers"], | |
| "max_position_size_pct": risk_engine.max_position_size_pct, | |
| "max_daily_drawdown_pct": risk_engine.max_daily_drawdown_pct, | |
| "stop_loss_pct": risk_engine.stop_loss_pct, | |
| "take_profit_pct": risk_engine.take_profit_pct, | |
| }, | |
| "pending_proposal": state["pending_proposal"], | |
| "logs": state["terminal_logs"] | |
| } | |
| })) | |
| while True: | |
| # Keep connection alive | |
| await websocket.receive_text() | |
| except WebSocketDisconnect: | |
| manager.disconnect(websocket) | |
| async def broadcast_state(): | |
| """Broadcast current portfolio state, positions and settings to all websockets.""" | |
| try: | |
| acc_info = alpaca.get_account_info() | |
| positions = alpaca.get_active_positions() | |
| await manager.broadcast({ | |
| "type": "status_update", | |
| "data": { | |
| "account": acc_info, | |
| "positions": positions, | |
| "pending_proposal": state["pending_proposal"] | |
| } | |
| }) | |
| except Exception as e: | |
| logger.error(f"Error broadcasting state: {e}") | |
| # Background Scanner Loop | |
| async def scan_market_loop(): | |
| logger.info("Starting agent scanner background loop...") | |
| while True: | |
| try: | |
| # Run scan cycle | |
| proposals, logs = agent.run_agent_scan(state["tickers"]) | |
| # Send logs to UI | |
| for log in logs: | |
| await log_and_broadcast(log) | |
| await asyncio.sleep(0.1) # brief pause to simulate real-time printing | |
| if proposals: | |
| proposal = proposals[0] # Handle first approved proposal | |
| if state["trading_mode"] == "autopilot": | |
| # Autonomous execution | |
| await log_and_broadcast(f"🤖 [Executor] Autopilot mode enabled. Executing proposal autonomously...") | |
| res = alpaca.execute_options_trade( | |
| symbol=proposal["symbol"], | |
| qty=proposal["qty"], | |
| side=proposal["side"], | |
| strategy_name=proposal["strategy"] | |
| ) | |
| if res["success"]: | |
| await log_and_broadcast( | |
| f"✅ [Executor] Order filled: {res['side'].upper()} {res['qty']}x {res['symbol']} " | |
| f"at ${res.get('price', proposal['premium']):.2f}. Order ID: {res['order_id']}" | |
| ) | |
| else: | |
| await log_and_broadcast(f"❌ [Executor] Autopilot order execution failed: {res.get('error')}") | |
| else: | |
| # Copilot mode (Queue for approval) | |
| await log_and_broadcast(f"🔔 [Strategist] Proposed trade added to Copilot Queue. Awaiting approval...") | |
| state["pending_proposal"] = proposal | |
| # Broadcast the trade proposal to trigger UI alerts | |
| await manager.broadcast({ | |
| "type": "proposal", | |
| "data": proposal | |
| }) | |
| # Periodic P&L/Status updates between scans | |
| await broadcast_state() | |
| except Exception as e: | |
| logger.error(f"Error in background scanner cycle: {e}") | |
| # Sleep for specified interval | |
| await asyncio.sleep(state["scan_interval"]) | |
| async def cache_optionable_assets_task(): | |
| logger.info("Background caching of optionable assets started...") | |
| if alpaca.is_mock: | |
| # Default mock popular optionable tickers | |
| state["optionable_assets"] = [ | |
| "AAPL", "MSFT", "NVDA", "AMD", "TSLA", "AMZN", "META", "GOOGL", "NFLX", | |
| "SPY", "QQQ", "IWM", "DIS", "COIN", "PLTR", "JPM", "WMT", "V", "MA", "UNH" | |
| ] | |
| logger.info(f"Loaded {len(state['optionable_assets'])} mock optionable tickers.") | |
| return | |
| try: | |
| # Run synchronous SDK call in executor to avoid blocking the asyncio event loop! | |
| loop = asyncio.get_event_loop() | |
| assets = await loop.run_in_executor(None, alpaca.trading_client.get_all_assets) | |
| opt_symbols = [ | |
| a.symbol for a in assets | |
| if a.tradable and a.attributes and "has_options" in a.attributes | |
| ] | |
| if opt_symbols: | |
| state["optionable_assets"] = sorted(opt_symbols) | |
| logger.info(f"Successfully cached {len(opt_symbols)} optionable assets from Alpaca.") | |
| else: | |
| logger.warning("Alpaca returned no optionable assets. Using defaults.") | |
| except Exception as e: | |
| logger.error(f"Failed to cache Alpaca optionable assets: {e}. Using defaults.") | |
| # Start background loop on startup | |
| async def startup_event(): | |
| asyncio.create_task(cache_optionable_assets_task()) | |
| asyncio.create_task(scan_market_loop()) | |
| # Serve static frontend files | |
| # Front-end folder must contain index.html, style.css, app.js | |
| frontend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "frontend")) | |
| if os.path.exists(frontend_dir): | |
| logger.info(f"Serving static frontend files from: {frontend_dir}") | |
| # Route for fallback/index.html | |
| async def read_index(): | |
| return FileResponse(os.path.join(frontend_dir, "index.html")) | |
| app.mount("/", StaticFiles(directory=frontend_dir), name="static") | |
| else: | |
| logger.error(f"Frontend directory '{frontend_dir}' not found! Backend only mode.") | |