Optimal Execution Algorithms: TWAP, VWAP & Market Impact › Putting It All Together: An Execution Simulator

class ExecutionSimulator:
    """
    Simulates execution of different algorithms against
    realistic price dynamics with market impact.
    """

    def __init__(
        self,
        initial_price: float,
        volatility: float,
        permanent_impact: float,
        temporary_impact: float,
        num_periods: int,
    ):
        self.S0 = initial_price
        self.sigma = volatility
        self.gamma = permanent_impact
        self.eta = temporary_impact
        self.num_periods = num_periods

    def simulate_execution(
        self, schedule: np.ndarray, seed: int = None
    ) -> dict:
        """
        Simulate execution and compute performance metrics.

        Returns execution prices, costs, and statistics.
        """
        if seed is not None:
            np.random.seed(seed)

        n = len(schedule)
        prices = [self.S0]
        execution_prices = []

        # Price follows Brownian motion with impact
        for i in range(n):
            shares = schedule[i]

            # Permanent impact shifts fundamental price
            permanent_shift = self.gamma * shares

            # Random price movement
            random_shock = self.sigma * np.random.randn() / np.sqrt(n)

            # New fundamental price
            new_price = prices[-1] + permanent_shift + random_shock
            prices.append(new_price)

            # Execution occurs at fundamental + temporary impact
            if shares > 0:
                temp_impact = self.eta * shares
                exec_price = new_price + temp_impact
                execution_prices.append((shares, exec_price))

        # Compute metrics
        total_shares = sum(s for s, _ in execution_prices)
        if total_shares > 0:
            vwap_exec = sum(s * p for s, p in execution_prices) / total_shares
        else:
            vwap_exec = self.S0

        implementation_shortfall = (vwap_exec - self.S0) * total_shares

        return {
            "prices": np.array(prices),
            "execution_prices": execution_prices,
            "average_price": vwap_exec,
            "implementation_shortfall": implementation_shortfall,
            "arrival_price": self.S0,
            "final_price": prices[-1],
        }


# Set up simulation parameters
sim = ExecutionSimulator(
    initial_price=100.0,
    volatility=2.0,  # $2 volatility over execution period
    permanent_impact=1e-5,  # Price moves $0.01 per 1000 shares permanently
    temporary_impact=2e-5,  # Additional $0.02 per 1000 shares temporarily
    num_periods=40,
)

# Generate different schedules
order_size = 100_000
num_periods = 40

schedules = {
    "TWAP": generate_twap_schedule(order_size, num_periods),
    "VWAP": generate_vwap_schedule(
        order_size, create_intraday_volume_profile(num_periods)
    ),
    "IS-Patient": implementation_shortfall_schedule(
        order_size, num_periods, 0.3
    ),
    "IS-Urgent": implementation_shortfall_schedule(
        order_size, num_periods, 0.8
    ),
}

import numpy as np

# Run multiple simulations for each schedule
num_simulations = 500
results = {name: [] for name in schedules}

for name, schedule in schedules.items():
    for seed in range(num_simulations):
        result = sim.simulate_execution(schedule, seed=seed)
        results[name].append(result["implementation_shortfall"])

import numpy as np

performance_metrics = []
for name in schedules.keys():
    mean_is = np.mean(results[name])
    std_is = np.std(results[name])
    # Negative because lower IS is better
    sharpe_like = -mean_is / std_is if std_is > 0 else 0
    performance_metrics.append((name, mean_is, std_is, sharpe_like))