Engineering
By Flowistic
Difficulty: Advanced
Read time: 35 min
Published: August 14, 2026
Last modified: August 14, 2026
Introduction
Most trading systems fail not because of bad algorithms, but because of infrastructure brittleness. A single market data feed lag, a dropped order, or a network glitch can wipe out profits.
RabbitMQ Streams is a distributed message broker designed for exactly these problems: guaranteed delivery, persistent storage, replay capability, and horizontal scaling. Unlike traditional request-response systems, Streams enable:
- Multiple consumers of the same market data without coupling
- Replay from history - rewind and reprocess all trades for debugging
- Backpressure handling - gracefully handle data spikes without dropping messages
- Fault tolerance - automatic failover and recovery
This guide shows how to build production-grade trading infrastructure on RabbitMQ Streams.
Part 1: RabbitMQ Streams Fundamentals
RabbitMQ Streams are append-only logs (like Kafka), but with RabbitMQ’s simplicity. Every message is persisted, and consumers can replay from any point in time.
import pika
from pika.adapters.blocking_connection import BlockingConnection
import json
from datetime import datetime
class MarketDataPublisher:
"""Publish tick data to RabbitMQ Streams"""
def __init__(self, rabbitmq_host: str = "localhost",
stream_name: str = "market_data"):
self.stream_name = stream_name
# Connect with stream plugin
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(
host=rabbitmq_host,
credentials=credentials
)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
# Declare stream (idempotent - safe to call multiple times)
self.channel.queue_declare(
queue=stream_name,
durable=True,
arguments={
'x-queue-type': 'stream',
# Retention: keep messages for 7 days
'x-max-age': '7d',
# Max size: 10GB per stream
'x-stream-max-length-bytes': 10737418240
}
)
def publish_tick(self, tick_data: dict):
"""Publish a single tick to the stream"""
message = json.dumps({
'timestamp': datetime.utcnow().isoformat(),
'ticker': tick_data['ticker'],
'price': tick_data['price'],
'volume': tick_data['volume'],
'bid': tick_data.get('bid'),
'ask': tick_data.get('ask'),
'bid_size': tick_data.get('bid_size'),
'ask_size': tick_data.get('ask_size'),
})
# Publish with confirmation
self.channel.basic_publish(
exchange='',
routing_key=self.stream_name,
body=message.encode(),
properties=pika.BasicProperties(
delivery_mode=2, # Persistent delivery
content_type='application/json'
)
)
def publish_batch(self, ticks: list):
"""Publish multiple ticks efficiently"""
for tick in ticks:
self.publish_tick(tick)
# Flush to ensure delivery
self.connection.process_data_events()
def close(self):
"""Close connection"""
self.connection.close()
Key concept: Unlike queues where messages disappear after consumption, Streams persist all messages. Consumers independently track their position in the stream—new consumers can start from the beginning or the tail.
Part 2: Stream Consumers with Offset Management
A single market data stream can feed multiple trading strategies. Each strategy independently consumes messages from wherever it wants in the stream.
from typing import Callable, Dict
import threading
class StreamConsumer:
"""Subscribe to market data stream and process ticks"""
def __init__(self, rabbitmq_host: str = "localhost",
stream_name: str = "market_data",
consumer_name: str = "strategy_1"):
self.stream_name = stream_name
self.consumer_name = consumer_name
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(host=rabbitmq_host)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
# Declare stream (idempotent)
self.channel.queue_declare(
queue=stream_name,
durable=True,
arguments={'x-queue-type': 'stream'}
)
def consume_from_beginning(self, callback: Callable):
"""Subscribe from the start of the stream"""
# Consumer name identifies this consumer for offset tracking
self.channel.basic_qos(prefetch_count=100)
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=self._make_callback(callback),
consumer_tag=self.consumer_name,
# Start from beginning
arguments={
'x-stream-offset': 'first'
}
)
print(f"Consuming from {self.stream_name} (from beginning)")
self.channel.start_consuming()
def consume_from_latest(self, callback: Callable):
"""Subscribe from latest message only (new ticks going forward)"""
self.channel.basic_qos(prefetch_count=100)
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=self._make_callback(callback),
consumer_tag=self.consumer_name,
# Start from latest
arguments={
'x-stream-offset': 'last'
}
)
print(f"Consuming from {self.stream_name} (from latest)")
self.channel.start_consuming()
def consume_from_timestamp(self, timestamp: str, callback: Callable):
"""Subscribe from specific timestamp (e.g., "2025-02-01T09:30:00")"""
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=self._make_callback(callback),
consumer_tag=self.consumer_name,
arguments={
'x-stream-offset': 'timestamp',
'x-stream-offset-timestamp': timestamp
}
)
print(f"Consuming from {self.stream_name} (from {timestamp})")
self.channel.start_consuming()
def _make_callback(self, user_callback: Callable):
"""Wrap user callback with message parsing and offset management"""
def callback(ch, method, properties, body):
try:
tick = json.loads(body.decode())
# Call user function
user_callback(tick)
# Acknowledge message (offset is saved automatically)
ch.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e:
print(f"Error processing message: {e}")
# NACK to retry
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
return callback
def close(self):
self.connection.close()
Why this matters: Two trading strategies consume the same market data stream independently. Strategy A processes all ticks from 9:30 AM. Strategy B starts at 2:00 PM. The stream keeps track of each consumer’s offset automatically.
Part 3: Event Sourcing - Complete Audit Trail
Every trade, order, and decision should be logged immutably. RabbitMQ Streams enables complete event sourcing: rebuild any historical state by replaying events.
from enum import Enum
from dataclasses import dataclass, asdict
class EventType(Enum):
MARKET_DATA = "market_data"
ORDER_CREATED = "order_created"
ORDER_FILLED = "order_filled"
ORDER_CANCELLED = "order_cancelled"
TRADE_EXECUTED = "trade_executed"
POSITION_UPDATED = "position_updated"
ERROR = "error"
@dataclass
class TradingEvent:
"""Immutable event in the audit log"""
event_type: EventType
timestamp: str
data: dict
def to_json(self) -> str:
return json.dumps(asdict(self))
class AuditLogPublisher:
"""Publish all trading events to audit stream"""
def __init__(self, rabbitmq_host: str = "localhost"):
self.stream_name = "trading_audit_log"
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(host=rabbitmq_host)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
# Create audit stream with longer retention
self.channel.queue_declare(
queue=self.stream_name,
durable=True,
arguments={
'x-queue-type': 'stream',
# Keep audit logs for 1 year
'x-max-age': '365d'
}
)
def log_market_data(self, ticker: str, price: float, volume: int):
"""Log incoming market data"""
event = TradingEvent(
event_type=EventType.MARKET_DATA,
timestamp=datetime.utcnow().isoformat(),
data={'ticker': ticker, 'price': price, 'volume': volume}
)
self._publish_event(event)
def log_order_created(self, order_id: str, ticker: str,
quantity: int, price: float, side: str):
"""Log order creation"""
event = TradingEvent(
event_type=EventType.ORDER_CREATED,
timestamp=datetime.utcnow().isoformat(),
data={
'order_id': order_id,
'ticker': ticker,
'quantity': quantity,
'price': price,
'side': side
}
)
self._publish_event(event)
def log_order_filled(self, order_id: str, filled_quantity: int,
filled_price: float):
"""Log order execution"""
event = TradingEvent(
event_type=EventType.ORDER_FILLED,
timestamp=datetime.utcnow().isoformat(),
data={
'order_id': order_id,
'filled_quantity': filled_quantity,
'filled_price': filled_price
}
)
self._publish_event(event)
def log_error(self, error_message: str, context: dict = None):
"""Log errors with full context"""
event = TradingEvent(
event_type=EventType.ERROR,
timestamp=datetime.utcnow().isoformat(),
data={'message': error_message, 'context': context or {}}
)
self._publish_event(event)
def _publish_event(self, event: TradingEvent):
"""Publish event to audit stream"""
self.channel.basic_publish(
exchange='',
routing_key=self.stream_name,
body=event.to_json().encode(),
properties=pika.BasicProperties(delivery_mode=2)
)
class AuditLogReplayer:
"""Replay audit logs to reconstruct historical state"""
def __init__(self, rabbitmq_host: str = "localhost"):
self.stream_name = "trading_audit_log"
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(host=rabbitmq_host)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
def replay_all_events(self, start_timestamp: str = None):
"""Replay entire audit log from beginning or timestamp"""
events = []
def collect_events(tick):
events.append(tick)
# Consumer with callback
if start_timestamp:
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=self._wrap_callback(collect_events),
arguments={'x-stream-offset': 'timestamp',
'x-stream-offset-timestamp': start_timestamp}
)
else:
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=self._wrap_callback(collect_events),
arguments={'x-stream-offset': 'first'}
)
self.channel.start_consuming()
return events
def get_position_at_timestamp(self, timestamp: str) -> dict:
"""Reconstruct trading position at any point in time"""
position = {'long': 0, 'short': 0, 'realized_pnl': 0}
def update_position(event_dict):
event_type = event_dict.get('event_type')
data = event_dict.get('data', {})
if event_type == EventType.ORDER_FILLED.value:
quantity = data.get('filled_quantity', 0)
side = data.get('side', 'BUY')
if side == 'BUY':
position['long'] += quantity
else:
position['short'] += quantity
events = self.replay_all_events(start_timestamp=timestamp)
for event in events:
event_dict = json.loads(event) if isinstance(event, str) else event
update_position(event_dict)
return position
def _wrap_callback(self, user_callback):
def callback(ch, method, properties, body):
event_dict = json.loads(body.decode())
user_callback(event_dict)
ch.basic_ack(delivery_tag=method.delivery_tag)
return callback
Power of event sourcing: At 3:47 PM, you discover a bug. Replay the audit log from 9:30 AM. See every trade that was executed based on the buggy logic. Calculate exact P&L impact. Perfect for compliance and debugging.
Part 4: Handling Backpressure & Flow Control
Market data can spike (earnings announcements, flash crashes). Naive systems drop messages. RabbitMQ Streams handle this with prefetch limits and consumer groups.
class BackpressureAwareConsumer:
"""Process stream messages with proper flow control"""
def __init__(self, rabbitmq_host: str = "localhost",
stream_name: str = "market_data",
prefetch_count: int = 100):
self.stream_name = stream_name
self.prefetch_count = prefetch_count
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(
host=rabbitmq_host,
connection_attempts=3,
retry_delay=2
)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
# Set prefetch: process max 100 messages before requiring ACK
# Higher prefetch = better throughput, but more memory
# Lower prefetch = lower latency
self.channel.basic_qos(prefetch_count=prefetch_count)
self.channel.queue_declare(
queue=stream_name,
durable=True,
arguments={'x-queue-type': 'stream'}
)
def process_stream_with_batching(self,
batch_callback,
batch_size: int = 100):
"""Process messages in batches for efficiency"""
batch = []
def callback(ch, method, properties, body):
nonlocal batch
tick = json.loads(body.decode())
batch.append(tick)
# Process when batch is full
if len(batch) >= batch_size:
try:
batch_callback(batch)
batch = []
ch.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e:
print(f"Batch processing error: {e}")
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
else:
# Keep the message until batch is ready
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=callback,
arguments={'x-stream-offset': 'last'}
)
self.channel.start_consuming()
def process_with_timeout_and_retry(self, callback,
max_retries: int = 3):
"""Handle slow processing with retry logic"""
def wrapped_callback(ch, method, properties, body):
tick = json.loads(body.decode())
retry_count = 0
while retry_count < max_retries:
try:
callback(tick)
ch.basic_ack(delivery_tag=method.delivery_tag)
break
except Exception as e:
retry_count += 1
print(f"Error (attempt {retry_count}): {e}")
if retry_count >= max_retries:
# Give up, send to dead letter queue
self._send_to_dlq(tick, str(e))
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
else:
# Retry
import time
time.sleep(0.5 * retry_count) # Exponential backoff
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=wrapped_callback,
arguments={'x-stream-offset': 'last'}
)
self.channel.start_consuming()
def _send_to_dlq(self, message: dict, error: str):
"""Send failed messages to dead letter queue"""
dlq_message = {
'original_message': message,
'error': error,
'timestamp': datetime.utcnow().isoformat()
}
self.channel.basic_publish(
exchange='',
routing_key='trading_dlq',
body=json.dumps(dlq_message).encode()
)
Flow control: When strategy processing slows down, RabbitMQ automatically throttles incoming messages. No data loss, no crashes.
Part 5: Consumer Groups - Scaling Strategies
Run multiple instances of the same strategy. Consumer groups automatically distribute messages so each is processed exactly once.
class ConsumerGroup:
"""Distribute work across multiple consumer instances"""
def __init__(self, rabbitmq_host: str = "localhost",
stream_name: str = "market_data",
group_name: str = "strategy_group"):
self.stream_name = stream_name
self.group_name = group_name
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(host=rabbitmq_host)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
# Declare stream
self.channel.queue_declare(
queue=stream_name,
durable=True,
arguments={'x-queue-type': 'stream'}
)
def create_consumer_in_group(self, consumer_id: str,
callback: Callable):
"""Join consumer group to process stream partitions"""
# RabbitMQ Streams automatically partition data across consumers
# Each message processed by exactly one consumer in the group
self.channel.basic_qos(prefetch_count=100)
def wrapped_callback(ch, method, properties, body):
tick = json.loads(body.decode())
try:
callback(tick)
ch.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e:
print(f"Consumer {consumer_id} error: {e}")
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
# Consumer tag identifies this instance in the group
self.channel.basic_consume(
queue=self.stream_name,
on_message_callback=wrapped_callback,
consumer_tag=f"{self.group_name}_{consumer_id}",
arguments={'x-stream-offset': 'last'}
)
print(f"Consumer {consumer_id} joined {self.group_name}")
self.channel.start_consuming()
# Example: Run multiple strategy instances in parallel
def deploy_strategy_instances(num_instances: int = 3):
"""Deploy multiple instances of trading strategy"""
import threading
def strategy_worker(instance_id: int):
group = ConsumerGroup(group_name="momentum_strategy")
def process_tick(tick):
# Trading logic
price = tick['price']
volume = tick['volume']
# Strategy calculation...
signal = calculate_signal(price, volume)
if signal:
print(f"Instance {instance_id}: Signal on {tick['ticker']}")
group.create_consumer_in_group(f"instance_{instance_id}", process_tick)
# Start workers in threads
threads = []
for i in range(num_instances):
t = threading.Thread(target=strategy_worker, args=(i,))
t.daemon = True
t.start()
threads.append(t)
# Keep main thread alive
for t in threads:
t.join()
Scaling: Start with 1 strategy instance. As volume grows, launch 2, 3, 10 instances. RabbitMQ automatically distributes work. Each message processed exactly once.
Part 6: Real-Time Aggregation Across Exchanges
Combine market data from multiple exchanges into a single stream for cross-exchange strategies.
import threading
from queue import Queue
class MultiExchangeAggregator:
"""Aggregate ticks from multiple exchanges into unified stream"""
def __init__(self, rabbitmq_host: str = "localhost"):
self.rabbitmq_host = rabbitmq_host
self.unified_stream = "unified_market_data"
# Setup publisher to unified stream
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(host=rabbitmq_host)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
self.channel.queue_declare(
queue=self.unified_stream,
durable=True,
arguments={'x-queue-type': 'stream'}
)
def aggregate_exchanges(self, exchange_names: list):
"""Subscribe to all exchanges and republish to unified stream"""
# Create consumer for each exchange feed
for exchange in exchange_names:
exchange_stream = f"market_data_{exchange.lower()}"
# Each exchange in its own thread
consumer_thread = threading.Thread(
target=self._consume_and_aggregate,
args=(exchange, exchange_stream)
)
consumer_thread.daemon = True
consumer_thread.start()
def _consume_and_aggregate(self, exchange: str, stream_name: str):
"""Consume from exchange stream and republish"""
credentials = pika.PlainCredentials("guest", "guest")
parameters = pika.ConnectionParameters(host=self.rabbitmq_host)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
def callback(ch, method, properties, body):
tick = json.loads(body.decode())
# Add exchange metadata
enriched_tick = {
**tick,
'exchange': exchange,
'received_at': datetime.utcnow().isoformat()
}
# Republish to unified stream
self.channel.basic_publish(
exchange='',
routing_key=self.unified_stream,
body=json.dumps(enriched_tick).encode(),
properties=pika.BasicProperties(delivery_mode=2)
)
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(
queue=stream_name,
on_message_callback=callback,
arguments={'x-stream-offset': 'last'}
)
print(f"Aggregating {exchange} from {stream_name}")
channel.start_consuming()
def subscribe_to_unified(self, callback: Callable):
"""Subscribe to unified stream"""
def wrapped_callback(ch, method, properties, body):
tick = json.loads(body.decode())
callback(tick)
ch.basic_ack(delivery_tag=method.delivery_tag)
consumer_conn = pika.BlockingConnection(
pika.ConnectionParameters(self.rabbitmq_host)
)
consumer_channel = consumer_conn.channel()
consumer_channel.basic_consume(
queue=self.unified_stream,
on_message_callback=wrapped_callback,
arguments={'x-stream-offset': 'last'}
)
consumer_channel.start_consuming()
Cross-exchange strategies: Subscribe to unified stream. Process ticks from NYSE, NASDAQ, CBOE in order. Execute arbitrage or statistical strategies across venues.
Best Practices for Production
- Set retention policies: Keep audit logs forever, market data for 30 days
- Monitor consumer lag: Alert if any strategy falls behind
- Test failover: Simulate broker crashes, verify recovery
- Use consumer groups: Scale horizontally as volume grows
- Implement dead letter queues: Never silently drop messages
- Log everything: Stream is immutable audit trail
- Version your messages: Add
versionfield for schema evolution
Guide Version: 1.0 (Draft) Last Updated: February 2025
Published: August 14, 2026
Author: Flowistic
Ready to optimize your system?
Let our team help you implement these strategies and achieve peak performance.
Schedule a Consultation