Building a Real-Time Candlestick Chart for Live Crypto Trading Using Python

Introduction
What Is a Candlestick Chart?
Why Real-Time Visualization Matters in Crypto Trading
Streaming Price Data: From Ticks to Candles
Efficient Candle Aggregation in Python
Complete Implementation with Live Simulation
Best Practices for Production Trading Dashboards
Conclusion

Introduction

In the volatile world of cryptocurrency trading, seconds determine profit or loss. Traders don’t just need data—they need instant visual insight into price action. The candlestick chart, with its intuitive display of open, high, low, and close prices, is the gold standard for technical analysis.

But most tutorials show static charts from historical CSV files. Real trading demands live, streaming candlesticks that update as new trades arrive.

This article shows you how to build a real-time candlestick chart engine in pure Python—no external brokers required—using a simulated crypto exchange feed and efficient in-memory aggregation.

What Is a Candlestick Chart?

A candlestick represents price movement over a fixed time interval (e.g., 1 minute). Each candle has four values:

Open: First traded price in the interval
High: Highest price reached
Low: Lowest price reached
Close: Last traded price

Green (or white) candles indicate price rose; red (or black) mean it fell. This visual language lets traders spot trends, reversals, and volatility at a glance.

Why Real-Time Visualization Matters in Crypto Trading

Imagine you’re monitoring SOL/USD during a major protocol upgrade. News breaks, and whales start dumping. Within seconds, the price drops 8%.

A static chart won’t help. But a live candlestick chart:

Shows the bearish red candle forming in real time
Reveals increasing volume and wick rejection
Lets you react before the move completes

For algo traders and manual traders alike, real-time candles are the eyes on the market.

Streaming Price Data: From Ticks to Candles

Crypto exchanges emit trade ticks—individual buy/sell events with timestamp, price, and size. To build candles, we must:

Receive ticks in real time (via WebSocket or simulated stream)
Group them into time buckets (e.g., every 60 seconds)
Compute OHLC for each bucket as it fills

The challenge: do this without missing ticks or blocking the UI.

Efficient Candle Aggregation in Python

We maintain:

Current active candle (mutable)
List of completed candles
Timer aligned to UTC intervals (e.g., minute boundaries)

When a new tick arrives:

If it belongs to the current candle → update OHLC
If it starts a new interval → finalize current candle, start new one

This avoids storing all ticks—only OHLC and timestamp are kept per candle.

Complete Implementation with Live Simulation

import time
import random
import threading
from datetime import datetime, timezone
from collections import deque
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
from matplotlib.dates import DateFormatter, date2num
import numpy as np
import sys

# Global flag to control all simulation and plot threads
TICK_SIM_RUNNING = True 

class RealTimeCandleAggregator:
    """
    Aggregates trade ticks into real-time OHLC candles.
    """
    def __init__(self, interval_sec: int = 10):
        self.interval_sec = interval_sec
        self.candles = deque(maxlen=50)  # Keep last 50 completed candles
        self.current_candle = None
        self._lock = threading.Lock()
        self.last_finalize_time = 0 # To track when the last full candle was finalized

    def _get_candle_start(self, timestamp: float) -> float:
        """Align timestamp to nearest interval boundary."""
        return (int(timestamp) // self.interval_sec) * self.interval_sec

    def add_tick(self, price: float, timestamp: float = None):
        """Add a new trade tick and update candles."""
        if timestamp is None:
            timestamp = time.time()
        
        candle_start = self._get_candle_start(timestamp)
        
        with self._lock:
            # Check if we need to start or finalize a candle
            if self.current_candle is None:
                self.current_candle = self._new_candle_data(candle_start, price)
                
            elif self.current_candle['time'] != candle_start:
                # Finalize current candle and append to deque
                self.candles.append(self.current_candle)
                self.last_finalize_time = self.current_candle['time']
                
                # Start new candle
                self.current_candle = self._new_candle_data(candle_start, price)
            
            else:
                # Update existing candle
                self.current_candle['high'] = max(self.current_candle['high'], price)
                self.current_candle['low'] = min(self.current_candle['low'], price)
                self.current_candle['close'] = price

    def _new_candle_data(self, start_time, price):
        return {
            'time': start_time,
            'open': price,
            'high': price,
            'low': price,
            'close': price
        }

    def get_data(self):
        """Return list of completed candles and the current forming candle."""
        with self._lock:
            completed = list(self.candles)
            current = self.current_candle.copy() if self.current_candle else None
            return completed, current

    def update_interval(self, new_interval: int):
        """Dynamically update the candle interval."""
        if new_interval > 0 and new_interval != self.interval_sec:
            with self._lock:
                print(f"\n[INFO] Changing interval from {self.interval_sec}s to {new_interval}s.")
                self.interval_sec = new_interval
                
                # Resetting is necessary for immediate application
                self.candles.clear()
                self.current_candle = None
                self.last_finalize_time = 0
                print("Aggregator reset to start new interval aggregation.")


# ----------------------------------------------------------------------
## Simulated Crypto Exchange Feed Thread
# ----------------------------------------------------------------------

def simulate_crypto_ticks(aggregator: RealTimeCandleAggregator):
    """Generate realistic SOL/USD price ticks."""
    price = 150.0
    while TICK_SIM_RUNNING:
        # Simulate small random walk with occasional volatility spike
        change = random.gauss(0, 0.3)
        if random.random() < 0.05:  # 5% chance of big move
            change += random.choice([-5, 5])
        price = max(1, price + change)
        
        aggregator.add_tick(price)
        time.sleep(0.1)  # Faster ticks: 10 per second


# ----------------------------------------------------------------------
## User Input Thread for Interval Change
# ----------------------------------------------------------------------

def user_input_handler(aggregator: RealTimeCandleAggregator):
    """Listens for user input to change the candle interval."""
    # Note: When run in a notebook, input() can be finicky and block output.
    # The clean shutdown is the priority here.
    while True:
        try:
            print("\n[INPUT] Enter new interval (sec) or 'exit': ", end="")
            # sys.stdin.flush() # Not always needed, but harmless
            user_input = input()
            
            if user_input.lower() == 'exit':
                global TICK_SIM_RUNNING
                TICK_SIM_RUNNING = False # Signal other threads to stop
                break
            
            new_interval = int(user_input)
            aggregator.update_interval(new_interval)
            
        except ValueError:
            print("[ERROR] Invalid input. Please enter an integer.")
        except EOFError:
            # Handles unexpected interruption in notebook/IDE
            break
        except Exception as e:
            print(f"[ERROR] An unexpected error occurred: {e}")
            break

# ----------------------------------------------------------------------
## Main Execution Block
# ----------------------------------------------------------------------

if __name__ == "__main__":
    initial_interval = 10
    aggregator = RealTimeCandleAggregator(interval_sec=initial_interval)
    
    # Start tick simulation (daemon=True means it dies when main program exits)
    tick_thread = threading.Thread(target=simulate_crypto_ticks, args=(aggregator,), daemon=True)
    
    # Start user input handler (daemon=False so main thread waits for it to exit)
    input_thread = threading.Thread(target=user_input_handler, args=(aggregator,), daemon=False)
    
    # Matplotlib objects need to be defined outside the try/finally if accessed later
    fig = None
    ani = None

    try:
        print(" Starting Real-Time Candlestick Chart for Crypto Trading...")
        print("Chart updates twice per second. Ticks run 10 times per second.")
        
        tick_thread.start()
        input_thread.start()

        # --- Matplotlib Setup ---
        fig, ax = plt.subplots(figsize=(14, 7))
        plt.ion() # Turn on interactive mode
        
        ax.xaxis.set_major_formatter(DateFormatter('%H:%M:%S'))
        ax.set_ylabel("Price (USD)")
        ax.grid(True, linestyle='--', alpha=0.7)
        
        # Text annotation for current candle data (must be defined once)
        current_info_text = ax.text(0.01, 1.05, '', transform=ax.transAxes, fontsize=10, verticalalignment='top')
        
        
        def animate(_):
            """Matplotlib animation update function."""
            global TICK_SIM_RUNNING
            if not TICK_SIM_RUNNING:
                plt.close(fig) # Closes the figure when flag is false
                return

            completed, current = aggregator.get_data()
            all_candles = completed + [current] if current else completed
            
            if not all_candles:
                return
            
            times = [datetime.fromtimestamp(c['time'], tz=timezone.utc) for c in all_candles]
            
            # Clear and redraw the chart
            ax.clear()
            
            # --- Drawing Candlesticks ---
            
            # Calculate width based on interval (handle edge case of 1 candle)
            width = date2num(times[1]) - date2num(times[0]) if len(times) > 1 else 0.00005 
            width *= 0.6
            
            for i, candle in enumerate(all_candles):
                t = times[i]
                o, h, l, c = candle['open'], candle['high'], candle['low'], candle['close']
                t_num = date2num(t) 
                color = 'green' if c >= o else 'red'
                is_current = candle == current
                
                # 1. Wick (High/Low)
                ax.plot([t_num, t_num], [l, h], color='black', linewidth=1, alpha=0.8)
                
                # 2. Body (Open/Close)
                body_bottom = min(o, c)
                body_height = abs(o - c)
                # Use alpha=0.5 for the current (forming) candle to distinguish it
                ax.bar(t_num, body_height, width=width, bottom=body_bottom, color=color, 
                       alpha=0.9 if not is_current else 0.5)
                
            # --- Update Interactive Elements ---
            if current:
                current_time = datetime.fromtimestamp(current['time'], tz=timezone.utc).strftime('%H:%M:%S')
                info = (
                    f"Interval: {aggregator.interval_sec}s | Current Candle @ {current_time}\n"
                    f"O: {current['open']:.2f} | H: {current['high']:.2f} | L: {current['low']:.2f} | C: {current['close']:.2f}"
                )
                # Re-add text (ax.clear() removes it)
                current_info_text = ax.text(0.01, 1.05, info, transform=ax.transAxes, fontsize=10, verticalalignment='top')
            
            # Re-apply labels and formatting (ax.clear() removes these)
            ax.set_title(f" Real-Time SOL/USD Candlestick Chart ({aggregator.interval_sec}-sec intervals)", y=1.02)
            ax.set_ylabel("Price (USD)")
            ax.xaxis.set_major_formatter(DateFormatter('%H:%M:%S'))
            ax.grid(True, linestyle='--', alpha=0.7)
            plt.xticks(rotation=45)
            plt.tight_layout(rect=[0, 0, 1, 0.98]) # Adjust layout for text box

        # IMPORTANT: Assign the animation to a variable 'ani' to prevent garbage collection 
        # (This resolves the UserWarning)
        ani = FuncAnimation(fig, animate, interval=500, cache_frame_data=False) 
        
        # plt.show(block=False) allows the main thread to continue execution (needed for input_thread.join())
        plt.show(block=False) 
        
        # Wait for the input thread to complete (i.e., user types 'exit')
        # This keeps the main process alive.
        input_thread.join()

    except KeyboardInterrupt:
        # User pressed Ctrl+C
        pass
        
    finally:
        # This code runs after the input_thread.join() completes or a KeyboardInterrupt
        print("\n\nShutting down simulation threads.")
        
        # Ensure the global flag is set to False for all threads
        TICK_SIM_RUNNING = False
        
        # Give the tick thread a moment to recognize the flag and exit
        if tick_thread.is_alive():
            tick_thread.join(timeout=1.0) 

        # The interactive environment ends cleanly here without sys.exit()
        print("Cleanup complete. Program terminated gracefully.")

Best Practices for Production Trading Dashboards

Use UTC timestamps to avoid timezone confusion
Prevent candle drift by aligning to exact interval boundaries
Handle late ticks with a small grace period (e.g., 500ms)
Offload rendering to a separate thread/process to avoid blocking aggregation
Add volume bars below the chart for full context
Persist candles to disk or database for backtesting

Conclusion

A real-time candlestick chart isn’t just a visualization—it’s a trader’s command center. By building your own aggregation engine, you gain full control over latency, reliability, and customization. This implementation gives you a foundation that scales from demo scripts to live trading systems. Whether you’re building a personal dashboard or a professional trading terminal, real-time candles put the market’s pulse at your fingertips. Code it. Watch it. Trade with confidence.

Table of Contents