Market Impact Model

Market Impact Model

import numpy as np
from dataclasses import dataclass

@dataclass
class MarketParams:
    daily_volume: float          # average daily volume in lots
    avg_spread_pips: float
    volatility_daily_pips: float
    symbol: str = ""

class MarketImpactModel:

    @staticmethod
    def almgren_chriss_impact(order_size_lots: float, market: MarketParams,
                               urgency: float = 0.5) -> dict:
        """
        Almgren-Chriss market impact model.
        Estimates permanent and temporary impact of an order.
        urgency: 0 (patient) to 1 (aggressive)
        """
        participation_rate = order_size_lots / max(market.daily_volume, 1)
        # Temporary impact (goes away after execution)
        temp_impact = market.avg_spread_pips * 0.5 + market.volatility_daily_pips * participation_rate * urgency * 2
        # Permanent impact (stays)
        perm_impact = market.volatility_daily_pips * np.sqrt(participation_rate) * 0.1
        total_impact = temp_impact + perm_impact
        return {
            "order_size_lots": order_size_lots,
            "participation_rate": round(participation_rate * 100, 2),
            "temporary_impact_pips": round(temp_impact, 2),
            "permanent_impact_pips": round(perm_impact, 2),
            "total_estimated_impact_pips": round(total_impact, 2),
            "cost_in_spread_multiples": round(total_impact / market.avg_spread_pips, 1),
            "recommendation": MarketImpactModel._execution_recommendation(participation_rate, urgency),
        }

    @staticmethod
    def optimal_execution_schedule(order_size_lots: float, market: MarketParams,
                                    execution_hours: float = 4) -> list[dict]:
        """TWAP-style execution schedule to minimize impact."""
        n_slices = max(int(execution_hours * 4), 1)  # One slice per 15 min
        base_size = order_size_lots / n_slices
        schedule = []
        for i in range(n_slices):
            # Vary size: slightly larger at open/close (more liquidity)
            hour = i / 4
            liquidity_factor = 1.2 if hour < 1 or hour > execution_hours - 1 else 0.9
            size = round(base_size * liquidity_factor, 2)
            schedule.append({"slice": i + 1, "lots": max(size, 0.01),
                            "minutes_from_start": i * 15})
        return schedule

    @staticmethod
    def _execution_recommendation(participation: float, urgency: float) -> str:
        if participation < 0.01:
            return "SMALL ORDER — execute immediately, impact negligible"
        if participation < 0.05:
            return "MODERATE — consider splitting into 3-5 slices over 1 hour"
        if participation < 0.15:
            return "LARGE — use TWAP over 2-4 hours, consider limit orders"
        return "VERY LARGE — use TWAP over full session, consider iceberg orders"

    @staticmethod
    def transaction_cost_analysis(trades: list[dict], market: MarketParams) -> dict:
        """Post-trade TCA: measure actual vs expected costs."""
        slippages = [t.get("slippage_pips", 0) for t in trades]
        return {
            "avg_slippage_pips": round(np.mean(slippages), 2),
            "max_slippage_pips": round(max(slippages), 2),
            "total_cost_pips": round(sum(slippages) + len(trades) * market.avg_spread_pips, 2),
            "cost_vs_benchmark": round(np.mean(slippages) / market.avg_spread_pips, 2),
        }