# Paper Slippage Model > "Implements slippage modeling and execution simulation for risk management and algorithmic trading execution." - Skill: `paulpas/paper-slippage-model` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/paper-slippage-model` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/paper-slippage-model/raw - Safety review: PASS (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/paper-slippage-model --- **Role:** Execution Quality Specialist — implements comprehensive slippage models to simulate real-world execution quality, accounting for market impact, liquidity constraints, and order flow dynamics. **Philosophy:** Realistic Simulation — paper trading must mirror live execution quality; ignoring slippage or using simplistic models produces misleading performance metrics that fail to predict live trading outcomes. ## Key Principles 1. **Slippage Sources**: Slippage arises from multiple sources: bid-ask spread, market impact, liquidity provision, and order type; all should be modeled accurately. 2. **Historical Analysis**: Slippage should be analyzed from historical execution data to calibrate models to actual market conditions and instrument characteristics. 3. **Order Size Dependence**: Slippage scales non-linearly with order size; small orders have minimal slippage while large orders face significant impact. 4. **Time-of-Day Effects**: Slippage varies significantly by time of day due to liquidity patterns; models must account for intraday liquidity cycles. 5. **VWAP Comparison**: Slippage should be measured relative to VWAP rather than just execution price to properly assess execution quality. ## Implementation Guidelines ### Structure - Core logic: `skills/paper-trading/slippage_model.py` - Historical analysis: `skills/paper-trading/slippage_analysis.py` - Tests: `skills/tests/test_slippage_model.py` ### Patterns to Follow - Implement slippage as a configurable model with multiple components - Support both deterministic and stochastic slippage models - Include historical slippage analysis and calibration - Provide execution simulation with realistic timing - Use vectorized operations for efficient batch simulations ## Adherence Checklist Before completing your task, verify: - [ ] **Bid-Ask Component**: Is the bid-ask spread properly modeled for each instrument? - [ ] **Market Impact**: Does the model include order-size-dependent market impact? - [ ] **Liquidity Adjustment**: Are slippage estimates adjusted for instrument liquidity? - [ ] **Time-of-Day Effects**: Does the model account for intraday liquidity variations? - [ ] **VWAP Reference**: Is slippage measured relative to VWAP, not just spot price? ## Code Examples ### Slippage Model with Multiple Components ```python from dataclasses import dataclass from typing import List, Tuple, Dict, Optional import numpy as np import pandas as pd from enum import Enum from datetime import datetime, time class SlippageComponent(Enum): """Components of slippage.""" BID_ASK = "bid_ask" MARKET_IMPACT = "market_impact" LIQUIDITY = "liquidity" TIME_OF_DAY = "time_of_day" ORDER_TYPE = "order_type" @dataclass class SlippageParameters: """Parameters for slippage model.""" bid_ask_spread: float # As fraction of price (e.g., 0.0001 = 10 bps) market_impact_coeff: float # Coefficient for order-size impact liquidity_factor: float # Inverse liquidity multiplier intraday_multiplier: float # Intraday liquidity multiplier order_type_modifier: float # Adjust based on order type @classmethod def default(cls) -> 'SlippageParameters': return cls( bid_ask_spread=0.0001, market_impact_coeff=0.00005, liquidity_factor=1.0, intraday_multiplier=1.0, order_type_modifier=1.0 ) @dataclass class SlippageResult: """Result of slippage calculation.""" total_slippage: float bid_ask_component: float market_impact_component: float liquidity_component: float time_of_day_component: float order_type_component: float execution_price: float reference_price: float slippage_bps: float execution_quality: str class SlippageModel: """ Comprehensive slippage model with multiple components. Models bid-ask spread, market impact, liquidity effects, and intraday variation. """ def __init__(self, parameters: Optional[SlippageParameters] = None, instrument_liquidity: float = 1.0): """ Initialize slippage model. Args: parameters: Slippage parameters (uses defaults if not provided) instrument_liquidity: Instrument liquidity factor (1.0 = average) """ self.parameters = parameters or SlippageParameters.default() self.instrument_liquidity = instrument_liquidity def calculate_slippage(self, order_size: float, order_type: str, reference_price: float, execution_time: Optional[datetime] = None, market_volume: Optional[float] = None, avg_daily_volume: float = 1000000) -> SlippageResult: """ Calculate total slippage for an order. Args: order_size: Size of the order (positive = buy, negative = sell) order_type: Type of order ('market', 'limit', 'stop') reference_price: Reference price (e.g., VWAP) execution_time: Optional execution time for intraday effects market_volume: Optional current market volume avg_daily_volume: Average daily volume for normalization Returns: SlippageResult with all components """ # Determine direction is_buy = order_size > 0 order_size_abs = abs(order_size) # 1. Bid-Ask Spread Component # Half spread is typical for market orders bid_ask_component = self.parameters.bid_ask_spread / 2 if order_type == 'limit': bid_ask_component *= 0.1 # Much less for limit orders # Direction adjustment if is_buy: bid_ask_component = abs(bid_ask_component) else: bid_ask_component = -abs(bid_ask_component) # 2. Market Impact Component # Order size impact scales non-linearly # Impact = coefficient * (order_size / ADV)^exponent volume_ratio = order_size_abs / avg_daily_volume if avg_daily_volume > 0 else 1 volume_ratio = min(volume_ratio, 1.0) # Cap at 100% of ADV exponent = 1.5 # Typical exponent for market impact market_impact = self.parameters.market_impact_coeff * ( volume_ratio ** exponent ) # Direction adjustment if is_buy: market_impact = abs(market_impact) else: market_impact = -abs(market_impact) # 3. Liquidity Component # Adjust based on instrument liquidity liquidity_component = ( self.parameters.liquidity_factor / self.instrument_liquidity - 1 ) * 0.0001 # Typical liquidity adjustment # Direction adjustment (larger orders in low liquidity face more impact) liquidity_component *= (1 + volume_ratio * 0.5) if is_buy: liquidity_component = abs(liquidity_component) else: liquidity_component = -abs(liquidity_component) # 4. Time-of-Day Component # Adjust based on intraday liquidity patterns time_of_day_component = 0.0 if execution_time is not None: hour = execution_time.hour minute = execution_time.minute # Calculate minutes from market open (9:30 AM EST = 570 minutes) market_open_minutes = 9 * 60 + 30 market_close_minutes = 16 * 60 current_minutes = hour * 60 + minute minutes_since_open = max(0, current_minutes - market_open_minutes) minutes_before_close = max(0, market_close_minutes - current_minutes) # Intraday pattern: higher slippage at open and close if minutes_since_open < 30: time_of_day_component = 0.0002 # Higher slippage at open elif minutes_before_close < 30: time_of_day_component = 0.00015 # Higher slippage at close else: # Midday has lower slippage time_of_day_component = 0.00005 # Apply intraday multiplier from parameters time_of_day_component *= self.parameters.intraday_multiplier # 5. Order Type Component order_component = 0.0 if order_type == 'market': order_component = 0.0 # Market orders already include spread elif order_type == 'limit': order_component = -0.00003 # Slight benefit for limit orders elif order_type == 'stop': order_component = 0.0001 # Higher slippage for stop orders # Apply order type modifier order_component *= self.parameters.order_type_modifier # Calculate total slippage total_slippage = ( bid_ask_component + market_impact + liquidity_component + time_of_day_component + order_component ) # Calculate execution price execution_price = reference_price * (1 + total_slippage) # Convert to basis points slippage_bps = total_slippage * 10000 # Determine execution quality if slippage_bps < -5: quality = "Excellent" elif slippage_bps < 0: quality = "Good" elif slippage_bps < 5: quality = "Average" elif slippage_bps < 10: quality = "Poor" else: quality = "Very Poor" return SlippageResult( total_slippage=total_slippage, bid_ask_component=bid_ask_component, market_impact_component=market_impact, liquidity_component=liquidity_component, time_of_day_component=time_of_day_component, order_type_component=order_component, execution_price=execution_price, reference_price=reference_price, slippage_bps=slippage_bps, execution_quality=quality ) def simulate_execution(self, order_size: float, reference_price: float, execution_time: Optional[datetime] = None, market_conditions: Optional[Dict] = None) -> Dict: """ Simulate full execution process with realistic timing. Args: order_size: Size of the order reference_price: Reference price for execution execution_time: Time of execution market_conditions: Optional market conditions dict Returns: Dictionary with execution simulation results """ market_conditions = market_conditions or {} # Default market conditions avg_daily_volume = market_conditions.get('adv', 1000000) instrument_liquidity = market_conditions.get('liquidity', 1.0) current_volume = market_conditions.get('current_volume', avg_daily_volume / 260) # Calculate slippage result = self.calculate_slippage( order_size=order_size, order_type=market_conditions.get('order_type', 'market'), reference_price=reference_price, execution_time=execution_time, market_volume=current_volume, avg_daily_volume=avg_daily_volume ) # Add execution timing execution_delay_ms = np.random.exponential(50) # Exponential delay return { "order_size": order_size, "reference_price": reference_price, "execution_price": result.execution_price, "slippage_bps": result.slippage_bps, "execution_quality": result.execution_quality, "execution_delay_ms": execution_delay_ms, "components": { "bid_ask": result.bid_ask_component * 10000, "market_impact": result.market_impact_component * 10000, "liquidity": result.liquidity_component * 10000, "time_of_day": result.time_of_day_component * 10000, "order_type": result.order_type_component * 10000 } } # Historical Slippage Analysis class SlippageAnalyzer: """ Analyze historical slippage data to calibrate models. """ def __init__(self, executions: pd.DataFrame, prices: pd.DataFrame): """ Initialize slippage analyzer. Args: executions: DataFrame of execution history prices: DataFrame of price data (VWAP, close, etc.) """ self.executions = executions self.prices = prices def analyze_slippage_distribution(self) -> Dict: """ Analyze historical slippage distribution. Returns: Dictionary with slippage statistics """ # Merge executions with prices merged = self.executions.merge( self.prices[['timestamp', 'vwap', 'close']], left_on='timestamp', right_on='timestamp', how='left' ) # Calculate slippage relative to VWAP merged['slippage'] = (merged['execution_price'] - merged['vwap']) / merged['vwap'] # Statistics total_slippage = merged['slippage'].mean() total_std = merged['slippage'].std() percentile_95 = merged['slippage'].quantile(0.95) percentile_5 = merged['slippage'].quantile(0.05) # By order type by_type = merged.groupby('order_type')['slippage'].agg(['mean', 'std', 'count']) # By hour of day merged['hour'] = merged['timestamp'].dt.hour by_hour = merged.groupby('hour')['slippage'].mean() return { "mean_slippage": total_slippage, "std_slippage": total_std, "p95_slippage": percentile_95, "p5_slippage": percentile_5, "by_order_type": by_type.to_dict(), "by_hour": by_hour.to_dict(), "n_executions": len(merged) } def fit_slippage_model(self) -> SlippageParameters: """ Fit slippage model parameters from historical data. Returns: Fitted SlippageParameters """ analysis = self.analyze_slippage_distribution() # Estimate bid-ask spread from limit order execution limit_orders = self.executions[self.executions['order_type'] == 'limit'] if len(limit_orders) > 0: limit_slippage = abs(limit_orders['slippage']).mean() bid_ask_spread = limit_slippage * 2 # Estimate from limit order execution else: bid_ask_spread = 0.0001 # Default # Estimate market impact coefficient # Slippage ~ coefficient * (order_size / ADV)^1.5 if 'order_size' in self.executions.columns and 'adv' in self.executions.columns: self.executions['size_ratio'] = abs(self.executions['order_size']) / self.executions['adv'] self.executions['size_ratio'] = self.executions['size_ratio'].clip(0, 1) # Simple regression estimation x = self.executions['size_ratio'] ** 1.5 y = abs(self.executions['slippage']) # Linear regression: y = a + b*x x_mean = x.mean() y_mean = y.mean() numerator = ((x - x_mean) * (y - y_mean)).sum() denominator = ((x - x_mean) ** 2).sum() slope = numerator / denominator if denominator > 0 else 0.0001 intercept = y_mean - slope * x_mean market_impact_coeff = max(slope, 0.00001) else: market_impact_coeff = 0.00005 # Default return SlippageParameters( bid_ask_spread=bid_ask_spread, market_impact_coeff=market_impact_coeff, liquidity_factor=1.0, intraday_multiplier=analysis['by_hour'].get(10, 1.0), # Mid-morning order_type_modifier=1.0 ) # Execution Simulation System class ExecutionSimulator: """ Simulate multiple executions with realistic slippage. Used for strategy backtesting with accurate execution assumptions. """ def __init__(self, slippage_model: SlippageModel, avg_daily_volume: float = 1000000, market_hours: Tuple[int, int] = (9, 16)): """ Initialize execution simulator. Args: slippage_model: Slippage model to use avg_daily_volume: Average daily volume for normalization market_hours: Tuple of (market_open_hour, market_close_hour) """ self.slippage_model = slippage_model self.avg_daily_volume = avg_daily_volume self.market_open = market_hours[0] self.market_close = market_hours[1] def simulate_order(self, size: float, reference_price: float, execution_time: Optional[datetime] = None, order_type: str = 'market', instrument_liquidity: float = 1.0) -> Dict: """ Simulate a single order execution. Args: size: Order size reference_price: Reference price execution_time: Optional execution time order_type: Order type instrument_liquidity: Instrument liquidity factor Returns: Dictionary with execution details """ # Set default time to current hour if not provided if execution_time is None: execution_time = datetime.now().replace( minute=0, second=0, microsecond=0 ) # Generate market volume for time of day hour = execution_time.hour minutes_since_open = max(0, hour - self.market_open) * 60 minutes_before_close = max(0, self.market_close - hour) * 60 # Intraday volume pattern if minutes_since_open < 30 or minutes_before_close < 30: current_volume = self.avg_daily_volume * 0.15 elif 30 <= minutes_since_open <= 360: current_volume = self.avg_daily_volume * 0.35 else: current_volume = self.avg_daily_volume * 0.20 market_conditions = { 'adv': self.avg_daily_volume, 'current_volume': current_volume, 'liquidity': instrument_liquidity, 'order_type': order_type } return self.slippage_model.simulate_execution( order_size=size, reference_price=reference_price, execution_time=execution_time, market_conditions=market_conditions ) def simulate_batch(self, orders: List[Dict], prices: pd.DataFrame) -> pd.DataFrame: """ Simulate batch of order executions. Args: orders: List of order dictionaries prices: DataFrame of price data Returns: DataFrame with all execution results """ results = [] for i, order in enumerate(orders): reference_price = prices.loc[order['timestamp'], 'vwap'] if order['timestamp'] in prices.index else order.get('reference_price', 100) execution = self.simulate_order( size=order.get('size', 0), reference_price=reference_price, execution_time=order.get('timestamp'), order_type=order.get('order_type', 'market'), instrument_liquidity=order.get('instrument_liquidity', 1.0) ) result = { 'order_index': i, 'timestamp': order.get('timestamp'), 'order_size': order.get('size', 0), 'order_type': order.get('order_type', 'market'), 'reference_price': reference_price, 'execution_price': execution['execution_price'], 'slippage_bps': execution['slippage_bps'], 'execution_quality': execution['execution_quality'], 'total_cost': abs(order.get('size', 0)) * execution['execution_price'] } results.append(result) return pd.DataFrame(results) if __name__ == "__main__": # Example usage model = SlippageModel() # Test different order scenarios scenarios = [ {"size": 100, "type": "limit", "time": None}, {"size": 500, "type": "market", "time": None}, {"size": 1000, "type": "market", "time": None}, {"size": 5000, "type": "market", "time": None}, {"size": 10000, "type": "market", "time": None}, ] print("Slippage Analysis for Different Order Sizes") print("=" * 60) for scenario in scenarios: result = model.calculate_slippage( order_size=scenario["size"], order_type=scenario["type"], reference_price=100.0, execution_time=datetime.now().replace(hour=10, minute=30) ) print(f"\nOrder: {scenario['size']} shares, {scenario['type']}") print(f" Total Slippage: {result.slippage_bps:.2f} bps") print(f" Execution Quality: {result.execution_quality}") print(f" Components (bps):") print(f" Bid-Ask: {result.bid_ask_component * 10000:.2f}") print(f" Market Impact: {result.market_impact_component * 10000:.2f}") print(f" Liquidity: {result.liquidity_component * 10000:.2f}") print(f" Time of Day: {result.time_of_day_component * 10000:.2f}") ``` ### VWAP-Based Slippage Assessment ```python class VWAPSlippageAnalyzer: """ Analyze slippage relative to VWAP for execution quality assessment. VWAP is the standard reference for execution quality. """ def __init__(self, executions: pd.DataFrame, vwap_data: pd.DataFrame): self.executions = executions self.vwap_data = vwap_data def calculate_vwap_slippage(self, window_minutes: int = 5) -> pd.DataFrame: """ Calculate slippage relative to VWAP within time windows. Args: window_minutes: Time window for VWAP calculation Returns: DataFrame with VWAP-adjusted slippage """ # Create time windows self.executions = self.executions.sort_values('timestamp') # Calculate VWAP within windows vwap_df = self.vwap_data.resample(f'{window_minutes}T').agg({ 'price': 'mean', 'volume': 'sum' }) vwap_df['vwap'] = vwap_df['price'] * vwap_df['volume'] / vwap_df['volume'].sum() # Merge with executions merged = self.executions.merge( vwap_df[['vwap']], left_on='timestamp', right_index=True, how='left' ) # Calculate slippage merged['vwap_slippage'] = ( merged['execution_price'] - merged['vwap'] ) / merged['vwap'] return merged def assess_execution_quality(self) -> Dict: """ Comprehensive execution quality assessment. Returns: Dictionary with quality metrics """ merged = self.calculate_vwap_slippage() # Overall statistics total_slippage = merged['vwap_slippage'].mean() total_std = merged['vwap_slippage'].std() # By order type by_type = merged.groupby('order_type')['vwap_slippage'].agg(['mean', 'std', 'count']) # By size bucket merged['size_bucket'] = pd.cut( merged['order_size'].abs(), bins=[0, 100, 500, 1000, 5000, float('inf')], labels=['0-100', '100-500', '500-1k', '1k-5k', '5k+'] ) by_size = merged.groupby('size_bucket')['vwap_slippage'].mean() # VWAP hit rate (executions on same side of VWAP as intended direction) merged['direction'] = np.where(merged['order_size'] > 0, 'buy', 'sell') merged['vwap_direction'] = np.where(merged['vwap_slippage'] > 0, 'negative', 'positive') merged['hit'] = (merged['direction'] == 'buy') == (merged['vwap_direction'] == 'positive') hit_rate = merged['hit'].mean() return { "total_average_slippage_bps": total_slippage * 10000, "total_std_slippage_bps": total_std * 10000, "execution_count": len(merged), "vwap_hit_rate": hit_rate, "by_order_type": by_type.to_dict(), "by_size_bucket": by_size.to_dict() } # Intraday Slippage Pattern Analysis class IntradaySlippagePatterns: """ Analyze and model intraday slippage patterns. Slippage varies significantly by time of day. """ def __init__(self, executions: pd.DataFrame): self.executions = executions def analyze_intraday_patterns(self) -> Dict: """ Analyze intraday slippage patterns. Returns: Dictionary with intraday patterns """ # Extract hour and minute self.executions['hour'] = self.executions['timestamp'].dt.hour self.executions['minute'] = self.executions['timestamp'].dt.minute # Calculate time since market open self.executions['minutes_since_open'] = ( self.executions['hour'] - 9 ) * 60 + (self.executions['minute'] - 30) # Ensure positive self.executions['minutes_since_open'] = self.executions['minutes_since_open'].clip(0) # By minute buckets self.executions['minute_bucket'] = self.executions['minutes_since_open'] // 5 * 5 # Calculate slippage if 'vwap' in self.executions.columns: self.executions['slippage'] = ( self.executions['execution_price'] - self.executions['vwap'] ) / self.executions['vwap'] # Intraday pattern pattern = self.executions.groupby('minute_bucket')['slippage'].agg(['mean', 'std', 'count']) # Peak times (highest slippage) peak_times = pattern['mean'].nlargest(3).index.tolist() # Lull times (lowest slippage) lull_times = pattern['mean'].nsmallest(3).index.tolist() # Volatility pattern vol_pattern = self.executions.groupby('minute_bucket')['slippage'].std() return { "pattern": pattern.to_dict(), "peak_slippage_minutes": peak_times, "lull_slippage_minutes": lull_times, "volatility_pattern": vol_pattern.to_dict(), "max_slippage_hour": pattern['mean'].idxmax(), "min_slippage_hour": pattern['mean'].idxmin() } ``` --- --- ## Constraints ### MUST DO - Implement commission structures that reflect real broker fee schedules including per-share, per-contract, and regulatory fees - Model market impact for paper orders: simulate price movement caused by your order based on order book depth - Use the same data feeds and latency characteristics as live trading to ensure paper results are realistic - Track slippage separately from commissions and calculate both fill-level and execution-level slippage metrics - Validate paper trading P&L against theoretical expectations at regular intervals to detect simulation bugs ### MUST NOT DO - Do not assume fills at the next available price without modeling order book dynamics or queue position - Avoid using perfect historical data for paper trading — add realistic noise and latency characteristics - Never use different data sources between backtesting, paper trading, and live execution without documenting the delta - Do not ignore dividend adjustments in equity paper trading — missed dividends create false P&L discrepancies ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [Understanding Slippage in Trading](https://www.investopedia.com/terms/s/slippage.asp) - [Slippage Modeling for Backtests](https://docs.quantconnect.com/tutorials/commission-models) - [Price Impact and Execution Quality](https://en.wikipedia.org/wiki/Market_impact) - [Reducing Slippage in Algorithmic Trading](https://www.investopedia.com/articles/trading/08/slippage.asp) - [Slippage Estimation Methods](https://docs.quantconnect.com/tutorials/live-trading-overview)