risk management - position sizing with kelly criterion in python

import pandas as pd
import numpy as np
from typing import Dict, List

class KellyCriterion:
    def __init__(self, trades_df: pd.DataFrame, fraction: float = 0.25):
        """
        Initialize Kelly Criterion calculator

        Args:
            trades_df: DataFrame with columns ['pnl', 'win']
            fraction: Kelly fraction (0.25 = quarter kelly)
        """
        self.trades = trades_df
        self.fraction = fraction
        self.metrics = self._calculate_metrics()

    def _calculate_metrics(self) -> Dict:
        """Calculate win rate and profit metrics"""
        wins = self.trades[self.trades['win'] == True]
        losses = self.trades[self.trades['win'] == False]

        total_trades = len(self.trades)
        win_count = len(wins)
        loss_count = len(losses)

        if total_trades == 0:
            return {'error': 'No trades provided'}

        win_rate = win_count / total_trades if total_trades > 0 else 0
        loss_rate = 1 - win_rate

        avg_win = wins['pnl'].mean() if len(wins) > 0 else 0
        avg_loss = abs(losses['pnl'].mean()) if len(losses) > 0 else 1

        # Profit factor
        total_wins = wins['pnl'].sum() if len(wins) > 0 else 0
        total_losses = abs(losses['pnl'].sum()) if len(losses) > 0 else 1
        profit_factor = total_wins / total_losses if total_losses > 0 else 0

        return {
            'win_rate': win_rate,
            'loss_rate': loss_rate,
            'avg_win': avg_win,
            'avg_loss': avg_loss,
            'profit_factor': profit_factor,
            'total_trades': total_trades
        }

    def calculate_kelly(self) -> float:
        """
        Calculate full Kelly percentage

        Returns:
            Kelly fraction as decimal (0.25 = 25% of capital)
        """
        if 'error' in self.metrics:
            return 0.0

        p = self.metrics['win_rate']
        q = self.metrics['loss_rate']

        # Avoid division by zero
        if self.metrics['avg_loss'] == 0:
            return 0.0

        b = self.metrics['avg_win'] / self.metrics['avg_loss']

        # Kelly formula: (bp - q) / b
        kelly_full = (b * p - q) / b

        # Ensure kelly is positive (negative kelly = don't trade)
        kelly_full = max(0, kelly_full)

        # Apply fraction (quarter kelly)
        kelly_fractional = kelly_full * self.fraction

        # Cap at 25% max (safety)
        kelly_fractional = min(kelly_fractional, 0.25)

        return kelly_fractional

    def get_position_size(self, account_balance: float) -> float:
        """
        Calculate position size based on Kelly

        Args:
            account_balance: Current account value

        Returns:
            Dollar amount to risk per trade
        """
        kelly_pct = self.calculate_kelly()
        position_size = account_balance * kelly_pct

        return position_size

    def get_report(self) -> Dict:
        """Generate detailed Kelly report"""
        kelly_full = self.calculate_kelly() / self.fraction
        kelly_fractional = self.calculate_kelly()

        return {
            'metrics': self.metrics,
            'kelly_full': kelly_full,
            'kelly_fractional': kelly_fractional,
            'kelly_fraction_used': self.fraction,
            'recommended_risk_pct': kelly_fractional * 100
        }

# Example usage
if __name__ == "__main__":
    # Create sample trade data
    trades_data = {
        'pnl': [520, -340, 680, 460, -280, 740, 380, -220],
        'win': [True, False, True, True, False, True, True, False]
    }
    trades_df = pd.DataFrame(trades_data)

    # Calculate Kelly
    kelly = KellyCriterion(trades_df, fraction=0.25)
    report = kelly.get_report()

    print(f"Win Rate: {report['metrics']['win_rate']:.1%}")
    print(f"Profit Factor: {report['metrics']['profit_factor']:.2f}")
    print(f"Full Kelly: {report['kelly_full']:.2%}")
    print(f"Quarter Kelly: {report['kelly_fractional']:.2%}")
    print(f"Recommended Risk: {report['recommended_risk_pct']:.2f}%")

    # Get position size for $450,000 account
    position_size = kelly.get_position_size(450000)
    print(f"Position Size: ${position_size:,.0f}")

class RollingKelly:
    def __init__(self, trades_df: pd.DataFrame, window: int = 30, fraction: float = 0.25):
        """
        Calculate rolling Kelly criterion

        Args:
            trades_df: DataFrame with datetime index, columns ['pnl', 'win']
            window: Rolling window size (number of trades)
            fraction: Kelly fraction
        """
        self.trades = trades_df.sort_index()
        self.window = window
        self.fraction = fraction

    def calculate_rolling(self) -> pd.DataFrame:
        """Calculate Kelly for each rolling window"""
        results = []

        for i in range(self.window, len(self.trades) + 1):
            window_trades = self.trades.iloc[i - self.window:i]

            kelly_calc = KellyCriterion(window_trades, self.fraction)
            kelly_pct = kelly_calc.calculate_kelly()

            results.append({
                'date': window_trades.index[-1],
                'kelly_pct': kelly_pct,
                'win_rate': kelly_calc.metrics['win_rate'],
                'profit_factor': kelly_calc.metrics['profit_factor']
            })

        return pd.DataFrame(results)

    def plot_rolling_kelly(self, save_path: str = None):
        """Plot rolling Kelly over time"""
        import matplotlib.pyplot as plt

        rolling_df = self.calculate_rolling()

        fig, axes = plt.subplots(3, 1, figsize=(12, 10))

        # Plot Kelly percentage
        axes[0].plot(rolling_df['date'], rolling_df['kelly_pct'] * 100)
        axes[0].axhline(y=5, color='r', linestyle='--', label='5% threshold')
        axes[0].set_ylabel('Kelly %')
        axes[0].set_title(f'Rolling Kelly Criterion ({self.window}-trade window)')
        axes[0].legend()
        axes[0].grid(True, alpha=0.3)

        # Plot win rate
        axes[1].plot(rolling_df['date'], rolling_df['win_rate'] * 100, color='green')
        axes[1].axhline(y=50, color='gray', linestyle='--')
        axes[1].set_ylabel('Win Rate %')
        axes[1].grid(True, alpha=0.3)

        # Plot profit factor
        axes[2].plot(rolling_df['date'], rolling_df['profit_factor'], color='orange')
        axes[2].axhline(y=1.0, color='r', linestyle='--', label='Breakeven')
        axes[2].set_ylabel('Profit Factor')
        axes[2].set_xlabel('Date')
        axes[2].legend()
        axes[2].grid(True, alpha=0.3)

        plt.tight_layout()

        if save_path:
            plt.savefig(save_path, dpi=150)
        else:
            plt.show()

# Example usage
if __name__ == "__main__":
    # Load historical trades
    trades_df = pd.read_csv('my_trades_2024.csv', index_col='date', parse_dates=True)

    # Calculate rolling Kelly
    rolling = RollingKelly(trades_df, window=30, fraction=0.25)
    rolling_results = rolling.calculate_rolling()

    # Plot
    rolling.plot_rolling_kelly(save_path='rolling_kelly_2024.png')

def monthly_kelly_review(account_balance: float, trades_csv: str):
    """
    Monthly Kelly review for position sizing adjustment

    Args:
        account_balance: Current account value
        trades_csv: Path to trades CSV file
    """
    # Load last 90 days of trades
    trades_df = pd.read_csv(trades_csv, parse_dates=['date'])
    trades_df = trades_df[trades_df['date'] > pd.Timestamp.now() - pd.Timedelta(days=90)]

    # Calculate Kelly
    kelly = KellyCriterion(trades_df, fraction=0.25)
    report = kelly.get_report()

    # Get recommended position size
    recommended_size = kelly.get_position_size(account_balance)

    # Current fixed size
    current_size = 1500

    # Analysis
    print("=== Monthly Kelly Review ===")
    print(f"Account Balance: ${account_balance:,.0f}")
    print(f"Current Position Size: ${current_size:,.0f}")
    print(f"")
    print(f"Last 90 Days Performance:")
    print(f"  Trades: {report['metrics']['total_trades']}")
    print(f"  Win Rate: {report['metrics']['win_rate']:.1%}")
    print(f"  Profit Factor: {report['metrics']['profit_factor']:.2f}")
    print(f"")
    print(f"Kelly Analysis:")
    print(f"  Full Kelly: {report['kelly_full']:.2%}")
    print(f"  Quarter Kelly: {report['kelly_fractional']:.2%}")
    print(f"  Recommended Size: ${recommended_size:,.0f}")
    print(f"")

    # Decision
    if recommended_size > current_size * 1.2:
        print("⚠️ RECOMMENDED: Increase position size")
        print(f"   New size: ${recommended_size:,.0f}")
    elif recommended_size < current_size * 0.8:
        print("⚠️ RECOMMENDED: Decrease position size")
        print(f"   New size: ${recommended_size:,.0f}")
    else:
        print("✓ Current position size within optimal range")

    return report

# Run monthly review
if __name__ == "__main__":
    monthly_kelly_review(
        account_balance=449490,
        trades_csv='trades_2025.csv'
    )

b = 580 / 260 = 2.23
p = 0.72
q = 0.28

kelly_full = (2.23 * 0.72 - 0.28) / 2.23 = 0.595 (59.5%!)

quarter_kelly = 0.595 * 0.25 = 0.149 (14.9%)

position_size = 449490 * 0.149 = $66,974