Risk Management in Algorithmic Trading: A Developer's Guide

Risk Management Is the Strategy

A mediocre strategy with good risk management will outperform a great strategy with poor risk management. This is the most counterintuitive truth in trading: your edge comes from how much you risk, not from how often you are right.

Position Sizing with ATR

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://tickatlas.com/v1"

class RiskManager:
    def __init__(self, balance: float, max_risk_pct: float = 1.0,
                 max_positions: int = 5, max_drawdown_pct: float = 15.0):
        self.initial_balance = balance
        self.balance = balance
        self.max_risk_pct = max_risk_pct
        self.max_positions = max_positions
        self.max_drawdown_pct = max_drawdown_pct
        self.open_positions = []
        self.peak_balance = balance

    def calculate_position_size(self, symbol: str, entry: float,
                                 stop_loss: float) -> float:
        """Calculate lot size based on fixed-percentage risk."""
        risk_amount = self.balance * (self.max_risk_pct / 100)
        stop_distance = abs(entry - stop_loss)

        if stop_distance == 0:
            return 0.0

        pip_factor = 100 if "JPY" in symbol else 10000
        stop_pips = stop_distance * pip_factor
        pip_value = 10.0  # USD per pip per standard lot

        lots = risk_amount / (stop_pips * pip_value)
        return round(lots, 2)

    def can_open_position(self) -> bool:
        """Check if we can open another position."""
        if len(self.open_positions) >= self.max_positions:
            return False
        if self.current_drawdown() >= self.max_drawdown_pct:
            return False
        return True

    def current_drawdown(self) -> float:
        """Calculate current drawdown from peak."""
        self.peak_balance = max(self.peak_balance, self.balance)
        dd = (self.peak_balance - self.balance) / self.peak_balance * 100
        return round(dd, 2)

    def total_risk_exposure(self) -> float:
        """Sum of risk across all open positions."""
        return sum(p.get("risk_pct", 0) for p in self.open_positions)

ATR-Based Dynamic Position Sizing

def dynamic_position_size(self, symbol: str, timeframe: str,
                          direction: str, atr_multiplier: float = 2.0) -> dict:
    """Calculate position size using live ATR data."""
    # Fetch ATR
    resp = requests.get(f"{BASE_URL}/indicator",
        headers={"X-API-Key": API_KEY},
        params={"symbol": symbol, "indicator": "ATR_14", "timeframe": timeframe})
    data = resp.json()["data"]

    atr = data["values"]["atr"]
    close = data["ohlcv"]["close"]

    stop_distance = atr * atr_multiplier
    if direction == "long":
        stop_loss = close - stop_distance
    else:
        stop_loss = close + stop_distance

    lots = self.calculate_position_size(symbol, close, stop_loss)

    return {
        "symbol": symbol,
        "direction": direction,
        "entry": close,
        "stop_loss": round(stop_loss, 5),
        "lots": lots,
        "risk_amount": round(self.balance * self.max_risk_pct / 100, 2),
        "atr": atr,
    }

Correlation-Aware Risk

Being long EURUSD, long GBPUSD, and long AUDUSD is not three separate bets -- it is essentially one large bet against the US dollar. Correlation-aware risk management counts correlated positions as partial duplicates.

CORRELATION_GROUPS = {
    "USD_WEAK": ["EURUSD", "GBPUSD", "AUDUSD", "NZDUSD"],
    "USD_STRONG": ["USDCAD", "USDJPY", "USDCHF"],
    "SAFE_HAVEN": ["XAUUSD", "USDJPY"],
}

def check_correlation_risk(self, new_symbol: str, new_direction: str) -> bool:
    """Check if a new position would over-concentrate risk."""
    # Find which group the new symbol belongs to
    for group_name, symbols in CORRELATION_GROUPS.items():
        if new_symbol in symbols:
            # Count how many open positions are in the same group
            same_group = [p for p in self.open_positions
                          if p["symbol"] in symbols
                          and p["direction"] == new_direction]
            if len(same_group) >= 2:
                return False  # Too much exposure to this group
    return True

Drawdown Circuit Breaker

class DrawdownBreaker:
    """Halt trading when drawdown exceeds thresholds."""

    def __init__(self):
        self.thresholds = [
            (10.0, "REDUCE"),   # 10% DD: halve position sizes
            (15.0, "PAUSE"),    # 15% DD: stop opening new trades
            (20.0, "FLATTEN"),  # 20% DD: close everything
        ]

    def evaluate(self, current_drawdown: float) -> str:
        action = "NORMAL"
        for threshold, level in self.thresholds:
            if current_drawdown >= threshold:
                action = level
        return action

# In the main trading loop:
breaker = DrawdownBreaker()
rm = RiskManager(balance=10000)

action = breaker.evaluate(rm.current_drawdown())
if action == "FLATTEN":
    close_all_positions()
elif action == "PAUSE":
    pass  # Do not open new trades
elif action == "REDUCE":
    rm.max_risk_pct = 0.5  # Halve risk per trade

The Risk Checklist

Related Reading