43. Smart Capital Allocation Engine¶

services/capital-allocation-center/
├── src/
│   ├── main.py
│   ├── manager/
│   │   └── strategy_fund_manager.py
│   ├── weighter/
│   │   ├── sector_signal_weighter.py
│   │   └── performance_weighter.py
│   ├── controller/
│   │   ├── risk_exposure_controller.py
│   │   └── target_exposure_controller.py
│   ├── allocator/
│   │   ├── dynamic_allocator.py
│   │   ├── risk_parity_allocator.py
│   │   └── momentum_allocator.py
│   ├── transition/
│   │   └── smooth_transition_manager.py
│   ├── api/
│   │   └── allocation_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class StrategyFundManager:
    def __init__(self):
        self.strategy_funds = {}
        self.strategy_performance = {}
        self.allocation_history = []

    def set_fund(self, strategy_id: str, fund_amount: float):
        """Set absolute fund amount for a strategy"""
        self.strategy_funds[strategy_id] = fund_amount
        self.record_allocation_change(strategy_id, fund_amount)

    def adjust_fund(self, strategy_id: str, adjustment_ratio: float):
        """Adjust fund amount by ratio"""
        if strategy_id in self.strategy_funds:
            new_amount = self.strategy_funds[strategy_id] * adjustment_ratio
            self.strategy_funds[strategy_id] = new_amount
            self.record_allocation_change(strategy_id, new_amount)

    def get_total_capital(self) -> float:
        """Get total allocated capital"""
        return sum(self.strategy_funds.values())

    def get_allocation_weights(self) -> dict:
        """Get current allocation weights"""
        total = self.get_total_capital()
        if total == 0:
            return {}
        return {strategy: amount / total 
                for strategy, amount in self.strategy_funds.items()}

    def record_allocation_change(self, strategy_id: str, new_amount: float):
        """Record allocation change for audit trail"""
        self.allocation_history.append({
            'timestamp': datetime.now(),
            'strategy_id': strategy_id,
            'amount': new_amount,
            'total_capital': self.get_total_capital()
        })

class SectorSignalWeighter:
    def __init__(self, signal_decay_factor=0.95):
        self.signal_decay_factor = signal_decay_factor
        self.signal_history = {}

    def compute_weights(self, rotation_signals: list, base_weights: dict) -> dict:
        """Compute adjusted weights based on sector rotation signals"""
        adjusted_weights = base_weights.copy()

        for signal in rotation_signals:
            sector = signal['sector']
            action = signal['action']
            confidence = signal.get('confidence', 1.0)

            if sector in adjusted_weights:
                if action == "BUY":
                    # Increase weight for bullish sectors
                    weight_multiplier = 1.0 + (0.3 * confidence)
                    adjusted_weights[sector] *= weight_multiplier
                elif action == "SELL":
                    # Decrease weight for bearish sectors
                    weight_multiplier = 1.0 - (0.2 * confidence)
                    adjusted_weights[sector] *= weight_multiplier

        # Normalize weights
        total_weight = sum(adjusted_weights.values())
        if total_weight > 0:
            adjusted_weights = {k: v / total_weight for k, v in adjusted_weights.items()}

        return adjusted_weights

    def apply_signal_decay(self, weights: dict) -> dict:
        """Apply signal decay to prevent excessive weight changes"""
        decayed_weights = {}
        for strategy, weight in weights.items():
            if strategy in self.signal_history:
                # Apply decay to recent signal impact
                decayed_weight = (weight * self.signal_decay_factor + 
                                self.signal_history[strategy] * (1 - self.signal_decay_factor))
                decayed_weights[strategy] = decayed_weight
            else:
                decayed_weights[strategy] = weight

        self.signal_history = decayed_weights.copy()
        return decayed_weights

class PerformanceWeighter:
    def __init__(self, lookback_period=30, momentum_factor=0.3):
        self.lookback_period = lookback_period
        self.momentum_factor = momentum_factor

    def compute_performance_weights(self, strategy_performance: dict) -> dict:
        """Compute weights based on strategy performance"""
        performance_scores = {}

        for strategy_id, performance in strategy_performance.items():
            # Calculate performance score using Sharpe ratio
            returns = performance.get('returns', [])
            if len(returns) >= self.lookback_period:
                recent_returns = returns[-self.lookback_period:]
                sharpe_ratio = np.mean(recent_returns) / np.std(recent_returns)
                performance_scores[strategy_id] = max(sharpe_ratio, 0)  # No negative weights

        # Normalize performance scores
        total_score = sum(performance_scores.values())
        if total_score > 0:
            performance_weights = {k: v / total_score for k, v in performance_scores.items()}
        else:
            # Equal weights if no positive performance
            strategy_count = len(strategy_performance)
            performance_weights = {k: 1.0 / strategy_count for k in strategy_performance.keys()}

        return performance_weights

    def apply_momentum_adjustment(self, current_weights: dict, 
                                performance_weights: dict) -> dict:
        """Apply momentum-based adjustment to current weights"""
        adjusted_weights = {}

        for strategy in current_weights:
            if strategy in performance_weights:
                # Blend current weight with performance weight
                current_weight = current_weights[strategy]
                performance_weight = performance_weights[strategy]

                adjusted_weight = (current_weight * (1 - self.momentum_factor) + 
                                 performance_weight * self.momentum_factor)
                adjusted_weights[strategy] = adjusted_weight
            else:
                adjusted_weights[strategy] = current_weights[strategy]

        return adjusted_weights

class RiskExposureController:
    def __init__(self, max_single_strategy_exposure=0.2, 
                 max_sector_exposure=0.3, max_total_risk=0.8):
        self.max_single_strategy_exposure = max_single_strategy_exposure
        self.max_sector_exposure = max_sector_exposure
        self.max_total_risk = max_total_risk

    def enforce_risk_limits(self, strategy_funds: dict, 
                          total_equity: float, 
                          strategy_risk_metrics: dict) -> dict:
        """Enforce risk limits on strategy allocations"""
        adjusted_funds = strategy_funds.copy()

        # Single strategy exposure limit
        for strategy, fund in adjusted_funds.items():
            exposure_ratio = fund / total_equity
            if exposure_ratio > self.max_single_strategy_exposure:
                max_fund = total_equity * self.max_single_strategy_exposure
                adjusted_funds[strategy] = max_fund

        # Sector exposure limit
        sector_exposures = self.calculate_sector_exposures(adjusted_funds)
        for sector, exposure in sector_exposures.items():
            if exposure > self.max_sector_exposure:
                # Reduce sector exposure proportionally
                reduction_factor = self.max_sector_exposure / exposure
                for strategy, fund in adjusted_funds.items():
                    if self.get_strategy_sector(strategy) == sector:
                        adjusted_funds[strategy] *= reduction_factor

        # Total risk limit
        total_risk = self.calculate_total_risk(adjusted_funds, strategy_risk_metrics)
        if total_risk > self.max_total_risk:
            # Scale down all allocations proportionally
            scale_factor = self.max_total_risk / total_risk
            adjusted_funds = {k: v * scale_factor for k, v in adjusted_funds.items()}

        return adjusted_funds

    def calculate_sector_exposures(self, strategy_funds: dict) -> dict:
        """Calculate total exposure by sector"""
        sector_exposures = {}
        total_funds = sum(strategy_funds.values())

        for strategy, fund in strategy_funds.items():
            sector = self.get_strategy_sector(strategy)
            sector_exposures[sector] = sector_exposures.get(sector, 0) + fund

        # Convert to ratios
        if total_funds > 0:
            sector_exposures = {k: v / total_funds for k, v in sector_exposures.items()}

        return sector_exposures

    def calculate_total_risk(self, strategy_funds: dict, 
                           strategy_risk_metrics: dict) -> float:
        """Calculate total portfolio risk"""
        total_risk = 0.0
        total_funds = sum(strategy_funds.values())

        for strategy, fund in strategy_funds.items():
            if strategy in strategy_risk_metrics:
                strategy_risk = strategy_risk_metrics[strategy].get('volatility', 0)
                weight = fund / total_funds if total_funds > 0 else 0
                total_risk += weight * strategy_risk

        return total_risk

    def get_strategy_sector(self, strategy_id: str) -> str:
        """Get sector classification for a strategy"""
        # This would typically come from strategy metadata
        sector_mapping = {
            'tech_strategy': 'Technology',
            'finance_strategy': 'Financial',
            'healthcare_strategy': 'Healthcare',
            'crypto_strategy': 'Cryptocurrency'
        }
        return sector_mapping.get(strategy_id, 'Other')

class DynamicAllocator:
    def __init__(self, fund_manager, sector_weighter, performance_weighter, 
                 risk_controller, transition_manager):
        self.fund_manager = fund_manager
        self.sector_weighter = sector_weighter
        self.performance_weighter = performance_weighter
        self.risk_controller = risk_controller
        self.transition_manager = transition_manager
        self.allocation_method = 'hybrid'  # 'risk_parity', 'momentum', 'hybrid'

    def reallocate(self, total_equity: float, 
                  sector_signals: list = None,
                  strategy_performance: dict = None,
                  force_rebalance: bool = False) -> dict:
        """Execute dynamic capital reallocation"""

        # Get current allocation
        current_weights = self.fund_manager.get_allocation_weights()

        # Compute new target weights
        if self.allocation_method == 'risk_parity':
            target_weights = self.compute_risk_parity_weights(strategy_performance)
        elif self.allocation_method == 'momentum':
            target_weights = self.compute_momentum_weights(sector_signals, strategy_performance)
        else:  # hybrid
            target_weights = self.compute_hybrid_weights(sector_signals, strategy_performance)

        # Apply risk controls
        target_funds = {k: v * total_equity for k, v in target_weights.items()}
        adjusted_funds = self.risk_controller.enforce_risk_limits(
            target_funds, total_equity, strategy_performance or {}
        )

        # Execute smooth transition
        if not force_rebalance:
            final_funds = self.transition_manager.smooth_transition(
                current_weights, adjusted_funds, total_equity
            )
        else:
            final_funds = adjusted_funds

        # Update fund manager
        for strategy, fund in final_funds.items():
            self.fund_manager.set_fund(strategy, fund)

        return final_funds

    def compute_risk_parity_weights(self, strategy_performance: dict) -> dict:
        """Compute risk parity weights"""
        risk_contributions = {}

        for strategy, performance in strategy_performance.items():
            volatility = performance.get('volatility', 0.1)
            risk_contributions[strategy] = 1.0 / volatility if volatility > 0 else 1.0

        # Normalize to weights
        total_contribution = sum(risk_contributions.values())
        if total_contribution > 0:
            weights = {k: v / total_contribution for k, v in risk_contributions.items()}
        else:
            strategy_count = len(strategy_performance)
            weights = {k: 1.0 / strategy_count for k in strategy_performance.keys()}

        return weights

    def compute_momentum_weights(self, sector_signals: list, 
                               strategy_performance: dict) -> dict:
        """Compute momentum-based weights"""
        # Start with performance weights
        performance_weights = self.performance_weighter.compute_performance_weights(
            strategy_performance
        )

        # Apply sector signal adjustments
        if sector_signals:
            adjusted_weights = self.sector_weighter.compute_weights(
                sector_signals, performance_weights
            )
        else:
            adjusted_weights = performance_weights

        return adjusted_weights

    def compute_hybrid_weights(self, sector_signals: list, 
                             strategy_performance: dict) -> dict:
        """Compute hybrid weights combining multiple approaches"""
        # Get weights from different methods
        risk_parity_weights = self.compute_risk_parity_weights(strategy_performance)
        momentum_weights = self.compute_momentum_weights(sector_signals, strategy_performance)

        # Combine with equal weight (can be made configurable)
        hybrid_weights = {}
        for strategy in risk_parity_weights:
            if strategy in momentum_weights:
                hybrid_weights[strategy] = (risk_parity_weights[strategy] * 0.5 + 
                                          momentum_weights[strategy] * 0.5)
            else:
                hybrid_weights[strategy] = risk_parity_weights[strategy]

        return hybrid_weights

class SmoothTransitionManager:
    def __init__(self, transition_period_days=5, max_daily_change=0.1):
        self.transition_period_days = transition_period_days
        self.max_daily_change = max_daily_change
        self.transition_schedules = {}

    def smooth_transition(self, current_weights: dict, target_weights: dict, 
                         total_equity: float) -> dict:
        """Execute smooth transition to target weights"""

        # Calculate required changes
        weight_changes = {}
        for strategy in set(current_weights.keys()) | set(target_weights.keys()):
            current_weight = current_weights.get(strategy, 0)
            target_weight = target_weights.get(strategy, 0)
            weight_changes[strategy] = target_weight - current_weight

        # Check if changes are significant enough to warrant transition
        total_change = sum(abs(change) for change in weight_changes.values())
        if total_change < 0.05:  # Less than 5% total change
            return target_weights

        # Create transition schedule
        transition_schedule = self.create_transition_schedule(
            current_weights, target_weights, total_equity
        )

        # Execute first step of transition
        next_weights = self.execute_transition_step(transition_schedule, 0)

        return next_weights

    def create_transition_schedule(self, current_weights: dict, 
                                 target_weights: dict, 
                                 total_equity: float) -> dict:
        """Create multi-day transition schedule"""
        schedule = {
            'current_weights': current_weights,
            'target_weights': target_weights,
            'total_equity': total_equity,
            'steps': []
        }

        # Calculate daily changes
        for day in range(self.transition_period_days):
            step_weights = {}
            for strategy in set(current_weights.keys()) | set(target_weights.keys()):
                current_weight = current_weights.get(strategy, 0)
                target_weight = target_weights.get(strategy, 0)

                # Linear interpolation
                progress = (day + 1) / self.transition_period_days
                step_weight = current_weight + (target_weight - current_weight) * progress

                # Apply daily change limit
                if day > 0:
                    prev_weight = schedule['steps'][day - 1].get(strategy, current_weight)
                    max_change = self.max_daily_change
                    if abs(step_weight - prev_weight) > max_change:
                        if step_weight > prev_weight:
                            step_weight = prev_weight + max_change
                        else:
                            step_weight = prev_weight - max_change

                step_weights[strategy] = step_weight

            schedule['steps'].append(step_weights)

        return schedule

    def execute_transition_step(self, schedule: dict, step_index: int) -> dict:
        """Execute a specific step in the transition schedule"""
        if step_index < len(schedule['steps']):
            return schedule['steps'][step_index]
        else:
            return schedule['target_weights']

from fastapi import APIRouter, HTTPException, BackgroundTasks
from typing import Dict, List, Optional
from pydantic import BaseModel

router = APIRouter()

class AllocationRequest(BaseModel):
    total_equity: float
    sector_signals: Optional[List[Dict]] = None
    strategy_performance: Optional[Dict] = None
    force_rebalance: bool = False
    allocation_method: str = "hybrid"

class AllocationResponse(BaseModel):
    allocation_id: str
    status: str
    current_allocation: Dict[str, float]
    target_allocation: Optional[Dict[str, float]] = None
    transition_schedule: Optional[List[Dict]] = None

@router.get("/allocation/current")
async def get_current_allocation():
    """Get current capital allocation"""
    return {
        "timestamp": datetime.now().isoformat(),
        "total_capital": fund_manager.get_total_capital(),
        "allocation": fund_manager.get_allocation_weights(),
        "allocation_history": fund_manager.allocation_history[-10:]  # Last 10 changes
    }

@router.post("/allocation/rebalance", response_model=AllocationResponse)
async def trigger_rebalance(request: AllocationRequest):
    """Trigger capital reallocation"""
    allocation_id = generate_allocation_id()

    try:
        new_allocation = dynamic_allocator.reallocate(
            total_equity=request.total_equity,
            sector_signals=request.sector_signals,
            strategy_performance=request.strategy_performance,
            force_rebalance=request.force_rebalance
        )

        return AllocationResponse(
            allocation_id=allocation_id,
            status="completed",
            current_allocation=new_allocation,
            target_allocation=new_allocation
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.get("/allocation/performance")
async def get_allocation_performance():
    """Get allocation performance metrics"""
    return {
        "timestamp": datetime.now().isoformat(),
        "total_return": calculate_total_return(),
        "risk_metrics": calculate_risk_metrics(),
        "sector_exposures": risk_controller.calculate_sector_exposures(
            fund_manager.strategy_funds
        )
    }

@router.get("/allocation/methods")
async def get_allocation_methods():
    """Get available allocation methods"""
    return {
        "methods": [
            "risk_parity",
            "momentum", 
            "hybrid",
            "equal_weight",
            "performance_weighted"
        ],
        "current_method": dynamic_allocator.allocation_method
    }

@router.put("/allocation/method/{method}")
async def set_allocation_method(method: str):
    """Set allocation method"""
    if method not in ["risk_parity", "momentum", "hybrid", "equal_weight", "performance_weighted"]:
        raise HTTPException(status_code=400, detail="Invalid allocation method")

    dynamic_allocator.allocation_method = method
    return {"message": f"Allocation method set to {method}"}

// CapitalAllocationView.tsx
interface AllocationData {
  strategyId: string;
  currentWeight: number;
  targetWeight: number;
  fundAmount: number;
  sector: string;
  performance: number;
}

interface AllocationMetrics {
  totalCapital: number;
  totalReturn: number;
  riskMetrics: Record<string, number>;
  sectorExposures: Record<string, number>;
}

const CapitalAllocationView: React.FC = () => {
  const [allocationData, setAllocationData] = useState<AllocationData[]>([]);
  const [metrics, setMetrics] = useState<AllocationMetrics | null>(null);
  const [allocationMethod, setAllocationMethod] = useState<string>('hybrid');

  const triggerRebalance = async () => {
    const response = await api.post('/allocation/rebalance', {
      total_equity: metrics?.totalCapital || 0,
      allocation_method: allocationMethod
    });
    // Update allocation data
  };

  return (
    <div className="capital-allocation-dashboard">
      {/* Allocation Overview */}
      <AllocationOverview metrics={metrics} />

      {/* Strategy Allocation Chart */}
      <StrategyAllocationChart data={allocationData} />

      {/* Sector Exposure Heatmap */}
      <SectorExposureHeatmap exposures={metrics?.sectorExposures} />

      {/* Allocation Controls */}
      <AllocationControls 
        method={allocationMethod}
        onMethodChange={setAllocationMethod}
        onRebalance={triggerRebalance}
      />

      {/* Historical Allocation Trends */}
      <AllocationTrends data={allocationData} />
    </div>
  );
};

# capital-allocation-config.yaml
allocation:
  method: "hybrid"
  rebalance_frequency: "daily"
  force_rebalance_threshold: 0.1

risk_limits:
  max_single_strategy_exposure: 0.2
  max_sector_exposure: 0.3
  max_total_risk: 0.8

transition:
  transition_period_days: 5
  max_daily_change: 0.1
  min_change_threshold: 0.05

performance:
  lookback_period: 30
  momentum_factor: 0.3
  signal_decay_factor: 0.95