Table of Contents
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
![PlantUML Diagram]()
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.")
![1]()
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.