import asyncio import json import logging import datetime from typing import Dict, List, Optional import websockets import plotly.graph_objects as go from plotly.subplots import make_subplots import dash from dash import html, dcc from dash.dependencies import Input, Output import pandas as pd import numpy as np from collections import deque import time from threading import Thread # Configure logging with more detailed format logging.basicConfig( level=logging.DEBUG, # Changed to DEBUG for more detailed logs format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler('realtime_chart.log') ] ) logger = logging.getLogger(__name__) class TradeTickStorage: """Store and manage raw trade ticks for display and candle formation""" def __init__(self, max_age_seconds: int = 300): # 5 minutes by default self.ticks = [] self.max_age_seconds = max_age_seconds logger.info(f"Initialized TradeTickStorage with max age: {max_age_seconds} seconds") def add_tick(self, tick: Dict): """Add a new trade tick to storage""" self.ticks.append(tick) logger.debug(f"Added tick: {tick}, total ticks: {len(self.ticks)}") # Clean up old ticks self._cleanup() def _cleanup(self): """Remove ticks older than max_age_seconds""" now = int(time.time() * 1000) # Current time in ms cutoff = now - (self.max_age_seconds * 1000) old_count = len(self.ticks) self.ticks = [tick for tick in self.ticks if tick['timestamp'] >= cutoff] if old_count > len(self.ticks): logger.debug(f"Cleaned up {old_count - len(self.ticks)} old ticks") def get_ticks_as_df(self) -> pd.DataFrame: """Return ticks as a DataFrame""" if not self.ticks: return pd.DataFrame() df = pd.DataFrame(self.ticks) if not df.empty: df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') return df def get_candles(self, interval_seconds: int = 1) -> pd.DataFrame: """Convert ticks to OHLCV candles at specified interval""" if not self.ticks: return pd.DataFrame() # Ensure ticks are up to date self._cleanup() # Convert to DataFrame df = self.get_ticks_as_df() if df.empty: return pd.DataFrame() # Use timestamp column for resampling df = df.set_index('timestamp') # Create interval string for resampling interval_str = f'{interval_seconds}S' # Resample to create OHLCV candles candles = df.resample(interval_str).agg({ 'price': ['first', 'max', 'min', 'last'], 'volume': 'sum' }) # Flatten MultiIndex columns candles.columns = ['open', 'high', 'low', 'close', 'volume'] # Reset index to get timestamp as column candles = candles.reset_index() logger.debug(f"Generated {len(candles)} candles from {len(self.ticks)} ticks") return candles class CandlestickData: def __init__(self, max_length: int = 300): self.timestamps = deque(maxlen=max_length) self.opens = deque(maxlen=max_length) self.highs = deque(maxlen=max_length) self.lows = deque(maxlen=max_length) self.closes = deque(maxlen=max_length) self.volumes = deque(maxlen=max_length) self.current_candle = { 'timestamp': None, 'open': None, 'high': None, 'low': None, 'close': None, 'volume': 0 } self.candle_interval = 1 # 1 second by default def update_from_trade(self, trade: Dict): timestamp = trade['timestamp'] price = trade['price'] volume = trade.get('volume', 0) # Round timestamp to nearest candle interval candle_timestamp = int(timestamp / (self.candle_interval * 1000)) * (self.candle_interval * 1000) if self.current_candle['timestamp'] != candle_timestamp: # Save current candle if it exists if self.current_candle['timestamp'] is not None: self.timestamps.append(self.current_candle['timestamp']) self.opens.append(self.current_candle['open']) self.highs.append(self.current_candle['high']) self.lows.append(self.current_candle['low']) self.closes.append(self.current_candle['close']) self.volumes.append(self.current_candle['volume']) logger.debug(f"New candle saved: {self.current_candle}") # Start new candle self.current_candle = { 'timestamp': candle_timestamp, 'open': price, 'high': price, 'low': price, 'close': price, 'volume': volume } logger.debug(f"New candle started: {self.current_candle}") else: # Update current candle if self.current_candle['high'] is None or price > self.current_candle['high']: self.current_candle['high'] = price if self.current_candle['low'] is None or price < self.current_candle['low']: self.current_candle['low'] = price self.current_candle['close'] = price self.current_candle['volume'] += volume logger.debug(f"Updated current candle: {self.current_candle}") def get_dataframe(self) -> pd.DataFrame: # Include current candle in the dataframe if it exists timestamps = list(self.timestamps) opens = list(self.opens) highs = list(self.highs) lows = list(self.lows) closes = list(self.closes) volumes = list(self.volumes) if self.current_candle['timestamp'] is not None: timestamps.append(self.current_candle['timestamp']) opens.append(self.current_candle['open']) highs.append(self.current_candle['high']) lows.append(self.current_candle['low']) closes.append(self.current_candle['close']) volumes.append(self.current_candle['volume']) df = pd.DataFrame({ 'timestamp': timestamps, 'open': opens, 'high': highs, 'low': lows, 'close': closes, 'volume': volumes }) if not df.empty: df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') return df class MEXCWebSocket: """MEXC-specific WebSocket implementation""" def __init__(self, symbol: str): self.symbol = symbol.replace('/', '').upper() self.ws = None self.running = False self.reconnect_delay = 1 self.max_reconnect_delay = 60 self.ping_interval = 20 self.last_ping_time = 0 self.message_count = 0 # MEXC WebSocket configuration self.ws_url = "wss://wbs-api.mexc.com/ws" self.ws_sub_params = [ f"spot@public.kline.v3.api@{self.symbol}@Min1" ] self.subscribe_msgs = [ { "method": "SUBSCRIPTION", "params": self.ws_sub_params } ] logger.info(f"Initialized MEXC WebSocket for symbol: {self.symbol}") logger.debug(f"Subscribe messages: {json.dumps(self.subscribe_msgs)}") async def connect(self): while True: try: logger.info(f"Attempting to connect to {self.ws_url}") self.ws = await websockets.connect(self.ws_url) logger.info("WebSocket connection established") # Subscribe to the streams for msg in self.subscribe_msgs: logger.info(f"Sending subscription message: {json.dumps(msg)}") await self.ws.send(json.dumps(msg)) # Wait for subscription confirmation response = await self.ws.recv() logger.info(f"Subscription response: {response}") if "Not Subscribed" in response: logger.error(f"Subscription error: {response}") await self.unsubscribe() await self.close() return False self.running = True self.reconnect_delay = 1 logger.info(f"Successfully connected to MEXC WebSocket for {self.symbol}") # Start ping task asyncio.create_task(self.ping_loop()) return True except Exception as e: logger.error(f"WebSocket connection error: {str(e)}") await self.unsubscribe() await asyncio.sleep(self.reconnect_delay) self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay) continue async def ping_loop(self): """Send ping messages to keep connection alive""" while self.running: try: current_time = time.time() if current_time - self.last_ping_time >= self.ping_interval: ping_msg = {"method": "PING"} logger.debug("Sending ping") await self.ws.send(json.dumps(ping_msg)) self.last_ping_time = current_time await asyncio.sleep(1) except Exception as e: logger.error(f"Error in ping loop: {str(e)}") break async def receive(self) -> Optional[Dict]: if not self.ws: return None try: message = await self.ws.recv() self.message_count += 1 if self.message_count % 10 == 0: logger.info(f"Received message #{self.message_count}") logger.debug(f"Raw message: {message[:200]}...") if isinstance(message, bytes): return None data = json.loads(message) # Handle PONG response if isinstance(data, dict) and data.get('msg') == 'PONG': logger.debug("Received pong") return None # Handle kline data if isinstance(data, dict) and 'data' in data and isinstance(data['data'], list): kline = data['data'][0] if len(kline) >= 6: kline_data = { 'timestamp': int(kline[0]), # Timestamp 'open': float(kline[1]), # Open 'high': float(kline[2]), # High 'low': float(kline[3]), # Low 'price': float(kline[4]), # Close 'volume': float(kline[5]), # Volume 'type': 'kline' } logger.info(f"Processed kline data: {kline_data}") return kline_data return None except websockets.exceptions.ConnectionClosed: logger.warning("WebSocket connection closed") self.running = False return None except json.JSONDecodeError as e: logger.error(f"JSON decode error: {str(e)}, message: {message[:200]}...") return None except Exception as e: logger.error(f"Error receiving message: {str(e)}") return None async def unsubscribe(self): """Unsubscribe from all channels""" if self.ws: for msg in self.subscribe_msgs: unsub_msg = { "method": "UNSUBSCRIPTION", "params": msg["params"] } try: await self.ws.send(json.dumps(unsub_msg)) except: pass async def close(self): """Close the WebSocket connection""" if self.ws: await self.unsubscribe() await self.ws.close() self.running = False logger.info("WebSocket connection closed") class BinanceWebSocket: """Binance WebSocket implementation for real-time tick data""" def __init__(self, symbol: str): self.symbol = symbol.replace('/', '').lower() self.ws = None self.running = False self.reconnect_delay = 1 self.max_reconnect_delay = 60 self.message_count = 0 # Binance WebSocket configuration self.ws_url = f"wss://stream.binance.com:9443/ws/{self.symbol}@trade" logger.info(f"Initialized Binance WebSocket for symbol: {self.symbol}") async def connect(self): while True: try: logger.info(f"Attempting to connect to {self.ws_url}") self.ws = await websockets.connect(self.ws_url) logger.info("WebSocket connection established") self.running = True self.reconnect_delay = 1 logger.info(f"Successfully connected to Binance WebSocket for {self.symbol}") return True except Exception as e: logger.error(f"WebSocket connection error: {str(e)}") await asyncio.sleep(self.reconnect_delay) self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay) continue async def receive(self) -> Optional[Dict]: if not self.ws: return None try: message = await self.ws.recv() self.message_count += 1 if self.message_count % 100 == 0: # Log every 100th message to avoid spam logger.info(f"Received message #{self.message_count}") logger.debug(f"Raw message: {message[:200]}...") data = json.loads(message) # Process trade data if 'e' in data and data['e'] == 'trade': trade_data = { 'timestamp': data['T'], # Trade time 'price': float(data['p']), # Price 'volume': float(data['q']), # Quantity 'type': 'trade' } logger.debug(f"Processed trade data: {trade_data}") return trade_data return None except websockets.exceptions.ConnectionClosed: logger.warning("WebSocket connection closed") self.running = False return None except json.JSONDecodeError as e: logger.error(f"JSON decode error: {str(e)}, message: {message[:200]}...") return None except Exception as e: logger.error(f"Error receiving message: {str(e)}") return None async def close(self): """Close the WebSocket connection""" if self.ws: await self.ws.close() self.running = False logger.info("WebSocket connection closed") class ExchangeWebSocket: """Generic WebSocket interface for cryptocurrency exchanges""" def __init__(self, symbol: str, exchange: str = "binance"): self.symbol = symbol self.exchange = exchange.lower() self.ws = None # Initialize the appropriate WebSocket implementation if self.exchange == "binance": self.ws = BinanceWebSocket(symbol) elif self.exchange == "mexc": self.ws = MEXCWebSocket(symbol) else: raise ValueError(f"Unsupported exchange: {exchange}") async def connect(self): """Connect to the exchange WebSocket""" return await self.ws.connect() async def receive(self) -> Optional[Dict]: """Receive data from the WebSocket""" return await self.ws.receive() async def close(self): """Close the WebSocket connection""" await self.ws.close() @property def running(self): """Check if the WebSocket is running""" return self.ws.running if self.ws else False class RealTimeChart: def __init__(self, symbol: str): self.symbol = symbol self.app = dash.Dash(__name__) self.candlestick_data = CandlestickData() self.tick_storage = TradeTickStorage(max_age_seconds=300) # Store 5 minutes of ticks logger.info(f"Initializing RealTimeChart for {symbol}") # Initialize the layout with improved styling self.app.layout = html.Div([ html.H1(f"{symbol} Real-Time Price", style={ 'textAlign': 'center', 'color': '#2c3e50', 'fontFamily': 'Arial, sans-serif', 'marginTop': '20px' }), html.Div([ html.Button('1s', id='btn-1s', n_clicks=0, style={'margin': '5px'}), html.Button('5s', id='btn-5s', n_clicks=0, style={'margin': '5px'}), html.Button('15s', id='btn-15s', n_clicks=0, style={'margin': '5px'}), html.Button('30s', id='btn-30s', n_clicks=0, style={'margin': '5px'}), html.Button('1m', id='btn-1m', n_clicks=0, style={'margin': '5px'}), ], style={'textAlign': 'center', 'margin': '10px'}), dcc.Store(id='interval-store', data={'interval': 1}), # Store for current interval dcc.Graph( id='live-chart', style={'height': '80vh'} ), dcc.Interval( id='interval-component', interval=500, # Update every 500ms for smoother display n_intervals=0 ) ], style={ 'backgroundColor': '#f8f9fa', 'padding': '20px' }) # Callback to update interval based on button clicks @self.app.callback( Output('interval-store', 'data'), [Input('btn-1s', 'n_clicks'), Input('btn-5s', 'n_clicks'), Input('btn-15s', 'n_clicks'), Input('btn-30s', 'n_clicks'), Input('btn-1m', 'n_clicks')], [dash.dependencies.State('interval-store', 'data')] ) def update_interval(n1, n5, n15, n30, n60, data): ctx = dash.callback_context if not ctx.triggered: return data button_id = ctx.triggered[0]['prop_id'].split('.')[0] if button_id == 'btn-1s': return {'interval': 1} elif button_id == 'btn-5s': return {'interval': 5} elif button_id == 'btn-15s': return {'interval': 15} elif button_id == 'btn-30s': return {'interval': 30} elif button_id == 'btn-1m': return {'interval': 60} return data # Callback to update the chart @self.app.callback( Output('live-chart', 'figure'), [Input('interval-component', 'n_intervals'), Input('interval-store', 'data')] ) def update_chart(n, interval_data): interval = interval_data.get('interval', 1) fig = make_subplots( rows=2, cols=1, shared_xaxis=True, vertical_spacing=0.03, subplot_titles=(f'{self.symbol} Price ({interval}s)', 'Volume'), row_heights=[0.7, 0.3] ) # Get candlesticks from tick storage df = self.tick_storage.get_candles(interval_seconds=interval) if not df.empty: # Add candlestick chart fig.add_trace( go.Candlestick( x=df['timestamp'], open=df['open'], high=df['high'], low=df['low'], close=df['close'], name='Price', increasing_line_color='#33CC33', # Green decreasing_line_color='#FF4136' # Red ), row=1, col=1 ) # Add volume bars colors = ['#33CC33' if close >= open else '#FF4136' for close, open in zip(df['close'], df['open'])] fig.add_trace( go.Bar( x=df['timestamp'], y=df['volume'], name='Volume', marker_color=colors ), row=2, col=1 ) # Add latest price line if len(df) > 0: latest_price = df['close'].iloc[-1] fig.add_shape( type="line", x0=df['timestamp'].min(), y0=latest_price, x1=df['timestamp'].max(), y1=latest_price, line=dict(color="yellow", width=1, dash="dash"), row=1, col=1 ) # Add price label fig.add_annotation( x=df['timestamp'].max(), y=latest_price, text=f"{latest_price:.2f}", showarrow=False, font=dict(size=14, color="yellow"), xshift=50, row=1, col=1 ) # Update layout with improved styling fig.update_layout( title_text=f"{self.symbol} Real-Time Data ({interval}s candles)", title_x=0.5, # Center the title xaxis_rangeslider_visible=False, height=800, template='plotly_dark', paper_bgcolor='rgba(0,0,0,0)', plot_bgcolor='rgba(0,0,0,0)', font=dict(family="Arial, sans-serif", size=12, color="#2c3e50"), showlegend=True, legend=dict( yanchor="top", y=0.99, xanchor="left", x=0.01 ) ) # Update axes styling fig.update_xaxes(showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)') fig.update_yaxes(showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)') return fig async def start_websocket(self): ws = ExchangeWebSocket(self.symbol) while True: # Keep trying to maintain connection if not await ws.connect(): logger.error(f"Failed to connect to exchange for {self.symbol}") await asyncio.sleep(5) continue try: while True: if not ws.running: logger.warning("WebSocket not running, breaking loop") break data = await ws.receive() if data: if data.get('type') == 'kline': # Use kline data directly for candlestick trade_data = { 'timestamp': data['timestamp'], 'price': data['price'], 'volume': data['volume'], 'open': data['open'], 'high': data['high'], 'low': data['low'] } else: # Use trade data trade_data = { 'timestamp': data['timestamp'], 'price': data['price'], 'volume': data['volume'] } logger.debug(f"Updating candlestick with data: {trade_data}") # Store raw tick in the tick storage self.tick_storage.add_tick(trade_data) # Also update the old candlestick data for backward compatibility self.candlestick_data.update_from_trade(trade_data) await asyncio.sleep(0.01) except Exception as e: logger.error(f"Error in WebSocket loop: {str(e)}") finally: await ws.close() logger.info("Waiting 5 seconds before reconnecting...") await asyncio.sleep(5) def run(self, host='localhost', port=8050): logger.info(f"Starting Dash server on {host}:{port}") self.app.run(debug=False, host=host, port=port) async def main(): symbols = ["ETH/USDT", "BTC/USDT"] logger.info(f"Starting application for symbols: {symbols}") charts = [] websocket_tasks = [] # Create a chart and websocket task for each symbol for symbol in symbols: chart = RealTimeChart(symbol) charts.append(chart) websocket_tasks.append(asyncio.create_task(chart.start_websocket())) # Run Dash in a separate thread to not block the event loop server_threads = [] for i, chart in enumerate(charts): port = 8050 + i # Use different ports for each chart thread = Thread(target=lambda c=chart, p=8050+i: c.run(port=p)) # Fix lambda capture thread.daemon = True thread.start() server_threads.append(thread) try: # Keep the main task running while True: await asyncio.sleep(1) except KeyboardInterrupt: logger.info("Shutting down...") except Exception as e: logger.error(f"Unexpected error: {str(e)}") finally: for task in websocket_tasks: task.cancel() try: await task except asyncio.CancelledError: pass if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("Application terminated by user")