shaibu01/Titan-Engine

1"""2TITAN OMNI-CORE: BINANCE PUBLIC EXECUTION NODE (v300.1 ULTIMATE ASYNC PATCH)3TYPE: ZMQ Broker -> Binance Public REST API + Internal Paper Engine4DESCRIPTION: High-Frequency Asynchronous Node. Bypasses ZMQ Slow-Joiner5             by aggressively pinging BARS during boot.6"""7 8import time9import json10import threading11import logging12import zmq13import asyncio14import aiohttp15 16# ==============================================================================17# CONFIGURATION18# ==============================================================================19HUB_IP = "127.0.0.1"20BINANCE_URL = "https://api.binance.com/api/v3"21 22# Map Titan internal symbols to Binance public tickers23ASSET_MAP = {24    "BTCUSD": "BTCUSDT",25    "ETHUSD": "ETHUSDT",26    "SOLUSD": "SOLUSDT",27    "XRPUSD": "XRPUSDT",28    "DOGEUSD": "DOGEUSDT",29    "ADAUSD": "ADAUSDT"30}31 32logging.basicConfig(level=logging.INFO, format='%(asctime)s [CRYPTO-NODE] %(message)s')33logger = logging.getLogger()34 35class CryptoPaperGateway:36    def __init__(self):37        logger.info("Initializing HFT Async Crypto Public API & Local Paper Engine...")38        39        # Internal Paper Trading State40        self.paper_equity = 1000000.041        self.paper_cash = 1000000.042        self.paper_positions = {}43        self.live_prices = {}44        self.state_lock = threading.Lock()45            46        # Setup ZeroMQ Bus47        self.context = zmq.Context()48        49        # Publisher (5557 - Core, Trainer, and UI listen here)50        self.zmq_pub = self.context.socket(zmq.PUB)51        self.zmq_pub.bind(f"tcp://{HUB_IP}:5557") 52        53        # Subscriber (5556 - Core sends commands and telemetry here)54        self.zmq_sub = self.context.socket(zmq.SUB)55        self.zmq_sub.bind(f"tcp://{HUB_IP}:5556")56        self.zmq_sub.setsockopt_string(zmq.SUBSCRIBE, "")57        58        logger.info("🚀 ZMQ Cloud Bridge Armed. Async Data pipelines open.")59 60    async def fetch_bars_for_symbol(self, session, titan_sym, binance_sym):61        """Asynchronous fetch for a single asset's historical BARS."""62        url = f"{BINANCE_URL}/klines?symbol={binance_sym}&interval=15m&limit=150"63        try:64            async with session.get(url, timeout=5) as response:65                if response.status == 200:66                    data = await response.json()67                    return titan_sym, {68                        "o": [float(k[1]) for k in data],69                        "h": [float(k[2]) for k in data],70                        "l": [float(k[3]) for k in data],71                        "c": [float(k[4]) for k in data],72                        "v": [float(k[5]) for k in data]73                    }74                return titan_sym, None75        except Exception as e:76            logger.debug(f"Async BARS Fetch Error for {binance_sym}: {e}")77            return titan_sym, None78 79    async def async_fetch_all_bars(self):80        """Fires all HTTP requests concurrently for microsecond latency."""81        async with aiohttp.ClientSession() as session:82            tasks = [self.fetch_bars_for_symbol(session, ts, bs) for ts, bs in ASSET_MAP.items()]83            results = await asyncio.gather(*tasks)84            85            bars_payload = {}86            for titan_sym, data in results:87                if data:88                    bars_payload[titan_sym] = data89            return bars_payload90 91    def fetch_and_broadcast_bars_loop(self):92        """The threaded loop that manages the async BARS event loop."""93        logger.info("📡 Commencing Asynchronous BARS Uplink for Neural Swarm...")94        loop = asyncio.new_event_loop()95        asyncio.set_event_loop(loop)96        97        boot_counter = 0 # Track how many times we've looped98        99        while True:100            start_t = time.time()101            bars_payload = loop.run_until_complete(self.async_fetch_all_bars())102            103            if bars_payload:104                self.zmq_pub.send_string(f"BARS {json.dumps(bars_payload)}")105                106            fetch_time = (time.time() - start_t) * 1000107            108            # 🚨 THE ZMQ SLOW-JOINER PATCH: Aggressive broadcast during boot109            if boot_counter < 30:110                logger.info(f"⚡ [BOOT PHASE] Async BARS Fetch completed in {fetch_time:.2f}ms. Re-broadcasting instantly...")111                time.sleep(1.0)112                boot_counter += 1113            else:114                # Settle into the normal 15-second rhythm once the system is stable115                if int(time.time()) % 60 == 0:116                    logger.info(f"⚡ Async BARS Fetch completed in {fetch_time:.2f}ms")117                time.sleep(15.0) 118 119    async def fetch_tick_for_symbol(self, session, titan_sym, binance_sym):120        """Asynchronous fetch for a single asset's live Order Book."""121        url = f"{BINANCE_URL}/ticker/bookTicker?symbol={binance_sym}"122        try:123            async with session.get(url, timeout=3) as response:124                if response.status == 200:125                    data = await response.json()126                    bid = float(data['bidPrice'])127                    ask = float(data['askPrice'])128                    mid = (bid + ask) / 2.0129                    return titan_sym, {"price": mid, "bid": bid, "ask": ask, "contract_size": 1.0}130                return titan_sym, None131        except Exception:132            return titan_sym, None133 134    async def async_fetch_all_ticks(self):135        """Fires all TICK requests concurrently."""136        async with aiohttp.ClientSession() as session:137            tasks = [self.fetch_tick_for_symbol(session, ts, bs) for ts, bs in ASSET_MAP.items()]138            results = await asyncio.gather(*tasks)139            140            tick_payload = {}141            for titan_sym, data in results:142                if data:143                    tick_payload[titan_sym] = data144                    with self.state_lock:145                        self.live_prices[titan_sym] = data["price"]146            return tick_payload147 148    def fetch_and_broadcast_ticks_loop(self):149        """The threaded loop that manages the async TICKS event loop."""150        loop = asyncio.new_event_loop()151        asyncio.set_event_loop(loop)152        153        while True:154            tick_payload = loop.run_until_complete(self.async_fetch_all_ticks())155            if tick_payload:156                self.zmq_pub.send_string(f"TICK {json.dumps(tick_payload)}")157            time.sleep(1.5) # Ultra-fast 1.5s tick updates for the UI158 159    def calculate_paper_telemetry(self):160        """Calculates simulated PnL and broadcasts to the Omni-Core."""161        while True:162            with self.state_lock:163                unrealized_pnl = 0.0164                positions_payload = {}165                166                for sym, pos in self.paper_positions.items():167                    current_price = self.live_prices.get(sym, pos['entry'])168                    qty = pos['qty']169                    entry = pos['entry']170                    171                    # Calculate PnL172                    pos_pnl = (current_price - entry) * qty173                    unrealized_pnl += pos_pnl174                    175                    pnl_pct = (current_price - entry) / entry if qty > 0 else (entry - current_price) / entry176                    177                    positions_payload[sym] = {178                        "qty": qty,179                        "entry": entry,180                        "current_price": current_price,181                        "pnl_pct": float(pnl_pct),182                        "strategy": pos['strategy']183                    }184                185                current_equity = self.paper_cash + unrealized_pnl + sum([abs(p['qty']*p['entry']) for p in self.paper_positions.values()])186                187                telem_payload = {188                    "equity": float(current_equity),189                    "margin_free": float(self.paper_cash),190                    "positions": positions_payload191                }192                193            self.zmq_pub.send_string(f"TELEMETRY {json.dumps(telem_payload)}")194            time.sleep(3.0)195 196    def execute_order(self, payload):197        """Internal Paper Trading Match Engine."""198        sym = payload.get("symbol", "")199        side = payload.get("side", "buy").lower() 200        qty = float(payload.get("lot_size", payload.get("qty", 1.0)))201        strategy = payload.get("strategy", "JARVIS_CORE")202 203        logger.info(f"🚀 PAPER EXECUTION: {side.upper()} {qty} {sym} [{strategy}]")204 205        with self.state_lock:206            price = self.live_prices.get(sym)207            if not price:208                logger.error(f"⚠️ Rejecting {sym} - No live price data available.")209                return210                211            notional = qty * price212            213            if side == "buy":214                if self.paper_cash >= notional:215                    self.paper_cash -= notional216                    if sym in self.paper_positions:217                        # Average down218                        old_qty = self.paper_positions[sym]['qty']219                        old_entry = self.paper_positions[sym]['entry']220                        new_qty = old_qty + qty221                        new_entry = ((old_qty * old_entry) + notional) / new_qty222                        self.paper_positions[sym]['qty'] = new_qty223                        self.paper_positions[sym]['entry'] = new_entry224                    else:225                        self.paper_positions[sym] = {'qty': qty, 'entry': price, 'strategy': strategy}226                else:227                    logger.error(f"⚠️ Insufficient Paper Cash for {sym} Buy.")228                    return229            230            elif side == "sell":231                if sym in self.paper_positions:232                    pos = self.paper_positions[sym]233                    if pos['qty'] >= qty:234                        # Close out position, realize PnL235                        realized_pnl = (price - pos['entry']) * qty236                        self.paper_cash += (notional + realized_pnl)237                        pos['qty'] -= qty238                        239                        if pos['qty'] <= 0.0001: 240                            del self.paper_positions[sym]241                    else:242                        logger.error(f"⚠️ Attempted to sell more {sym} than owned.")243                        return244                        245        logger.info(f"✅ PAPER FILL CONFIRMED: {sym} @ ${price:.2f}")246        247        confirm_payload = {248            "time": time.strftime("%H:%M:%S"),249            "action": side.upper(), 250            "symbol": sym,251            "qty": qty, 252            "strategy": strategy, 253            "pnl": "0.0%"254        }255        self.zmq_pub.send_string(f"CONFIRM {json.dumps(confirm_payload)}")256 257    def listen_for_executions(self):258        logger.info("🎧 Listening for Omni-Core EXECUTE commands on tcp://127.0.0.1:5556...")259        while True:260            try:261                msg = self.zmq_sub.recv_string()262                263                # ZMQ BROKER LOGIC: Instantly forward EVERYTHING received on 5556 264                # out to 5557 so the Streamlit UI can receive CLOUD_TELEM265                self.zmq_pub.send_string(msg)266                267                topic, data_str = msg.split(" ", 1)268                if topic == "EXECUTE":269                    payload = json.loads(data_str)270                    threading.Thread(target=self.execute_order, args=(payload,), daemon=True).start()271            except Exception:272                time.sleep(0.01)273 274if __name__ == "__main__":275    gateway = CryptoPaperGateway()276    277    threading.Thread(target=gateway.fetch_and_broadcast_bars_loop, daemon=True).start()278    threading.Thread(target=gateway.fetch_and_broadcast_ticks_loop, daemon=True).start()279    threading.Thread(target=gateway.calculate_paper_telemetry, daemon=True).start()280    281    gateway.listen_for_executions()