import os import json import uuid import datetime from fastapi import FastAPI, WebSocket, WebSocketDisconnect import psycopg2 from psycopg2 import pool # Función auxiliar para cargar variables desde .env def load_dotenv(): env_path = os.path.join(os.path.dirname(__file__), ".env") if os.path.exists(env_path): with open(env_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: key, value = line.split("=", 1) os.environ[key.strip()] = value.strip() # Cargar variables de entorno load_dotenv() app = FastAPI( title="Cloud Trade Copier V2 Backend", description="Backend asíncrono WebSocket para el Copiador Cloud V2 en tiempo real", version="2.0.0" ) # Inicializar Pool de conexiones de base de datos db_pool = None try: db_pool = psycopg2.pool.ThreadedConnectionPool( minconn=1, maxconn=20, host=os.getenv("DB_HOST", "localhost"), port=int(os.getenv("DB_PORT", "5432")), database=os.getenv("DB_NAME", "metacopia"), user=os.getenv("DB_USER", "postgres"), password=os.getenv("DB_PASSWORD", "postgres") ) print("Database connection pool initialized successfully") except Exception as e: print(f"WARNING: Error initializing database connection pool: {e}") # Registro de conexiones WebSockets activas por nivel de riesgo (1, 2, 3) active_followers = {1: set(), 2: set(), 3: set()} active_masters = {1: set(), 2: set(), 3: set()} def validate_follower(account_number: int): """Valida la licencia de la cuenta seguidora en la base de datos.""" if not db_pool: # Fallback de desarrollo si la base de datos no está activa if account_number == 107674535: #661023 return True, 2, None return False, None, "Database pool not initialized. Please register account 661023 for offline test." conn = None try: conn = db_pool.getconn() with conn.cursor() as cur: cur.execute( "SELECT is_active, expiration_date, assigned_leader FROM followers WHERE account_number = %s", (account_number,) ) row = cur.fetchone() if not row: return False, None, f"La cuenta {account_number} no está registrada" is_active, expiration_date, assigned_leader = row if not is_active: return False, None, "La cuenta está inactiva" if expiration_date < datetime.datetime.now(): return False, None, f"Licencia expirada el {expiration_date.strftime('%Y-%m-%d %H:%M:%S')}" if assigned_leader not in [1, 2, 3]: return False, None, "Nivel de líder inválido en la base de datos" return True, assigned_leader, None except Exception as e: print(f"Error DB en validate_follower: {e}") return False, None, f"Error de base de datos: {str(e)}" finally: if conn: db_pool.putconn(conn) def db_insert_pending_trade(trade_id: str, master_ticket: int, follower_account: int, symbol: str): """Inserta una operación copiada con estado inicial PENDING en la base de datos.""" if not db_pool: return conn = None try: conn = db_pool.getconn() with conn.cursor() as cur: cur.execute( """ INSERT INTO copied_trades (id, master_ticket, follower_account, symbol, status) VALUES (%s, %s, %s, %s, %s) """, (trade_id, master_ticket, follower_account, symbol, "PENDING") ) conn.commit() except Exception as e: print(f"Error DB en db_insert_pending_trade: {e}") finally: if conn: db_pool.putconn(conn) def db_update_copied_trade(trade_id: str, follower_ticket: int, status: str, latency_ms: int): """Actualiza los resultados y latencias del trade copiado.""" if not db_pool: return conn = None try: conn = db_pool.getconn() with conn.cursor() as cur: # Si el ticket es 0 o None, lo dejamos como NULL en la DB f_ticket = follower_ticket if follower_ticket and follower_ticket > 0 else None cur.execute( """ UPDATE copied_trades SET follower_ticket = %s, status = %s, execution_latency_ms = %s WHERE id = %s """, (f_ticket, status, latency_ms, trade_id) ) conn.commit() except Exception as e: print(f"Error DB en db_update_copied_trade: {e}") finally: if conn: db_pool.putconn(conn) @app.get("/") def read_root(): return { "status": "online", "service": "Cloud Trade Copier V2 Backend", "active_connections": { "level_1": {"masters": len(active_masters[1]), "followers": len(active_followers[1])}, "level_2": {"masters": len(active_masters[2]), "followers": len(active_followers[2])}, "level_3": {"masters": len(active_masters[3]), "followers": len(active_followers[3])} } } @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() client_info = None try: while True: data = await websocket.receive_text() try: message = json.loads(data) except json.JSONDecodeError: await websocket.send_text(json.dumps({"status": "error", "message": "Formato JSON inválido"})) continue msg_type = message.get("type") # --- APRETÓN DE MANOS (HANDSHAKE) CLIENTE / SEGUIDOR --- if msg_type == "handshake": account_number_raw = message.get("account_number") if account_number_raw is None: await websocket.send_text(json.dumps({"status": "rejected", "message": "Falta account_number"})) await websocket.close() return try: account_number = int(account_number_raw) except ValueError: await websocket.send_text(json.dumps({"status": "rejected", "message": "account_number inválido"})) await websocket.close() return # Validar licencia contra PostgreSQL valid, assigned_leader, error_msg = validate_follower(account_number) if not valid: await websocket.send_text(json.dumps({"status": "rejected", "message": error_msg})) await websocket.close() return # Registrar cliente client_info = { "role": "follower", "account_number": account_number, "leader_level": assigned_leader } # Guardamos como atributo dinámico del websocket websocket.follower_account = account_number active_followers[assigned_leader].add(websocket) await websocket.send_text(json.dumps({ "status": "approved", "leader_level": assigned_leader, "message": "Handshake aprobado exitosamente" })) print(f"[Follower Connected] Cuenta: {account_number} registrado en nivel {assigned_leader}") # --- REGISTRO DEL EA MAESTRO --- elif msg_type == "register_master": leader_level_raw = message.get("leader_level") if leader_level_raw is None: await websocket.send_text(json.dumps({"status": "error", "message": "Falta leader_level"})) await websocket.close() return try: leader_level = int(leader_level_raw) except ValueError: await websocket.send_text(json.dumps({"status": "error", "message": "leader_level debe ser entero"})) await websocket.close() return if leader_level not in [1, 2, 3]: await websocket.send_text(json.dumps({"status": "error", "message": "leader_level debe ser 1, 2 o 3"})) await websocket.close() return client_info = { "role": "master", "leader_level": leader_level } active_masters[leader_level].add(websocket) await websocket.send_text(json.dumps({ "status": "approved", "leader_level": leader_level, "message": f"Master registrado en el nivel {leader_level}" })) print(f"[Master Connected] Registrado en nivel {leader_level}") # --- RETRANSMISIÓN DE SEÑALES DESDE EL MAESTRO --- elif msg_type == "trade_signal": if not client_info or client_info.get("role") != "master": await websocket.send_text(json.dumps({"status": "error", "message": "No autorizado: solo Masters registrados"})) continue leader_level = client_info["leader_level"] master_ticket = message.get("master_ticket") action = message.get("action") # "open", "close" o "modify" symbol = message.get("symbol") cmd = message.get("cmd") # "buy" o "sell" lots = message.get("lots") entry_price = message.get("entry_price") sl = message.get("sl", 0.0) tp = message.get("tp", 0.0) # Obtener seguidores activos de este nivel followers = active_followers[leader_level] if not followers: print(f"[Signal Route] Señal del maestro de nivel {leader_level} recibida, pero sin seguidores activos.") continue print(f"[Signal Route] Retransmitiendo señal del maestro (Nivel {leader_level}) para {symbol} ({action}) a {len(followers)} seguidores...") # Broadcast asíncrono for f_ws in list(followers): f_account = getattr(f_ws, "follower_account", None) if not f_account: continue trade_uuid = str(uuid.uuid4()) # Registrar como PENDING en la base de datos (solo para aperturas o cierres, las modificaciones de SL/TP son de gestión de límites) if action in ["open", "close"]: db_insert_pending_trade(trade_uuid, master_ticket, f_account, symbol) # Enviar la señal al cliente try: await f_ws.send_text(json.dumps({ "type": "trade_signal", "trade_id": trade_uuid, "master_ticket": master_ticket, "action": action, "symbol": symbol, "cmd": cmd, "lots": lots, "entry_price": entry_price, "sl": sl, "tp": tp })) except Exception as e: print(f"Error enviando señal al seguidor {f_account}: {e}") active_followers[leader_level].discard(f_ws) # --- CONFIRMACIÓN DE OPERACIONES DESDE EL CLIENTE --- elif msg_type == "trade_confirm": if not client_info or client_info.get("role") != "follower": await websocket.send_text(json.dumps({"status": "error", "message": "No autorizado: solo Followers registrados"})) continue trade_id = message.get("trade_id") follower_ticket = message.get("follower_ticket") status = message.get("status") # EXECUTED, SLIPPAGE_REJECTED, FAILED latency_ms = message.get("execution_latency_ms") if not trade_id or not status: await websocket.send_text(json.dumps({"status": "error", "message": "trade_id y status son requeridos"})) continue # Actualizar auditoría en PostgreSQL db_update_copied_trade(trade_id, follower_ticket, status, latency_ms) print(f"[Signal Confirm] Cuenta {client_info['account_number']}: Trade {trade_id} -> {status} ({latency_ms}ms)") except WebSocketDisconnect: if client_info: role = client_info["role"] leader_level = client_info["leader_level"] if role == "follower": active_followers[leader_level].discard(websocket) print(f"[Follower Disconnected] Cuenta: {client_info['account_number']}") elif role == "master": active_masters[leader_level].discard(websocket) print(f"[Master Disconnected] Nivel: {leader_level}") else: print("[Client Disconnected] Desconectado antes del handshake")