import pandas as pd
import numpy as np
from enum import Enum
class PositionSizingMethod(Enum):
    FIXED_FRACTIONAL = 1
    KELLY_CRITERION = 2
    VOLATILITY_ADJUSTED = 3
    EQUAL_RISK = 4
    DYNAMIC_DRAWDOWN = 5
class PositionSizingEngine:
    """
    Production-grade position sizing system
    """
def __init__(self, account_balance=100000, method=PositionSizingMethod.FIXED_FRACTIONAL):
        self.initial_balance = account_balance
        self.current_balance = account_balance
        self.peak_balance = account_balance
        self.method = method
        self.open_positions = {}
        self.trade_history = []
def calculate_position_size(self, symbol, entry_price, stop_loss, signal_metadata):
        """
        Main method: calculates position size based on configured method
        """
if self.method == PositionSizingMethod.FIXED_FRACTIONAL:
            size = self._fixed_fractional(entry_price, stop_loss)
elif self.method == PositionSizingMethod.KELLY_CRITERION:
            size = self._kelly_criterion(entry_price, stop_loss, signal_metadata)
elif self.method == PositionSizingMethod.VOLATILITY_ADJUSTED:
            size = self._volatility_adjusted(entry_price, stop_loss,
                                           signal_metadata.get('volatility'))
elif self.method == PositionSizingMethod.EQUAL_RISK:
            size = self._equal_risk(entry_price, stop_loss)
elif self.method == PositionSizingMethod.DYNAMIC_DRAWDOWN:
            size = self._dynamic_drawdown(entry_price, stop_loss)
# Apply maximum position constraints
        size = self._apply_constraints(symbol, size, entry_price)
return size
def _fixed_fractional(self, entry_price, stop_loss, risk_pct=0.02):
        """
        Risk fixed percentage of account per trade
        """
risk_amount = self.current_balance * risk_pct
        stop_distance = abs(entry_price - stop_loss)
if stop_distance == 0:
            return 0
size = risk_amount / stop_distance
        return size
def _kelly_criterion(self, entry_price, stop_loss, signal_metadata):
        """
        Kelly Criterion: optimal position sizing
        f = (p × w - (1-p) × l) / w
        """
# Extract statistics from backtests
        win_rate = signal_metadata.get('win_rate', 0.55)
        avg_win = signal_metadata.get('avg_win_pct', 0.02)
        avg_loss = signal_metadata.get('avg_loss_pct', 0.01)
# Kelly fraction
        if avg_win > 0:
            f = (win_rate  avg_win - (1 - win_rate)  avg_loss) / avg_win
        else:
            return 0
# Safety: use 25% of Kelly
        safe_f = max(0, min(f, 0.25))  # Cap at 25% of Kelly
# Convert to position size
        risk_amount = self.current_balance * safe_f
        stop_distance = abs(entry_price - stop_loss)
size = risk_amount / stop_distance if stop_distance > 0 else 0
        return size
def _volatility_adjusted(self, entry_price, stop_loss, volatility=None):
        """
        Scale position size inversely to current volatility
        """
# Base size
        base_size = self._fixed_fractional(entry_price, stop_loss, risk_pct=0.02)
if volatility is None:
            return base_size
# Volatility scaling
        vol_percentile = volatility['percentile']  # 0-100
if vol_percentile > 75:  # High volatility
            multiplier = 0.50
        elif vol_percentile > 50:  # Above average
            multiplier = 0.75
        else:  # Low volatility
            multiplier = 1.0
return base_size * multiplier
def _equal_risk(self, entry_price, stop_loss):
        """
        Size each position to contribute equally to portfolio risk
        """
# Total account risk budget: 10% per new trade
        num_open = len(self.open_positions) + 1
        risk_per_position = self.current_balance * 0.10 / num_open
stop_distance = abs(entry_price - stop_loss)
        size = risk_per_position / stop_distance if stop_distance > 0 else 0
return size
def _dynamic_drawdown(self, entry_price, stop_loss):
        """
        Reduce sizing if account is in drawdown
        """
drawdown = (self.peak_balance - self.current_balance) / self.peak_balance
# Scaling based on drawdown
        if drawdown < 0.05:
            multiplier = 1.0
        elif drawdown < 0.10:
            multiplier = 0.75
        elif drawdown < 0.15:
            multiplier = 0.50
        elif drawdown < 0.20:
            multiplier = 0.25
        else:
            multiplier = 0.0  # Stop trading
base_size = self._fixed_fractional(entry_price, stop_loss, risk_pct=0.02)
        return base_size * multiplier
def _apply_constraints(self, symbol, size, entry_price):
        """
        Apply maximum position constraints
        """
# Constraint 1: Maximum single position = 5% of account
        max_position_value = self.current_balance * 0.05
        if size * entry_price > max_position_value:
            size = max_position_value / entry_price
# Constraint 2: Maximum leverage = 2.0x
        portfolio_value = sum([p['size'] * p['entry_price'] for p in self.open_positions.values()])
        max_leverage_value = self.current_balance * 2.0
if portfolio_value + (size * entry_price) > max_leverage_value:
            available = max_leverage_value - portfolio_value
            size = available / entry_price if entry_price > 0 else 0
# Constraint 3: Minimum size (avoid micro-positions)
        if size * entry_price < 100:  # Less than $100
            size = 0
return max(0, size)
def add_position(self, symbol, entry_price, size, stop_loss, atr):
        """
        Track new opened position
        """
self.open_positions[symbol] = {
            'entry_price': entry_price,
            'size': size,
            'notional': size * entry_price,
            'stop_loss': stop_loss,
            'risk_amount': abs(entry_price - stop_loss) * size,
            'entry_time': pd.Timestamp.now(),
            'atr': atr
        }
print(f"Opened {symbol}: {size:.2f} shares @ ${entry_price:.2f}")
def close_position(self, symbol, exit_price, reason='TARGET_HIT'):
        """
        Close position and record trade
        """
if symbol not in self.open_positions:
            return
position = self.open_positions[symbol]
        pnl = (exit_price - position['entry_price']) * position['size']
        pnl_pct = (exit_price - position['entry_price']) / position['entry_price']
        duration = (pd.Timestamp.now() - position['entry_time']).days
self.trade_history.append({
            'symbol': symbol,
            'entry_price': position['entry_price'],
            'exit_price': exit_price,
            'size': position['size'],
            'pnl': pnl,
            'pnl_pct': pnl_pct,
            'duration_days': duration,
            'reason': reason
        })
# Update balance
        self.current_balance += pnl
        if self.current_balance > self.peak_balance:
            self.peak_balance = self.current_balance
del self.open_positions[symbol]
print(f"Closed {symbol}: P&L = ${pnl:+.2f} ({pnl_pct:+.2%})")
def get_portfolio_metrics(self):
        """
        Calculate current portfolio statistics
        """
trades_df = pd.DataFrame(self.trade_history)
if len(trades_df) == 0:
            return {
                'total_trades': 0,
                'win_rate': 0,
                'avg_win': 0,
                'avg_loss': 0,
                'profit_factor': 0,
                'sharpe_ratio': 0
            }
wins = len(trades_df[trades_df['pnl'] > 0])
        losses = len(trades_df[trades_df['pnl'] <= 0])
total_win = trades_df[trades_df['pnl'] > 0]['pnl'].sum()
        total_loss = abs(trades_df[trades_df['pnl'] <= 0]['pnl'].sum())
return {
            'total_trades': len(trades_df),
            'win_rate': wins / len(trades_df) if len(trades_df) > 0 else 0,
            'avg_win': trades_df[trades_df['pnl'] > 0]['pnl_pct'].mean(),
            'avg_loss': trades_df[trades_df['pnl'] <= 0]['pnl_pct'].mean(),
            'profit_factor': total_win / total_loss if total_loss > 0 else 0,
            'current_balance': self.current_balance,
            'total_return': (self.current_balance - self.initial_balance) / self.initial_balance,
            'max_drawdown': (self.peak_balance - min([self.initial_balance] +
                                                     [self.current_balance]) /
                           self.peak_balance)
        }

class AlgorithmicTradingBot:
    """
    Complete trading bot with integrated position sizing
    """
def __init__(self, broker_api, initial_capital=100000):
        self.broker = broker_api
        self.position_sizer = PositionSizingEngine(
            account_balance=initial_capital,
            method=PositionSizingMethod.VOLATILITY_ADJUSTED
        )
        self.data_manager = DataManager()
        self.signal_generator = SignalGenerator()
def process_trading_signals(self, symbols_to_scan):
        """
        Main trading loop: generate signals and execute with proper sizing
        """
for symbol in symbols_to_scan:
            try:
                # Get latest data
                data = self.data_manager.fetch_latest(symbol)
                atr = self.data_manager.calculate_atr(data)
                volatility = self.data_manager.calculate_volatility(data)
# Generate signal
                signal, confidence = self.signal_generator.analyze(symbol, data)
if signal is None:
                    continue
# Calculate position size
                entry_price = data['close'].iloc[-1]
                stop_loss = entry_price - (2 * atr)
signal_metadata = {
                    'volatility': volatility,
                    'win_rate': 0.60,
                    'avg_win_pct': 0.02,
                    'avg_loss_pct': 0.01,
                    'confidence': confidence
                }
size = self.position_sizer.calculate_position_size(
                    symbol, entry_price, stop_loss, signal_metadata
                )
# Execute trade if size meets minimum threshold
                if size > 0:
                    # Place orders
                    entry_order = self.broker.buy(symbol, size, entry_price)
                    stop_order = self.broker.stop_loss(symbol, size, stop_loss)
                    target_order = self.broker.profit_target(
                        symbol, size, entry_price + (3 * atr)
                    )
# Track position
                    self.position_sizer.add_position(
                        symbol, entry_price, size, stop_loss, atr
                    )
print(f"Trade executed: {symbol}, Size: {size:.2f}, "
                          f"Entry: ${entry_price:.2f}, Stop: ${stop_loss:.2f}")
except Exception as e:
                print(f"Error processing {symbol}: {e}")
def monitor_open_positions(self):
        """
        Monitor open positions and close on targets/stops
        """
for symbol in list(self.position_sizer.open_positions.keys()):
            try:
                current_price = self.broker.get_current_price(symbol)
                position = self.position_sizer.open_positions[symbol]
# Check stop loss
                if current_price <= position['stop_loss']:
                    self.broker.sell(symbol, position['size'], current_price)
                    self.position_sizer.close_position(symbol, current_price, 'STOP_LOSS')
# Check profit target
                elif current_price >= (position['entry_price'] + (3 * position['atr'])):
                    self.broker.sell(symbol, position['size'], current_price)
                    self.position_sizer.close_position(symbol, current_price, 'PROFIT_TARGET')
except Exception as e:
                print(f"Error monitoring {symbol}: {e}")
def run_backtest(self, symbols, start_date, end_date):
        """
        Backtest system with position sizing
        """
# Fetch historical data
        historical_data = {}
        for symbol in symbols:
            historical_data[symbol] = self.data_manager.fetch_historical(
                symbol, start_date, end_date
            )
# Simulate trading
        for date in pd.date_range(start_date, end_date):
            self.process_trading_signals(symbols)
            self.monitor_open_positions()
# Report metrics
        metrics = self.position_sizer.get_portfolio_metrics()
return metrics
Usage
bot = AlgorithmicTradingBot(broker_api, initial_capital=100000)
Paper trade first
metrics = bot.run_backtest(
    symbols=['AAPL', 'MSFT', 'GOOGL', 'AMZN'],
    start_date='2025-01-01',
    end_date='2026-03-15'
)
print(f"\nBacktest Results:")
print(f"Total Return: {metrics['total_return']:.2%}")
print(f"Win Rate: {metrics['win_rate']:.2%}")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")

def optimize_position_sizing_parameters(symbol, historical_data, parameter_ranges):
    """
    Grid search to find optimal position sizing parameters
    """
best_sharpe = -np.inf
    best_params = {}
for risk_pct in parameter_ranges['risk_percentages']:
        for vol_multiplier in parameter_ranges['vol_multipliers']:
            for leverage_limit in parameter_ranges['leverage_limits']:
# Simulate with parameters
                sizer = PositionSizingEngine(method=PositionSizingMethod.VOLATILITY_ADJUSTED)
for date, row in historical_data.iterrows():
                    # Generate size with current parameters
                    entry = row['close']
                    stop = entry - (2 * row['atr'])
size = sizer._volatility_adjusted(entry, stop, row['volatility'])
                    # Apply constraints...
# Calculate Sharpe ratio
                metrics = sizer.get_portfolio_metrics()
                sharpe = metrics['sharpe_ratio']
if sharpe > best_sharpe:
                    best_sharpe = sharpe
                    best_params = {
                        'risk_pct': risk_pct,
                        'vol_multiplier': vol_multiplier,
                        'leverage_limit': leverage_limit
                    }
return best_params, best_sharpe
Optimize parameters
params, sharpe = optimize_position_sizing_parameters(
    'AAPL',
    historical_data,
    parameter_ranges={
        'risk_percentages': [0.01, 0.02, 0.03],
        'vol_multipliers': [0.5, 0.75, 1.0, 1.25],
        'leverage_limits': [1.5, 2.0, 2.5, 3.0]
    }
)
print(f"Optimal parameters: {params}")
print(f"Achievable Sharpe: {sharpe:.2f}")