# File: /Users/bpa/Desktop/ForesixaQuant/mobile_gateway/backend/consigliere_server.py # Configuration: Consigliere Mobile Gateway Core Enterprise v5 (Schema Aligned) import argparse import asyncio import hmac import json import sqlite3 from pathlib import Path from aiohttp import web, WSMsgType MAX_REQUEST_BYTES = 16384 class HardenedConsigliereGateway: def __init__(self, root_path, token): self.root_path = Path(root_path) self.token = token.encode("utf-8") self.db_path = self.root_path / "trading_live.db" self.access_json_path = self.root_path / "mobile_access.json" self.connected_sockets = set() def verify_token_constant_time(self, inbound_token_str): return hmac.compare_digest(inbound_token_str.encode("utf-8"), self.token) def fetch_aligned_ledger_rows(self): """Queries all 9 columns from the live table to match the production schema model.""" db_uri = f"file:{self.db_path}?mode=ro&uri=true" conn = sqlite3.connect(db_uri, uri=True) cursor = conn.cursor() # Querying the exact column order: id, timestamp, symbol, side, qty, price, allocation_spent, status, alpaca_order_id cursor.execute("SELECT id, timestamp, symbol, side, qty, price, allocation_spent, status, alpaca_order_id FROM corporate_trade_ledger ORDER BY id DESC LIMIT 5") trades = cursor.fetchall() conn.close() return [ { "id": t[0], "timestamp": t[1], "symbol": t[2], "side": t[3], "qty": str(t[4]), "price": str(t[5]), "allocation_spent": str(t[6]), "status": t[7], "alpaca_order_id": t[8] } for t in trades ] async def handle_snapshot_lookup(self, request): auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith("Bearer "): return web.json_response({"error": "Unauthorized"}, status=401) inbound_token = auth_header.split(" ")[1].strip() if not self.verify_token_constant_time(inbound_token): return web.json_response({"error": "Unauthorized"}, status=401) try: trade_matrices = self.fetch_aligned_ledger_rows() return web.json_response({"status": "SNAPSHOT_PROCESSED_SUCCESS", "active_ledgers": trade_matrices}) except Exception as e: return web.json_response({"error": "Database error", "details": str(e)}, status=500) async def handle_websocket_stream(self, request): ws = web.WebSocketResponse(autoclose=True, autoping=True, heartbeat=15.0) await ws.prepare(request) self.connected_sockets.add(ws) print("📶 [WEBSOCKET CONNECTED] Secure phone data stream channel opened.") try: async for msg in ws: if msg.type == WSMsgType.TEXT: data = json.loads(msg.data) if data.get("type") == "auth": inbound_token = data.get("token", "") if self.verify_token_constant_time(inbound_token): await ws.send_str(json.dumps({"type": "auth_result", "status": "success"})) # Fetch and stream the fully aligned 9-column data template instantly trade_matrices = self.fetch_aligned_ledger_rows() await ws.send_str(json.dumps({"type": "snapshot", "active_ledgers": trade_matrices})) if self.access_json_path.exists(): with open(self.access_json_path, "r") as f: access_data = json.load(f) access_data["pairing_active"] = True with open(self.access_json_path, "w") as f: json.dump(access_data, f, indent=2) else: await ws.send_str(json.dumps({"type": "auth_result", "status": "failed"})) await ws.close() elif data.get("type") == "ping": await ws.send_str(json.dumps({"type": "pong"})) finally: if ws in self.connected_sockets: self.connected_sockets.remove(ws) if self.access_json_path.exists(): with open(self.access_json_path, "r") as f: access_data = json.load(f) access_data["pairing_active"] = False with open(self.access_json_path, "w") as f: json.dump(access_data, f, indent=2) print("🛑 [WEBSOCKET DISCONNECTED] Secure phone data stream channel closed.") return ws async def handle_root_health(self, request): try: with open(self.access_json_path, "r") as f: data = json.load(f) pairing_status = data.get("pairing_active", False) except: pairing_status = False return web.json_response({"gateway_status": "ONLINE", "pairing_active": pairing_status}) def setup_routing_matrix(gateway_instance): app = web.Application() async def cors_middleware(app, handler): async def middleware_handler(request): if request.method == "OPTIONS": response = web.Response(status=204) else: response = await handler(request) response.headers["Access-Control-Allow-Origin"] = "https://chatgpt.site" response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS" response.headers["Access-Control-Allow-Headers"] = "Authorization, Content-Type" return response return middleware_handler app.middlewares.append(cors_middleware) app.router.add_get("/", gateway_instance.handle_root_health) app.router.add_get("/api/snapshot", gateway_instance.handle_snapshot_lookup) app.router.add_get("/api/stream", gateway_instance.handle_websocket_stream) return app def main(): p = argparse.ArgumentParser() p.add_argument('--root', type=Path, default=Path('/Users/bpa/Desktop/ForesixaQuant')) p.add_argument('--host', default='127.0.0.1') p.add_argument('--port', type=int, default=8765) args = p.parse_args() with open(args.root / "mobile_access.json", "r") as f: config = json.load(f) active_token = config.get("token") gateway = HardenedConsigliereGateway(args.root, active_token) app = setup_routing_matrix(gateway) print(f"\n============================================================") print(f" CONSIGLIERE MOBILE APP GATEWAY: INITIALIZED SUCCESSFUL ") print(f"============================================================") print(f" [BOOT] Listening for incoming secure syncs at: http://{args.host}:{args.port}") print(f" [DATA] Multi-dimensional mapping target: {args.root} (Read-Only Schema-Aligned)") print(f"============================================================\n") runner = web.AppRunner(app) asyncio.get_event_loop().run_until_complete(runner.setup()) site = web.TCPSite(runner, args.host, args.port) asyncio.get_event_loop().run_until_complete(site.start()) print(" [BOOT] Sockets Bound Successfully. Production Pairing Rails Active.") asyncio.get_event_loop().run_forever() if __name__ == "__main__": main()