import numpy as np
import pandas as pd
import backtrader as bt
from scipy.optimize import minimize
from scipy.stats import norm

import pandas as pd
import numpy as np
from abc import ABC, abstractmethod
class PositionSizer(ABC):
    """Abstract base class for position sizing strategies"""
@abstractmethod
    def calculate_size(self, **kwargs) -> float:
        """Return position size in shares/contracts"""
        pass
class FixedFractionalSizer(PositionSizer):
    """Risk fixed fraction of account per trade"""
def __init__(self, risk_fraction=0.02):
        self.risk_fraction = risk_fraction
def calculate_size(self, account_size, entry_price, stop_loss_price):
        """
        Calculate shares to risk exactly risk_fraction of account
        Formula: Shares = (Account × Risk%) / (Entry - Stop)
        """
        risk_amount = account_size * self.risk_fraction
        stop_distance = abs(entry_price - stop_loss_price)
if stop_distance == 0:
            raise ValueError("Stop distance cannot be zero")
return risk_amount / stop_distance
class VolatilityAdjustedSizer(PositionSizer):
    """Scale position inversely to volatility (ATR-based)"""
def __init__(self, target_risk_dollars=1000, lookback=20):
        self.target_risk_dollars = target_risk_dollars
        self.lookback = lookback
def calculate_size(self, current_price, volatility_metric):
        """
        Size inversely to volatility
        Lower volatility → larger position
        """
        # Normalize volatility
        normalized_vol = volatility_metric / np.mean(volatility_metric[-self.lookback:])
        volatility_adjustment = 1.0 / normalized_vol
# Base position size
        base_size = self.target_risk_dollars / (current_price * 0.02)
return base_size * volatility_adjustment
class KellySizer(PositionSizer):
    """Calculate optimal sizing via Kelly Criterion"""
def __init__(self, kelly_fraction=0.25, max_position_pct=0.05):
        self.kelly_fraction = kelly_fraction
        self.max_position_pct = max_position_pct
def calculate_size(self, account_size, trade_history, entry_price, stop_price):
        """
        Calculate Kelly-optimal position size from historical trades
        """
        if len(trade_history) < 10:
            # Insufficient data, use conservative 1%
            return (account_size * 0.01) / abs(entry_price - stop_price)
returns = np.array([t['return'] for t in trade_history[-50:]])
        win_rate = (returns > 0).sum() / len(returns)
if win_rate < 0.4:
            return (account_size * 0.01) / abs(entry_price - stop_price)
avg_win = np.mean(returns[returns > 0]) if (returns > 0).any() else 1
        avg_loss = abs(np.mean(returns[returns < 0])) if (returns < 0).any() else 1
b = avg_win / avg_loss
        kelly_fraction_opt = (win_rate * b - (1 - win_rate)) / b
# Apply fractional Kelly and cap
        kelly_safe = max(0, min(kelly_fraction_opt * self.kelly_fraction, self.max_position_pct))
return (account_size * kelly_safe) / abs(entry_price - stop_price)

class PositionSizingBacktest:
    """Full backtesting engine with modular position sizing"""
def __init__(self, prices, signals, sizer, initial_capital=100000, commission=0.001):
        self.prices = prices.values if isinstance(prices, pd.Series) else prices
        self.signals = signals
        self.sizer = sizer
        self.capital = initial_capital
        self.initial_capital = initial_capital
        self.commission = commission
self.trades = []
        self.equity_curve = [initial_capital]
        self.positions = []
def run(self):
        """Execute backtest"""
        position = None
for i in range(1, len(self.signals)):
            signal = self.signals[i]
            price = self.prices[i]
# Close existing position if signal flips
            if position and signal != position['signal']:
                self._close_position(position, price, i)
                position = None
# Enter new position
            if signal != 0 and not position:
                position = self._open_position(signal, price, i)
# Close final position
        if position:
            self._close_position(position, self.prices[-1], len(self.prices) - 1)
return self.equity_curve, self.trades
def _open_position(self, signal, entry_price, bar):
        """Open a new trade"""
        stop_loss = entry_price  0.97 if signal == 1 else entry_price  1.03
# Calculate position size
        try:
            shares = self.sizer.calculate_size(
                account_size=self.capital,
                entry_price=entry_price,
                stop_loss_price=stop_loss,
                trade_history=self.trades,
                current_price=entry_price,
                volatility_metric=np.std(self.prices[max(0, bar-20):bar])
            )
        except TypeError:
            # Handle different sizer signatures
            shares = self.sizer.calculate_size(
                account_size=self.capital,
                entry_price=entry_price,
                stop_loss_price=stop_loss
            )
shares = max(1, int(shares))  # Ensure minimum 1 share
return {
            'entry': entry_price,
            'stop': stop_loss,
            'shares': shares,
            'signal': signal,
            'entry_bar': bar,
            'value': entry_price * shares
        }
def _close_position(self, position, exit_price, bar):
        """Close a position and record trade"""
        # Apply commission
        entry_cost = position['value'] * (1 + self.commission)
        exit_value = exit_price  position['shares']  (1 - self.commission)
pnl = exit_value - entry_cost if position['signal'] == 1 else \
              entry_cost - exit_value
self.capital += pnl
self.trades.append({
            'entry': position['entry'],
            'exit': exit_price,
            'shares': position['shares'],
            'pnl': pnl,
            'return': pnl / self.capital,
            'bars_held': bar - position['entry_bar']
        })
self.equity_curve.append(self.capital)
def metrics(self):
        """Calculate performance metrics"""
        if not self.trades:
            return {}
returns = np.array([t['return'] for t in self.trades])
total_return = (self.capital - self.initial_capital) / self.initial_capital
        annual_return = total_return * 252 / len(self.trades)
        sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252) if np.std(returns) > 0 else 0
cumulative = np.cumprod(1 + returns)
        running_max = np.maximum.accumulate(cumulative)
        drawdown = (cumulative - running_max) / running_max
        max_dd = np.min(drawdown)
return {
            'total_return': total_return,
            'annual_return': annual_return,
            'sharpe_ratio': sharpe,
            'max_drawdown': max_dd,
            'win_rate': (returns > 0).sum() / len(returns),
            'num_trades': len(self.trades),
            'avg_trade_pnl': np.mean(returns) * self.capital
        }

import backtrader as bt
class BacktraderSizer(bt.Sizer):
    """Custom Backtrader sizer using our position sizing logic"""
def __init__(self, risk_fraction=0.02):
        self.risk_fraction = risk_fraction
def _getsizing(self, comminfo, cash, price, dtdetail):
        """Calculate position size for Backtrader"""
        # Risk fixed percentage of current cash
        stop_distance = price * 0.03  # 3% stop loss
size = (cash * self.risk_fraction) / stop_distance
# Cap to avoid over-leveraging
        max_notional = cash * 0.5
        notional = size * price
if notional > max_notional:
            size = max_notional / price
return int(size)
class MyStrategy(bt.Strategy):
    """Example strategy using custom position sizer"""
def __init__(self):
        self.sma20 = bt.indicators.SimpleMovingAverage(self.data.close, period=20)
        self.sma50 = bt.indicators.SimpleMovingAverage(self.data.close, period=50)
def next(self):
        if self.sma20[0] > self.sma50[0] and not self.position:
            self.buy(size=self.getsizing())
elif self.sma20[0] < self.sma50[0] and self.position:
            self.close()
Run backtest
cerebro = bt.Cerebro()
cerebro.addstrategy(MyStrategy)
cerebro.broker.add_cash(100000)
cerebro.addsizer(BacktraderSizer(risk_fraction=0.02))
Add data and run
data = bt.feeds.YahooFinanceData(dataname='AAPL', fromdate=..., todate=...)
cerebro.adddata(data)
result = cerebro.run()