Build a Real-Time Crypto Trading Dashboard with Python, WebSockets, and React

Binance.""" self.client = await AsyncClient.create() self.bsm = BinanceSocketManager(self.client) logger.info(f"Collector initialized for {self.symbol}")

async def run(self, output_queue: asyncio.Queue) -> None:
    """Main loop: listen to trades, compute signals, push to queue."""
    if not self.bsm:
        raise RuntimeError("Collector not initialized")

    trade_socket = self.bsm.trade_socket(self.symbol)
    
    async with trade_socket as socket_cm:
        while True:
            try:
                raw_msg = await socket_cm.recv()
                trade = self._parse_trade(raw_msg)
                
                self.price_history.append(trade['price'])
                signal = self._evaluate_momentum(trade['price'])
                
                if signal:
                    self.signal_buffer.append(signal)
                
                # Push aggregated state to queue
                await output_queue.put({
                    'current_price': trade['price'],
                    'volume': trade['qty'],
                    'active_signals': list(self.signal_buffer)
                })
            except Exception as e:
                logger.error(f"Stream error: {e}")
                # In production, implement reconnection logic here
                await asyncio.sleep(1)

def _parse_trade(self, msg: Dict[str, Any]) -> Dict[str, float]:
    """Extract relevant fields from Binance trade payload."""
    return {
        'price': float(msg['p']),
        'qty': float(msg['q']),
        'timestamp': msg['T']
    }

def _evaluate_momentum(self, current_price: float) -> Dict[str, Any] | None:
    """Calculate deviation from rolling average."""
    if len(self.price_history) < self.window_size:
        return None

    avg_price = sum(self.price_history) / len(self.price_history)
    deviation = ((current_price - avg_price) / avg_price) * 100

    if abs(deviation) > self.threshold_pct:
        return {
            'type': 'MOMENTUM_SPIKE',
            'deviation': deviation,
            'price': current_price,
            'direction': 'BULLISH' if deviation > 0 else 'BEARISH'
        }
    return None

### Step 2: FastAPI Bridge Server

Create `server.py`. This service exposes the collected data via REST endpoints, draining the async queue on each request to ensure the frontend receives the latest state.

```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import asyncio
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="Market Data Bridge", version="1.0.0")

# Configure CORS for frontend access
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],  # Restrict in production
    allow_methods=["GET"],
    allow_headers=["*"],
)

# Shared state and queue
data_queue = asyncio.Queue()
state = {
    "price_history": [],
    "current_price": None,
    "signals": []
}

class MarketSnapshot(BaseModel):
    current_price: Optional[float]
    history: List[float]
    signals: List[dict]

@app.on_event("startup")
async def startup_event():
    """Initialize collector and background task."""
    # In a real app, import and instantiate MarketDataCollector here
    # collector = MarketDataCollector("BTCUSDT")
    # await collector.initialize()
    # asyncio.create_task(collector.run(data_queue))
    pass

@app.get("/api/v1/market-data", response_model=MarketSnapshot)
async def get_market_data() -> MarketSnapshot:
    """Drain queue and return latest state."""
    # Non-blocking drain
    while not data_queue.empty():
        update = await data_queue.get()
        state["current_price"] = update["current_price"]
        state["price_history"].append(update["current_price"])
        state["signals"] = update["active_signals"]
    
    # Keep history bounded
    if len(state["price_history"]) > 100:
        state["price_history"] = state["price_history"][-100:]

    return MarketSnapshot(
        current_price=state["current_price"],
        history=state["price_history"],
        signals=state["signals"]
    )

@app.get("/health")
async def health_check():
    return {"status": "ok"}

Step 3: React Frontend Dashboard

Create TradingDashboard.tsx. This component uses a custom hook to poll the bridge server and renders price data and signals.

import { useState, useEffect, useCallback } from 'react';
import './App.css';

interface Signal {
  type: string;
  deviation: number;
  price: number;
  direction: string;
}

interface MarketState {
  price: number | null;
  history: number[];
  signals: Signal[];
  isLoading: boolean;
  error: string | null;
}

const useMarketStream = (pollIntervalMs: number = 1500) => {
  const [state, setState] = useState<MarketState>({
    price: null,
    history: [],
    signals: [],
    isLoading: true,
    error: null,
  });

  const fetchSnapshot = useCallback(async () => {
    try {
      const response = await fetch('http://localhost:8000/api/v1/market-data');
      if (!response.ok) throw new Error('Network response was not ok');
      const data = await response.json();
      
      setState({
        price: data.current_price,
        history: data.history,
        signals: data.signals,
        isLoading: false,
        error: null,
      });
    } catch (err) {
      setState(prev => ({
        ...prev,
        isLoading: false,
        error: err instanceof Error ? err.message : 'Unknown error',
      }));
    }
  }, []);

  useEffect(() => {
    fetchSnapshot();
    const timer = setInterval(fetchSnapshot, pollIntervalMs);
    return () => clearInterval(timer);
  }, [fetchSnapshot, pollIntervalMs]);

  return state;
};

export const TradingDashboard = () => {
  const { price, history, signals, isLoading, error } = useMarketStream(1000);

  if (isLoading) return <div className="loader">Initializing Stream...</div>;
  if (error) return <div className="error">Connection Failed: {error}</div>;

  return (
    <div className="dashboard">
      <header className="header">
        <h1>Live Asset Monitor</h1>
        <div className="price-card">
          <span className="label">BTCUSDT</span>
          <span className="value">
            {price ? `$${price.toFixed(2)}` : '---'}
          </span>
        </div>
      </header>

      <section className="signals-panel">
        <h2>Signal Alerts</h2>
        {signals.length === 0 ? (
          <p className="no-signals">No active signals</p>
        ) : (
          <ul className="signal-list">
            {signals.map((sig, idx) => (
              <li 
                key={idx} 
                className={`signal-item ${sig.direction.toLowerCase()}`}
              >
                <span className="signal-type">{sig.type}</span>
                <span className="signal-deviation">
                  {sig.deviation > 0 ? '+' : ''}{sig.deviation.toFixed(2)}%
                </span>
              </li>
            ))}
          </ul>
        )}
      </section>
    </div>
  );
};

Pitfall Guide

1. WebSocket Disconnection Hell

Explanation: Network instability causes WebSocket connections to drop silently. Without reconnection logic, the pipeline stops ingesting data, and the frontend serves stale information. Fix: Implement exponential backoff with jitter in the collector loop. Monitor connection state and trigger re-initialization on failure.

2. Blocking the Event Loop

Explanation: Performing heavy computation or synchronous I/O inside the async WebSocket loop blocks other tasks, causing latency spikes and potential message loss. Fix: Keep the ingestion loop lightweight. Offload complex analysis to separate worker tasks or use asyncio.to_thread for CPU-bound operations.

3. Unbounded Memory Growth

Explanation: Appending to lists indefinitely during high-frequency trading leads to memory exhaustion and eventual process crash. Fix: Use collections.deque with maxlen for all buffers. In the FastAPI state, periodically trim history arrays.

4. CORS Misconfiguration in Production

Explanation: Using allow_origins=["*"] exposes the API to cross-origin requests from any domain, creating security vulnerabilities. Fix: Restrict CORS to specific frontend origins. Use environment variables to configure allowed origins dynamically.

5. Race Conditions in UI State

Explanation: React state updates triggered by rapid polling can cause race conditions, leading to inconsistent UI rendering or stale data display. Fix: Use functional state updates or libraries like Zustand/Redux for state management. Ensure cleanup functions in useEffect cancel pending requests.

6. Latency Blindness

Explanation: Developers often assume "real-time" implies instant delivery. Without measuring timestamps, latency issues go undetected. Fix: Log end-to-end latency by comparing trade timestamps with processing time. Implement metrics collection for pipeline health.

7. Ignoring Rate Limits

Explanation: While WebSockets have fewer rate limits, REST endpoints and API keys still have quotas. Excessive polling or API calls can trigger bans. Fix: Implement client-side rate limiting. Use WebSocket streams for real-time data and REST only for historical snapshots or account data.

Production Bundle

Action Checklist

• Implement Reconnection Logic: Add exponential backoff and jitter to the WebSocket collector.
• Add Structured Logging: Replace print statements with JSON-formatted logs for better observability.
• Containerize Deployment: Use Docker to package the collector and API server for consistent environments.
• Secure API Keys: Store credentials in environment variables; never hardcode them.
• Add Health Checks: Expose /health endpoints for orchestration tools to monitor service status.
• Implement Graceful Shutdown: Handle SIGTERM signals to close connections and flush buffers cleanly.
• Add Metrics Collection: Integrate Prometheus/Grafana to monitor latency, queue depth, and error rates.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Single Symbol MVP	Python + FastAPI + Polling	Rapid development, low infrastructure overhead	Low
Multi-Symbol Scale	Kafka + Go/Rust + WebSocket	High throughput, low latency, fault tolerance	Medium
Enterprise Grade	Managed Data Feed + Database	Reliability, compliance, historical analysis	High
Mobile/Remote Access	WebSocket + Auth Proxy	Secure, real-time updates over unstable networks	Medium

Configuration Template

.env

# API Configuration
BINANCE_API_KEY=your_api_key_here
BINANCE_API_SECRET=your_api_secret_here
ALLOWED_ORIGINS=http://localhost:5173,https://yourdomain.com

# Collector Settings
SYMBOL=BTCUSDT
WINDOW_SIZE=20
THRESHOLD_PCT=0.3

# Server Settings
PORT=8000
LOG_LEVEL=INFO

docker-compose.yml

version: '3.8'
services:
  collector:
    build: ./collector
    env_file: .env
    restart: unless-stopped
    depends_on:
      - api

  api:
    build: ./api
    ports:
      - "8000:8000"
    env_file: .env
    restart: unless-stopped

  frontend:
    build: ./frontend
    ports:
      - "5173:5173"
    depends_on:
      - api

Quick Start Guide

1. Install Dependencies:

   pip install fastapi uvicorn python-binance pydantic
   npm create vite@latest frontend -- --template react-ts
   cd frontend && npm install
   

2. Start Backend:

   uvicorn server:app --reload --port 8000
   

3. Start Frontend:

   npm run dev
   

4. Verify Pipeline: Open http://localhost:5173 and confirm live price updates and signal alerts appear within seconds. Check /health endpoint for service status.