50. Smart Slippage Estimation System¶

services/slippage-estimation-center/
├── src/
│   ├── main.py
│   ├── fetcher/
│   │   ├── depth_fetcher.py
│   │   ├── trade_flow_fetcher.py
│   │   └── market_data_aggregator.py
│   ├── simulator/
│   │   ├── fill_simulator.py
│   │   ├── impact_model.py
│   │   └── execution_simulator.py
│   ├── calculator/
│   │   ├── slippage_calculator.py
│   │   ├── cost_analyzer.py
│   │   └── risk_metrics.py
│   ├── grader/
│   │   ├── slippage_grader.py
│   │   ├── risk_classifier.py
│   │   └── threshold_manager.py
│   ├── optimizer/
│   │   ├── execution_optimizer.py
│   │   ├── algorithm_selector.py
│   │   └── path_finder.py
│   ├── api/
│   │   ├── slippage_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class DepthFetcher:
    def __init__(self, market_data_provider, cache_manager):
        self.market_data_provider = market_data_provider
        self.cache_manager = cache_manager
        self.depth_cache = {}

    async def fetch_depth(self, exchange: str, symbol: str) -> Dict:
        """Fetch real-time market depth for slippage estimation"""

        cache_key = f"{exchange}_{symbol}_depth"

        # Check cache first
        cached_depth = self.cache_manager.get(cache_key)
        if cached_depth and self.is_cache_valid(cached_depth):
            return cached_depth["data"]

        # Fetch fresh data
        try:
            order_book = await self.market_data_provider.get_order_book(exchange, symbol)

            # Process and structure depth data
            processed_depth = self.process_order_book(order_book)

            # Cache the result
            self.cache_manager.set(cache_key, {
                "data": processed_depth,
                "timestamp": datetime.now().isoformat()
            })

            return processed_depth

        except Exception as e:
            # Return cached data if available, otherwise raise
            if cached_depth:
                return cached_depth["data"]
            raise e

    def process_order_book(self, order_book: Dict) -> Dict:
        """Process raw order book into structured depth data"""

        bids = order_book.get("bids", [])
        asks = order_book.get("asks", [])

        # Calculate cumulative depth
        cumulative_bids = self.calculate_cumulative_depth(bids, "bid")
        cumulative_asks = self.calculate_cumulative_depth(asks, "ask")

        # Calculate depth metrics
        depth_metrics = self.calculate_depth_metrics(bids, asks)

        return {
            "symbol": order_book.get("symbol"),
            "exchange": order_book.get("exchange"),
            "timestamp": order_book.get("timestamp"),
            "bids": cumulative_bids,
            "asks": cumulative_asks,
            "metrics": depth_metrics,
            "spread": self.calculate_spread(bids, asks),
            "mid_price": self.calculate_mid_price(bids, asks)
        }

    def calculate_cumulative_depth(self, orders: List, side: str) -> List[Dict]:
        """Calculate cumulative depth for each price level"""

        cumulative_depth = []
        cumulative_quantity = 0.0

        for price, quantity in orders:
            cumulative_quantity += quantity
            cumulative_depth.append({
                "price": price,
                "quantity": quantity,
                "cumulative_quantity": cumulative_quantity,
                "side": side
            })

        return cumulative_depth

    def calculate_depth_metrics(self, bids: List, asks: List) -> Dict:
        """Calculate depth-related metrics"""

        # Calculate total depth at different levels
        bid_depth_1 = sum(qty for _, qty in bids[:1]) if bids else 0
        bid_depth_5 = sum(qty for _, qty in bids[:5]) if bids else 0
        bid_depth_10 = sum(qty for _, qty in bids[:10]) if bids else 0

        ask_depth_1 = sum(qty for _, qty in asks[:1]) if asks else 0
        ask_depth_5 = sum(qty for _, qty in asks[:5]) if asks else 0
        ask_depth_10 = sum(qty for _, qty in asks[:10]) if asks else 0

        # Calculate depth imbalance
        total_bid_depth = sum(qty for _, qty in bids[:10]) if bids else 0
        total_ask_depth = sum(qty for _, qty in asks[:10]) if asks else 0
        total_depth = total_bid_depth + total_ask_depth

        depth_imbalance = (total_bid_depth - total_ask_depth) / total_depth if total_depth > 0 else 0

        return {
            "bid_depth_1": bid_depth_1,
            "bid_depth_5": bid_depth_5,
            "bid_depth_10": bid_depth_10,
            "ask_depth_1": ask_depth_1,
            "ask_depth_5": ask_depth_5,
            "ask_depth_10": ask_depth_10,
            "total_depth": total_depth,
            "depth_imbalance": depth_imbalance,
            "depth_ratio": total_bid_depth / total_ask_depth if total_ask_depth > 0 else 1.0
        }

    def calculate_spread(self, bids: List, asks: List) -> float:
        """Calculate current bid-ask spread"""

        if not bids or not asks:
            return 0.0

        best_bid = bids[0][0]
        best_ask = asks[0][0]

        return (best_ask - best_bid) / best_bid

    def calculate_mid_price(self, bids: List, asks: List) -> float:
        """Calculate mid price"""

        if not bids or not asks:
            return 0.0

        best_bid = bids[0][0]
        best_ask = asks[0][0]

        return (best_bid + best_ask) / 2

    def is_cache_valid(self, cached_data: Dict) -> bool:
        """Check if cached data is still valid"""

        cache_time = datetime.fromisoformat(cached_data["timestamp"])
        current_time = datetime.now()

        # Cache valid for 1 second for high-frequency data
        return (current_time - cache_time).total_seconds() < 1.0

class TradeFlowFetcher:
    def __init__(self, market_data_provider):
        self.market_data_provider = market_data_provider
        self.trade_history = {}

    async def fetch_trade_flow(self, exchange: str, symbol: str, 
                             time_window: int = 300) -> Dict:
        """Fetch recent trade flow for impact estimation"""

        try:
            # Fetch recent trades
            recent_trades = await self.market_data_provider.get_recent_trades(
                exchange, symbol, limit=1000
            )

            # Filter trades within time window
            cutoff_time = datetime.now() - timedelta(seconds=time_window)
            filtered_trades = [
                trade for trade in recent_trades
                if datetime.fromisoformat(trade["timestamp"]) > cutoff_time
            ]

            # Analyze trade flow
            trade_flow = self.analyze_trade_flow(filtered_trades)

            return trade_flow

        except Exception as e:
            return self.get_default_trade_flow()

    def analyze_trade_flow(self, trades: List[Dict]) -> Dict:
        """Analyze trade flow characteristics"""

        if not trades:
            return self.get_default_trade_flow()

        # Calculate trade flow metrics
        total_volume = sum(trade["quantity"] for trade in trades)
        total_trades = len(trades)
        avg_trade_size = total_volume / total_trades if total_trades > 0 else 0

        # Calculate volume-weighted average price
        vwap = sum(trade["price"] * trade["quantity"] for trade in trades) / total_volume if total_volume > 0 else 0

        # Calculate trade flow density (trades per second)
        if len(trades) >= 2:
            time_span = (datetime.fromisoformat(trades[-1]["timestamp"]) - 
                        datetime.fromisoformat(trades[0]["timestamp"])).total_seconds()
            trade_density = total_trades / time_span if time_span > 0 else 0
        else:
            trade_density = 0

        # Calculate buy/sell imbalance
        buy_volume = sum(trade["quantity"] for trade in trades if trade.get("side") == "buy")
        sell_volume = sum(trade["quantity"] for trade in trades if trade.get("side") == "sell")
        total_volume_side = buy_volume + sell_volume

        buy_ratio = buy_volume / total_volume_side if total_volume_side > 0 else 0.5

        # Calculate price volatility
        prices = [trade["price"] for trade in trades]
        price_volatility = self.calculate_price_volatility(prices)

        return {
            "total_volume": total_volume,
            "total_trades": total_trades,
            "avg_trade_size": avg_trade_size,
            "vwap": vwap,
            "trade_density": trade_density,
            "buy_ratio": buy_ratio,
            "sell_ratio": 1 - buy_ratio,
            "price_volatility": price_volatility,
            "time_window": len(trades),
            "timestamp": datetime.now().isoformat()
        }

    def calculate_price_volatility(self, prices: List[float]) -> float:
        """Calculate price volatility"""

        if len(prices) < 2:
            return 0.0

        returns = []
        for i in range(1, len(prices)):
            if prices[i-1] > 0:
                returns.append((prices[i] - prices[i-1]) / prices[i-1])

        if not returns:
            return 0.0

        return np.std(returns)

    def get_default_trade_flow(self) -> Dict:
        """Return default trade flow when no data available"""

        return {
            "total_volume": 0,
            "total_trades": 0,
            "avg_trade_size": 0,
            "vwap": 0,
            "trade_density": 0,
            "buy_ratio": 0.5,
            "sell_ratio": 0.5,
            "price_volatility": 0,
            "time_window": 0,
            "timestamp": datetime.now().isoformat()
        }

class FillSimulator:
    def __init__(self, impact_model):
        self.impact_model = impact_model

    def simulate_fill(self, depth: Dict, order_size: float, side: str, 
                     order_type: str = "market") -> Dict:
        """Simulate order execution and calculate expected fill price"""

        if side == "buy":
            levels = depth["asks"]
            current_price = depth["asks"][0]["price"] if depth["asks"] else 0
        else:
            levels = depth["bids"]
            current_price = depth["bids"][0]["price"] if depth["bids"] else 0

        if not levels:
            return self.get_default_fill_result(order_size, current_price)

        # Simulate fill based on order type
        if order_type == "market":
            fill_result = self.simulate_market_fill(levels, order_size, side)
        elif order_type == "limit":
            fill_result = self.simulate_limit_fill(levels, order_size, side)
        else:
            fill_result = self.simulate_market_fill(levels, order_size, side)

        # Apply market impact model
        impact_adjusted_result = self.apply_market_impact(
            fill_result, order_size, side, depth
        )

        return impact_adjusted_result

    def simulate_market_fill(self, levels: List[Dict], order_size: float, 
                           side: str) -> Dict:
        """Simulate market order fill"""

        filled_quantity = 0.0
        total_cost = 0.0
        fills = []

        for level in levels:
            available_quantity = level["quantity"]
            price = level["price"]

            if filled_quantity + available_quantity >= order_size:
                # Partial fill at this level
                remaining_quantity = order_size - filled_quantity
                fill_cost = remaining_quantity * price

                fills.append({
                    "price": price,
                    "quantity": remaining_quantity,
                    "level": levels.index(level) + 1
                })

                total_cost += fill_cost
                filled_quantity = order_size
                break
            else:
                # Full fill at this level
                fill_cost = available_quantity * price

                fills.append({
                    "price": price,
                    "quantity": available_quantity,
                    "level": levels.index(level) + 1
                })

                total_cost += fill_cost
                filled_quantity += available_quantity

        # Calculate average fill price
        avg_fill_price = total_cost / filled_quantity if filled_quantity > 0 else 0

        return {
            "filled_quantity": filled_quantity,
            "total_cost": total_cost,
            "avg_fill_price": avg_fill_price,
            "fills": fills,
            "fill_ratio": filled_quantity / order_size if order_size > 0 else 0,
            "unfilled_quantity": max(0, order_size - filled_quantity)
        }

    def simulate_limit_fill(self, levels: List[Dict], order_size: float, 
                          side: str) -> Dict:
        """Simulate limit order fill (simplified)"""

        # For limit orders, assume partial fill at best price
        best_level = levels[0]
        best_price = best_level["price"]
        available_quantity = min(order_size, best_level["quantity"])

        return {
            "filled_quantity": available_quantity,
            "total_cost": available_quantity * best_price,
            "avg_fill_price": best_price,
            "fills": [{
                "price": best_price,
                "quantity": available_quantity,
                "level": 1
            }],
            "fill_ratio": available_quantity / order_size if order_size > 0 else 0,
            "unfilled_quantity": max(0, order_size - available_quantity)
        }

    def apply_market_impact(self, fill_result: Dict, order_size: float, 
                          side: str, depth: Dict) -> Dict:
        """Apply market impact model to adjust fill prices"""

        # Get market impact parameters
        impact_params = self.impact_model.get_impact_parameters(
            order_size, side, depth
        )

        # Calculate impact adjustment
        impact_adjustment = self.calculate_impact_adjustment(
            fill_result, impact_params
        )

        # Apply adjustment
        adjusted_fill_price = fill_result["avg_fill_price"] + impact_adjustment
        adjusted_total_cost = adjusted_fill_price * fill_result["filled_quantity"]

        return {
            **fill_result,
            "avg_fill_price": adjusted_fill_price,
            "total_cost": adjusted_total_cost,
            "impact_adjustment": impact_adjustment,
            "impact_percentage": impact_adjustment / fill_result["avg_fill_price"] if fill_result["avg_fill_price"] > 0 else 0
        }

    def calculate_impact_adjustment(self, fill_result: Dict, 
                                  impact_params: Dict) -> float:
        """Calculate market impact adjustment"""

        # Simple linear impact model
        base_impact = impact_params.get("base_impact", 0.0001)
        size_multiplier = impact_params.get("size_multiplier", 0.1)

        order_size = fill_result["filled_quantity"]
        impact_adjustment = base_impact * (1 + size_multiplier * order_size)

        return impact_adjustment

    def get_default_fill_result(self, order_size: float, current_price: float) -> Dict:
        """Return default fill result when no depth data available"""

        return {
            "filled_quantity": 0,
            "total_cost": 0,
            "avg_fill_price": current_price,
            "fills": [],
            "fill_ratio": 0,
            "unfilled_quantity": order_size,
            "impact_adjustment": 0,
            "impact_percentage": 0
        }

class SlippageCalculator:
    def __init__(self, slippage_grader):
        self.slippage_grader = slippage_grader

    def calculate_slippage(self, order_request: Dict, fill_result: Dict) -> Dict:
        """Calculate comprehensive slippage metrics"""

        # Extract order parameters
        side = order_request["side"]
        order_size = order_request["order_size"]
        current_price = order_request.get("current_price", 0)

        # Get fill results
        expected_fill_price = fill_result["avg_fill_price"]
        filled_quantity = fill_result["filled_quantity"]

        # Calculate slippage metrics
        slippage_pct = self.calculate_slippage_percentage(
            current_price, expected_fill_price, side
        )

        slippage_cost = self.calculate_slippage_cost(
            current_price, expected_fill_price, filled_quantity, side
        )

        # Calculate additional metrics
        market_impact = fill_result.get("impact_percentage", 0)
        total_cost_pct = slippage_pct + market_impact

        # Get risk grade
        risk_grade = self.slippage_grader.grade_slippage(total_cost_pct)

        return {
            "order_size": order_size,
            "filled_quantity": filled_quantity,
            "current_price": current_price,
            "expected_fill_price": expected_fill_price,
            "slippage_pct": slippage_pct,
            "slippage_cost": slippage_cost,
            "market_impact_pct": market_impact,
            "total_cost_pct": total_cost_pct,
            "risk_grade": risk_grade,
            "fill_ratio": fill_result["fill_ratio"],
            "unfilled_quantity": fill_result["unfilled_quantity"],
            "timestamp": datetime.now().isoformat()
        }

    def calculate_slippage_percentage(self, current_price: float, 
                                    expected_fill_price: float, side: str) -> float:
        """Calculate slippage percentage"""

        if current_price <= 0 or expected_fill_price <= 0:
            return 0.0

        if side == "buy":
            # For buy orders, slippage is positive when fill price > current price
            slippage_pct = (expected_fill_price - current_price) / current_price
        else:
            # For sell orders, slippage is positive when fill price < current price
            slippage_pct = (current_price - expected_fill_price) / current_price

        return max(0, slippage_pct)  # Slippage is always non-negative

    def calculate_slippage_cost(self, current_price: float, expected_fill_price: float,
                              filled_quantity: float, side: str) -> float:
        """Calculate absolute slippage cost"""

        slippage_pct = self.calculate_slippage_percentage(
            current_price, expected_fill_price, side
        )

        return current_price * filled_quantity * slippage_pct

    def estimate_slippage_for_size(self, exchange: str, symbol: str, 
                                 side: str, order_size: float) -> Dict:
        """Estimate slippage for a given order size"""

        # This would integrate with depth fetcher and fill simulator
        # For now, return a simplified estimation

        # Base slippage model
        base_slippage = 0.0005  # 5 basis points base slippage

        # Size-based adjustment
        size_multiplier = min(2.0, order_size / 10000)  # Cap at 2x for large orders

        # Market condition adjustment (simplified)
        market_volatility = 1.0  # Would be calculated from market data

        estimated_slippage = base_slippage * size_multiplier * market_volatility

        return {
            "estimated_slippage_pct": estimated_slippage,
            "estimated_slippage_cost": estimated_slippage * order_size,
            "confidence_level": "medium",
            "assumptions": {
                "base_slippage": base_slippage,
                "size_multiplier": size_multiplier,
                "market_volatility": market_volatility
            }
        }

class SlippageGrader:
    def __init__(self, threshold_manager):
        self.threshold_manager = threshold_manager

    def grade_slippage(self, slippage_pct: float) -> str:
        """Grade slippage based on percentage"""

        thresholds = self.threshold_manager.get_thresholds()

        if slippage_pct <= thresholds["good"]:
            return "Good"
        elif slippage_pct <= thresholds["caution"]:
            return "Caution"
        else:
            return "Risky"

    def get_detailed_grade(self, slippage_data: Dict) -> Dict:
        """Get detailed slippage grade with recommendations"""

        slippage_pct = slippage_data.get("slippage_pct", 0)
        order_size = slippage_data.get("order_size", 0)
        side = slippage_data.get("side", "buy")

        # Basic grade
        grade = self.grade_slippage(slippage_pct)

        # Generate recommendations
        recommendations = self.generate_recommendations(
            grade, slippage_pct, order_size, side
        )

        # Calculate risk score
        risk_score = self.calculate_risk_score(slippage_data)

        return {
            "grade": grade,
            "risk_score": risk_score,
            "recommendations": recommendations,
            "slippage_pct": slippage_pct,
            "timestamp": datetime.now().isoformat()
        }

    def generate_recommendations(self, grade: str, slippage_pct: float,
                               order_size: float, side: str) -> List[str]:
        """Generate execution recommendations based on slippage grade"""

        recommendations = []

        if grade == "Good":
            recommendations.extend([
                "Proceed with market order",
                "Consider limit order for better price",
                "Monitor market conditions"
            ])
        elif grade == "Caution":
            recommendations.extend([
                "Consider splitting order into smaller chunks",
                "Use TWAP/VWAP execution algorithm",
                "Set conservative limit prices",
                "Monitor market depth changes"
            ])
        else:  # Risky
            recommendations.extend([
                "Split order into multiple smaller orders",
                "Use advanced execution algorithms (TWAP/VWAP/Smart Order Routing)",
                "Consider alternative execution venues",
                "Review order timing and market conditions",
                "Set strict price limits"
            ])

        # Add size-specific recommendations
        if order_size > 10000:
            recommendations.append("Consider institutional execution services")

        return recommendations

    def calculate_risk_score(self, slippage_data: Dict) -> float:
        """Calculate comprehensive risk score (0-100)"""

        slippage_pct = slippage_data.get("slippage_pct", 0)
        order_size = slippage_data.get("order_size", 0)
        market_impact = slippage_data.get("market_impact_pct", 0)

        # Base risk from slippage percentage
        slippage_risk = min(100, slippage_pct * 10000)  # Scale slippage to 0-100

        # Size risk (larger orders = higher risk)
        size_risk = min(50, order_size / 1000)  # Cap at 50 points

        # Market impact risk
        impact_risk = min(30, market_impact * 1000)  # Cap at 30 points

        # Total risk score
        total_risk = slippage_risk + size_risk + impact_risk

        return min(100, total_risk)

@router.post("/slippage/estimate")
async def estimate_slippage(request: SlippageRequest):
    """Estimate slippage for a potential order"""

    # Fetch market data
    depth = await depth_fetcher.fetch_depth(request.exchange, request.symbol)
    trade_flow = await trade_flow_fetcher.fetch_trade_flow(request.exchange, request.symbol)

    # Simulate fill
    fill_result = fill_simulator.simulate_fill(
        depth, request.order_size, request.side, request.order_type
    )

    # Calculate slippage
    slippage_result = slippage_calculator.calculate_slippage(request.dict(), fill_result)

    # Get detailed grade
    grade_result = slippage_grader.get_detailed_grade(slippage_result)

    return {
        "request": request.dict(),
        "market_data": {
            "depth": depth,
            "trade_flow": trade_flow
        },
        "fill_simulation": fill_result,
        "slippage_analysis": slippage_result,
        "risk_assessment": grade_result
    }

@router.get("/slippage/history/{symbol}")
async def get_slippage_history(symbol: str, timeframe: str = "1d"):
    """Get historical slippage data for analysis"""

    history = slippage_calculator.get_slippage_history(symbol, timeframe)
    return {"symbol": symbol, "timeframe": timeframe, "history": history}

@router.get("/slippage/thresholds")
async def get_slippage_thresholds():
    """Get current slippage grading thresholds"""

    thresholds = threshold_manager.get_thresholds()
    return {"thresholds": thresholds}

const SlippageEstimationView: React.FC = () => {
  const [estimationRequest, setEstimationRequest] = useState<SlippageRequest | null>(null);
  const [estimationResult, setEstimationResult] = useState<SlippageResult | null>(null);
  const [slippageHistory, setSlippageHistory] = useState<SlippageHistory[]>([]);

  return (
    <div className="slippage-estimation-dashboard">
      {/* Estimation Form */}
      <SlippageEstimationForm 
        onSubmit={handleEstimationRequest}
        request={estimationRequest}
      />

      {/* Estimation Results */}
      {estimationResult && (
        <SlippageResultPanel 
          result={estimationResult}
          onOptimize={handleOptimization}
        />
      )}

      {/* Risk Assessment */}
      {estimationResult && (
        <RiskAssessmentPanel 
          riskAssessment={estimationResult.risk_assessment}
        />
      )}

      {/* Market Depth Visualization */}
      {estimationResult && (
        <MarketDepthPanel 
          depth={estimationResult.market_data.depth}
          orderSize={estimationRequest?.order_size}
        />
      )}

      {/* Slippage History */}
      <SlippageHistoryPanel 
        history={slippageHistory}
        symbol={estimationRequest?.symbol}
      />

      {/* Execution Recommendations */}
      {estimationResult && (
        <ExecutionRecommendationsPanel 
          recommendations={estimationResult.risk_assessment.recommendations}
        />
      )}
    </div>
  );
};