position sizing with kelly criterion - python implementation

import numpy as np
from dataclasses import dataclass
from typing import Optional

@dataclass
class TradeSetup:
    symbol: str
    direction: str  # 'long' or 'short'
    entry_price: float
    stop_loss: float
    take_profit: float
    confidence: float  # 0.0 to 1.0
    strategy_win_rate: float
    strategy_avg_win: float
    strategy_avg_loss: float

class KellyPositionSizer:
    def __init__(
        self,
        account_size: float,
        max_position_pct: float = 0.05,  # 5% max
        kelly_fraction: float = 0.25,     # quarter kelly
        min_position_pct: float = 0.005,  # 0.5% min
        max_daily_risk: float = 0.02      # 2% daily max
    ):
        self.account_size = account_size
        self.max_position_pct = max_position_pct
        self.kelly_fraction = kelly_fraction
        self.min_position_pct = min_position_pct
        self.max_daily_risk = max_daily_risk
        self.daily_risk_used = 0.0

    def calculate_kelly(self, setup: TradeSetup) -> float:
        """Calculate raw kelly percentage"""
        p = setup.strategy_win_rate
        w = setup.strategy_avg_win
        l = setup.strategy_avg_loss

        if w <= 0 or l <= 0:
            return 0.0

        kelly = (p * w - (1 - p) * l) / w

        # kelly can be negative (don't trade)
        return max(0.0, kelly)

    def calculate_position_size(
        self,
        setup: TradeSetup,
        current_positions: int = 0
    ) -> dict:
        """Calculate position size with all adjustments"""

        # 1. raw kelly
        raw_kelly = self.calculate_kelly(setup)

        if raw_kelly <= 0:
            return {
                'shares': 0,
                'dollars': 0,
                'risk_pct': 0,
                'reason': 'negative_edge'
            }

        # 2. apply fractional kelly
        fractional_kelly = raw_kelly * self.kelly_fraction

        # 3. confidence adjustment
        confidence_adjusted = fractional_kelly * setup.confidence

        # 4. position count adjustment (reduce size with more positions)
        position_factor = 1.0 / (1 + current_positions * 0.2)
        adjusted_pct = confidence_adjusted * position_factor

        # 5. apply bounds
        bounded_pct = np.clip(
            adjusted_pct,
            self.min_position_pct,
            self.max_position_pct
        )

        # 6. daily risk check
        risk_per_trade = self._calculate_risk(setup, bounded_pct)
        remaining_daily_risk = self.max_daily_risk - self.daily_risk_used

        if risk_per_trade > remaining_daily_risk:
            # scale down to fit daily budget
            scale_factor = remaining_daily_risk / risk_per_trade
            bounded_pct *= scale_factor
            risk_per_trade = remaining_daily_risk

        # 7. calculate final values
        position_dollars = self.account_size * bounded_pct
        shares = int(position_dollars / setup.entry_price)

        return {
            'shares': shares,
            'dollars': shares * setup.entry_price,
            'risk_pct': risk_per_trade,
            'position_pct': bounded_pct,
            'raw_kelly': raw_kelly,
            'confidence_factor': setup.confidence,
            'reason': 'calculated'
        }

    def _calculate_risk(self, setup: TradeSetup, position_pct: float) -> float:
        """Calculate risk as percentage of account"""
        position_value = self.account_size * position_pct
        risk_per_share = abs(setup.entry_price - setup.stop_loss)
        shares = position_value / setup.entry_price
        total_risk = shares * risk_per_share
        return total_risk / self.account_size

    def record_trade(self, risk_pct: float):
        """Record trade risk for daily tracking"""
        self.daily_risk_used += risk_pct

    def reset_daily_risk(self):
        """Reset daily risk counter (call at market open)"""
        self.daily_risk_used = 0.0


class VolatilityAdjustedSizer(KellyPositionSizer):
    """Kelly sizer with volatility adjustment"""

    def __init__(self, *args, target_vol: float = 0.02, **kwargs):
        super().__init__(*args, **kwargs)
        self.target_vol = target_vol

    def calculate_position_size(
        self,
        setup: TradeSetup,
        current_vol: float,  # current realized volatility
        current_positions: int = 0
    ) -> dict:

        # get base kelly size
        base_result = super().calculate_position_size(setup, current_positions)

        if base_result['shares'] == 0:
            return base_result

        # volatility adjustment
        vol_ratio = self.target_vol / current_vol if current_vol > 0 else 1.0
        vol_adjusted_pct = base_result['position_pct'] * vol_ratio

        # re-apply bounds after vol adjustment
        vol_adjusted_pct = np.clip(
            vol_adjusted_pct,
            self.min_position_pct,
            self.max_position_pct
        )

        position_dollars = self.account_size * vol_adjusted_pct
        shares = int(position_dollars / setup.entry_price)

        return {
            **base_result,
            'shares': shares,
            'dollars': shares * setup.entry_price,
            'position_pct': vol_adjusted_pct,
            'vol_adjustment': vol_ratio
        }

# initialize sizer
sizer = VolatilityAdjustedSizer(
    account_size=460000,
    max_position_pct=0.05,
    kelly_fraction=0.25,
    target_vol=0.015
)

# define setup
setup = TradeSetup(
    symbol='SPX',
    direction='short',
    entry_price=5850,
    stop_loss=5900,
    take_profit=5750,
    confidence=0.75,
    strategy_win_rate=0.68,
    strategy_avg_win=450,
    strategy_avg_loss=280
)

# calculate size
result = sizer.calculate_position_size(
    setup,
    current_vol=0.012,
    current_positions=2
)

print(f"Position: ${result['dollars']:,.0f}")
print(f"Risk: {result['risk_pct']:.2%}")
print(f"Kelly raw: {result['raw_kelly']:.2%}")