51. Smart Order Splitting System¶

services/order-splitting-center/
├── src/
│   ├── main.py
│   ├── manager/
│   │   ├── split_strategy_manager.py
│   │   ├── algorithm_selector.py
│   │   └── strategy_optimizer.py
│   ├── executor/
│   │   ├── split_executor.py
│   │   ├── child_order_manager.py
│   │   ├── order_router.py
│   │   └── execution_engine.py
│   ├── monitor/
│   │   ├── execution_monitor.py
│   │   ├── progress_tracker.py
│   │   ├── performance_analyzer.py
│   │   └── benchmark_calculator.py
│   ├── handler/
│   │   ├── anomaly_handler.py
│   │   ├── risk_controller.py
│   │   ├── emergency_handler.py
│   │   └── circuit_breaker.py
│   ├── strategies/
│   │   ├── twap_strategy.py
│   │   ├── vwap_strategy.py
│   │   ├── pov_strategy.py
│   │   ├── adaptive_strategy.py
│   │   └── opportunistic_strategy.py
│   ├── api/
│   │   ├── splitting_api.py
│   │   ├── config.py
│   │   └── requirements.txt
│   ├── Dockerfile
│   └── tests/

class SplitStrategyManager:
    def __init__(self, algorithm_selector, strategy_optimizer):
        self.algorithm_selector = algorithm_selector
        self.strategy_optimizer = strategy_optimizer
        self.strategies = {
            "TWAP": TwapStrategy(),
            "VWAP": VwapStrategy(),
            "POV": PovStrategy(),
            "ADAPTIVE": AdaptiveStrategy(),
            "OPPORTUNISTIC": OpportunisticStrategy()
        }

    def create_split_plan(self, parent_order: Dict) -> Dict:
        """Create comprehensive split execution plan"""

        # Select optimal strategy
        strategy_name = self.algorithm_selector.select_strategy(parent_order)
        strategy = self.strategies.get(strategy_name)

        if not strategy:
            raise ValueError(f"Unknown strategy: {strategy_name}")

        # Generate child orders
        child_orders = strategy.generate_child_orders(parent_order)

        # Optimize execution plan
        optimized_plan = self.strategy_optimizer.optimize_plan(
            parent_order, child_orders, strategy_name
        )

        return {
            "parent_order_id": parent_order["order_id"],
            "strategy": strategy_name,
            "child_orders": optimized_plan["child_orders"],
            "execution_schedule": optimized_plan["schedule"],
            "risk_parameters": optimized_plan["risk_params"],
            "created_at": datetime.now().isoformat()
        }

    def get_strategy(self, strategy_name: str):
        """Get specific splitting strategy"""

        return self.strategies.get(strategy_name)

    def validate_order(self, parent_order: Dict) -> bool:
        """Validate parent order for splitting"""

        required_fields = ["order_id", "symbol", "side", "quantity", "order_type"]

        for field in required_fields:
            if field not in parent_order:
                return False

        # Check minimum size for splitting
        min_split_size = 1000  # Minimum order size for splitting
        if parent_order["quantity"] < min_split_size:
            return False

        return True

    def get_available_strategies(self) -> List[Dict]:
        """Get list of available splitting strategies"""

        strategies = []
        for name, strategy in self.strategies.items():
            strategies.append({
                "name": name,
                "description": strategy.get_description(),
                "parameters": strategy.get_parameters(),
                "suitable_for": strategy.get_suitable_conditions()
            })

        return strategies

class TwapStrategy:
    def __init__(self):
        self.name = "TWAP"
        self.description = "Time-Weighted Average Price execution"

    def generate_child_orders(self, parent_order: Dict) -> List[Dict]:
        """Generate TWAP child orders"""

        total_quantity = parent_order["quantity"]
        duration_minutes = parent_order.get("duration_minutes", 60)
        interval_seconds = parent_order.get("interval_seconds", 60)

        # Calculate number of slices
        total_intervals = int((duration_minutes * 60) / interval_seconds)
        slice_quantity = total_quantity / total_intervals

        # Generate child orders
        child_orders = []
        start_time = datetime.now()

        for i in range(total_intervals):
            # Calculate execution time for this slice
            execution_time = start_time + timedelta(seconds=i * interval_seconds)

            # Adjust slice size for rounding
            if i == total_intervals - 1:
                # Last slice gets remaining quantity
                actual_quantity = total_quantity - (slice_quantity * i)
            else:
                actual_quantity = slice_quantity

            child_order = {
                "parent_order_id": parent_order["order_id"],
                "child_order_id": f"{parent_order['order_id']}_child_{i+1}",
                "symbol": parent_order["symbol"],
                "side": parent_order["side"],
                "quantity": actual_quantity,
                "order_type": parent_order.get("child_order_type", "market"),
                "execution_time": execution_time.isoformat(),
                "slice_number": i + 1,
                "total_slices": total_intervals,
                "strategy": "TWAP"
            }

            child_orders.append(child_order)

        return child_orders

    def get_description(self) -> str:
        return "Executes orders evenly over time to minimize market impact"

    def get_parameters(self) -> List[str]:
        return ["duration_minutes", "interval_seconds", "child_order_type"]

    def get_suitable_conditions(self) -> List[str]:
        return ["large_orders", "low_urgency", "stable_markets"]

class VwapStrategy:
    def __init__(self, market_data_provider):
        self.market_data_provider = market_data_provider
        self.name = "VWAP"
        self.description = "Volume-Weighted Average Price execution"

    def generate_child_orders(self, parent_order: Dict) -> List[Dict]:
        """Generate VWAP child orders based on historical volume patterns"""

        total_quantity = parent_order["quantity"]
        symbol = parent_order["symbol"]
        duration_minutes = parent_order.get("duration_minutes", 60)

        # Get historical volume profile
        volume_profile = self.get_volume_profile(symbol, duration_minutes)

        # Calculate slice quantities based on volume profile
        child_orders = []
        start_time = datetime.now()

        for i, volume_weight in enumerate(volume_profile):
            slice_quantity = total_quantity * volume_weight
            execution_time = start_time + timedelta(minutes=i)

            child_order = {
                "parent_order_id": parent_order["order_id"],
                "child_order_id": f"{parent_order['order_id']}_child_{i+1}",
                "symbol": parent_order["symbol"],
                "side": parent_order["side"],
                "quantity": slice_quantity,
                "order_type": parent_order.get("child_order_type", "market"),
                "execution_time": execution_time.isoformat(),
                "slice_number": i + 1,
                "total_slices": len(volume_profile),
                "volume_weight": volume_weight,
                "strategy": "VWAP"
            }

            child_orders.append(child_order)

        return child_orders

    def get_volume_profile(self, symbol: str, duration_minutes: int) -> List[float]:
        """Get historical volume profile for symbol"""

        # Fetch historical volume data
        historical_data = self.market_data_provider.get_historical_volume(
            symbol, duration_minutes
        )

        # Calculate volume weights for each time period
        total_volume = sum(period["volume"] for period in historical_data)

        if total_volume == 0:
            # Default to equal distribution if no historical data
            num_periods = max(1, duration_minutes // 5)  # 5-minute periods
            return [1.0 / num_periods] * num_periods

        volume_weights = []
        for period in historical_data:
            weight = period["volume"] / total_volume
            volume_weights.append(weight)

        return volume_weights

    def get_description(self) -> str:
        return "Executes orders proportionally to historical volume patterns"

    def get_parameters(self) -> List[str]:
        return ["duration_minutes", "child_order_type", "volume_profile_source"]

    def get_suitable_conditions(self) -> List[str]:
        return ["large_orders", "predictable_volume", "liquid_markets"]

class PovStrategy:
    def __init__(self, market_data_provider):
        self.market_data_provider = market_data_provider
        self.name = "POV"
        self.description = "Percentage of Volume execution"

    def generate_child_orders(self, parent_order: Dict) -> List[Dict]:
        """Generate POV child orders"""

        total_quantity = parent_order["quantity"]
        pov_percentage = parent_order.get("pov_percentage", 10.0)  # 10% of volume
        max_duration_minutes = parent_order.get("max_duration_minutes", 120)

        # Get current market volume
        current_volume = self.get_current_market_volume(parent_order["symbol"])

        # Calculate target execution rate
        target_execution_rate = current_volume * (pov_percentage / 100.0)

        # Generate adaptive child orders
        child_orders = []
        remaining_quantity = total_quantity
        current_time = datetime.now()

        while remaining_quantity > 0:
            # Calculate next slice size based on current market conditions
            slice_size = min(remaining_quantity, target_execution_rate)

            child_order = {
                "parent_order_id": parent_order["order_id"],
                "child_order_id": f"{parent_order['order_id']}_child_{len(child_orders)+1}",
                "symbol": parent_order["symbol"],
                "side": parent_order["side"],
                "quantity": slice_size,
                "order_type": "market",
                "execution_time": current_time.isoformat(),
                "pov_percentage": pov_percentage,
                "target_rate": target_execution_rate,
                "strategy": "POV"
            }

            child_orders.append(child_order)
            remaining_quantity -= slice_size

            # Update execution time for next slice
            current_time += timedelta(minutes=1)

            # Check if we've exceeded max duration
            if (current_time - datetime.now()).total_seconds() > max_duration_minutes * 60:
                break

        return child_orders

    def get_current_market_volume(self, symbol: str) -> float:
        """Get current market volume for symbol"""

        # Fetch recent volume data
        recent_trades = self.market_data_provider.get_recent_trades(symbol, limit=100)

        # Calculate average volume per minute
        if recent_trades:
            total_volume = sum(trade["quantity"] for trade in recent_trades)
            time_span = (datetime.fromisoformat(recent_trades[-1]["timestamp"]) - 
                        datetime.fromisoformat(recent_trades[0]["timestamp"])).total_seconds()

            if time_span > 0:
                volume_per_minute = (total_volume / time_span) * 60
                return volume_per_minute

        # Default volume if no data available
        return 1000.0

    def get_description(self) -> str:
        return "Executes orders as a percentage of market volume"

    def get_parameters(self) -> List[str]:
        return ["pov_percentage", "max_duration_minutes"]

    def get_suitable_conditions(self) -> List[str]:
        return ["large_orders", "active_markets", "volume_based_execution"]

class SplitExecutor:
    def __init__(self, child_order_manager, order_router, execution_engine):
        self.child_order_manager = child_order_manager
        self.order_router = order_router
        self.execution_engine = execution_engine
        self.active_executions = {}

    async def execute_split_plan(self, split_plan: Dict) -> Dict:
        """Execute the complete split order plan"""

        parent_order_id = split_plan["parent_order_id"]
        child_orders = split_plan["child_orders"]
        execution_schedule = split_plan["execution_schedule"]

        # Initialize execution tracking
        execution_id = f"exec_{parent_order_id}_{int(time.time())}"
        self.active_executions[execution_id] = {
            "parent_order_id": parent_order_id,
            "total_child_orders": len(child_orders),
            "executed_orders": 0,
            "filled_quantity": 0,
            "total_quantity": sum(child["quantity"] for child in child_orders),
            "start_time": datetime.now(),
            "status": "running"
        }

        try:
            # Execute child orders according to schedule
            execution_results = await self.execute_scheduled_orders(
                child_orders, execution_schedule, execution_id
            )

            # Update execution status
            self.active_executions[execution_id]["status"] = "completed"
            self.active_executions[execution_id]["end_time"] = datetime.now()
            self.active_executions[execution_id]["results"] = execution_results

            return {
                "execution_id": execution_id,
                "status": "completed",
                "results": execution_results,
                "summary": self.generate_execution_summary(execution_id)
            }

        except Exception as e:
            # Handle execution failure
            self.active_executions[execution_id]["status"] = "failed"
            self.active_executions[execution_id]["error"] = str(e)

            # Cancel remaining orders
            await self.cancel_remaining_orders(execution_id)

            raise e

    async def execute_scheduled_orders(self, child_orders: List[Dict], 
                                     schedule: Dict, execution_id: str) -> List[Dict]:
        """Execute child orders according to schedule"""

        execution_results = []

        for child_order in child_orders:
            try:
                # Check if it's time to execute this order
                execution_time = datetime.fromisoformat(child_order["execution_time"])
                current_time = datetime.now()

                if current_time < execution_time:
                    # Wait until execution time
                    wait_seconds = (execution_time - current_time).total_seconds()
                    if wait_seconds > 0:
                        await asyncio.sleep(wait_seconds)

                # Execute the child order
                result = await self.execute_child_order(child_order, execution_id)
                execution_results.append(result)

                # Update execution tracking
                self.active_executions[execution_id]["executed_orders"] += 1
                if result["status"] == "filled":
                    self.active_executions[execution_id]["filled_quantity"] += result["filled_quantity"]

            except Exception as e:
                # Handle individual order failure
                result = {
                    "child_order_id": child_order["child_order_id"],
                    "status": "failed",
                    "error": str(e)
                }
                execution_results.append(result)

        return execution_results

    async def execute_child_order(self, child_order: Dict, execution_id: str) -> Dict:
        """Execute individual child order"""

        try:
            # Route order to appropriate venue
            venue = self.order_router.select_venue(child_order)

            # Place order through execution engine
            order_result = await self.execution_engine.place_order(
                symbol=child_order["symbol"],
                side=child_order["side"],
                quantity=child_order["quantity"],
                order_type=child_order["order_type"],
                venue=venue
            )

            return {
                "child_order_id": child_order["child_order_id"],
                "status": order_result["status"],
                "filled_quantity": order_result.get("filled_quantity", 0),
                "avg_price": order_result.get("avg_price", 0),
                "venue": venue,
                "execution_time": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "child_order_id": child_order["child_order_id"],
                "status": "failed",
                "error": str(e)
            }

    async def cancel_remaining_orders(self, execution_id: str):
        """Cancel all remaining orders for an execution"""

        # This would integrate with order management system
        # to cancel pending orders
        pass

    def generate_execution_summary(self, execution_id: str) -> Dict:
        """Generate execution summary"""

        execution = self.active_executions[execution_id]

        if execution["status"] == "completed":
            duration = (execution["end_time"] - execution["start_time"]).total_seconds()
            fill_rate = execution["filled_quantity"] / execution["total_quantity"]

            return {
                "execution_id": execution_id,
                "status": "completed",
                "total_orders": execution["total_child_orders"],
                "executed_orders": execution["executed_orders"],
                "filled_quantity": execution["filled_quantity"],
                "total_quantity": execution["total_quantity"],
                "fill_rate": fill_rate,
                "duration_seconds": duration,
                "start_time": execution["start_time"].isoformat(),
                "end_time": execution["end_time"].isoformat()
            }
        else:
            return {
                "execution_id": execution_id,
                "status": execution["status"],
                "error": execution.get("error", "Unknown error")
            }

class ExecutionMonitor:
    def __init__(self, progress_tracker, performance_analyzer):
        self.progress_tracker = progress_tracker
        self.performance_analyzer = performance_analyzer
        self.active_monitors = {}

    def start_monitoring(self, execution_id: str, split_plan: Dict):
        """Start monitoring an execution"""

        self.active_monitors[execution_id] = {
            "split_plan": split_plan,
            "start_time": datetime.now(),
            "progress": 0.0,
            "performance_metrics": {},
            "alerts": []
        }

    def update_progress(self, execution_id: str, current_status: Dict):
        """Update execution progress"""

        if execution_id not in self.active_monitors:
            return

        monitor = self.active_monitors[execution_id]

        # Update progress
        total_quantity = monitor["split_plan"]["child_orders"][0]["total_quantity"]
        filled_quantity = current_status.get("filled_quantity", 0)
        monitor["progress"] = filled_quantity / total_quantity if total_quantity > 0 else 0

        # Calculate performance metrics
        performance_metrics = self.performance_analyzer.calculate_metrics(
            execution_id, current_status
        )
        monitor["performance_metrics"] = performance_metrics

        # Check for alerts
        alerts = self.check_alerts(execution_id, current_status, performance_metrics)
        monitor["alerts"].extend(alerts)

    def check_alerts(self, execution_id: str, current_status: Dict, 
                    performance_metrics: Dict) -> List[Dict]:
        """Check for execution alerts"""

        alerts = []

        # Check progress alerts
        progress = current_status.get("progress", 0)
        expected_progress = self.calculate_expected_progress(execution_id)

        if progress < expected_progress * 0.8:  # 20% behind schedule
            alerts.append({
                "type": "progress_lag",
                "severity": "warning",
                "message": f"Execution is {((expected_progress - progress) / expected_progress * 100):.1f}% behind schedule",
                "timestamp": datetime.now().isoformat()
            })

        # Check performance alerts
        slippage = performance_metrics.get("slippage", 0)
        if slippage > 0.002:  # 20 basis points
            alerts.append({
                "type": "high_slippage",
                "severity": "warning",
                "message": f"High slippage detected: {slippage:.4f}",
                "timestamp": datetime.now().isoformat()
            })

        return alerts

    def calculate_expected_progress(self, execution_id: str) -> float:
        """Calculate expected progress based on time elapsed"""

        monitor = self.active_monitors[execution_id]
        start_time = monitor["start_time"]
        current_time = datetime.now()

        # Calculate expected progress based on time
        elapsed_seconds = (current_time - start_time).total_seconds()
        total_duration = monitor["split_plan"].get("duration_minutes", 60) * 60

        return min(1.0, elapsed_seconds / total_duration)

    def get_execution_status(self, execution_id: str) -> Dict:
        """Get current execution status"""

        if execution_id not in self.active_monitors:
            return {"status": "not_found"}

        monitor = self.active_monitors[execution_id]

        return {
            "execution_id": execution_id,
            "progress": monitor["progress"],
            "performance_metrics": monitor["performance_metrics"],
            "alerts": monitor["alerts"],
            "start_time": monitor["start_time"].isoformat(),
            "elapsed_time": (datetime.now() - monitor["start_time"]).total_seconds()
        }

@router.post("/split/execute")
async def execute_split_order(request: SplitOrderRequest):
    """Execute a large order using splitting strategy"""

    # Validate request
    if not split_strategy_manager.validate_order(request.dict()):
        raise HTTPException(status_code=400, detail="Invalid order parameters")

    # Create split plan
    split_plan = split_strategy_manager.create_split_plan(request.dict())

    # Start monitoring
    execution_monitor.start_monitoring(split_plan["parent_order_id"], split_plan)

    # Execute split plan
    result = await split_executor.execute_split_plan(split_plan)

    return result

@router.get("/split/status/{execution_id}")
async def get_execution_status(execution_id: str):
    """Get execution status and progress"""

    status = execution_monitor.get_execution_status(execution_id)
    return status

@router.get("/split/strategies")
async def get_available_strategies():
    """Get available splitting strategies"""

    strategies = split_strategy_manager.get_available_strategies()
    return {"strategies": strategies}

@router.post("/split/cancel/{execution_id}")
async def cancel_execution(execution_id: str):
    """Cancel ongoing execution"""

    result = await split_executor.cancel_execution(execution_id)
    return {"status": "cancelled", "execution_id": execution_id}

const OrderSplittingView: React.FC = () => {
  const [splitOrderForm, setSplitOrderForm] = useState<SplitOrderForm | null>(null);
  const [activeExecutions, setActiveExecutions] = useState<Execution[]>([]);
  const [availableStrategies, setAvailableStrategies] = useState<Strategy[]>([]);

  return (
    <div className="order-splitting-dashboard">
      {/* Split Order Form */}
      <SplitOrderForm 
        onSubmit={handleSplitOrder}
        strategies={availableStrategies}
        form={splitOrderForm}
      />

      {/* Active Executions */}
      <ActiveExecutionsPanel 
        executions={activeExecutions}
        onCancel={handleCancelExecution}
      />

      {/* Execution Progress */}
      {activeExecutions.map(execution => (
        <ExecutionProgressPanel 
          key={execution.execution_id}
          execution={execution}
        />
      ))}

      {/* Child Orders List */}
      <ChildOrdersPanel 
        executions={activeExecutions}
      />

      {/* Performance Metrics */}
      <PerformanceMetricsPanel 
        executions={activeExecutions}
      />

      {/* Alerts and Warnings */}
      <AlertsPanel 
        executions={activeExecutions}
      />
    </div>
  );
};