Optimal Order Execution for
AI Trading Agents

1. The Execution Problem

Most AI trading agents invest enormous effort in signal generation — predicting price direction, sizing positions, and timing entries. But once the decision is made, many agents submit a single market order and call it done. That's leaving money on the table.

For small orders relative to market depth, the naive approach works fine. But as order size grows relative to available liquidity, three forces erode your edge before a single unit of profit is locked in.

1.1 Market Impact

Market impact is the price movement caused by your own order. When you buy, you consume the best ask offers in the order book. If your order is large enough, you exhaust the first price level and begin filling at progressively worse prices — temporary impact. Additionally, other market participants infer from your activity that a well-informed agent is buying, causing them to update their own quotes upward — permanent impact.

The square-root market impact model (Almgren-Chriss framework) quantifies this:

Where x is total order size, V is average daily volume, T is execution horizon, and σ is volatility. The function f is roughly proportional to the square root of participation rate. A 10% participation rate typically costs 3x more than a 1% rate.

1.2 Timing Risk

Every millisecond your order remains unexecuted is a millisecond of price exposure. If you're trying to buy and the market moves up before you finish, you've paid more than your decision price. This is timing risk — the variance of final execution price around the arrival midpoint.

Timing risk grows with the square root of execution time and with volatility. An execution stretched over 60 minutes in a 30% annualized volatility asset has roughly 30% / sqrt(252 * 24) * sqrt(60/60) ≈ 0.36% of one-sigma price uncertainty. For a $100K position, that's ±$360 of luck.

The tradeoff is fundamental: executing faster reduces timing risk but increases market impact. Executing slower reduces impact but increases timing risk. Optimal execution algorithms find the path that minimizes total cost — impact plus timing risk — given your urgency.

1.3 Adverse Selection

Adverse selection is the subtlest of the three forces. Limit orders get filled when someone on the other side decides to take liquidity — and that someone often has more information than you. When your buy limit order fills, it's frequently because a well-informed seller just decided to dump.

This means passive (limit order) execution isn't free. The bid-ask spread you "save" by posting a limit order is partially offset by the adverse selection cost of who fills against you. Sophisticated execution engines measure the post-fill price drift to quantify this cost and adapt their passive/aggressive split accordingly.

2. TWAP: Time-Weighted Average Price

TWAP is the simplest meaningful execution algorithm. It slices your total order into equal pieces and executes them at regular time intervals throughout a defined window. If you want to buy 1,000 units over 60 minutes with 6 slices, you submit one 167-unit order every 10 minutes.

The benchmark for TWAP is the arithmetic mean of all mid-prices sampled during the execution window. If your executed average price beats this benchmark, you "won" your TWAP execution.

2.1 When TWAP Works Best

• No volume data: When you can't reliably predict volume profiles (thin markets, new instruments), equal time slicing avoids the risk of front-loading into a low-volume period.
• Trending markets: In a strong trend, you want to participate at each time interval regardless of volume concentration. TWAP keeps you in step with the trend clock.
• Regulatory/benchmark contexts: Some institutional mandates specify TWAP as the execution benchmark, making it the right tool by definition.
• Simple agent logic: TWAP is easy to implement, audit, and explain. For an AI agent still building its execution track record, TWAP provides a clean baseline to measure future improvements against.

2.2 TWAP Formula and Benchmark

Where P_i is the mid-price at each interval i of N total sampling points. Your execution quality is measured as:

A negative slippage on a buy order means you beat the benchmark — you paid less than the time-weighted average. A positive slippage means you paid more.

2.3 TWAP Variations

Pure TWAP is rarely used in production without modifications:

• Randomized TWAP: Add ±20% randomness to slice size and timing to avoid the predictability that allows front-runners to exploit your pattern.
• Conditional TWAP: Skip or reduce slices when spread is abnormally wide (signal of poor liquidity), resume when spread normalizes.
• Urgency-weighted TWAP: Accelerate slice cadence as the execution window closes if remaining quantity is large — ensuring full fill by deadline.

3. VWAP: Volume-Weighted Average Price

VWAP improves on TWAP by aligning your order flow with market volume patterns. Instead of equal time slices, you execute larger child orders during high-volume periods (when your market impact is proportionally smaller) and smaller orders during low-volume periods.

The insight: if the market naturally processes 30% of its daily volume between 9am and 10am, your order should also be 30% executed during that window. This keeps your participation rate constant throughout the day, minimizing impact per unit traded.

3.1 Volume Profiles in Crypto Markets

Traditional equity VWAP relies on predictable intraday volume profiles — the classic U-shape with high volume at open and close. Crypto markets have different patterns:

• UTC morning spike: Asian session open brings concentrated volume from 0:00-2:00 UTC, especially for BTC/ETH.
• US open surge: 13:30-15:00 UTC typically sees the highest sustained volume globally.
• Weekend depression: Volume drops 30-50% on weekends, spreads widen, and microstructure deteriorates.
• News-driven spikes: Economic data, Fed decisions, and major protocol events cause abrupt volume surges that can temporarily dominate the daily profile.

Purple Flea's trading API provides historical volume data by minute, enabling agents to build dynamic volume profiles that adapt to the current market regime rather than relying on stale averages.

3.2 VWAP Formula

Where P_i is the transaction price and V_i is the volume at each transaction. Your order slice sizes are:

The quality of your VWAP execution depends entirely on the accuracy of your volume forecast. Agents that build better volume prediction models consistently outperform naive VWAP implementors.

3.3 Dynamic VWAP Adjustment

Static VWAP uses a fixed historical volume profile computed before execution begins. Dynamic VWAP continuously updates the remaining schedule based on actual realized volumes. If volume is higher than expected in early intervals, the remaining slices are reduced proportionally.

The adaptive formula for remaining quantity at interval k:

4. Implementation Shortfall & Arrival Price

Implementation shortfall (IS), also known as arrival price benchmarking, is the most theoretically rigorous of the three main execution frameworks. It measures the total cost of execution relative to the price at the moment you made the trading decision — your arrival price.

The concept was formalized by Robert Kissell and Robert Almgren and is now the standard benchmark for institutional execution quality measurement.

4.1 Components of Implementation Shortfall

Total IS has four additive components:

1. Delay cost: Price drift between signal generation and order submission. If you decide to buy at $100 but take 30 seconds to submit (price is now $100.05), you've incurred $0.05 of delay cost.
2. Market impact: The price movement caused by your own execution. This is the permanent component — prices that don't revert after your order completes.
3. Timing risk: The realized variance of execution prices around the expected trajectory. This component has zero expectation but non-zero variance.
4. Opportunity cost: The cost of not completing the order. If you planned to buy 1,000 units but only filled 850 because prices moved away, the missed 150 units carry an opportunity cost.

4.2 IS vs TWAP/VWAP Benchmarks

The key philosophical difference: TWAP and VWAP benchmark against prices that occur during your execution window. Implementation shortfall benchmarks against the price that existed when you decided to trade. This matters enormously.

If you're executing a large buy order over 2 hours while prices drift steadily upward, TWAP will show near-zero slippage (you matched the time-weighted price). But IS will show large positive cost — you paid far more than the arrival price. IS captures the real cost of being a slow mover in a fast market.

4.3 IS-Optimal Execution Schedule

Almgren and Chriss (2000) derived the IS-optimal execution schedule analytically. Under their model assumptions (linear market impact, quadratic utility), the optimal trajectory is an exponential decay:

Where X is initial order size, T is execution horizon, and κ is a risk-aversion parameter derived from impact and volatility. High risk aversion (urgency) leads to front-loaded execution; low risk aversion leads to more evenly spread execution.

5. Python OrderExecutionEngine

Below is a production-quality OrderExecutionEngine class that implements TWAP, VWAP, and an adaptive IS-minimizing algorithm. It integrates directly with Purple Flea's trading API for sliced order submission.

import asyncio
import time
import math
import statistics
from dataclasses import dataclass, field
from typing import List, Optional, Literal
import aiohttp

# Purple Flea Trading API base URL
PF_API = "https://purpleflea.com/api/trading"
API_KEY = "pf_live_YOUR_KEY_HERE"  # replace with your key


@dataclass
class ExecutionSlice:
    """Represents a single child order in an execution schedule."""
    index: int
    quantity: float
    target_time: float       # Unix timestamp for submission
    submitted: bool = False
    fill_price: Optional[float] = None
    fill_quantity: float = 0.0
    order_id: Optional[str] = None
    latency_ms: Optional[float] = None


@dataclass
class ExecutionReport:
    """Post-execution quality report."""
    algorithm: str
    symbol: str
    side: str
    total_quantity: float
    filled_quantity: float
    arrival_price: float
    executed_avg_price: float
    twap_benchmark: float
    slices: List[ExecutionSlice] = field(default_factory=list)

    @property
    def fill_rate(self) -> float:
        return self.filled_quantity / self.total_quantity

    @property
    def implementation_shortfall_bps(self) -> float:
        """IS vs arrival price, in basis points (buy order)."""
        direction = 1 if self.side == "buy" else -1
        return direction * (
            (self.executed_avg_price - self.arrival_price) / self.arrival_price
        ) * 10000

    @property
    def twap_slippage_bps(self) -> float:
        """Slippage vs TWAP benchmark, in basis points."""
        direction = 1 if self.side == "buy" else -1
        return direction * (
            (self.executed_avg_price - self.twap_benchmark) / self.twap_benchmark
        ) * 10000

    def summary(self) -> dict:
        return {
            "algorithm": self.algorithm,
            "symbol": self.symbol,
            "side": self.side,
            "fill_rate": f"{self.fill_rate:.1%}",
            "avg_price": f"{self.executed_avg_price:.6f}",
            "arrival_price": f"{self.arrival_price:.6f}",
            "IS_bps": f"{self.implementation_shortfall_bps:.2f}",
            "twap_slippage_bps": f"{self.twap_slippage_bps:.2f}",
            "n_slices": len(self.slices),
        }


class OrderExecutionEngine:
    """
    Algorithmic order execution engine for AI trading agents.
    Supports TWAP, VWAP, and adaptive IS-minimizing execution.

    Usage:
        engine = OrderExecutionEngine(api_key="pf_live_...")
        report = await engine.twap(
            symbol="BTC-USDT",
            side="buy",
            quantity=0.5,
            duration_minutes=30,
            n_slices=6
        )
        print(report.summary())
    """

    def __init__(self, api_key: str, base_url: str = PF_API):
        self.api_key = api_key
        self.base_url = base_url
        self.session: Optional[aiohttp.ClientSession] = None
        self._price_samples: List[float] = []

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(headers={
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })
        return self

    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()

    # ─── Internal helpers ──────────────────────────────────────────

    async def _get_mid_price(self, symbol: str) -> float:
        """Fetch current mid price from Purple Flea orderbook."""
        url = f"{self.base_url}/orderbook/{symbol}"
        async with self.session.get(url) as resp:
            data = await resp.json()
            best_bid = float(data["bids"][0][0])
            best_ask = float(data["asks"][0][0])
            mid = (best_bid + best_ask) / 2
            self._price_samples.append(mid)
            return mid

    async def _submit_order(
        self,
        symbol: str,
        side: str,
        quantity: float,
        order_type: str = "market",
        limit_price: Optional[float] = None
    ) -> ExecutionSlice:
        """Submit a single child order to Purple Flea trading API."""
        payload = {
            "symbol": symbol,
            "side": side,
            "quantity": quantity,
            "type": order_type,
        }
        if limit_price:
            payload["price"] = limit_price

        t0 = time.time()
        async with self.session.post(
            f"{self.base_url}/orders", json=payload
        ) as resp:
            data = await resp.json()
            latency_ms = (time.time() - t0) * 1000

        return ExecutionSlice(
            index=0,
            quantity=quantity,
            target_time=t0,
            submitted=True,
            fill_price=float(data.get("fill_price", 0)),
            fill_quantity=float(data.get("filled_quantity", 0)),
            order_id=data.get("order_id"),
            latency_ms=latency_ms,
        )

    def _compute_vwap_benchmark(self) -> float:
        """Compute TWAP from sampled prices during execution."""
        if not self._price_samples:
            return 0.0
        return statistics.mean(self._price_samples)

    def _weighted_avg_fill(self, slices: List[ExecutionSlice]) -> float:
        """Compute quantity-weighted average fill price."""
        total_value = sum(s.fill_price * s.fill_quantity for s in slices if s.fill_price)
        total_qty = sum(s.fill_quantity for s in slices)
        return total_value / total_qty if total_qty > 0 else 0.0

    # ─── TWAP ──────────────────────────────────────────────────────

    async def twap(
        self,
        symbol: str,
        side: str,
        quantity: float,
        duration_minutes: float,
        n_slices: int = 10,
        randomize: bool = True,
        order_type: str = "market",
    ) -> ExecutionReport:
        """
        Execute using Time-Weighted Average Price algorithm.

        Args:
            symbol: Trading pair (e.g. "BTC-USDT")
            side: "buy" or "sell"
            quantity: Total quantity to execute
            duration_minutes: Execution window in minutes
            n_slices: Number of child orders (default 10)
            randomize: Add ±20% noise to slice size and timing
            order_type: "market" or "limit"
        """
        self._price_samples = []
        interval_seconds = (duration_minutes * 60) / n_slices
        base_slice_qty = quantity / n_slices

        # Record arrival price
        arrival_price = await self._get_mid_price(symbol)
        slices: List[ExecutionSlice] = []

        for i in range(n_slices):
            # Determine slice quantity with optional randomization
            if randomize:
                noise = 1.0 + ((__import__('random').random() - 0.5) * 0.4)
                # Keep running total in check
                remaining = quantity - sum(s.quantity for s in slices)
                slices_left = n_slices - i
                slice_qty = min(remaining, base_slice_qty * noise)
                if i == n_slices - 1:
                    slice_qty = remaining  # last slice fills remainder
            else:
                slice_qty = base_slice_qty

            # Sample price before submitting
            mid = await self._get_mid_price(symbol)

            # Submit child order
            lp = mid * (1.001 if side == "buy" else 0.999) if order_type == "limit" else None
            s = await self._submit_order(symbol, side, slice_qty, order_type, lp)
            s.index = i
            s.target_time = time.time()
            slices.append(s)

            print(f"TWAP slice {i+1}/{n_slices}: qty={slice_qty:.4f} fill={s.fill_price:.4f}")

            # Wait for next interval (skip wait after last slice)
            if i < n_slices - 1:
                wait = interval_seconds
                if randomize:
                    wait *= 1.0 + ((__import__('random').random() - 0.5) * 0.2)
                await asyncio.sleep(wait)

        avg_fill = self._weighted_avg_fill(slices)
        twap_benchmark = self._compute_vwap_benchmark()

        return ExecutionReport(
            algorithm="TWAP",
            symbol=symbol,
            side=side,
            total_quantity=quantity,
            filled_quantity=sum(s.fill_quantity for s in slices),
            arrival_price=arrival_price,
            executed_avg_price=avg_fill,
            twap_benchmark=twap_benchmark,
            slices=slices,
        )

    # ─── VWAP ──────────────────────────────────────────────────────

    async def vwap(
        self,
        symbol: str,
        side: str,
        quantity: float,
        duration_minutes: float,
        volume_profile: Optional[List[float]] = None,
        dynamic: bool = True,
    ) -> ExecutionReport:
        """
        Execute using Volume-Weighted Average Price algorithm.

        Args:
            symbol: Trading pair
            side: "buy" or "sell"
            quantity: Total quantity
            duration_minutes: Execution window in minutes
            volume_profile: List of relative volume weights per interval.
                           If None, fetches from Purple Flea historical data.
            dynamic: Continuously update schedule based on realized volumes
        """
        self._price_samples = []

        # Fetch historical volume profile if not provided
        if volume_profile is None:
            volume_profile = await self._fetch_volume_profile(symbol, n_intervals=12)

        n_slices = len(volume_profile)
        total_weight = sum(volume_profile)
        interval_seconds = (duration_minutes * 60) / n_slices

        arrival_price = await self._get_mid_price(symbol)
        slices: List[ExecutionSlice] = []
        remaining_qty = quantity
        remaining_weights = volume_profile.copy()

        for i, weight in enumerate(volume_profile):
            # Dynamic recalculation of remaining schedule
            if dynamic and i > 0:
                remaining_weights = volume_profile[i:]
                total_remaining_weight = sum(remaining_weights)
                slice_qty = remaining_qty * (weight / total_remaining_weight)
            else:
                slice_qty = quantity * (weight / total_weight)

            slice_qty = min(slice_qty, remaining_qty)

            mid = await self._get_mid_price(symbol)
            s = await self._submit_order(symbol, side, slice_qty)
            s.index = i
            s.target_time = time.time()
            slices.append(s)
            remaining_qty -= s.fill_quantity

            print(f"VWAP slice {i+1}/{n_slices}: weight={weight:.3f} qty={slice_qty:.4f}")

            if remaining_qty <= 0:
                break
            if i < n_slices - 1:
                await asyncio.sleep(interval_seconds)

        avg_fill = self._weighted_avg_fill(slices)
        twap_benchmark = self._compute_vwap_benchmark()

        return ExecutionReport(
            algorithm="VWAP",
            symbol=symbol,
            side=side,
            total_quantity=quantity,
            filled_quantity=sum(s.fill_quantity for s in slices),
            arrival_price=arrival_price,
            executed_avg_price=avg_fill,
            twap_benchmark=twap_benchmark,
            slices=slices,
        )

    async def _fetch_volume_profile(self, symbol: str, n_intervals: int) -> List[float]:
        """Fetch normalized intraday volume profile from Purple Flea API."""
        url = f"{self.base_url}/volume-profile/{symbol}?intervals={n_intervals}"
        async with self.session.get(url) as resp:
            data = await resp.json()
            # Returns list of relative volume weights
            return data.get("profile", [1.0] * n_intervals)

    # ─── Adaptive (IS-minimizing) ───────────────────────────────────

    async def adaptive(
        self,
        symbol: str,
        side: str,
        quantity: float,
        duration_minutes: float,
        risk_aversion: float = 0.5,
        n_slices: int = 12,
        volatility: Optional[float] = None,
    ) -> ExecutionReport:
        """
        IS-minimizing adaptive execution (Almgren-Chriss inspired).

        Uses exponential decay schedule with dynamic kappa adjustment
        based on realized volatility and spread width.

        Args:
            symbol: Trading pair
            side: "buy" or "sell"
            quantity: Total quantity
            duration_minutes: Max execution window
            risk_aversion: 0.0 = patient (minimize impact), 1.0 = urgent (minimize timing risk)
            n_slices: Number of child orders
            volatility: Override annualized volatility estimate
        """
        self._price_samples = []

        # Fetch vol if not provided
        if volatility is None:
            volatility = await self._estimate_volatility(symbol)

        arrival_price = await self._get_mid_price(symbol)

        # Compute kappa (urgency factor) from risk aversion and vol
        interval_seconds = (duration_minutes * 60) / n_slices
        sigma_per_interval = volatility / math.sqrt(365 * 24 * 3600 / interval_seconds)
        kappa = risk_aversion * sigma_per_interval * 10  # simplified kappa

        # Generate IS-optimal schedule (exponential decay)
        T = n_slices
        schedule = []
        for t in range(1, T + 1):
            if kappa > 0.001:
                weight = math.sinh(kappa * (T - t + 1)) - math.sinh(kappa * (T - t))
            else:
                weight = 1.0  # degenerate to TWAP for very low kappa
            schedule.append(weight)

        total_weight = sum(schedule)
        slices: List[ExecutionSlice] = []
        remaining_qty = quantity

        for i, weight in enumerate(schedule):
            slice_qty = min(remaining_qty, quantity * weight / total_weight)

            # Check spread — if too wide, delay this slice
            mid = await self._get_mid_price(symbol)
            spread_bps = await self._get_spread_bps(symbol)

            if spread_bps > 50 and remaining_qty / quantity > 0.2:
                print(f"Slice {i+1}: spread {spread_bps:.1f}bps too wide — pausing")
                await asyncio.sleep(interval_seconds / 2)
                continue

            s = await self._submit_order(symbol, side, slice_qty)
            s.index = i
            s.target_time = time.time()
            slices.append(s)
            remaining_qty -= s.fill_quantity

            print(
                f"Adaptive slice {i+1}/{n_slices}: "
                f"qty={slice_qty:.4f} spread={spread_bps:.1f}bps fill={s.fill_price:.4f}"
            )

            if remaining_qty <= 0:
                break
            if i < n_slices - 1:
                await asyncio.sleep(interval_seconds)

        avg_fill = self._weighted_avg_fill(slices)
        twap_benchmark = self._compute_vwap_benchmark()

        return ExecutionReport(
            algorithm="Adaptive-IS",
            symbol=symbol,
            side=side,
            total_quantity=quantity,
            filled_quantity=sum(s.fill_quantity for s in slices),
            arrival_price=arrival_price,
            executed_avg_price=avg_fill,
            twap_benchmark=twap_benchmark,
            slices=slices,
        )

    async def _estimate_volatility(self, symbol: str) -> float:
        """Estimate annualized volatility from recent 1-min candles."""
        url = f"{self.base_url}/candles/{symbol}?interval=1m&limit=60"
        async with self.session.get(url) as resp:
            candles = await resp.json()
        closes = [float(c["close"]) for c in candles]
        returns = [math.log(closes[i] / closes[i-1]) for i in range(1, len(closes))]
        vol_per_min = statistics.stdev(returns)
        return vol_per_min * math.sqrt(365 * 24 * 60)

    async def _get_spread_bps(self, symbol: str) -> float:
        """Return current bid-ask spread in basis points."""
        url = f"{self.base_url}/orderbook/{symbol}"
        async with self.session.get(url) as resp:
            data = await resp.json()
        best_bid = float(data["bids"][0][0])
        best_ask = float(data["asks"][0][0])
        mid = (best_bid + best_ask) / 2
        return ((best_ask - best_bid) / mid) * 10000

6. Integration with Purple Flea Trading API

Purple Flea's trading API provides the primitives needed for sophisticated order execution: order submission with configurable types, live orderbook depth, historical candles, and volume profile endpoints.

6.1 Authentication

All trading API calls require a bearer token obtained after agent registration. Tokens are prefixed pf_live_ and carry per-agent rate limits based on your volume tier.

import asyncio
from order_execution_engine import OrderExecutionEngine

async def main():
    async with OrderExecutionEngine(
        api_key="pf_live_YOUR_KEY_HERE"
    ) as engine:

        # TWAP: buy 0.1 BTC over 20 minutes, 10 slices
        report = await engine.twap(
            symbol="BTC-USDT",
            side="buy",
            quantity=0.1,
            duration_minutes=20,
            n_slices=10,
            randomize=True,
        )
        print(report.summary())
        # {algorithm: TWAP, fill_rate: 100.0%, IS_bps: 4.2, ...}

        # VWAP: sell 500 USDT of ETH aligned to volume profile
        report2 = await engine.vwap(
            symbol="ETH-USDT",
            side="sell",
            quantity=0.2,
            duration_minutes=60,
            dynamic=True,
        )
        print(report2.summary())

        # Adaptive: urgency=0.8 (60% front-loaded)
        report3 = await engine.adaptive(
            symbol="SOL-USDT",
            side="buy",
            quantity=10.0,
            duration_minutes=15,
            risk_aversion=0.8,
        )
        print(report3.summary())

asyncio.run(main())

6.2 Key API Endpoints

• POST /api/trading/orders — submit a child order (market or limit)
• GET /api/trading/orderbook/{symbol} — live best bid/ask with depth
• GET /api/trading/candles/{symbol} — OHLCV candles (1m, 5m, 1h)
• GET /api/trading/volume-profile/{symbol} — intraday volume weights
• GET /api/trading/orders/{id} — order status and fill details
• GET /api/wallet/balance — wallet balance for pre-trade checks

7. Measuring Execution Quality

Building execution algorithms is only half the work. Without rigorous quality measurement, you can't improve. Purple Flea's wallet API exposes transaction history that enables post-trade analysis.

7.1 Slippage Reporting via Wallet API

import aiohttp
import statistics
from dataclasses import dataclass
from typing import List

@dataclass
class SlippageReport:
    symbol: str
    period_days: int
    n_executions: int
    avg_is_bps: float         # average implementation shortfall
    p50_is_bps: float         # median IS
    p95_is_bps: float         # 95th pctile IS (worst executions)
    avg_fill_rate: float      # what fraction of orders fully filled
    algo_breakdown: dict      # IS by algorithm type

class ExecutionAnalytics:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {"Authorization": f"Bearer {api_key}"}

    async def fetch_execution_history(self, days: int = 30) -> List[dict]:
        """Pull recent trade history from Purple Flea wallet API."""
        url = f"https://purpleflea.com/api/wallet/transactions?days={days}&type=trade"
        async with aiohttp.ClientSession(headers=self.headers) as s:
            async with s.get(url) as resp:
                data = await resp.json()
                return data.get("transactions", [])

    def compute_slippage_report(
        self, trades: List[dict], symbol: str = None
    ) -> SlippageReport:
        """
        Compute execution quality stats from trade history.

        Each trade dict should include:
          - symbol, side, fill_price, arrival_price, quantity, algorithm
        """
        if symbol:
            trades = [t for t in trades if t["symbol"] == symbol]

        is_list = []
        fill_rates = []
        algo_is: dict = {}

        for t in trades:
            direction = 1 if t["side"] == "buy" else -1
            is_bps = direction * (
                (t["fill_price"] - t["arrival_price"]) / t["arrival_price"]
            ) * 10000
            is_list.append(is_bps)

            fill_rates.append(t.get("fill_rate", 1.0))

            algo = t.get("algorithm", "UNKNOWN")
            algo_is.setdefault(algo, []).append(is_bps)

        is_list.sort()
        algo_breakdown = {
            algo: round(statistics.mean(vals), 2)
            for algo, vals in algo_is.items()
        }

        return SlippageReport(
            symbol=symbol or "ALL",
            period_days=30,
            n_executions=len(trades),
            avg_is_bps=round(statistics.mean(is_list), 2),
            p50_is_bps=round(statistics.median(is_list), 2),
            p95_is_bps=round(is_list[int(len(is_list) * 0.95)], 2),
            avg_fill_rate=round(statistics.mean(fill_rates), 4),
            algo_breakdown=algo_breakdown,
        )

7.2 Key Quality Metrics

• Implementation Shortfall (bps): The headline metric. Lower is better for buy orders. Industry benchmark for algorithmic execution is under 10 bps for liquid pairs.
• Fill Rate: What fraction of your intended quantity actually traded. A 95% fill rate with good IS beats a 100% fill rate with poor IS only if the unfilled 5% becomes opportunity cost.
• Participation Rate: Your executed volume divided by total market volume during the window. Above 5-10%, your own activity starts affecting your price.
• Spread Capture: For limit-based strategies, the fraction of executions that hit at the midpoint or better vs paying the spread.
• Schedule Adherence: For VWAP, how closely actual fills tracked the target quantity curve. Divergence indicates either poor volume forecasting or unusual market conditions.

8. When to Use Each Algorithm

The choice of execution algorithm depends on four variables: order size relative to market volume, urgency, market conditions (trending vs mean-reverting), and available execution time. Here's a decision framework:

8.1 The Urgency Dial

Most real-world executions lie somewhere on a continuum between "pure impact minimization" (extremely patient, like central bank FX intervention) and "pure urgency" (immediate fill at any cost, like event-driven liquidation). Your optimal point on this dial depends on:

• Signal decay rate: If your alpha signal has a 5-minute half-life, executing over 60 minutes captures almost no signal benefit. Use higher urgency.
• Opportunity cost of unfilled quantity: If missing the trade entirely is worse than paying 20 extra bps, weight urgency accordingly.
• Market depth relative to order size: Deep markets tolerate higher urgency with lower impact. Thin markets punish it severely.
• Correlation between signal and execution speed: In momentum strategies, faster execution amplifies positive returns. In mean-reversion, patience is rewarded.

8.2 Dynamic Algorithm Switching

Advanced agents don't commit to a single algorithm for the duration of an order. They switch dynamically based on real-time market conditions:

1. Start with TWAP as baseline during the first 10-15% of the execution window while assessing market microstructure.
2. If volume profile matches historical expectations, switch to VWAP for the remaining execution.
3. If the bid-ask spread widens more than 2x normal, pause and wait for liquidity to improve.
4. If more than 70% of the window has elapsed with less than 60% fill rate, increase urgency — shift to more aggressive market orders for the remainder.
5. If a strong contra-trend signal emerges mid-execution, abort and reverse — the opportunity cost of completing the original order has reversed sign.

Conclusion

Order execution is where AI trading agents lose or preserve the value of their signals. TWAP provides a clean, auditable baseline. VWAP captures liquidity patterns to reduce per-unit impact costs. Implementation shortfall benchmarking reveals the true cost of execution against the most honest price: the moment you made your decision.

The OrderExecutionEngine above gives your agent a production-ready framework that integrates with Purple Flea's trading API. Start with TWAP for small orders while building your execution track record, graduate to VWAP as you scale, and deploy adaptive IS-minimizing execution once your order sizes warrant the additional complexity.

Every basis point saved in execution is a basis point added to your edge — compound that over thousands of trades, and optimal execution becomes one of the highest-return improvements an AI trading agent can make.