Position Sizing & Leverage: Kelly Criterion Strategy › Practical Position Sizing Implementation

from dataclasses import dataclass
from typing import Dict, List

import numpy as np


@dataclass
class StrategyParams:
    """Parameters for a trading strategy."""

    name: str
    expected_return: float  # Annualized expected return
    volatility: float  # Annualized volatility
    max_weight: float = 0.5  # Maximum portfolio weight
    min_weight: float = -0.5  # Minimum portfolio weight (negative = short)


@dataclass
class PortfolioConstraints:
    """Constraints for the overall portfolio."""

    max_gross_leverage: float = 2.0  # Max sum of absolute weights
    max_net_leverage: float = 1.0  # Max sum of signed weights
    target_volatility: float = 0.15  # Target portfolio volatility
    kelly_fraction: float = 0.5  # Fractional Kelly multiplier


class PositionSizer:
    """
    Position sizing engine that combines Kelly criterion, risk budgeting,
    and practical constraints.
    """

    def __init__(
        self,
        strategies: List[StrategyParams],
        constraints: PortfolioConstraints,
        correlation_matrix: np.ndarray,
    ):
        self.strategies = {s.name: s for s in strategies}
        self.constraints = constraints
        self.n_strategies = len(strategies)
        self.correlation = correlation_matrix

        # Build covariance matrix
        vols = np.array([s.volatility for s in strategies])
        self.cov_matrix = np.outer(vols, vols) * correlation_matrix

    def kelly_weights(self) -> Dict[str, float]:
        """Calculate Kelly-optimal weights accounting for correlations."""
        exp_returns = np.array(
            [s.expected_return for s in self.strategies.values()]
        )

        # Full Kelly: w* = Sigma^{-1} * mu
        cov_inv = np.linalg.inv(self.cov_matrix)
        raw_weights = cov_inv @ exp_returns

        # Apply fractional Kelly
        raw_weights *= self.constraints.kelly_fraction

        return {name: w for name, w in zip(self.strategies.keys(), raw_weights)}

    def apply_constraints(self, weights: Dict[str, float]) -> Dict[str, float]:
        """Apply portfolio and position-level constraints."""
        constrained = {}

        # First, apply position-level constraints
        for name, w in weights.items():
            strategy = self.strategies[name]
            constrained[name] = np.clip(
                w, strategy.min_weight, strategy.max_weight
            )

        # Check gross leverage constraint
        gross_leverage = sum(abs(w) for w in constrained.values())
        if gross_leverage > self.constraints.max_gross_leverage:
            scale = self.constraints.max_gross_leverage / gross_leverage
            constrained = {k: v * scale for k, v in constrained.items()}

        # Check net leverage constraint
        net_leverage = sum(constrained.values())
        if abs(net_leverage) > self.constraints.max_net_leverage:
            # Scale down all positions proportionally
            scale = self.constraints.max_net_leverage / abs(net_leverage)
            constrained = {k: v * scale for k, v in constrained.items()}

        return constrained

    def volatility_scale(self, weights: Dict[str, float]) -> float:
        """Calculate scale factor to hit target volatility."""
        w = np.array(list(weights.values()))
        port_vol = np.sqrt(w @ self.cov_matrix @ w)

        if port_vol > 0:
            return self.constraints.target_volatility / port_vol
        return 1.0

    def compute_positions(
        self, volatility_target: bool = True
    ) -> Dict[str, float]:
        """
        Compute final position sizes.

        Parameters:
        - volatility_target: If True, scale positions to hit target volatility

        Returns:
        - Dictionary of strategy name to weight
        """
        # Start with Kelly-optimal weights
        weights = self.kelly_weights()

        # Apply constraints
        weights = self.apply_constraints(weights)

        # Optionally scale to target volatility
        if volatility_target:
            scale = self.volatility_scale(weights)
            # Don't scale up beyond max leverage
            if scale > 1:
                max_scale = self.constraints.max_gross_leverage / sum(
                    abs(w) for w in weights.values()
                )
                scale = min(scale, max_scale)
            weights = {k: v * scale for k, v in weights.items()}
            # Re-apply constraints after scaling
            weights = self.apply_constraints(weights)

        return weights

    def risk_decomposition(
        self, weights: Dict[str, float]
    ) -> Dict[str, Dict[str, float]]:
        """Decompose portfolio risk by strategy."""
        w = np.array(list(weights.values()))
        port_var = w @ self.cov_matrix @ w
        port_vol = np.sqrt(port_var)

        # Risk contribution of each strategy
        marginal_contrib = self.cov_matrix @ w / port_vol
        risk_contrib = w * marginal_contrib

        result = {}
        for i, name in enumerate(weights.keys()):
            result[name] = {
                "weight": weights[name],
                "marginal_risk_contrib": marginal_contrib[i],
                "total_risk_contrib": risk_contrib[i],
                "pct_of_risk": risk_contrib[i] / port_vol * 100,
            }

        return result

# Define strategies
strategies = [
    StrategyParams(
        "Momentum",
        expected_return=0.12,
        volatility=0.20,
        max_weight=0.6,
        min_weight=-0.1,
    ),
    StrategyParams(
        "MeanRev",
        expected_return=0.08,
        volatility=0.15,
        max_weight=0.5,
        min_weight=-0.1,
    ),
    StrategyParams(
        "Factor",
        expected_return=0.10,
        volatility=0.12,
        max_weight=0.5,
        min_weight=0.0,
    ),
]

# Correlation matrix
corr = np.array([[1.0, -0.3, 0.2], [-0.3, 1.0, 0.1], [0.2, 0.1, 1.0]])

# Define constraints
constraints = PortfolioConstraints(
    max_gross_leverage=1.5,
    max_net_leverage=1.0,
    target_volatility=0.12,
    kelly_fraction=0.5,
)

# Create position sizer
sizer = PositionSizer(strategies, constraints, corr)

# Compute positions
positions = sizer.compute_positions(volatility_target=True)
risk_decomp = sizer.risk_decomposition(positions)

# Calculate portfolio summary statistics
port_return = sum(positions[s.name] * s.expected_return for s in strategies)
weights_array = np.array(list(positions.values()))

# Reconstruct full covariance for portfolio vol calc
vols = np.array([s.volatility for s in strategies])
full_cov = np.outer(vols, vols) * corr
port_vol = np.sqrt(weights_array @ full_cov @ weights_array)
port_sharpe = port_return / port_vol
gross_leverage = sum(abs(w) for w in positions.values())
net_leverage = sum(positions.values())

def drawdown_adjusted_sizing(
    base_weight: float,
    current_dd: float,
    dd_threshold: float = 0.10,
    max_reduction: float = 0.5,
) -> float:
    """
    Reduce position size based on current drawdown.

    Parameters:
    - base_weight: Normal position size
    - current_dd: Current drawdown (positive number, e.g., 0.15 = 15% drawdown)
    - dd_threshold: Drawdown level at which reduction begins
    - max_reduction: Maximum reduction factor at extreme drawdowns

    Returns:
    - Adjusted position size
    """
    if current_dd <= dd_threshold:
        return base_weight

    # Linear reduction from threshold to 2x threshold
    reduction_range = dd_threshold  # Full reduction at 2x threshold
    excess_dd = current_dd - dd_threshold
    reduction_pct = min(excess_dd / reduction_range, 1.0) * max_reduction

    return base_weight * (1 - reduction_pct)


# Example: show position adjustment across drawdown levels
drawdowns = np.linspace(0, 0.30, 50)
base_position = 1.0
adjusted_positions = [
    drawdown_adjusted_sizing(base_position, dd) for dd in drawdowns
]