The moment of truth: connecting your strategy to a real broker, handling orders, monitoring fills, and making the psychological leap from simulation to live capital.
| Broker | Paper Trading | Commission | Data | Python SDK | Best For |
|---|---|---|---|---|---|
| Alpaca | Free, unlimited | $0 | Free real-time | Excellent | Beginners, US equities |
| Interactive Brokers | Free with account | $0.005/share | Paid real-time | Good (ib_insync) | Pros, global markets |
| Tradier | Free sandbox | $0 | Free delayed | Good | Options trading |
| TD Ameritrade | ThinkOrSwim paper | $0 | Free real-time | Community | US equities + options |
We use Alpaca for this series because: (1) free paper trading with real-time data, (2) zero commission, (3) excellent Python SDK, (4) modern REST + WebSocket API, (5) fractional shares supported. Sign up at alpaca.markets, create a paper trading account, and save your API key and secret in .env.
import os from dotenv import load_dotenv from alpaca.trading.client import TradingClient from alpaca.data.historical import StockHistoricalDataClient load_dotenv() # Paper trading (IMPORTANT: paper_trading=True) trading_client = TradingClient( api_key=os.getenv("ALPACA_API_KEY"), secret_key=os.getenv("ALPACA_SECRET_KEY"), paper=True # ALWAYS True until you're ready ) # Data client (market data) data_client = StockHistoricalDataClient( api_key=os.getenv("ALPACA_API_KEY"), secret_key=os.getenv("ALPACA_SECRET_KEY") ) # Verify connection account = trading_client.get_account() print(f"Account Status: {account.status}") print(f"Cash: ${float(account.cash):,.2f}") print(f"Portfolio Value: ${float(account.portfolio_value):,.2f}") print(f"Buying Power: ${float(account.buying_power):,.2f}")
from alpaca.trading.requests import ( MarketOrderRequest, LimitOrderRequest, StopOrderRequest, StopLimitOrderRequest, TakeProfitRequest, StopLossRequest ) from alpaca.trading.enums import ( OrderSide, TimeInForce, OrderClass ) def market_order(client, symbol, qty, side='buy'): """Simple market order — fills immediately at best price.""" order = MarketOrderRequest( symbol=symbol, qty=qty, side=OrderSide.BUY if side == 'buy' else OrderSide.SELL, time_in_force=TimeInForce.DAY ) return client.submit_order(order) def limit_order(client, symbol, qty, limit_price, side='buy'): """Limit order — only fills at limit_price or better.""" order = LimitOrderRequest( symbol=symbol, qty=qty, side=OrderSide.BUY if side == 'buy' else OrderSide.SELL, time_in_force=TimeInForce.GTC, # Good til canceled limit_price=limit_price ) return client.submit_order(order) def bracket_order(client, symbol, qty, limit_price, take_profit, stop_loss): """ Bracket order — entry + take profit + stop loss. The holy grail of automated trading: one call, three orders, fully managed by the broker. """ order = LimitOrderRequest( symbol=symbol, qty=qty, side=OrderSide.BUY, time_in_force=TimeInForce.GTC, limit_price=limit_price, order_class=OrderClass.BRACKET, take_profit=TakeProfitRequest(limit_price=take_profit), stop_loss=StopLossRequest(stop_price=stop_loss) ) return client.submit_order(order)
| Order Type | Fills At | Risk | Use When |
|---|---|---|---|
| Market | Best available | Slippage | Need guaranteed fill (liquid stocks) |
| Limit | Price or better | May not fill | Want price control |
| Stop | Market after trigger | Slippage after trigger | Stop losses |
| Bracket | Entry + TP + SL | Complex | Full trade management |
Compute signals from latest data
Check risk limits, drawdown, exposure
Calculate position size (Kelly/ATR)
Submit bracket order via API
Wait for fill, handle partial/reject
Record everything to trade journal
import logging import time from datetime import datetime logger = logging.getLogger('executor') class ExecutionEngine: """Live trading execution engine.""" def __init__(self, trading_client, data_client, strategy, risk_manager, journal): self.client = trading_client self.data = data_client self.strategy = strategy self.risk = risk_manager self.journal = journal self.positions = {} def run_daily(self): """Main daily execution cycle. Call this once per day after market close (for next-day execution) or at market open.""" logger.info("=== Execution cycle started ===") # Step 1: Get latest data data = self._fetch_latest_data() # Step 2: Generate signals signals = self.strategy.generate_signals(data) logger.info(f"Signals: {signals.to_dict()}") # Step 3: Check risk constraints account = self.client.get_account() equity = float(account.portfolio_value) risk_check = self.risk.check(equity, signals) if risk_check['action'] == 'HALT': logger.warning(f"HALT: {risk_check['message']}") self._close_all_positions() return # Step 4: Calculate target positions targets = self._compute_targets( signals, equity, risk_check['size_multiplier'] ) # Step 5: Execute trades (reconcile current vs target) current = self._get_current_positions() orders = self._reconcile(current, targets) for order in orders: try: result = self._submit_order(order) self.journal.log_order(order, result) logger.info(f"Order filled: {order}") except Exception as e: logger.error(f"Order failed: {order} — {e}") self.journal.log_error(order, e) # Step 6: Log summary self.journal.log_cycle(signals, targets, orders) logger.info("=== Execution cycle complete ===") def _reconcile(self, current, targets): """Compute orders needed to move from current to target.""" orders = [] all_symbols = set(current.keys()) | set(targets.keys()) for sym in all_symbols: curr_qty = current.get(sym, 0) tgt_qty = targets.get(sym, 0) diff = tgt_qty - curr_qty if abs(diff) > 0: orders.append({ 'symbol': sym, 'qty': abs(diff), 'side': 'buy' if diff > 0 else 'sell', 'reason': f'rebalance {curr_qty} -> {tgt_qty}' }) return orders
In live trading, orders can be partially filled (you asked for 100 shares, got 73), rejected (insufficient buying power, market closed), or cancelled (you cancelled, or it expired). Always check the order status after submission. For partial fills, decide whether to submit a new order for the remainder or accept the partial position. Never assume an order filled — always verify.
class LiveMonitor: """Real-time portfolio monitoring.""" def __init__(self, trading_client): self.client = trading_client def dashboard(self) -> dict: """Get current portfolio snapshot.""" account = self.client.get_account() positions = self.client.get_all_positions() portfolio = { 'equity': float(account.portfolio_value), 'cash': float(account.cash), 'buying_power': float(account.buying_power), 'daily_pnl': float(account.equity) - float(account.last_equity), 'positions': [] } for pos in positions: portfolio['positions'].append({ 'symbol': pos.symbol, 'qty': int(pos.qty), 'avg_entry': float(pos.avg_entry_price), 'current_price': float(pos.current_price), 'unrealized_pnl': float(pos.unrealized_pl), 'pnl_pct': float(pos.unrealized_plpc) * 100, 'market_value': float(pos.market_value), }) return portfolio def print_dashboard(self): """Pretty-print the dashboard.""" d = self.dashboard() print(f"\n{'='*60}") print(f" PORTFOLIO | {datetime.now():%Y-%m-%d %H:%M}") print(f"{'='*60}") print(f" Equity: ${d['equity']:>12,.2f}") print(f" Cash: ${d['cash']:>12,.2f}") print(f" Daily PnL: ${d['daily_pnl']:>12,.2f}") print(f"{'─'*60}") for p in d['positions']: color = '+' if p['unrealized_pnl'] >= 0 else '' print(f" {p['symbol']:<6} {p['qty']:>4} sh" f" @ ${p['avg_entry']:>8.2f}" f" -> ${p['current_price']:>8.2f}" f" {color}{p['pnl_pct']:>+6.1f}%" f" ${p['unrealized_pnl']:>+10,.2f}") print(f"{'='*60}\n")
Every trade should be logged with: symbol, direction, entry/exit price, size, reason, signal strength, outcome. This is your data for improving the strategy. No journal = no improvement.
import json from datetime import datetime from pathlib import Path class TradeJournal: """Persistent trade logging.""" def __init__(self, path: str = "trades/journal.jsonl"): self.path = Path(path) self.path.parent.mkdir(parents=True, exist_ok=True) def log_trade(self, trade: dict): """Append a trade to the journal.""" trade['timestamp'] = datetime.now().isoformat() with open(self.path, 'a') as f: f.write(json.dumps(trade) + '\n') def get_trades(self, last_n: int = 50) -> list: """Read last N trades.""" if not self.path.exists(): return [] with open(self.path) as f: lines = f.readlines() return [json.loads(l) for l in lines[-last_n:]] def summary(self) -> dict: """Compute journal statistics.""" trades = self.get_trades(last_n=9999) if not trades: return {} pnls = [t.get('pnl', 0) for t in trades if 'pnl' in t] wins = [p for p in pnls if p > 0] losses = [p for p in pnls if p <= 0] return { 'total_trades': len(pnls), 'win_rate': len(wins) / len(pnls) if pnls else 0, 'avg_pnl': sum(pnls) / len(pnls) if pnls else 0, 'best_trade': max(pnls) if pnls else 0, 'worst_trade': min(pnls) if pnls else 0, }
The code is identical. The data is the same. But the moment real money is at stake, everything changes. Here's how to make the transition safely.
1) Run paper for at least 3 months with real-time signals. 2) Achieve a Sharpe > 0.5 out-of-sample. 3) Experience at least one drawdown and verify your circuit breaker works. 4) Compare paper fills vs expected fills — if slippage is >20bps consistently, your backtest is unrealistic. 5) Verify every component: signal generation, sizing, order submission, fill handling, logging. 6) Have a kill switch (manual and automated).
Phase 1 (Weeks 1-4): Live with 10% of target capital. This is your real-money paper trade. Expect to find bugs and execution issues. Phase 2 (Weeks 5-8): Scale to 50% if Phase 1 metrics match paper trading. Phase 3 (Week 9+): Full capital — but only if Phase 2 confirms the edge is real. If at any phase the live Sharpe is <50% of paper Sharpe, stop and investigate.