From strategy idea to live trading: my full automation stack
Every algorithmic trader has a graveyard of dead scripts. Mine was a messy collection of Jupyter Notebooks, raw CSV files, and fragile cron jobs running on a cheap VPS.
For a long time, my workflow looked like this: historical data was pulled ad-hoc via Pandas, strategies were backtested in vectorized Python, and execution was handled by a cron-scheduled script that spun up, authenticated with the broker, placed market orders, and shut down.
Then came a high-volatility Tuesday. My execution script initiated a connection to Interactive Brokers, hit an unhandled socket timeout during a sudden market-wide margin-call cascade, and quietly hung. The script did not crash; it simply sat in an infinite loop waiting for an API response that never arrived. It left a leveraged $85,000 long position in NQ futures unhedged and unmonitored during a 180-point drop.
By the time I checked my phone, I had taken a $4,200 loss on a trade that my backtest had flagged for exit three hours prior.
That disaster forced me to stop treating my infrastructure as a collection of scripts and start building a robust, production-grade trading automation stack. I needed a system that separated data ingestion, signal generation, risk management, and order routing into decoupled, resilient services. This is the exact blueprint of the live trading stack I run today.
Architectural Philosophy: Strict Decoupling
The core failure of my early setups was tight coupling. If the broker API went down, my strategy loop crashed. If my database disk filled up, my execution engine halted.
To fix this, I moved to an event-driven architecture built on a simple rule: the signal generator must never block on order execution, and the risk engine must sit as a gatekeeper between both.
Here is the data flow of my current production stack:
flowchart LR feed["Data Feed"] --> ingest["Ingestion Engine"] ingest --> db["QuestDB"] ingest --> signal["Signal Generator"] signal --> risk["Risk Engine"] risk --> execution["Execution Worker"] execution --> broker["Broker API"]
The Component Stack
- Data Ingestion: A lightweight Python service running on
asynciothat streams L1/L2 data from WebSockets and writes directly to QuestDB (for time-series metrics) and broadcasts internal ticks via Redis Pub/Sub. - State Management: Redis stores the real-time system state (active positions, unfilled orders, account balances, and rate limits).
- Signal Generator (The Strategy): A stateful service that listens to Redis Pub/Sub, updates its internal indicators, and publishes target portfolios (e.g., “I want to be long 2 units of spy”) instead of direct buy/sell orders.
- Risk Engine: A stateless validator that evaluates every target portfolio change against hard leverage limits, drawdowns, and correlation thresholds.
- Execution Worker: A dedicated worker that reads approved target positions, compares them against current actual positions, and uses a TWAP/VWAP algo to route orders to the broker API.
The Core Code
Let us look at the actual code that runs this setup. This is a production-grade template of my core execution and risk validation pipeline.
1. The Risk Engine (The Gatekeeper)
This module validates that any generated signal complies with strict portfolio parameters before it can ever touch a broker API. It checks maximum position sizing, sector exposure limits, and daily loss thresholds.
import logging
from typing import Dict, Any, Tuple
logger = logging.getLogger("RiskEngine")
class RiskEngine:
def __init__(self, max_position_usd: float, max_leverage: float, max_daily_loss: float):
self.max_position_usd = max_position_usd
self.max_leverage = max_leverage
self.max_daily_loss = max_daily_loss
def validate_order(
self,
symbol: str,
proposed_size: float,
current_price: float,
portfolio_state: Dict[str, Any]
) -> Tuple[bool, str]:
"""
Validates whether a proposed order complies with pre-trade risk controls.
Returns (is_approved, reason).
"""
# 1. Zero check
if proposed_size == 0:
return False, "Proposed order size is zero."
proposed_value_usd = abs(proposed_size * current_price)
total_equity = portfolio_state.get("net_liquidation_value", 0.0)
current_leverage = portfolio_state.get("current_leverage", 0.0)
daily_pnl = portfolio_state.get("daily_pnl", 0.0)
# 2. Hard daily loss limit check
if daily_pnl <= –self.max_daily_loss:
return False, f"Daily loss limit breached: {daily_pnl:.2f} <= -{self.max_daily_loss}"
# 3. Maximum single position sizing check
if proposed_value_usd > self.max_position_usd:
return False, f"Order size ${proposed_value_usd:.2f} exceeds max allocation of ${self.max_position_usd:.2f}"
# 4. Leverage headroom check
added_leverage = proposed_value_usd / total_equity if total_equity > 0 else 999.0
if (current_leverage + added_leverage) > self.max_leverage:
return False, f"Order violates max leverage limit. Projected leverage: {current_leverage + added_leverage:.2f}"
return True, "Approved"
2. The Resilient Live Execution Worker
This worker runs an asynchronous event loop that processes desired targets and manages order lifecycles via the ib_insync library. It includes defensive disconnection recovery.
import asyncio
import logging
from ib_insync import IB, MarketOrder, OrderStatus
from risk_engine import RiskEngine
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("ExecutionWorker")
class LiveExecutionWorker:
def __init__(self, host: str, port: int, client_id: int, risk_engine: RiskEngine):
self.host = host
self.port = port
self.client_id = client_id
self.ib = IB()
self.risk_engine = risk_engine
self.active_orders = {}
async def connect_with_retry(self):
while not self.ib.isConnected():
try:
logger.info(f"Attempting connection to IB Gateway at {self.host}:{self.port}…")
await self.ib.connectAsync(self.host, self.port, clientId=self.client_id)
logger.info("Successfully connected to IB Gateway.")
self.ib.orderStatusEvent += self.on_order_status
except Exception as e:
logger.error(f"Connection failed: {e}. Retrying in 5 seconds…")
await asyncio.sleep(5)
def on_order_status(self, trade):
"""Callback to monitor orders in real-time."""
status = trade.orderStatus.status
symbol = trade.contract.symbol
logger.info(f"Order status update for {symbol}: {status} | Filled: {trade.orderStatus.filled}")
if status in [OrderStatus.Filled, OrderStatus.Cancelled, OrderStatus.Failed]:
self.active_orders.pop(trade.order.orderId, None)
async def execute_target_portfolio(self, symbol: str, target_qty: float, current_price: float, portfolio_state: dict):
"""
Adjusts the portfolio to match the target quantity, checking risk controls first.
"""
await self.connect_with_retry()
# Determine active position
positions = {p.contract.symbol: p.position for p in self.ib.positions()}
current_qty = positions.get(symbol, 0.0)
order_qty = target_qty – current_qty
if order_qty == 0:
logger.info(f"No execution needed. Current quantity matches target for {symbol}.")
return
# Risk Validation
is_approved, reason = self.risk_engine.validate_order(
symbol=symbol,
proposed_size=order_qty,
current_price=current_price,
portfolio_state=portfolio_state
)
if not is_approved:
logger.warning(f"Order rejected by Risk Engine: {reason}")
return
# Resolve Contract
contracts = await self.ib.reqContractDetailsAsync(MarketOrder(symbol, "BUY").contract) # Dummy contract for resolution
if not contracts:
logger.error(f"Could not resolve contract for symbol: {symbol}")
return
contract = contracts[0].contract
# Create Order
action = "BUY" if order_qty > 0 else "SELL"
order = MarketOrder(action, abs(order_qty))
logger.info(f"Placing order: {action} {abs(order_qty)} shares of {symbol}")
trade = self.ib.placeOrder(contract, order)
self.active_orders[trade.order.orderId] = trade
async def shutdown(self):
logger.info("Shutting down worker safely…")
self.ib.disconnect()
# Quick smoke test execution block
if __name__ == "__main__":
async def main():
risk_config = RiskEngine(max_position_usd=50000.0, max_leverage=2.0, max_daily_loss=1000.0)
worker = LiveExecutionWorker(host="127.0.0.1", port=4001, client_id=1, risk_engine=risk_config)
mock_portfolio_state = {
"net_liquidation_value": 100000.0,
"current_leverage": 0.5,
"daily_pnl": –150.0
}
# This will trigger a connection loop until TWS or IB Gateway is live on port 4001
try:
await worker.connect_with_retry()
# Mock executing a target position of 100 units of SPY (assumed price $500)
await worker.execute_target_portfolio(symbol="SPY", target_qty=100.0, current_price=500.0, portfolio_state=mock_portfolio_state)
finally:
await worker.shutdown()
asyncio.run(main())
3. Schema Management for Traceability
I use a local PostgreSQL instance as my relational logger. While tick data goes to QuestDB, every state change, order submission, and execution response must go to PostgreSQL with strict relational safety.
CREATE TABLE IF NOT EXISTS signal_history (
id SERIAL PRIMARY KEY,
timestamp TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
symbol VARCHAR(12) NOT NULL,
signal_type VARCHAR(16) NOT NULL, — e.g., 'mean_reversion_buy'
target_position NUMERIC(12, 4) NOT NULL,
current_price NUMERIC(12, 4) NOT NULL
);
CREATE TABLE IF NOT EXISTS risk_evaluation_log (
id SERIAL PRIMARY KEY,
timestamp TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
symbol VARCHAR(12) NOT NULL,
proposed_size NUMERIC(12, 4) NOT NULL,
approved BOOLEAN NOT NULL,
rejection_reason VARCHAR(255)
);
CREATE TABLE IF NOT EXISTS executions (
id SERIAL PRIMARY KEY,
timestamp TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
order_id INT UNIQUE NOT NULL,
symbol VARCHAR(12) NOT NULL,
action VARCHAR(4) NOT NULL, — 'BUY' or 'SELL'
quantity NUMERIC(12, 4) NOT NULL,
average_fill_price NUMERIC(12, 4),
execution_status VARCHAR(32) NOT NULL
);
— Ensure index on timestamps for auditing performance and slippage
CREATE INDEX idx_executions_timestamp ON executions (timestamp DESC);
CREATE INDEX idx_signal_history_timestamp ON signal_history (timestamp DESC);
Performance and Slippage Results
The impact of rewriting my infrastructure was immediate and quantifiable.
After running this decoupled system live for six months across both equity and futures markets, I compared its performance directly to my previous cron-based architecture.
| Metric | Old Stack (Cron + Monolithic Scripts) | New Stack (Asyncio + Decoupled Workers) |
|---|---|---|
| System p99 Latency | 850ms to 1200ms | 42ms |
| Average Slippage (SPY) | 1.8 bps | 0.4 bps |
| Average Slippage (ES Futures) | 4.2 bps | 0.8 bps |
| Connection Drop Recovery Time | Infinite (Manual restart required) | < 5 seconds |
| Execution Crash Incidence | 3 – 4 times per quarter | 0 times over 180 days |
The drop in slippage is primarily due to moving from slow, blocking REST client initializations during execution to maintaining a warm, authenticated WebSocket stream via ib_insync. Rather than logging in and requesting data when a signal occurs, my worker evaluates signals inside a hot loop and fires off orders instantly.
Lessons Learned the Hard Way
State Belongs in Redis, Not in Python Memory
In an early iteration of this stack, I kept track of active positions in a local dictionary variable inside the execution class. When a transient network hiccup caused the Docker container to restart, that local dictionary evaporated. The script came back online, assumed it had no open positions, and immediately fired duplicate buy orders.
Lesson: Store your target and actual states in an out-of-process store like Redis. If your execution engine crashes and restarts, it should be able to reconstruct its exact state in milliseconds.
Always Have a “Dead Man’s Switch”
Even with automated connection recovery, API endpoints fail, or VMs lock up. I now run an external watchdog script on a completely separate cloud provider (DigitalOcean) that queries my primary VPS (Hetzner). If the primary fails to update a Redis heartbeat key every 30 seconds, the watchdog immediately sends a CLOSE ALL request via a secondary broker API endpoint and fires off a high-priority SMS alert.
Decouple Testing from Live Broker APIs
Never test your execution flow on live paper accounts using the same code path you use for live accounts without an interface layer. I wrote an abstract BaseBrokerInterface class that allows me to seamlessly swap out the LiveExecutionWorker with a mock engine that acts exactly like the live broker but logs executions to a local JSON file. Tests run on every deployment without touching real-world infrastructure.