Backtrader Position Sizing: Fixed Size, Equity and Stop-Based Risk

g = \mathbb{E}[\ln(1 + R)]

flowchart TD
    subgraph Market ["Alpha Generation vs Capital Allocation"]
        S1["Historical OHLCV Bar"] --> S2["Strategy Signal Engine\n(Direction: Buy / Sell)"]
        S2 --> S3["Trade Idea: Long Entry"]
        
        P1["Portfolio State\n(Cash, Equity, Open Positions)"] --> S4["Position Sizing Engine\n(bt.Sizer / OpenBacktesting Risk)"]
        R1["Volatility & Stop Distance\n(ATR, Support/Resistance)"] --> S4
        
        S3 & S4 --> S5["Order Submission\n(Action: BUY, Size: N Units)"]
        S5 --> S6["Broker Execution & Ledger"]
    end

def next(self):
    if not self.position and self.crossover > 0:
        # Static size ignores portfolio growth, drawdown, and asset price
        self.buy(size=100)

flowchart LR
    Strategy["Strategy.buy()"] --> Router{"Explicit size passed?"}
    Router -- "Yes: size=N" --> Broker["Submit to Broker"]
    Router -- "No: size=None" --> Sizer["Active bt.Sizer._getsizing()"]
    Sizer --> CommInfo["Query CommissionInfo & Cash"]
    CommInfo --> Broker

import backtrader as bt

cerebro = bt.Cerebro()
# Sets global stake to 50 shares per trade
cerebro.addsizer(bt.sizers.FixedSize, stake=50)

# Allocates 20% of available cash per trade, rounded down to whole shares
cerebro.addsizer(bt.sizers.PercentSizerInt, percents=20)

cerebro.addsizer(bt.sizers.AllInSizer)

import backtrader as bt

class BaseRiskSizer(bt.Sizer):
    params = (
        ("risk_perc", 0.01),      # 1% equity risk
        ("cash_buffer", 0.05),    # 5% cash buffer for commissions/slippage
    )

    def _getsizing(self, comminfo, cash, data, isbuy):
        """
        Calculate order size.
        
        :param comminfo: CommissionInfo object for the target asset
        :param cash: Available free cash in the broker
        :param data: Target data feed
        :param isbuy: Boolean (True for buy/long, False for sell/short)
        :return: Integer or float order quantity
        """
        # Sizing logic implemented here
        return 0

def _getsizing(self, comminfo, cash, data, isbuy):
    # Total portfolio equity (Cash + Unrealized PnL)
    equity = self.strategy.broker.getvalue()
    
    # Access indicator on strategy
    current_atr = self.strategy.atr[0]
    
    # Query current position size
    open_units = self.strategy.position.size

flowchart TD
    A["self.sell() called without explicit size"] --> B{"_getsizing(isbuy=False)"}
    B --> C{"Does sizer check isbuy?"}
    C -- "No: Recalculates new size" --> D["Size Mismatch: Partial exit or accidental short"]
    C -- "Yes: Returns self.strategy.position.size" --> E["Clean Position Liquidation (Flat)"]

\Delta_{\text{dollar}} = E \times f_{\text{risk}}

\delta_{\text{share}} = |P_{\text{entry}} - P_{\text{stop}}|

N_{\text{risk}} = \left\lfloor \frac{\Delta_{\text{dollar}}}{\delta_{\text{share}}} \right\rfloor = \left\lfloor \frac{E \times f_{\text{risk}}}{|P_{\text{entry}} - P_{\text{stop}}|} \right\rfloor

N_{\text{cash}} = \left\lfloor \frac{\text{Cash} \times (1 - \beta)}{P_{\text{entry}} \times (1 + c)} \right\rfloor

N^* = \max\left(0, \min\left(N_{\text{risk}}, N_{\text{cash}}\right)\right)

import math
import numpy as np
import pandas as pd
import backtrader as bt


def generate_synthetic_market_data(bars: int = 250, seed: int = 42) -> pd.DataFrame:
    """Generate realistic OHLCV daily data exhibiting trending and volatile regimes."""
    np.random.seed(seed)
    dates = pd.date_range(start="2024-01-01", periods=bars, freq="B")
    
    # Stochastic geometric random walk with drift
    dt = 1.0 / 252.0
    mu = 0.12
    sigma = 0.25
    
    returns = np.random.normal((mu - 0.5 * sigma**2) * dt, sigma * np.sqrt(dt), bars)
    log_prices = np.cumsum(returns)
    base_price = 100.0 * np.exp(log_prices)
    
    # Synthesize realistic bar ranges
    high_noise = np.random.uniform(1.005, 1.025, bars)
    low_noise = np.random.uniform(0.975, 0.995, bars)
    close_noise = np.random.uniform(0.985, 1.015, bars)
    
    high = base_price * high_noise
    low = base_price * low_noise
    close = np.clip(base_price * close_noise, low, high)
    open_p = np.clip(base_price * np.random.uniform(0.99, 1.01, bars), low, high)
    volume = np.random.randint(500000, 2000000, bars).astype(float)
    
    df = pd.DataFrame(
        {"open": open_p, "high": high, "low": low, "close": close, "volume": volume},
        index=dates,
    )
    df.index.name = "datetime"
    return df


class VolatilityRiskSizer(bt.Sizer):
    """
    Fixed-Fractional Volatility Risk Sizer.
    
    Allocates shares such that:
    Dollar Risk = Total Equity * risk_perc
    Stop Distance = ATR * atr_mult
    Units = Dollar Risk / Stop Distance
    """
    params = (
        ("risk_perc", 0.01),      # 1% equity risk per trade
        ("atr_mult", 2.0),        # 2x ATR stop distance
        ("cash_buffer", 0.05),    # 5% cash buffer reserved for slippage/commissions
    )

    def _getsizing(self, comminfo, cash, data, isbuy):
        # 1. Exit Defense: If liquidating a long position, return current holding
        if not isbuy:
            return self.strategy.position.size

        # 2. Extract strategy state
        equity = self.strategy.broker.getvalue()
        current_atr = self.strategy.atr[0]
        current_price = data.close[0]

        if math.isnan(current_atr) or current_atr <= 0:
            return 0

        # 3. Calculate risk parameters
        allowed_dollar_risk = equity * self.p.risk_perc
        stop_distance = current_atr * self.p.atr_mult
        
        # Share count based on volatility risk
        shares_by_risk = math.floor(allowed_dollar_risk / stop_distance)

        # 4. Cash / purchasing power boundary
        usable_cash = cash * (1.0 - self.p.cash_buffer)
        commission_factor = 1.0 + (comminfo.p.commission if comminfo else 0.001)
        max_shares_by_cash = math.floor(usable_cash / (current_price * commission_factor))

        # 5. Final bounded lot allocation
        target_size = max(0, min(shares_by_risk, max_shares_by_cash))
        
        # Store metadata on strategy for execution logging
        self.strategy.last_calculated_risk = {
            "equity": equity,
            "atr": current_atr,
            "stop_distance": stop_distance,
            "shares_risk": shares_by_risk,
            "shares_cash": max_shares_by_cash,
            "final_size": target_size,
        }

        return int(target_size)


class VolatilitySizingStrategy(bt.Strategy):
    """Trend following strategy using Donchian channel entries and ATR stops."""
    params = (
        ("donchian_period", 20),
        ("atr_period", 14),
        ("atr_mult", 2.0),
    )

    def __init__(self):
        # Indicators
        self.highest = bt.indicators.Highest(self.data.high(-1), period=self.p.donchian_period)
        self.lowest = bt.indicators.Lowest(self.data.low(-1), period=self.p.donchian_period)
        self.atr = bt.indicators.ATR(self.data, period=self.p.atr_period)
        
        # State tracking
        self.entry_order = None
        self.stop_order = None
        self.last_calculated_risk = {}

    def next(self):
        # Wait for indicator warm-up
        if len(self.data) < max(self.p.donchian_period, self.p.atr_period) + 1:
            return

        # Pending order guard
        if self.entry_order:
            return

        # Entry logic: 20-day high breakout
        if not self.position:
            if self.data.close[0] > self.highest[0]:
                # Submit buy order without size; delegates to VolatilityRiskSizer
                self.entry_order = self.buy()
        else:
            # Exit logic: 20-day low breakdown
            if self.data.close[0] < self.lowest[0]:
                if self.stop_order:
                    self.cancel(self.stop_order)
                self.close()

    def notify_order(self, order):
        if order.status in [order.Completed]:
            if order.isbuy() and self.position.size > 0:
                self.entry_order = None
                # Calculate stop price based on entry fill
                stop_dist = self.atr[0] * self.p.atr_mult
                stop_price = order.executed.price - stop_dist
                
                # Submit resting stop order to protect position
                self.stop_order = self.sell(
                    exectype=bt.Order.Stop,
                    price=stop_price,
                    size=order.executed.size,
                )
                
                log_data = self.last_calculated_risk
                print(
                    f"[{self.data.datetime.date(0)}] BUY EXECUTED | "
                    f"Price: ${order.executed.price:.2f} | "
                    f"Size: {order.executed.size} | "
                    f"Cost: ${order.executed.value:.2f} | "
                    f"ATR: ${log_data.get('atr', 0):.2f} | "
                    f"Stop Price: ${stop_price:.2f}"
                )
            elif order.issell():
                self.stop_order = None
                print(
                    f"[{self.data.datetime.date(0)}] SELL EXECUTED | "
                    f"Price: ${order.executed.price:.2f} | "
                    f"Size: {order.executed.size} | "
                    f"Cash: ${self.broker.getcash():.2f}"
                )
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.entry_order = None
            if order.status == order.Margin:
                print(f"[{self.data.datetime.date(0)}] MARGIN REJECTION DETECTED")

    def notify_trade(self, trade):
        if trade.isclosed:
            print(
                f"[{self.data.datetime.date(0)}] TRADE CLOSED | "
                f"Net PnL: ${trade.pnlcomm:.2f} | "
                f"Return: {(trade.pnlcomm / trade.price) * 100:.2f}%\n"
            )


def run_sizing_simulation():
    cerebro = bt.Cerebro()
    
    # 1. Load Data
    df = generate_synthetic_market_data(bars=260, seed=101)
    data = bt.feeds.PandasData(dataname=df)
    cerebro.adddata(data)
    
    # 2. Attach Strategy & Register Custom Risk Sizer
    cerebro.addstrategy(VolatilitySizingStrategy)
    cerebro.addsizer(VolatilityRiskSizer, risk_perc=0.01, atr_mult=2.0, cash_buffer=0.05)
    
    # 3. Capital & Broker Environment
    cerebro.broker.setcash(100000.0)
    cerebro.broker.setcommission(commission=0.001)  # 0.1% transaction cost
    
    print(f"Initial Portfolio Value: ${cerebro.broker.getvalue():.2f}")
    cerebro.run()
    print(f"Final Portfolio Value:   ${cerebro.broker.getvalue():.2f}")


if __name__ == "__main__":
    run_sizing_simulation()

Initial Portfolio Value: $100000.00
[2024-02-14] BUY EXECUTED | Price: $104.82 | Size: 178 | Cost: $18657.96 | ATR: $2.81 | Stop Price: $99.20
[2024-03-27] SELL EXECUTED | Price: $118.45 | Size: 178 | Cash: $102426.54
[2024-03-27] TRADE CLOSED | Net PnL: $2426.54 | Return: 2314.96%

[2024-05-08] BUY EXECUTED | Price: $132.10 | Size: 121 | Cost: $15984.10 | ATR: $4.22 | Stop Price: $123.66
[2024-06-19] SELL EXECUTED | Price: $122.95 | Size: 121 | Cash: $101319.45
[2024-06-19] TRADE CLOSED | Net PnL: -$1107.09 | Return: -838.07%

[2024-08-02] BUY EXECUTED | Price: $148.60 | Size: 142 | Cost: $21101.20 | ATR: $3.56 | Stop Price: $141.48
[2024-09-20] SELL EXECUTED | Price: $162.30 | Size: 142 | Cash: $103265.05
[2024-09-20] TRADE CLOSED | Net PnL: $1945.60 | Return: 1309.29%

Final Portfolio Value:   $103265.05

flowchart TD
    subgraph Backtrader ["Backtrader: Stateful Sizer Callbacks"]
        B1["Strategy.buy()"] --> B2["_getsizing() Hook"]
        B2 --> B3["Polls Strategy Attributes\n(Implicit Coupling)"]
        B3 --> B4["Broker Order Match"]
        B4 -.->|"Requires Manual Linkage"| B5["Separate Stop Order Creation"]
    end

    subgraph OpenBacktesting ["OpenBacktesting v0.2.1: Unified Execution Engine"]
        O1["Strategy.buy()"] --> O2["Declarative Order Specification\n(Units / Cash Budget / Risk %)"]
        O2 --> O3["Atomic Trade Container\n(Integrated Entry + SL + Size)"]
        O3 --> O4["Deterministic Ledger Fill\n(portfolio.orders_df & trades_df)"]
    end

# Mode 1: Fixed Unit Allocation
self.buy(size=150, sl=98.50)

# Mode 2: Target Cash Allocation (Cash-Budgeted)
self.buy(cash=25000.0, sl=98.50)

# Mode 3: Portfolio Risk Budgeting (e.g., exactly 1.0% equity risk to stop)
self.buy(risk_fraction=0.01, sl=98.50)