Risk Management Guide¶

This guide covers risk management principles, position sizing strategies, and risk controls for the DSTA trading system.

Overview¶

Risk management is the most important factor in trading success. Even the best strategy will fail without proper risk controls.

Core Principles¶

Capital Preservation: Protecting capital is more important than making profits
Consistent Sizing: Position sizes should be systematic, not emotional
Limit Losses: Cut losses quickly, let winners run
Diversification: Don't put all eggs in one basket
Know Your Risk: Always know maximum potential loss

Risk Hierarchy¶

1. Account Risk    → Total portfolio value at risk
2. Position Risk   → Risk per individual position  
3. Trade Risk      → Risk per individual trade
4. Execution Risk  → Slippage, gaps, liquidity

Position Sizing Strategies¶

Position sizing determines how much capital to allocate to each trade. DSTA supports multiple methods.

1. Fixed Dollar Amount¶

Method: Allocate fixed dollar amount per trade.

When to Use: - Simple, easy to understand - Good for beginners - Stable account size

Formula: Position Size = Fixed Amount

Example:

from backtesting.portfolio import BacktestPortfolio, PositionSizingMethod
from decimal import Decimal

portfolio = BacktestPortfolio(
    bars=bars,
    events_queue=events_queue,
    start_date=start_date,
    initial_capital=Decimal('100000'),
    position_sizing=PositionSizingMethod.FIXED,
    position_size_value=Decimal('5000')  # $5,000 per trade
)

Pros: - Simple to implement - Predictable risk per trade - Easy to track

Cons: - Doesn't adjust for account growth - May be too conservative when account grows - May be too aggressive when account shrinks

2. Fixed Percentage of Capital¶

Method: Risk a fixed percentage of total capital per trade.

When to Use: - Most common method - Adjusts automatically with account size - Good balance of growth and safety

Formula: Position Size = Total Capital × Percentage

Example:

portfolio = BacktestPortfolio(
    bars=bars,
    events_queue=events_queue,
    start_date=start_date,
    initial_capital=Decimal('100000'),
    position_sizing=PositionSizingMethod.PERCENT_CAPITAL,
    position_size_value=Decimal('0.02')  # Risk 2% of capital per trade
)

# With $100,000 capital and 2% risk:
# Position Size = $100,000 × 0.02 = $2,000 at risk
# If stop loss is 5%: $2,000 / 0.05 = $40,000 position

Pros: - Scales with account size - Protects against large drawdowns - Industry standard

Cons: - Requires stop-loss calculation - More complex than fixed dollar

Recommended Percentages: - Conservative: 1% - Moderate: 2% - Aggressive: 3-5% - Never exceed 5% per single trade

3. Volatility-Based Sizing (ATR)¶

Method: Size positions based on volatility using Average True Range (ATR).

When to Use: - Assets with varying volatility - Want consistent risk despite price swings - More sophisticated traders

Formula: Position Size = (Capital × Risk%) / (ATR × ATR Multiplier)

Example:

from backtesting.strategies.base import BaseStrategy

class VolatilityBasedStrategy(BaseStrategy):
    def calculate_position_size(self, symbol):
        """Calculate position size based on ATR."""
        # Get ATR (14-period)
        atr = self.calculate_atr(symbol, period=14)
        if atr is None:
            return Decimal('0')

        # Risk management parameters
        account_value = self.portfolio.get_total_value()
        risk_per_trade = account_value * Decimal('0.02')  # 2% risk
        atr_multiplier = Decimal('2.0')  # Stop at 2× ATR

        # Calculate position size
        risk_per_unit = atr * atr_multiplier
        position_size = risk_per_trade / risk_per_unit

        return position_size

# Example calculation:
# Account: $100,000
# Risk: 2% = $2,000
# Current ATR: $500
# ATR Multiplier: 2.0
# Stop Loss: $500 × 2.0 = $1,000
# Position Size: $2,000 / $1,000 = 2 units

Pros: - Adapts to market volatility - Consistent risk across different market conditions - Professional approach

Cons: - More complex to implement - Requires volatility calculation - May miss opportunities in low-volatility periods

4. Kelly Criterion¶

Method: Mathematically optimal position sizing based on win rate and risk-reward.

When to Use: - Have reliable statistics (win rate, avg win/loss) - Want to maximize long-term growth - Sophisticated traders

Formula: Kelly % = W - [(1 - W) / R]

Where: - W = Win rate (as decimal) - R = Risk/reward ratio (avg win / avg loss)

Example:

from decimal import Decimal

def calculate_kelly_percentage(win_rate, avg_win, avg_loss):
    """
    Calculate Kelly percentage.

    Args:
        win_rate: Win rate as decimal (e.g., 0.6 for 60%)
        avg_win: Average winning trade amount
        avg_loss: Average losing trade amount (positive number)

    Returns:
        Kelly percentage (0.0 to 1.0)
    """
    if avg_loss == 0:
        return 0

    risk_reward_ratio = avg_win / avg_loss
    kelly = win_rate - ((1 - win_rate) / risk_reward_ratio)

    # Apply fractional Kelly (more conservative)
    fractional_kelly = kelly * 0.5  # Use 50% of Kelly (safer)

    # Cap at maximum
    return max(0, min(fractional_kelly, 0.25))  # Never exceed 25%

# Example:
# Win rate: 60% (0.6)
# Avg win: $1,000
# Avg loss: $500
# Risk/Reward: 1000/500 = 2.0
# Kelly: 0.6 - (0.4 / 2.0) = 0.6 - 0.2 = 0.4 (40%!)
# Fractional Kelly (50%): 0.4 × 0.5 = 0.2 (20%)
# 
# Risk 20% of capital per trade (very aggressive!)

Important Notes: - Full Kelly is too aggressive - use fractional Kelly (25-50%) - Requires accurate statistics (100+ trades minimum) - Small estimation errors can lead to large losses - Most professionals use ¼ Kelly or less

Pros: - Mathematically optimal for long-term growth - Maximizes compound growth rate

Cons: - Very aggressive (large drawdowns) - Requires accurate statistics - Sensitive to estimation errors - Not suitable for most traders

5. Custom Position Sizing¶

Method: Implement your own position sizing logic.

Example:

class CustomPositionSizing:
    """Custom position sizing based on multiple factors."""

    def __init__(self, base_risk_pct=0.02):
        self.base_risk_pct = Decimal(str(base_risk_pct))

    def calculate_position_size(
        self,
        capital,
        signal_strength,
        volatility,
        market_regime,
        correlation_score
    ):
        """
        Calculate position size based on multiple factors.

        Args:
            capital: Total account value
            signal_strength: Signal confidence (0.0 to 1.0)
            volatility: Current market volatility (0.0 to 1.0)
            market_regime: 'bull', 'bear', or 'sideways'
            correlation_score: Portfolio correlation (0.0 to 1.0)

        Returns:
            Position size in dollars
        """
        # Start with base risk
        risk_pct = self.base_risk_pct

        # Adjust for signal strength
        risk_pct *= Decimal(str(signal_strength))

        # Reduce in high volatility
        if volatility > 0.7:
            risk_pct *= Decimal('0.5')

        # Adjust for market regime
        if market_regime == 'bear':
            risk_pct *= Decimal('0.75')  # More conservative
        elif market_regime == 'bull':
            risk_pct *= Decimal('1.25')  # More aggressive

        # Reduce if portfolio is highly correlated
        if correlation_score > 0.7:
            risk_pct *= Decimal('0.8')

        # Calculate position size
        position_size = capital * risk_pct

        # Apply limits
        max_position = capital * Decimal('0.1')  # Never exceed 10%
        return min(position_size, max_position)

Risk Limits and Controls¶

1. Maximum Position Size¶

Limit the maximum size of any single position.

class Portfolio:
    MAX_POSITION_PCT = Decimal('0.20')  # 20% max per position

    def check_position_size(self, position_value):
        """Ensure position doesn't exceed maximum."""
        total_value = self.get_total_value()
        max_position = total_value * self.MAX_POSITION_PCT

        if position_value > max_position:
            logger.warning(
                f"Position size ${position_value} exceeds maximum ${max_position}"
            )
            return max_position

        return position_value

2. Maximum Daily Loss¶

Stop trading if daily loss exceeds threshold.

class Portfolio:
    MAX_DAILY_LOSS_PCT = Decimal('0.05')  # 5% max daily loss

    def __init__(self, ...):
        super().__init__(...)
        self.daily_start_value = self.initial_capital
        self.trading_halted = False

    def check_daily_loss(self):
        """Check if daily loss limit exceeded."""
        current_value = self.get_total_value()
        daily_loss = (self.daily_start_value - current_value) / self.daily_start_value

        if daily_loss >= self.MAX_DAILY_LOSS_PCT:
            self.trading_halted = True
            logger.critical(
                f"Daily loss limit exceeded: {daily_loss:.2%}. Trading halted."
            )
            return True

        return False

    def reset_daily_loss(self):
        """Reset daily loss tracking (call at start of each day)."""
        self.daily_start_value = self.get_total_value()
        self.trading_halted = False

3. Maximum Drawdown Limit¶

Monitor drawdown and reduce position sizing if exceeded.

class Portfolio:
    MAX_DRAWDOWN_PCT = Decimal('0.20')  # 20% max drawdown
    DRAWDOWN_REDUCE_FACTOR = Decimal('0.5')  # Cut positions in half

    def __init__(self, ...):
        super().__init__(...)
        self.peak_value = self.initial_capital
        self.in_drawdown_protection = False

    def check_drawdown(self):
        """Check and adjust for drawdown."""
        current_value = self.get_total_value()

        # Update peak
        if current_value > self.peak_value:
            self.peak_value = current_value
            self.in_drawdown_protection = False

        # Calculate drawdown
        drawdown = (self.peak_value - current_value) / self.peak_value

        # Activate protection if needed
        if drawdown >= self.MAX_DRAWDOWN_PCT and not self.in_drawdown_protection:
            self.in_drawdown_protection = True
            logger.warning(
                f"Drawdown protection activated: {drawdown:.2%}. "
                f"Reducing position sizes by {self.DRAWDOWN_REDUCE_FACTOR:.0%}"
            )

        return drawdown

    def get_adjusted_position_size(self, base_size):
        """Adjust position size based on drawdown protection."""
        if self.in_drawdown_protection:
            return base_size * self.DRAWDOWN_REDUCE_FACTOR
        return base_size

4. Maximum Number of Positions¶

Limit portfolio concentration.

class Portfolio:
    MAX_CONCURRENT_POSITIONS = 10

    def can_open_position(self):
        """Check if we can open another position."""
        num_positions = len([qty for qty in self.current_holdings.values() if qty != 0])

        if num_positions >= self.MAX_CONCURRENT_POSITIONS:
            logger.info(
                f"Cannot open position: already at maximum ({self.MAX_CONCURRENT_POSITIONS})"
            )
            return False

        return True

5. Correlation Limit¶

Avoid highly correlated positions.

import numpy as np

class Portfolio:
    MAX_CORRELATION = 0.7

    def check_correlation(self, new_symbol, lookback_days=30):
        """Check correlation between new symbol and existing positions."""
        # Get current positions
        current_symbols = [s for s, qty in self.current_holdings.items() if qty != 0]

        if not current_symbols:
            return True  # No positions, no correlation risk

        # Get price history for all symbols
        new_prices = self.get_price_history(new_symbol, lookback_days)

        for existing_symbol in current_symbols:
            existing_prices = self.get_price_history(existing_symbol, lookback_days)

            # Calculate correlation
            correlation = np.corrcoef(new_prices, existing_prices)[0, 1]

            if abs(correlation) > self.MAX_CORRELATION:
                logger.warning(
                    f"High correlation detected: {new_symbol} vs {existing_symbol} "
                    f"({correlation:.2f}). Skipping trade."
                )
                return False

        return True

Stop-Loss Strategies¶

Stop-losses limit losses on individual trades.

1. Fixed Percentage Stop-Loss¶

Method: Exit when price moves against you by a fixed percentage.

class FixedPercentageStopLoss:
    """Fixed percentage stop-loss."""

    def __init__(self, stop_loss_pct=0.05):
        self.stop_loss_pct = Decimal(str(stop_loss_pct))

    def calculate_stop_loss(self, entry_price, direction):
        """
        Calculate stop-loss price.

        Args:
            entry_price: Entry price
            direction: 'LONG' or 'SHORT'

        Returns:
            Stop-loss price
        """
        if direction == 'LONG':
            # For long: stop below entry
            stop_loss = entry_price * (1 - self.stop_loss_pct)
        else:
            # For short: stop above entry
            stop_loss = entry_price * (1 + self.stop_loss_pct)

        return stop_loss

# Example:
# Entry: $100
# Stop Loss: 5%
# Long Stop: $100 × (1 - 0.05) = $95
# Short Stop: $100 × (1 + 0.05) = $105

2. ATR-Based Stop-Loss¶

Method: Set stop-loss based on volatility (ATR).

class ATRStopLoss:
    """ATR-based stop-loss."""

    def __init__(self, atr_multiplier=2.0):
        self.atr_multiplier = Decimal(str(atr_multiplier))

    def calculate_stop_loss(self, entry_price, atr, direction):
        """
        Calculate stop-loss based on ATR.

        Args:
            entry_price: Entry price
            atr: Current Average True Range
            direction: 'LONG' or 'SHORT'

        Returns:
            Stop-loss price
        """
        stop_distance = atr * self.atr_multiplier

        if direction == 'LONG':
            stop_loss = entry_price - stop_distance
        else:
            stop_loss = entry_price + stop_distance

        return stop_loss

# Example:
# Entry: $100
# ATR: $2
# Multiplier: 2.0
# Stop Distance: $2 × 2.0 = $4
# Long Stop: $100 - $4 = $96

Advantages: - Adapts to market volatility - Wider stops in volatile markets (less likely to be stopped out) - Tighter stops in calm markets (better risk management)

3. Trailing Stop-Loss¶

Method: Stop-loss that follows price upward (for longs) but never moves down.

class TrailingStopLoss:
    """Trailing stop-loss."""

    def __init__(self, trail_pct=0.10):
        self.trail_pct = Decimal(str(trail_pct))
        self.stop_loss = None

    def initialize(self, entry_price, direction):
        """Initialize stop-loss at entry."""
        if direction == 'LONG':
            self.stop_loss = entry_price * (1 - self.trail_pct)
        else:
            self.stop_loss = entry_price * (1 + self.trail_pct)

    def update(self, current_price, direction):
        """Update trailing stop-loss."""
        if direction == 'LONG':
            # For long: move stop up, never down
            new_stop = current_price * (1 - self.trail_pct)
            if self.stop_loss is None or new_stop > self.stop_loss:
                self.stop_loss = new_stop
        else:
            # For short: move stop down, never up
            new_stop = current_price * (1 + self.trail_pct)
            if self.stop_loss is None or new_stop < self.stop_loss:
                self.stop_loss = new_stop

        return self.stop_loss

# Example (Long position):
# Entry: $100, Trail: 10%
# Initial Stop: $90
# Price rises to $120
# New Stop: $120 × 0.9 = $108 (moved up)
# Price falls to $115
# Stop stays at $108 (doesn't move down)
# Price falls to $108 → Stopped out with $8 profit

4. Support/Resistance Stop-Loss¶

Method: Place stop-loss below support (long) or above resistance (short).

class SupportResistanceStopLoss:
    """Stop-loss based on support/resistance levels."""

    def __init__(self, buffer_pct=0.01):
        self.buffer_pct = Decimal(str(buffer_pct))

    def calculate_stop_loss(self, entry_price, support_level, direction):
        """
        Calculate stop-loss based on support/resistance.

        Args:
            entry_price: Entry price
            support_level: Support level (for longs) or resistance (for shorts)
            direction: 'LONG' or 'SHORT'

        Returns:
            Stop-loss price
        """
        # Add small buffer below support / above resistance
        if direction == 'LONG':
            stop_loss = support_level * (1 - self.buffer_pct)
        else:
            stop_loss = support_level * (1 + self.buffer_pct)

        return stop_loss

# Example:
# Entry: $102
# Support: $98
# Buffer: 1%
# Stop: $98 × 0.99 = $97.02

5. Time-Based Stop-Loss¶

Method: Exit after a certain time period regardless of price.

from datetime import datetime, timedelta

class TimeBasedStopLoss:
    """Time-based stop-loss."""

    def __init__(self, max_hold_days=7):
        self.max_hold_days = max_hold_days

    def should_exit(self, entry_date, current_date):
        """Check if max hold time exceeded."""
        days_in_trade = (current_date - entry_date).days

        if days_in_trade >= self.max_hold_days:
            logger.info(
                f"Time stop triggered: {days_in_trade} days "
                f"(max: {self.max_hold_days})"
            )
            return True

        return False

Portfolio Risk Metrics¶

Value at Risk (VaR)¶

Definition: Maximum expected loss over a given time period at a given confidence level.

import numpy as np
from scipy import stats

def calculate_var(returns, confidence_level=0.95, time_horizon=1):
    """
    Calculate Value at Risk.

    Args:
        returns: Array of returns
        confidence_level: Confidence level (0.95 = 95%)
        time_horizon: Time horizon in days

    Returns:
        VaR as percentage
    """
    # Parametric VaR (assumes normal distribution)
    mean = np.mean(returns)
    std = np.std(returns)
    z_score = stats.norm.ppf(1 - confidence_level)

    var_parametric = -(mean + z_score * std) * np.sqrt(time_horizon)

    # Historical VaR (uses actual distribution)
    var_historical = -np.percentile(returns, (1 - confidence_level) * 100)

    return {
        'parametric': var_parametric * 100,  # As percentage
        'historical': var_historical * 100
    }

# Example:
# Returns: [-0.02, 0.01, -0.01, 0.03, -0.005, ...]
# Confidence: 95%
# Time Horizon: 1 day
# 
# Interpretation: "We are 95% confident we won't lose more than X% in one day"

Conditional Value at Risk (CVaR)¶

Definition: Expected loss given that VaR has been exceeded.

def calculate_cvar(returns, confidence_level=0.95):
    """
    Calculate Conditional Value at Risk (Expected Shortfall).

    Args:
        returns: Array of returns
        confidence_level: Confidence level (0.95 = 95%)

    Returns:
        CVaR as percentage
    """
    # Get VaR threshold
    var = np.percentile(returns, (1 - confidence_level) * 100)

    # Calculate average of returns below VaR
    cvar = -returns[returns <= var].mean()

    return cvar * 100

# Example:
# 95% VaR: -3.5%
# CVaR: -5.2%
# 
# Interpretation: "When we lose more than VaR, we expect to lose 5.2% on average"

Portfolio Beta¶

Definition: Measure of portfolio volatility relative to market.

def calculate_beta(portfolio_returns, market_returns):
    """
    Calculate portfolio beta.

    Args:
        portfolio_returns: Array of portfolio returns
        market_returns: Array of market (benchmark) returns

    Returns:
        Beta value
    """
    # Calculate covariance and variance
    covariance = np.cov(portfolio_returns, market_returns)[0, 1]
    market_variance = np.var(market_returns)

    beta = covariance / market_variance

    return beta

# Interpretation:
# Beta = 1.0: Moves with market
# Beta > 1.0: More volatile than market
# Beta < 1.0: Less volatile than market
# Beta < 0.0: Moves opposite to market

Risk-Adjusted Returns¶

Sharpe Ratio¶

def sharpe_ratio(returns, risk_free_rate=0.0, periods_per_year=252):
    """Calculate Sharpe ratio."""
    excess_returns = returns - risk_free_rate / periods_per_year
    return (np.mean(excess_returns) / np.std(excess_returns)) * np.sqrt(periods_per_year)

Sortino Ratio¶

def sortino_ratio(returns, risk_free_rate=0.0, periods_per_year=252):
    """Calculate Sortino ratio (only penalizes downside volatility)."""
    excess_returns = returns - risk_free_rate / periods_per_year
    downside_returns = excess_returns[excess_returns < 0]
    downside_std = np.std(downside_returns)

    return (np.mean(excess_returns) / downside_std) * np.sqrt(periods_per_year)

Calmar Ratio¶

def calmar_ratio(returns, periods_per_year=252):
    """Calculate Calmar ratio (return / max drawdown)."""
    # Annualized return
    total_return = (1 + returns).prod() - 1
    years = len(returns) / periods_per_year
    annualized_return = (1 + total_return) ** (1 / years) - 1

    # Max drawdown
    cumulative = (1 + returns).cumprod()
    running_max = np.maximum.accumulate(cumulative)
    drawdown = (cumulative - running_max) / running_max
    max_drawdown = abs(drawdown.min())

    return annualized_return / max_drawdown if max_drawdown != 0 else 0

Best Practices¶

1. Never Risk More Than You Can Afford to Lose¶

# ❌ Bad - Risking everything
position_size = account_value  # 100% in one trade!

# ✅ Good - Limited risk
max_risk_per_trade = account_value * 0.02  # 2% max

2. Use Position Sizing Consistently¶

# ❌ Bad - Emotional sizing
if feeling_confident:
    position_size = large_size
else:
    position_size = small_size

# ✅ Good - Systematic sizing
position_size = calculate_position_size(
    capital=account_value,
    risk_pct=0.02,
    stop_loss_pct=0.05
)

3. Always Use Stop-Losses¶

# ❌ Bad - No stop-loss
place_order(symbol, quantity, price)

# ✅ Good - With stop-loss
place_order(symbol, quantity, price, stop_loss=stop_price)

4. Diversify Across Multiple Positions¶

# ❌ Bad - All capital in one position
if signal:
    buy(symbol, quantity=max_quantity)

# ✅ Good - Limit per position
max_position_pct = 0.20  # 20% max per position
max_quantity = (account_value * max_position_pct) / price

5. Monitor and Adjust¶

def daily_risk_check(portfolio):
    """Daily risk monitoring."""
    # Check drawdown
    drawdown = portfolio.calculate_drawdown()
    if drawdown > 0.15:
        logger.warning(f"Drawdown at {drawdown:.2%} - consider reducing risk")

    # Check correlation
    avg_correlation = portfolio.calculate_avg_correlation()
    if avg_correlation > 0.7:
        logger.warning(f"High correlation ({avg_correlation:.2f}) - diversify!")

    # Check position sizes
    for symbol, value in portfolio.position_values.items():
        pct = value / portfolio.total_value
        if pct > 0.25:
            logger.warning(f"{symbol} is {pct:.2%} of portfolio - too concentrated")

Resources¶

Strategy Development: See docs/STRATEGY_DEVELOPMENT.md
Backtesting: See docs/BACKTESTING_GUIDE.md
Portfolio Management: See src/backtesting/portfolio.py
Risk Metrics: See src/backtesting/performance.py

Support¶

For questions about risk management:

Review this documentation
Check portfolio implementation in src/backtesting/portfolio.py
Examine risk metric calculations in src/backtesting/performance.py
Open an issue with the risk-management label

Risk Management Guide¶

Overview¶

Core Principles¶

Risk Hierarchy¶

Position Sizing Strategies¶

1. Fixed Dollar Amount¶

2. Fixed Percentage of Capital¶

3. Volatility-Based Sizing (ATR)¶

4. Kelly Criterion¶

5. Custom Position Sizing¶

Risk Limits and Controls¶

1. Maximum Position Size¶

2. Maximum Daily Loss¶

3. Maximum Drawdown Limit¶

4. Maximum Number of Positions¶

5. Correlation Limit¶

Stop-Loss Strategies¶

1. Fixed Percentage Stop-Loss¶

2. ATR-Based Stop-Loss¶

3. Trailing Stop-Loss¶

4. Support/Resistance Stop-Loss¶

5. Time-Based Stop-Loss¶

Portfolio Risk Metrics¶

Value at Risk (VaR)¶

Conditional Value at Risk (CVaR)¶

Portfolio Beta¶

Risk-Adjusted Returns¶

Sharpe Ratio¶

Sortino Ratio¶

Calmar Ratio¶

Best Practices¶

1. Never Risk More Than You Can Afford to Lose¶

2. Use Position Sizing Consistently¶

3. Always Use Stop-Losses¶

4. Diversify Across Multiple Positions¶

5. Monitor and Adjust¶

Resources¶

Further Reading¶

Support¶