P(ruin) = e^(-2bp/s²)

P(ruin) = e^(-2 × 1 × 0.55 / 1²) = e^(-1.1) = 0.333 = 33.3% chance of ruin

Minimum win probability needed to avoid ruin at f% risk:
p > 0.5 + (f × (1+b)) / (2b)
Where b = ratio of average win to average loss
f = fraction of capital risked
Example: Risking 5% with 1.5:1 profit factor
p > 0.5 + (0.05 × 2.5) / (2 × 1.5) = 0.542

def validate_strategy_for_position_sizing(
    win_rate,
    profit_factor,
    sample_size=50
):
    """
    Ensure strategy has adequate statistical edge before position sizing
    """
    required_sample = 30  # Minimum trades for meaningful statistics
if sample_size < required_sample:
        raise ValueError(f"Insufficient sample size: {sample_size} < {required_sample}")
if win_rate < 0.40:
        raise ValueError(f"Win rate {win_rate:.1%} too low; expect continuous losses")
# Calculate required win rate for safety
    required_wr = 0.5 + (0.02  (1 + profit_factor)) / (2  profit_factor)
if win_rate < required_wr:
        print(f"WARNING: {win_rate:.1%} win rate < {required_wr:.1%} required for 2% risk")
        print("Reduce position sizing to 1% or improve strategy")
return True

class SafePositionSizer:
    """Position sizer with safety circuit breakers"""
def __init__(
        self,
        initial_capital=100000,
        base_risk=0.02,
        max_risk=0.03,
        drawdown_limit=0.20
    ):
        self.initial_capital = initial_capital
        self.current_capital = initial_capital
        self.base_risk = base_risk
        self.max_risk = max_risk
        self.drawdown_limit = drawdown_limit
        self.peak_capital = initial_capital
        self.trades = []
def calculate_current_drawdown(self):
        """Calculate current drawdown percentage"""
        return (self.current_capital - self.peak_capital) / self.peak_capital
def adjust_risk_for_drawdown(self):
        """Reduce risk sizing during drawdowns"""
        drawdown = self.calculate_current_drawdown()
if drawdown < -self.drawdown_limit:
            # In severe drawdown: stop trading
            return 0.0
elif drawdown < -0.15:
            # Deep drawdown: 50% position reduction
            return self.base_risk * 0.5
elif drawdown < -0.10:
            # Moderate drawdown: 75% of normal
            return self.base_risk * 0.75
else:
            # Normal conditions
            return self.base_risk
def adjust_risk_for_volatility(self, recent_volatility, historical_volatility):
        """Reduce risk if volatility spikes"""
        vol_ratio = recent_volatility / historical_volatility
if vol_ratio > 2.0:
            # Extreme volatility: use 50% normal risk
            return self.base_risk * 0.5
elif vol_ratio > 1.5:
            # High volatility: use 75% normal risk
            return self.base_risk * 0.75
else:
            return self.base_risk
def adjust_risk_for_consecutive_losses(self, recent_trades, num_recent=10):
        """Reduce risk after consecutive losses (gambler's ruin protection)"""
        if len(recent_trades) < num_recent:
            return self.base_risk
recent_returns = recent_trades[-num_recent:]
        consecutive_losses = 0
        current_streak = 0
for ret in recent_returns:
            if ret < 0:
                current_streak += 1
                consecutive_losses = max(consecutive_losses, current_streak)
            else:
                current_streak = 0
# After 3 consecutive losses: reduce risk by 50%
        if consecutive_losses >= 3:
            return self.base_risk * 0.5
# After 5+ consecutive losses: reduce risk by 75%
        if consecutive_losses >= 5:
            return 0.005
return self.base_risk
def calculate_safe_position_size(
        self,
        entry_price,
        stop_loss_price,
        recent_volatility=0.02,
        historical_volatility=0.015,
        recent_trades=[]
    ):
        """Comprehensive safe position sizing"""
# Apply all safety adjustments
        drawdown_risk = self.adjust_risk_for_drawdown()
if drawdown_risk == 0:
            print("ERROR: In severe drawdown. Stop trading immediately.")
            return 0
vol_risk = self.adjust_risk_for_volatility(recent_volatility, historical_volatility)
        loss_risk = self.adjust_risk_for_consecutive_losses(recent_trades)
# Use minimum of all risk adjustments (most conservative)
        effective_risk = min(drawdown_risk, vol_risk, loss_risk)
# Ensure within bounds
        effective_risk = min(effective_risk, self.max_risk)
# Calculate position size
        risk_amount = self.current_capital * effective_risk
        stop_distance = abs(entry_price - stop_loss_price)
position_size = risk_amount / stop_distance
return {
            'position_size': position_size,
            'effective_risk': effective_risk,
            'drawdown_adjustment': drawdown_risk,
            'vol_adjustment': vol_risk,
            'loss_adjustment': loss_risk
        }
def record_trade(self, pnl):
        """Update capital tracking"""
        self.current_capital += pnl
        self.peak_capital = max(self.peak_capital, self.current_capital)
        self.trades.append(pnl)

class PortfolioRiskController:
    """Manage risk across multiple positions"""
def __init__(self, account_size=100000, max_portfolio_risk=0.05):
        self.account_size = account_size
        self.max_portfolio_risk = max_portfolio_risk  # 5% max daily risk
        self.positions = {}
def get_total_market_exposure(self):
        """Calculate total notional exposure"""
        total = sum(
            pos['shares'] * pos['current_price']
            for pos in self.positions.values()
        )
        return total
def get_total_position_risk(self):
        """Calculate total dollar risk from all positions"""
        total_risk = sum(
            abs(pos['shares'] * (pos['entry'] - pos['stop']))
            for pos in self.positions.values()
        )
        return total_risk
def can_open_new_position(self, position_risk):
        """Check if adding new position exceeds limits"""
        current_risk = self.get_total_position_risk()
        proposed_risk = current_risk + position_risk
max_risk = self.account_size * self.max_portfolio_risk
if proposed_risk > max_risk:
            print(f"Position would exceed limit: {proposed_risk:,.0f} > {max_risk:,.0f}")
            return False
# Also cap total exposure to 50% of account
        total_exposure = self.get_total_market_exposure()
        if total_exposure + position_risk  50 > self.account_size  0.5:
            print("Total exposure would exceed 50% account")
            return False
return True
def scale_position_for_portfolio_constraint(self, requested_size, position_risk):
        """Scale down position if portfolio constraint violated"""
        if self.can_open_new_position(position_risk):
            return requested_size
# Reduce size proportionally
        max_risk = self.account_size * self.max_portfolio_risk
        current_risk = self.get_total_position_risk()
        available_risk = max_risk - current_risk
scale_factor = available_risk / position_risk if position_risk > 0 else 0
        return requested_size * scale_factor