Spread Slippage Cost Analyzer

Spread & Slippage Cost Analyzer

import pandas as pd
import numpy as np

class CostAnalyzer:

    @staticmethod
    def spread_statistics(ticks: pd.DataFrame) -> dict:
        """Comprehensive spread analysis from tick data."""
        spread = ticks["ask"] - ticks["bid"]
        spread_pips = spread * 10000
        return {
            "avg_spread_pips": round(spread_pips.mean(), 2),
            "median_spread_pips": round(spread_pips.median(), 2),
            "min_spread_pips": round(spread_pips.min(), 2),
            "max_spread_pips": round(spread_pips.max(), 2),
            "std_spread_pips": round(spread_pips.std(), 2),
            "p95_spread_pips": round(spread_pips.quantile(0.95), 2),
            "spread_widening_events": int((spread_pips > spread_pips.quantile(0.95)).sum()),
            "pct_time_above_2x_avg": round((spread_pips > 2 * spread_pips.mean()).mean() * 100, 1),
        }

    @staticmethod
    def spread_by_session(ticks: pd.DataFrame) -> dict:
        """Spread behavior per session — find when spreads are tightest."""
        ticks = ticks.copy()
        ticks["spread_pips"] = (ticks["ask"] - ticks["bid"]) * 10000
        ticks["hour"] = ticks.index.hour
        ticks["session"] = ticks["hour"].apply(lambda h:
            "tokyo" if h < 7 else "london" if h < 13 else "overlap" if h < 16 else "ny_late" if h < 22 else "off")
        return ticks.groupby("session")["spread_pips"].agg(["mean", "median", "max"]).round(2).to_dict()

    @staticmethod
    def slippage_analysis(trades: pd.DataFrame) -> dict:
        """Analyze actual slippage from trade execution data."""
        if "expected_price" not in trades.columns or "actual_price" not in trades.columns:
            return {"error": "Need expected_price and actual_price columns"}
        trades = trades.copy()
        trades["slippage"] = (trades["actual_price"] - trades["expected_price"]).abs() * 10000
        return {
            "avg_slippage_pips": round(trades["slippage"].mean(), 2),
            "max_slippage_pips": round(trades["slippage"].max(), 2),
            "pct_positive_slippage": round((trades["actual_price"] > trades["expected_price"]).mean() * 100, 1),
            "total_slippage_cost_pips": round(trades["slippage"].sum(), 1),
            "slippage_per_lot": round(trades["slippage"].mean() * 10, 2),  # USD per lot
        }

    @staticmethod
    def cost_impact_on_strategy(avg_spread: float, avg_slippage: float,
                                 trades_per_year: int, avg_profit_per_trade: float) -> dict:
        """Calculate what % of profit goes to costs."""
        cost_per_trade = avg_spread + avg_slippage
        annual_cost = cost_per_trade * trades_per_year
        annual_profit_gross = avg_profit_per_trade * trades_per_year
        cost_pct = cost_per_trade / max(abs(avg_profit_per_trade), 0.01) * 100
        return {
            "cost_per_trade_pips": round(cost_per_trade, 2),
            "annual_cost_pips": round(annual_cost, 1),
            "cost_as_pct_of_profit": round(cost_pct, 1),
            "net_profit_ratio": round(1 - cost_pct / 100, 3),
            "verdict": "ACCEPTABLE" if cost_pct < 30 else "HIGH — consider fewer trades or tighter broker" if cost_pct < 60 else "CRITICAL — costs eating profits",
        }

    @staticmethod
    def broker_comparison(broker_data: list[dict]) -> pd.DataFrame:
        """Compare multiple brokers on cost metrics."""
        return pd.DataFrame(broker_data).sort_values("avg_spread_pips")