""" Real Market Data Provider for TheStarFX. Connects to Binance public WebSocket streams for real-time crypto and forex data. Falls back to simulation for symbols not available on Binance. """ import asyncio import json import logging import random from datetime import datetime, timezone from collections import defaultdict logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Symbol mappings to Binance # --------------------------------------------------------------------------- SYMBOL_MAPPINGS = { # Forex "EUR/USD": {"binance": "EURUSDT", "channel": "forex:majors", "stype": "forex", "dec": 5, "spread": 0.000020}, "GBP/USD": {"binance": "GBPUSDT", "channel": "forex:majors", "stype": "forex", "dec": 5, "spread": 0.000025}, "USD/JPY": {"binance": "JPYUSDT", "channel": "forex:majors", "stype": "forex-jpy", "dec": 3, "spread": 0.0020}, "USD/CHF": {"binance": None, "channel": "forex:majors", "stype": "forex", "dec": 5, "spread": 0.000022, "sim_price": 0.90118, "vol": 0.000070}, "AUD/USD": {"binance": "AUDUSDT", "channel": "forex:majors", "stype": "forex", "dec": 5, "spread": 0.000018}, "USD/CAD": {"binance": None, "channel": "forex:majors", "stype": "forex", "dec": 5, "spread": 0.000020, "sim_price": 1.36452, "vol": 0.000070}, "NZD/USD": {"binance": "NZDUSDT", "channel": "forex:majors", "stype": "forex", "dec": 5, "spread": 0.000018}, "EUR/GBP": {"binance": "EURGBP", "channel": "forex:minors", "stype": "forex", "dec": 5, "spread": 0.000025}, "EUR/JPY": {"binance": "EURJPY", "channel": "forex:minors", "stype": "forex-jpy", "dec": 3, "spread": 0.0030}, "GBP/JPY": {"binance": "GBPJPY", "channel": "forex:minors", "stype": "forex-jpy", "dec": 3, "spread": 0.0035}, "XAU/USD": {"binance": "XAUUSDT", "channel": "forex:majors", "stype": "commodity", "dec": 2, "spread": 0.25}, # Crypto "BTC/USD": {"binance": "BTCUSDT", "channel": "crypto:leaders", "stype": "crypto", "dec": 2, "spread": 8.0}, "ETH/USD": {"binance": "ETHUSDT", "channel": "crypto:leaders", "stype": "crypto", "dec": 2, "spread": 0.80}, "SOL/USD": {"binance": "SOLUSDT", "channel": "crypto:leaders", "stype": "crypto", "dec": 3, "spread": 0.05}, "BNB/USD": {"binance": "BNBUSDT", "channel": "crypto:leaders", "stype": "crypto", "dec": 3, "spread": 0.15}, "XRP/USD": {"binance": "XRPUSDT", "channel": "crypto:leaders", "stype": "crypto", "dec": 5, "spread": 0.0002}, "ADA/USD": {"binance": "ADAUSDT", "channel": "crypto:leaders", "stype": "crypto", "dec": 5, "spread": 0.0002}, "DOGE/USD": {"binance": "DOGEUSDT", "channel": "crypto:leaders", "stype": "crypto", "dec": 5, "spread": 0.0001}, } # --------------------------------------------------------------------------- # Candle state — tracks running OHLC per symbol for 1-minute candles # --------------------------------------------------------------------------- class CandleState: def __init__(self, price: float): self.open = price self.high = price self.low = price self.close = price self.volume = 0.0 self.ts = _minute_floor() def update(self, price: float, volume: float = 0.0): self.high = max(self.high, price) self.low = min(self.low, price) self.close = price self.volume += volume def to_dict(self, symbol: str) -> dict: return { "type": "candle", "symbol": symbol, "interval": "1m", "open": self.open, "high": self.high, "low": self.low, "close": self.close, "volume": round(self.volume, 4), "timestamp": self.ts, } def _minute_floor() -> str: """Return current UTC time floored to the minute as ISO string.""" now = datetime.now(timezone.utc) floored = now.replace(second=0, microsecond=0) return floored.isoformat() def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() async def binance_ticker_stream(queue): """Connect to Binance's mini ticker stream and push updates to queue.""" import websockets binance_symbols = [ data["binance"].lower() for data in SYMBOL_MAPPINGS.values() if data["binance"] is not None ] if not binance_symbols: return stream_url = f"wss://stream.binance.com:9443/ws/{'/'.join([f'{sym}@ticker' for sym in binance_symbols])}" logger.info(f"[Binance] Subscribing to symbols: {', '.join(binance_symbols)}") while True: try: logger.info(f"[Binance] Connecting to {stream_url}") async with websockets.connect(stream_url) as websocket: logger.info("[Binance] Connected successfully") async for message in websocket: data = json.loads(message) symbol = data.get("s") logger.debug(f"[Binance] Received update for symbol: {symbol}") await queue.put(data) except Exception as e: logger.error(f"[Binance] Stream error: {e}") logger.info("[Binance] Reconnecting in 5 seconds...") await asyncio.sleep(5) async def market_data_loop(): """Main market data loop that processes Binance updates and broadcasts.""" from channels.layers import get_channel_layer channel_layer = get_channel_layer() if channel_layer is None: logger.error("[Market] Channel layer is None — cannot broadcast. Check CHANNEL_LAYERS in settings.") return # Queue to receive data from Binance stream data_queue = asyncio.Queue() # Start Binance stream in background binance_task = asyncio.create_task(binance_ticker_stream(data_queue)) # State tracking open_prices = {} candles = {} last_prices = {} last_minute = _minute_floor() tick = 0 logger.info("[Market] Market data provider started.") logger.info(f"[Market] Tracking {len(SYMBOL_MAPPINGS)} pairs: {', '.join(SYMBOL_MAPPINGS.keys())}") while True: try: # Process Binance data if available try: binance_data = await asyncio.wait_for(data_queue.get(), timeout=0.1) symbol = None binance_symbol = binance_data.get("s") # Find our symbol from Binance ticker for our_sym, data in SYMBOL_MAPPINGS.items(): if data["binance"] and data["binance"].upper() == binance_symbol: symbol = our_sym break if symbol: logger.debug(f"[Market] Processing update for {symbol} (Binance: {binance_symbol})") sym_data = SYMBOL_MAPPINGS[symbol] price = float(binance_data["c"]) # Last trade price volume = float(binance_data["v"]) # 24h volume last_prices[symbol] = price if symbol not in open_prices: open_prices[symbol] = price candles[symbol] = CandleState(price) logger.info(f"[Market] First update for {symbol}, initial price: {price}") candles[symbol].update(price, volume) else: logger.debug(f"[Market] No mapping for Binance symbol {binance_symbol}") except asyncio.TimeoutError: pass # No data, just continue # Simulate prices for symbols with sim_price for symbol, sym_data in SYMBOL_MAPPINGS.items(): if sym_data["binance"] is None and "sim_price" in sym_data: if symbol not in last_prices: last_prices[symbol] = sym_data["sim_price"] open_prices[symbol] = last_prices[symbol] candles[symbol] = CandleState(last_prices[symbol]) logger.info(f"[Market] Starting simulation for {symbol} at price: {last_prices[symbol]}") # Gaussian random walk change = random.gauss(0, sym_data["vol"]) last_prices[symbol] = max(0.00001, last_prices[symbol] + change) candles[symbol].update(last_prices[symbol], abs(random.gauss(0.5, 0.3))) tick += 1 now_iso = _now_iso() current_minute = _minute_floor() # Minute boundary: emit completed candle, reset if current_minute != last_minute: for symbol, candle in candles.items(): msg = candle.to_dict(symbol) try: await channel_layer.group_send( "market_stream", {"type": "market.update", "message": msg}, ) except Exception as e: logger.warning(f"[Market] Candle send failed for {symbol}: {e}") for symbol in last_prices: candles[symbol] = CandleState(last_prices[symbol]) open_prices[symbol] = last_prices[symbol] last_minute = current_minute # Broadcast quotes for all symbols we have data for if tick % 2 == 0: # Broadcast every 1 second (0.5 * 2 = 1s) for symbol, price in last_prices.items(): sym_data = SYMBOL_MAPPINGS[symbol] half_spread = sym_data["spread"] / 2 bid = round(price - half_spread, sym_data["dec"]) ask = round(price + half_spread, sym_data["dec"]) mid = round(price, sym_data["dec"]) spread = round(sym_data["spread"], sym_data["dec"]) change_pct = 0.0 if symbol in open_prices and open_prices[symbol] != 0: change_pct = round( (price - open_prices[symbol]) / open_prices[symbol] * 100, 4 ) quote_msg = { "type": "quote", "channel": sym_data["channel"], "symbol": symbol, "bid": bid, "ask": ask, "mid": mid, "change_pct": change_pct, "spread": spread, "timestamp": now_iso, "source": "binance" if sym_data["binance"] else "simulation", } try: await channel_layer.group_send( "market_stream", {"type": "market.update", "message": quote_msg}, ) except Exception as e: logger.warning(f"[Market] Quote send failed for {symbol}: {e}") # Heartbeat every 15 seconds if tick % 30 == 0: try: await channel_layer.group_send( "market_stream", { "type": "market.update", "message": {"type": "heartbeat", "timestamp": now_iso}, }, ) except Exception as e: logger.warning(f"[Market] Heartbeat send failed: {e}") await asyncio.sleep(0.5) # Run loop every 0.5 seconds except Exception as e: logger.error(f"[Market] Error in main loop: {e}") await asyncio.sleep(1) # Backward compatibility: keep the same name market_simulator_loop = market_data_loop