# Execution Twap > "Time-Weighted Average Price algorithm for executing large orders with" minimal market impact - Skill: `paulpas/execution-twap` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/execution-twap` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/execution-twap/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/execution-twap --- **Role:** Execute large orders over time to minimize market impact and achieve good prices **Philosophy:** TWAP provides a disciplined approach to execution; the algorithm's predictability reduces slippage ## Key Principles 1. **Time-Based Execution**: Divide order into equal time slices 2. **Order Sizing**: Calculate slice size based on remaining time and quantity 3. **Slippage Monitoring**: Track execution quality vs. benchmark 4. **Adaptive Rate Control**: Adjust execution rate based on market conditions 5. **Completion Tracking**: Ensure all slices are executed within time window ## Implementation Guidelines ### Structure - Core logic: execution/twap.py - Order manager: execution/order_manager.py - Tests: tests/test_twap.py ### Patterns to Follow - Use asyncio for time-based scheduling - Implement completion callbacks - Support pause/resume functionality - Track execution metrics ## Adherence Checklist Before completing your task, verify: - [ ] Time slices are calculated correctly - [ ] Slippage is monitored throughout execution - [ ] Order completion is guaranteed - [ ] Execution metrics are recorded - [ ] Rate limiting is applied Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory. ## Python Implementation ```python import asyncio import time from typing import Dict, List, Optional, Callable from dataclasses import dataclass, field from datetime import datetime from enum import Enum import logging class TWAPState(Enum): PENDING = "pending" ACTIVE = "active" PAUSED = "paused" COMPLETED = "completed" CANCELLED = "cancelled" @dataclass class TWAPSlice: """A single TWAP execution slice.""" slice_index: int quantity: float price_target: Optional[float] = None executed: bool = False exec_quantity: float = 0.0 exec_price: Optional[float] = None exec_time: Optional[float] = None @dataclass class TWAPConfig: """Configuration for TWAP strategy.""" duration_seconds: float num_slices: int start_time: Optional[float] = None end_time: Optional[float] = None aggressive_first: bool = False aggressive_last: bool = False @dataclass class TWAPMetrics: """Metrics for TWAP execution.""" start_time: float end_time: Optional[float] = None total_quantity: float = 0.0 filled_quantity: float = 0.0 avg_fill_price: float = 0.0 slippage_bps: Optional[float] = None slices_completed: int = 0 total_slices: int = 0 avg_slice_duration: float = 0.0 class TWAPExecutor: """Executes orders using Time-Weighted Average Price strategy.""" def __init__( self, order_id: str, symbol: str, side: str, # 'buy' or 'sell' quantity: float, config: TWAPConfig, order_executor: Any ): self.order_id = order_id self.symbol = symbol self.side = side self.quantity = quantity self.config = config self.executor = order_executor self.state = TWAPState.PENDING self.slices: List[TWAPSlice] = [] self.metrics = TWAPMetrics(start_time=time.time()) self._execution_task: Optional[asyncio.Task] = None self._callbacks: List[Callable] = [] async def start(self) -> bool: """Start TWAP execution.""" if self.state != TWAPState.PENDING: return False self.state = TWAPState.ACTIVE self._initialize_slices() # Schedule execution self._execution_task = asyncio.create_task(self._execute_slices()) return True async def stop(self) -> bool: """Stop TWAP execution.""" if self.state != TWAPState.ACTIVE: return False self.state = TWAPState.PAUSED if self._execution_task: self._execution_task.cancel() try: await self._execution_task except asyncio.CancelledError: pass return True async def resume(self) -> bool: """Resume TWAP execution.""" if self.state != TWAPState.PAUSED: return False self.state = TWAPState.ACTIVE self._execution_task = asyncio.create_task(self._execute_slices()) return True async def cancel(self) -> Dict[str, Any]: """Cancel TWAP execution.""" if self.state not in [TWAPState.ACTIVE, TWAPState.PAUSED]: return {"success": False, "error": "Cannot cancel in current state"} self.state = TWAPState.CANCELLED if self._execution_task: self._execution_task.cancel() try: await self._execution_task except asyncio.CancelledError: pass # Cancel remaining slices results = {"cancelled_slices": [], "filled_quantity": self.metrics.filled_quantity} for slice_obj in self.slices[self.metrics.slices_completed:]: results["cancelled_slices"].append(slice_obj.slice_index) return results def _initialize_slices(self): """Initialize TWAP slices.""" slice_duration = self.config.duration_seconds / self.config.num_slices for i in range(self.config.num_slices): # Calculate slice quantity base_quantity = self.quantity / self.config.num_slices # Apply aggression if configured if self.config.aggressive_first and i == 0: base_quantity *= 1.5 elif self.config.aggressive_last and i == self.config.num_slices - 1: base_quantity *= 1.5 # Ensure we fill the total quantity if i == self.config.num_slices - 1: remaining = self.quantity - sum(s.quantity for s in self.slices) base_quantity = remaining self.slices.append(TWAPSlice( slice_index=i, quantity=base_quantity, price_target=None )) self.metrics.total_quantity = self.quantity self.metrics.total_slices = len(self.slices) async def _execute_slices(self): """Execute all TWAP slices.""" slice_duration = self.config.duration_seconds / self.config.num_slices for slice_obj in self.slices: if self.state != TWAPState.ACTIVE: break # Wait for slice start time if self.state == TWAPState.PENDING: self.state = TWAPState.ACTIVE start_time = time.time() slice_start = start_time # Calculate wait time for this slice expected_slice_start = self.metrics.start_time + (slice_obj.slice_index * slice_duration) wait_time = expected_slice_start - time.time() if wait_time > 0: await asyncio.sleep(wait_time) # Execute slice exec_result = await self._execute_single_slice(slice_obj) if exec_result: slice_obj.executed = True slice_obj.exec_quantity = exec_result["quantity"] slice_obj.exec_price = exec_result["price"] slice_obj.exec_time = time.time() self.metrics.slices_completed += 1 # Update metrics self.metrics.filled_quantity += slice_obj.exec_quantity if self.metrics.filled_quantity > 0: self.metrics.avg_fill_price = ( (self.metrics.avg_fill_price * (self.metrics.filled_quantity - slice_obj.exec_quantity) + slice_obj.exec_price * slice_obj.exec_quantity) / self.metrics.filled_quantity ) # Calculate slippage if benchmark available if slice_obj.price_target: if self.side == "buy": self.metrics.slippage_bps = ( (self.metrics.avg_fill_price - slice_obj.price_target) / slice_obj.price_target * 10000 ) else: self.metrics.slippage_bps = ( (slice_obj.price_target - self.metrics.avg_fill_price) / slice_obj.price_target * 10000 ) # Trigger callbacks for callback in self._callbacks: await callback(slice_obj) # Wait for next slice current_time = time.time() elapsed = current_time - slice_start if elapsed < slice_duration: await asyncio.sleep(slice_duration - elapsed) self.state = TWAPState.COMPLETED self.metrics.end_time = time.time() async def _execute_single_slice(self, slice_obj: TWAPSlice) -> Optional[Dict[str, Any]]: """Execute a single TWAP slice.""" # Create order for slice order = self.executor.create_order( symbol=self.symbol, side=self.side, quantity=slice_obj.quantity, order_type="market" if self.side == "buy" else "market" ) # Wait for fill start_time = time.time() timeout = slice_obj.quantity / 100 # Simplified timeout while time.time() - start_time < timeout: fill = self._check_slice_fill(slice_obj) if fill: return fill await asyncio.sleep(0.1) return None def _check_slice_fill(self, slice_obj: TWAPSlice) -> Optional[Dict[str, Any]]: """Check if slice is filled.""" # In real implementation, would check exchange forfills return {"quantity": slice_obj.quantity, "price": 0.0} # Placeholder def register_callback(self, callback: Callable): """Register callback for slice completion.""" self._callbacks.append(callback) def get_state(self) -> TWAPState: """Get current TWAP state.""" return self.state def get_progress(self) -> Dict[str, Any]: """Get TWAP progress.""" return { "state": self.state.value, "filled_quantity": self.metrics.filled_quantity, "total_quantity": self.metrics.total_quantity, "fill_percentage": (self.metrics.filled_quantity / self.metrics.total_quantity * 100) if self.metrics.total_quantity > 0 else 0, "slices_completed": self.metrics.slices_completed, "total_slices": self.metrics.total_slices, "avg_fill_price": self.metrics.avg_fill_price, "slippage_bps": self.metrics.slippage_bps } def get_metrics(self) -> TWAPMetrics: """Get TWAP metrics.""" return self.metrics ``` ### Usage Example ```python # Create TWAP executor with a 6-hour execution window twap = TWAPExecutor( symbol="BTC/USDT", side=OrderSide.BUY, total_quantity=10.0, duration_seconds=21600, # 6 hours exchange=binance_adapter ) # Start execution twap.start() # Monitor progress while not twap.is_complete(): progress = twap.get_progress() print(f"Progress: {progress['fill_percentage']:.1f}% complete") time.sleep(60) # Get final metrics metrics = twap.get_metrics() print(f"Total filled: {metrics.filled_quantity} @ {metrics.avg_fill_price} BPS slippage: {metrics.slippage_bps}") ``` --- --- ## Constraints ### MUST DO - Implement slippage modeling that accounts for market impact proportional to order size relative to average daily volume - Include pre-trade risk checks (position limits, exposure caps) before any order is submitted to an exchange - Log all execution decisions with timestamps, prices, quantities, and benchmark deviations for post-trade analysis - Support both aggressive (market/limit) and passive (maker) order types with configurable preference based on market regime - Implement circuit breaker logic: pause execution if price moves >X% from decision price or volume drops below threshold ### MUST NOT DO - Do not submit orders without verifying account equity, available margin, and symbol trading status first - Avoid static time-based slicing (e.g., 'divide order by 60 minutes') without considering actual market volume patterns - Never disable pre-trade risk checks for 'backtesting' or 'development' — use a separate simulation environment instead - Do not assume exchange API uptime — always implement retry logic with exponential backoff and fallback routing - Avoid rounding quantity to lot sizes using integer division without checking partial-fill policy compatibility ## 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. - [TWAP Execution Tutorial](https://docs.quantconnect.com/tutorials/time-weighted-average-price) - [Time-Weighted Average Price Strategy](https://www.investopedia.com/terms/t/twap.asp) - [Algorithmic Execution Algorithms Guide](https://www.investopedia.com/articles/trading/10/vwap-metric-trading.asp) - [TWAP vs VWAP Comparison](https://www.investopedia.com/terms/t/twap.asp) - [Execution Quality Analysis](https://arxiv.org/abs/1805.01469)