2025-01-15•
The Execution Layer: Order Books, VWAP, and Why Your Backtest Is Lying to You
Build a Python order book simulator, implement a VWAP execution engine, model realistic slippage, and run a transaction cost analysis — everything the backtest conveniently forgot to tell you.
quantpythonalgo-tradingmarket-microstructureexecution
Your backtested Sharpe ratio is beautiful. Smooth equity curve. Controlled drawdowns. The kind of thing you'd frame and hang above your desk.
Then you go live.
The first week, slippage quietly eats 8% of your expected returns. The second week, your large orders move the price before they fill. By week three, you're journaling things like "the model is fundamentally sound, the market is just irrational right now."
The model was fine. The market wasn't irrational. You just skipped the execution layer entirely.
This post fixes that. We're building four things:
- A Python order book simulator (Level 2 depth, bid/ask queues, order matching engine)
- A VWAP execution algorithm that minimises market impact
- A realistic slippage model for backtesting
- A transaction cost analysis (TCA) framework to measure your strategy's true execution quality
By the end of this, you'll have the tooling to know exactly how much your execution is costing you — before it costs you money.
Part 1: The Order Book Simulator
The order book is a sorted ledger of all pending buy and sell orders. Every price discovery event you've ever seen — every "the stock gapped up at open" or "liquidity dried up" — is just the order book doing its thing.
Let's build one from scratch.
Data Structures First
We need two sorted structures — bids sorted descending (highest buyer first), asks sorted ascending (lowest seller first):
from sortedcontainers import SortedDict
from collections import deque
from dataclasses import dataclass, field
from typing import Literal
import time
import uuid
@dataclass
class Order:
order_id: str
side: Literal["buy", "sell"]
price: float
quantity: int
timestamp: float = field(default_factory=time.time)
@dataclass
class Fill:
order_id: str
fill_price: float
fill_qty: int
timestamp: float = field(default_factory=time.time)
class OrderBook:
def __init__(self, ticker: str):
self.ticker = ticker
# Bids: price -> deque of orders (highest price = best bid)
self.bids: SortedDict = SortedDict(lambda x: -x)
# Asks: price -> deque of orders (lowest price = best ask)
self.asks: SortedDict = SortedDict()
self.fills: list[Fill] = []
self.trade_history: list[dict] = []
@property
def best_bid(self) -> float | None:
return next(iter(self.bids), None)
@property
def best_ask(self) -> float | None:
return next(iter(self.asks), None)
@property
def spread(self) -> float | None:
if self.best_bid and self.best_ask:
return round(self.best_ask - self.best_bid, 2)
return None
@property
def mid_price(self) -> float | None:
if self.best_bid and self.best_ask:
return round((self.best_bid + self.best_ask) / 2, 2)
return None
The Matching Engine
This is where orders actually get married to each other. A buy order at ₹1452 matches against any sell order priced ≤ ₹1452, starting from the lowest ask:
def add_limit_order(self, side: str, price: float, quantity: int) -> list[Fill]:
order = Order(
order_id=str(uuid.uuid4())[:8],
side=side,
price=price,
quantity=quantity
)
fills = []
if side == "buy":
fills = self._match_buy(order)
# If not fully filled, rest on the book
if order.quantity > 0:
if price not in self.bids:
self.bids[price] = deque()
self.bids[price].append(order)
elif side == "sell":
fills = self._match_sell(order)
if order.quantity > 0:
if price not in self.asks:
self.asks[price] = deque()
self.asks[price].append(order)
return fills
def _match_buy(self, order: Order) -> list[Fill]:
fills = []
for ask_price in list(self.asks.keys()):
if ask_price > order.price or order.quantity == 0:
break
queue = self.asks[ask_price]
while queue and order.quantity > 0:
resting = queue[0]
fill_qty = min(order.quantity, resting.quantity)
fill = Fill(order.order_id, ask_price, fill_qty)
fills.append(fill)
self._record_trade(ask_price, fill_qty)
order.quantity -= fill_qty
resting.quantity -= fill_qty
if resting.quantity == 0:
queue.popleft()
if not queue:
del self.asks[ask_price]
return fills
def _match_sell(self, order: Order) -> list[Fill]:
fills = []
for bid_price in list(self.bids.keys()):
if bid_price < order.price or order.quantity == 0:
break
queue = self.bids[bid_price]
while queue and order.quantity > 0:
resting = queue[0]
fill_qty = min(order.quantity, resting.quantity)
fill = Fill(order.order_id, bid_price, fill_qty)
fills.append(fill)
self._record_trade(bid_price, fill_qty)
order.quantity -= fill_qty
resting.quantity -= fill_qty
if resting.quantity == 0:
queue.popleft()
if not queue:
del self.bids[bid_price]
return fills
def add_market_order(self, side: str, quantity: int) -> list[Fill]:
# Market order = limit order at an absurdly aggressive price
price = float('inf') if side == "buy" else 0.0
return self.add_limit_order(side, price, quantity)
def _record_trade(self, price: float, qty: int):
self.trade_history.append({
"price": price,
"quantity": qty,
"timestamp": time.time()
})
def depth_snapshot(self, levels: int = 5) -> dict:
bids = [(p, sum(o.quantity for o in q))
for p, q in list(self.bids.items())[:levels]]
asks = [(p, sum(o.quantity for o in q))
for p, q in list(self.asks.items())[:levels]]
return {"bids": bids, "asks": asks,
"spread": self.spread, "mid": self.mid_price}
Watching an Order Walk the Book
# Seed the book with resting liquidity
book = OrderBook("NIFTY50-FUT")
# Add some sell-side liquidity (asks)
for price, qty in [(1452.0, 800), (1452.5, 450), (1453.0, 200)]:
book.add_limit_order("sell", price, qty)
# Add some buy-side liquidity (bids)
for price, qty in [(1451.5, 600), (1451.0, 300), (1450.5, 150)]:
book.add_limit_order("buy", price, qty)
print("Before large order:")
print(book.depth_snapshot())
# Now send a large market buy: 1200 shares
# This will eat through 800@1452, 400@1452.5
fills = book.add_market_order("buy", 1200)
print("\nFills received:")
for f in fills:
print(f" {f.fill_qty} shares @ ₹{f.fill_price}")
total_cost = sum(f.fill_price * f.fill_qty for f in fills)
total_qty = sum(f.fill_qty for f in fills)
avg_fill = total_cost / total_qty
print(f"\nAverage fill price: ₹{avg_fill:.2f}")
print(f"Expected (best ask): ₹1452.00")
print(f"Slippage: ₹{avg_fill - 1452.00:.2f} ({((avg_fill/1452)-1)*10000:.1f} bps)")
Output: Your 1200-share buy fills at ₹1452.17 average — 11.7 bps of slippage just from walking three price levels. On a 100-trade-per-day strategy, that compounds fast.
Part 2: The VWAP Execution Algorithm
VWAP (Volume Weighted Average Price) is the benchmark that most institutional desks use to evaluate execution quality. The goal: execute your order at a price close to or better than the day's VWAP.
The implementation: slice your large order into child orders that match the market's intraday volume distribution. Trade more when everyone else trades more. Be invisible.
Historical Volume Profile
Markets aren't uniformly active throughout the day. Volume follows a predictable U-shape — heavy at open, light at midday, heavy at close:
import numpy as np
import pandas as pd
def generate_intraday_volume_profile(
total_volume: int,
intervals: int = 78, # 5-minute bars in a 6.5-hour session
seed: int = 42
) -> np.ndarray:
"""
Approximate the empirical U-shaped intraday volume curve.
First 20% of day: ~30% of volume (open auction spillover)
Middle 60% of day: ~40% of volume
Last 20% of day: ~30% of volume (close auction buildup)
"""
np.random.seed(seed)
t = np.linspace(0, 1, intervals)
# U-shape base: high open, low midday, high close
base = 0.3 * np.exp(-8 * t) + 0.15 + 0.3 * np.exp(-8 * (1 - t))
# Add realistic noise (volume is never perfectly smooth)
noise = np.random.lognormal(0, 0.2, intervals)
profile = base * noise
# Normalise to total volume
profile = (profile / profile.sum() * total_volume).astype(int)
# Fix rounding so sum matches exactly
diff = total_volume - profile.sum()
profile[intervals // 2] += diff
return profile
The VWAP Slicer
class VWAPExecutor:
def __init__(
self,
target_quantity: int,
side: str,
volume_profile: np.ndarray,
participation_rate: float = 0.10, # target 10% of each interval's volume
urgency: float = 0.0 # 0 = pure VWAP, 1 = get done ASAP
):
self.target = target_quantity
self.side = side
self.profile = volume_profile
self.participation = participation_rate
self.urgency = urgency
self.remaining = target_quantity
self.child_orders: list[dict] = []
self.fills: list[dict] = []
def generate_schedule(self) -> pd.DataFrame:
"""
Build the execution schedule: how many shares to send per interval.
With urgency > 0, front-load the schedule.
"""
n = len(self.profile)
intervals = np.arange(n)
# Base schedule: proportional to volume profile
base_schedule = self.profile / self.profile.sum() * self.target
if self.urgency > 0:
# Urgency shifts weight toward earlier intervals
urgency_weight = np.exp(-self.urgency * intervals / n)
urgency_schedule = (
urgency_weight / urgency_weight.sum() * self.target
)
schedule = (
(1 - self.urgency) * base_schedule +
self.urgency * urgency_schedule
)
else:
schedule = base_schedule
# Cap each interval at participation rate * that interval's volume
max_per_interval = self.profile * self.participation
schedule = np.minimum(schedule, max_per_interval)
# Redistribute unfilled quantity to later intervals
total_scheduled = schedule.sum()
if total_scheduled < self.target:
shortfall = self.target - total_scheduled
# Add shortfall proportionally to remaining capacity
extra_capacity = max_per_interval - schedule
if extra_capacity.sum() > 0:
schedule += (extra_capacity / extra_capacity.sum()) * shortfall
return pd.DataFrame({
"interval": intervals,
"market_volume": self.profile,
"target_qty": schedule.astype(int),
"participation_pct": (schedule / self.profile * 100).round(1)
})
def simulate_execution(
self,
market_prices: np.ndarray,
market_volumes: np.ndarray,
impact_factor: float = 0.1
) -> dict:
"""
Simulate VWAP execution against a price/volume series.
Returns fills, costs, and vs-VWAP performance.
"""
schedule = self.generate_schedule()
fills = []
total_cost = 0
total_filled = 0
for i, row in schedule.iterrows():
qty = int(row["target_qty"])
if qty == 0 or total_filled >= self.target:
continue
qty = min(qty, self.target - total_filled)
# Market impact: your order nudges the price
participation = qty / max(market_volumes[i], 1)
impact_bps = impact_factor * np.sqrt(participation) * 10000
direction = 1 if self.side == "buy" else -1
exec_price = market_prices[i] * (
1 + direction * impact_bps / 10000
)
# Add slippage noise (±2 bps)
noise = np.random.normal(0, 0.0002)
exec_price *= (1 + direction * noise)
fills.append({
"interval": i,
"quantity": qty,
"exec_price": round(exec_price, 2),
"impact_bps": round(impact_bps, 2)
})
total_cost += exec_price * qty
total_filled += qty
avg_exec_price = total_cost / total_filled
market_vwap = np.average(market_prices, weights=market_volumes)
# Implementation shortfall vs arrival price
arrival_price = market_prices[0]
shortfall_bps = (
(avg_exec_price - arrival_price) / arrival_price * 10000
* (1 if self.side == "buy" else -1)
)
return {
"fills": fills,
"avg_exec_price": round(avg_exec_price, 4),
"market_vwap": round(market_vwap, 4),
"vs_vwap_bps": round(
(avg_exec_price - market_vwap) / market_vwap * 10000, 2
),
"implementation_shortfall_bps": round(shortfall_bps, 2),
"total_filled": total_filled,
"total_cost": round(total_cost, 2)
}
Running the Simulation
np.random.seed(42)
intervals = 78
# Simulate a trending day: price drifts up 0.8% over session
prices = 1450 * np.cumprod(1 + np.random.normal(0.0001, 0.001, intervals))
volume_profile = generate_intraday_volume_profile(total_volume=2_000_000)
# Execute 10,000 shares (0.5% of ADV) — meaningful size
executor = VWAPExecutor(
target_quantity=10_000,
side="buy",
volume_profile=volume_profile,
participation_rate=0.08,
urgency=0.0
)
result = executor.simulate_execution(prices, volume_profile)
print(f"Average execution price : ₹{result['avg_exec_price']:.4f}")
print(f"Market VWAP : ₹{result['market_vwap']:.4f}")
print(f"vs. VWAP : {result['vs_vwap_bps']:.1f} bps")
print(f"Implementation shortfall: {result['implementation_shortfall_bps']:.1f} bps")
print(f"Total cost : ₹{result['total_cost']:,.0f}")
⚠️ On a trending day, even a perfect VWAP execution will show positive implementation shortfall — you're buying into a rising market and paying more than the arrival price. That's not a failure. It's physics. The benchmark to beat is VWAP, not clairvoyance.
Part 3: Realistic Slippage Modeling for Backtesting
Most backtesters use one of three slippage models:
- None — optimistic to the point of delusion
- Fixed bps — better, but ignores order size and liquidity
- Proportional to spread — decent for liquid stocks, breaks on everything else
The correct model accounts for order size, volatility, and liquidity. Here it is:
class SlippageModel:
"""
A tiered slippage model combining:
1. Half-spread cost (you pay to cross)
2. Market impact (square-root law)
3. Timing slippage (latency + price drift)
"""
def __init__(
self,
spread_bps: float = 5.0, # typical bid-ask spread in bps
impact_coeff: float = 0.1, # calibrate to your market
latency_ms: float = 50.0, # your system's order-to-fill latency
volatility_per_ms: float = 0.00005 # per-ms price volatility
):
self.spread_bps = spread_bps
self.impact_coeff = impact_coeff
self.latency_ms = latency_ms
self.vol_per_ms = volatility_per_ms
def estimate(
self,
order_size: int,
avg_daily_volume: int,
side: str,
is_market_order: bool = True
) -> dict:
direction = 1 if side == "buy" else -1
# Component 1: Crossing the spread
# Market orders pay full spread, limit orders pay ~0
spread_cost = (self.spread_bps / 2) if is_market_order else 0.0
# Component 2: Market impact (Almgren-Chriss square-root law)
participation = order_size / avg_daily_volume
impact_bps = self.impact_coeff * np.sqrt(participation) * 10000
# Component 3: Timing slippage
# Price drifts during the latency window
timing_bps = (
self.vol_per_ms * self.latency_ms * np.sqrt(self.latency_ms) * 10000
)
total_bps = direction * (spread_cost + impact_bps + timing_bps)
return {
"spread_cost_bps": round(spread_cost, 3),
"market_impact_bps": round(impact_bps, 3),
"timing_slippage_bps": round(timing_bps, 3),
"total_slippage_bps": round(total_bps, 3),
"total_slippage_pct": round(total_bps / 100, 4)
}
def apply_to_price(
self,
price: float,
order_size: int,
avg_daily_volume: int,
side: str
) -> float:
est = self.estimate(order_size, avg_daily_volume, side)
return round(price * (1 + est["total_slippage_pct"] / 100), 4)
Plugging Into a Simple Backtest Loop
def backtest_with_slippage(
signals: pd.DataFrame, # columns: date, signal (+1/-1/0), price
adv: float = 500_000, # average daily volume
order_size: int = 1_000,
spread_bps: float = 5.0,
impact_coeff: float = 0.1
) -> pd.DataFrame:
"""
Runs a signal series through the slippage model.
Compares ideal (no slippage) vs realistic execution.
"""
slippage = SlippageModel(spread_bps=spread_bps, impact_coeff=impact_coeff)
results = []
for _, row in signals.iterrows():
if row["signal"] == 0:
continue
side = "buy" if row["signal"] > 0 else "sell"
ideal_price = row["price"]
real_price = slippage.apply_to_price(
ideal_price, order_size, adv, side
)
cost_breakdown = slippage.estimate(order_size, adv, side)
results.append({
"date": row["date"],
"side": side,
"ideal_price": ideal_price,
"real_price": real_price,
"total_slippage_bps": cost_breakdown["total_slippage_bps"],
"spread_component": cost_breakdown["spread_cost_bps"],
"impact_component": cost_breakdown["market_impact_bps"],
"timing_component": cost_breakdown["timing_slippage_bps"],
})
return pd.DataFrame(results)
Part 4: Transaction Cost Analysis (TCA)
TCA is how you hold yourself accountable after the fact. You ran your strategy live. Now what? You measure the gap between what you expected and what you got — and you decompose it into fixable vs unfixable parts.
class TransactionCostAnalyzer:
def __init__(self, fills: pd.DataFrame, signals: pd.DataFrame):
"""
fills: actual execution data
columns: date, ticker, side, qty, exec_price
signals: what your model wanted
columns: date, ticker, side, qty, signal_price, arrival_price
"""
self.fills = fills
self.signals = signals
self.merged = pd.merge(fills, signals, on=["date", "ticker", "side"])
def implementation_shortfall(self) -> pd.DataFrame:
"""
IS = (avg_exec - arrival_price) / arrival_price * 10000
The total cost of the decision to trade.
Positive = you paid more than arrival price (bad for buys).
"""
df = self.merged.copy()
direction = df["side"].map({"buy": 1, "sell": -1})
df["IS_bps"] = (
(df["exec_price"] - df["arrival_price"]) /
df["arrival_price"] * 10000 * direction
)
return df[["date", "ticker", "side", "qty", "IS_bps"]]
def vs_vwap(self) -> pd.DataFrame:
"""
How did you do relative to the day's VWAP?
Requires market_vwap column in fills.
"""
df = self.merged.copy()
direction = df["side"].map({"buy": 1, "sell": -1})
df["vs_vwap_bps"] = (
(df["exec_price"] - df["market_vwap"]) /
df["market_vwap"] * 10000 * direction
)
return df[["date", "ticker", "side", "qty", "vs_vwap_bps"]]
def cost_summary(self) -> dict:
is_df = self.implementation_shortfall()
total_notional = (self.fills["exec_price"] * self.fills["qty"]).sum()
avg_is = is_df["IS_bps"].mean()
total_cost_bps = avg_is
total_cost_currency = total_notional * (avg_is / 10000)
breakdown = {
"total_trades": len(self.fills),
"total_notional": round(total_notional, 2),
"avg_implementation_shortfall_bps": round(avg_is, 2),
"total_execution_cost": round(total_cost_currency, 2),
"worst_trade_bps": round(is_df["IS_bps"].max(), 2),
"best_trade_bps": round(is_df["IS_bps"].min(), 2),
"pct_trades_beating_arrival": round(
(is_df["IS_bps"] < 0).mean() * 100, 1
)
}
return breakdown
def diagnose(self) -> str:
summary = self.cost_summary()
avg_is = summary["avg_implementation_shortfall_bps"]
pct_beating = summary["pct_trades_beating_arrival"]
lines = ["=== TCA Diagnosis ===\n"]
if avg_is < 5:
lines.append("✓ Execution quality: GOOD (< 5 bps avg IS)")
elif avg_is < 15:
lines.append("⚠ Execution quality: MODERATE (5–15 bps avg IS)")
lines.append(" → Review order sizing and urgency settings")
else:
lines.append("✗ Execution quality: POOR (> 15 bps avg IS)")
lines.append(" → Significantly impacting strategy alpha")
lines.append(" → Consider reducing order size or using limit orders")
if pct_beating > 50:
lines.append(f"✓ {pct_beating}% of trades beat arrival price")
else:
lines.append(
f"⚠ Only {pct_beating}% of trades beat arrival price "
f"— possible adverse selection"
)
return "\n".join(lines)
Running a Full TCA Report
# Simulate some fills and signals for demonstration
np.random.seed(42)
n_trades = 50
dates = pd.date_range("2024-01-01", periods=n_trades, freq="B")
arrival_prices = 1450 + np.cumsum(np.random.normal(0, 5, n_trades))
# Simulate execution with some slippage noise
exec_prices = arrival_prices + np.random.normal(0.5, 0.3, n_trades)
market_vwap = arrival_prices + np.random.normal(0.2, 0.2, n_trades)
fills_df = pd.DataFrame({
"date": dates,
"ticker": "NIFTY50-FUT",
"side": np.random.choice(["buy", "sell"], n_trades),
"qty": np.random.randint(500, 2000, n_trades),
"exec_price": exec_prices,
"market_vwap": market_vwap
})
signals_df = fills_df[["date", "ticker", "side", "qty"]].copy()
signals_df["signal_price"] = arrival_prices
signals_df["arrival_price"] = arrival_prices
analyzer = TransactionCostAnalyzer(fills_df, signals_df)
summary = analyzer.cost_summary()
for k, v in summary.items():
print(f"{k:45s}: {v}")
print("\n")
print(analyzer.diagnose())
Putting It All Together
Here's the full pipeline — signal fires, VWAP executor slices the order, slippage model adjusts fills, TCA measures the outcome:
def full_execution_pipeline(
signal_price: float,
target_qty: int,
side: str,
adv: int,
intraday_prices: np.ndarray,
intraday_volumes: np.ndarray
) -> dict:
slippage_model = SlippageModel(spread_bps=4.0, impact_coeff=0.08)
executor = VWAPExecutor(
target_quantity=target_qty,
side=side,
volume_profile=intraday_volumes,
participation_rate=0.08,
urgency=0.2
)
exec_result = executor.simulate_execution(intraday_prices, intraday_volumes)
slippage_est = slippage_model.estimate(target_qty, adv, side)
return {
"signal_price": signal_price,
"avg_exec_price": exec_result["avg_exec_price"],
"vs_vwap_bps": exec_result["vs_vwap_bps"],
"implementation_shortfall_bps": exec_result["implementation_shortfall_bps"],
"slippage_estimate_bps": slippage_est["total_slippage_bps"],
"spread_component_bps": slippage_est["spread_cost_bps"],
"impact_component_bps": slippage_est["market_impact_bps"]
}
# Run it
volume_profile = generate_intraday_volume_profile(2_000_000)
prices = 1450 * np.cumprod(1 + np.random.normal(0.00005, 0.0008, 78))
output = full_execution_pipeline(
signal_price=prices[0],
target_qty=8_000,
side="buy",
adv=2_000_000,
intraday_prices=prices,
intraday_volumes=volume_profile
)
for k, v in output.items():
print(f"{k:40s}: {v}")
What You Now Know (That Most Backtests Don't)
The execution layer isn't optional. It's where alpha goes to quietly disappear.
Here's what changed in your toolkit today:
- You can simulate order book dynamics and see exactly how your order walks the book
- You have a VWAP executor that participates proportionally with market volume
- Your backtest now has a slippage model that decomposes into spread, impact, and timing
- You can run TCA on live trades and diagnose whether execution is killing your strategy
The SlippageModel, VWAPExecutor, and TransactionCostAnalyzer classes above are ready to drop into your existing backtesting framework. Calibrate spread_bps and impact_coeff to your target instruments. Run the TCA against your first 30 days of live trades.
If your implementation shortfall is above 15 bps on average, your execution is meaningfully eating into alpha and needs attention before anything else.
If it's below 5 bps, your backtest-to-live performance gap is probably coming from somewhere else — and that's a different post entirely.
Next on Quant Tech Dispatch: Adverse selection — why market makers always seem to know when your limit order is about to get filled at the worst possible moment.