Automating Bollinger Bands In Python
Introduction
Python's data science stack -- pandas for time series manipulation, numpy for vectorized computation, and matplotlib for visualization -- makes it the ideal language for implementing Bollinger Band trading systems. This article provides production-quality Python code for the complete workflow: computing Bollinger Bands, generating trading signals, running rigorous backtests, and connecting to a broker API for live execution. Every function includes type hints, docstrings, and edge-case handling suitable for real trading.
Core Implementation
Bollinger Band Calculator
import pandas as pd
import numpy as np
from typing import Tuple, Optional
from dataclasses import dataclass
@dataclass
class BollingerState:
"""Snapshot of Bollinger Band values at a point in time."""
middle: float
upper: float
lower: float
bandwidth: float
z_score: float
percent_b: float # %B indicator
class BollingerBandCalculator:
"""
Production Bollinger Band calculator with multiple output modes.
"""
def __init__(self, period: int = 20, num_std: float = 2.0,
ma_type: str = 'sma'):
"""
Parameters
----------
period : int - Lookback window
num_std : float - Standard deviation multiplier
ma_type : str - 'sma' for simple, 'ema' for exponential
"""
self.period = period
self.num_std = num_std
self.ma_type = ma_type
def calculate(self, close: pd.Series) -> pd.DataFrame:
"""
Calculate all Bollinger Band components.
Returns DataFrame with: middle, upper, lower, bandwidth,
z_score, percent_b
"""
if self.ma_type == 'ema':
middle = close.ewm(span=self.period, adjust=False).mean()
# For EMA bands, use rolling std (common convention)
std = close.rolling(self.period).std()
else:
middle = close.rolling(self.period).mean()
std = close.rolling(self.period).std()
upper = middle + self.num_std * std
lower = middle - self.num_std * std
result = pd.DataFrame({
'middle': middle,
'upper': upper,
'lower': lower,
'std': std,
'bandwidth': (upper - lower) / middle,
'z_score': (close - middle) / std,
'percent_b': (close - lower) / (upper - lower)
}, index=close.index)
return result
def get_state(self, close: pd.Series) -> Optional[BollingerState]:
"""Get current (latest) Bollinger Band state."""
bb = self.calculate(close)
if bb.iloc[-1].isna().any():
return None
row = bb.iloc[-1]
return BollingerState(
middle=row['middle'],
upper=row['upper'],
lower=row['lower'],
bandwidth=row['bandwidth'],
z_score=row['z_score'],
percent_b=row['percent_b']
)
The %B Indicator
The %B indicator normalizes price position within the bands to a [0, 1] scale:
$$
\%B = \frac{Close - Lower}{Upper - Lower}
$$
| %B Value | Meaning |
|----------|---------|
| > 1.0 | Price above upper band |
| 1.0 | Price at upper band |
| 0.5 | Price at middle band |
| 0.0 | Price at lower band |
| < 0.0 | Price below lower band |
%B is especially useful for cross-sectional comparisons: a %B of 0.1 on AAPL and 0.1 on MSFT means both are equally oversold relative to their own recent ranges.
Signal Generation Engine
from enum import Enum
from typing import Dict, List
class SignalType(Enum):
LONG = 1
SHORT = -1
FLAT = 0
@dataclass
class TradeSignal:
timestamp: pd.Timestamp
symbol: str
signal: SignalType
z_score: float
confidence: float
entry_price: float
stop_loss: float
take_profit: float
class BBSignalGenerator:
"""
Generate trading signals from Bollinger Bands with multiple strategies.
"""
def __init__(self, bb_calc: BollingerBandCalculator,
strategy: str = 'mean_reversion'):
self.bb = bb_calc
self.strategy = strategy
def mean_reversion_signals(self, df: pd.DataFrame,
entry_z: float = 2.0,
exit_z: float = 0.5,
stop_atr_mult: float = 2.0) -> pd.DataFrame:
"""
Mean reversion: buy oversold, sell overbought.
"""
bb = self.bb.calculate(df['close'])
# ATR for stop loss calculation
atr = self._compute_atr(df, period=14)
signals = pd.DataFrame(index=df.index)
signals['z_score'] = bb['z_score']
signals['bandwidth'] = bb['bandwidth']
# Entry conditions
signals['buy_signal'] = bb['z_score'] < -entry_z
signals['sell_signal'] = bb['z_score'] > entry_z
# Exit conditions
signals['exit_long'] = bb['z_score'] > -exit_z
signals['exit_short'] = bb['z_score'] < exit_z
# Build position column using state machine
signals['position'] = 0
position = 0
for i in range(1, len(signals)):
if position == 0:
if signals['buy_signal'].iloc[i]:
position = 1
elif signals['sell_signal'].iloc[i]:
position = -1
elif position == 1:
if signals['exit_long'].iloc[i]:
position = 0
elif position == -1:
if signals['exit_short'].iloc[i]:
position = 0
signals.iloc[i, signals.columns.get_loc('position')] = position
# Shift to avoid look-ahead
signals['position'] = signals['position'].shift(1).fillna(0)
# Stop loss levels
signals['stop_long'] = df['close'] - stop_atr_mult * atr
signals['stop_short'] = df['close'] + stop_atr_mult * atr
return signals
def breakout_signals(self, df: pd.DataFrame,
squeeze_lookback: int = 120,
squeeze_pct: float = 10) -> pd.DataFrame:
"""
Bollinger squeeze breakout: enter when price breaks out of
tight bands after a period of low volatility.
"""
bb = self.bb.calculate(df['close'])
signals = pd.DataFrame(index=df.index)
signals['bandwidth'] = bb['bandwidth']
signals['z_score'] = bb['z_score']
# Detect squeeze: bandwidth below Nth percentile
bw_threshold = bb['bandwidth'].rolling(squeeze_lookback).quantile(squeeze_pct / 100)
signals['in_squeeze'] = bb['bandwidth'] <= bw_threshold
# Detect breakout from squeeze
was_squeeze = signals['in_squeeze'].rolling(5).max() > 0
signals['breakout_up'] = was_squeeze & (df['close'] > bb['upper'])
signals['breakout_down'] = was_squeeze & (df['close'] < bb['lower'])
signals['position'] = 0
signals.loc[signals['breakout_up'], 'position'] = 1
signals.loc[signals['breakout_down'], 'position'] = -1
# Hold breakout positions for N bars
holding_period = 10
for i in range(len(signals)):
if signals['position'].iloc[i] != 0:
end = min(i + holding_period, len(signals))
signals.iloc[i:end, signals.columns.get_loc('position')] = signals['position'].iloc[i]
signals['position'] = signals['position'].shift(1).fillna(0)
return signals
@staticmethod
def _compute_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
"""Average True Range."""
high_low = df['high'] - df['low']
high_close = abs(df['high'] - df['close'].shift())
low_close = abs(df['low'] - df['close'].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
return tr.rolling(period).mean()
Backtesting Framework
class BBBacktester:
"""
Full-featured backtester for Bollinger Band strategies.
"""
def __init__(self, initial_capital: float = 100_000,
commission_pct: float = 0.001,
slippage_pct: float = 0.0005,
position_size_pct: float = 0.95):
self.capital = initial_capital
self.commission = commission_pct
self.slippage = slippage_pct
self.size_pct = position_size_pct
def run(self, signals: pd.DataFrame, prices: pd.Series) -> pd.DataFrame:
"""Execute backtest with position tracking."""
results = pd.DataFrame(index=signals.index)
results['close'] = prices
results['position'] = signals['position']
results['return'] = prices.pct_change()
# Strategy returns
results['strategy_return'] = results['position'] * results['return']
# Costs on position changes
results['turnover'] = results['position'].diff().abs()
results['cost'] = results['turnover'] * (self.commission + self.slippage)
results['net_return'] = results['strategy_return'] - results['cost']
# Equity
results['equity'] = self.capital * (1 + results['net_return']).cumprod()
results['benchmark'] = self.capital * (1 + results['return']).cumprod()
# Drawdown
results['peak'] = results['equity'].cummax()
results['drawdown'] = (results['equity'] - results['peak']) / results['peak']
return results
def metrics(self, results: pd.DataFrame) -> dict:
"""Comprehensive performance metrics."""
r = results['net_return'].dropna()
years = len(r) / 252
ann_ret = r.mean() * 252
ann_vol = r.std() * np.sqrt(252)
sharpe = ann_ret / ann_vol if ann_vol > 0 else 0
# Trade analysis
positions = results['position']
trade_starts = positions.diff().abs() > 0
n_trades = trade_starts.sum() // 2
# Underwater analysis
max_dd = results['drawdown'].min()
dd_duration = 0
max_dd_duration = 0
for dd in results['drawdown']:
if dd < 0:
dd_duration += 1
max_dd_duration = max(max_dd_duration, dd_duration)
else:
dd_duration = 0
return {
'annual_return': f"{ann_ret:.2%}",
'annual_volatility': f"{ann_vol:.2%}",
'sharpe_ratio': round(sharpe, 2),
'max_drawdown': f"{max_dd:.2%}",
'max_dd_duration_days': max_dd_duration,
'total_trades': int(n_trades),
'trades_per_year': round(n_trades / years, 1),
'total_cost': f"{results['cost'].sum():.2%}",
'time_in_market': f"{(positions != 0).mean():.1%}",
'final_equity': f"${results['equity'].iloc[-1]:,.0f}"
}
Multi-Symbol Portfolio
class BBPortfolio:
"""
Run Bollinger Band strategy across a portfolio of symbols.
"""
def __init__(self, symbols: List[str], strategy: str = 'mean_reversion',
max_positions: int = 10):
self.symbols = symbols
self.strategy = strategy
self.max_positions = max_positions
self.bb_calc = BollingerBandCalculator()
self.signal_gen = BBSignalGenerator(self.bb_calc, strategy)
def run_portfolio(self, data: Dict[str, pd.DataFrame]) -> pd.DataFrame:
"""
Generate signals for all symbols and combine into portfolio returns.
"""
all_signals = {}
for symbol in self.symbols:
df = data.get(symbol)
if df is None or len(df) < 50:
continue
if self.strategy == 'mean_reversion':
sigs = self.signal_gen.mean_reversion_signals(df)
else:
sigs = self.signal_gen.breakout_signals(df)
all_signals[symbol] = sigs[['position']].rename(
columns={'position': symbol}
)
# Combine into portfolio
positions = pd.concat(all_signals.values(), axis=1).fillna(0)
# Limit concurrent positions
active = positions.abs().sum(axis=1)
scale = np.where(active > self.max_positions,
self.max_positions / active, 1.0)
positions = positions.multiply(scale, axis=0)
# Equal weight per position
n_active = (positions != 0).sum(axis=1).replace(0, 1)
weights = positions.div(n_active, axis=0)
# Compute portfolio returns
returns = pd.DataFrame()
for symbol in positions.columns:
if symbol in data:
ret = data[symbol]['close'].pct_change()
returns[symbol] = weights[symbol] * ret
portfolio_return = returns.sum(axis=1)
result = pd.DataFrame({
'portfolio_return': portfolio_return,
'n_positions': (positions != 0).sum(axis=1),
'gross_exposure': positions.abs().sum(axis=1),
'equity': 100_000 * (1 + portfolio_return).cumprod()
})
return result
Live Trading Integration
class BBLiveTrader:
"""
Connect Bollinger Band signals to a broker for live execution.
"""
def __init__(self, signal_gen: BBSignalGenerator,
symbols: List[str], risk_per_trade: float = 0.01):
self.signal_gen = signal_gen
self.symbols = symbols
self.risk = risk_per_trade
def generate_orders(self, data: Dict[str, pd.DataFrame],
portfolio_value: float,
current_positions: Dict[str, int]) -> List[dict]:
"""Generate orders for all symbols."""
orders = []
for symbol in self.symbols:
df = data.get(symbol)
if df is None:
continue
signals = self.signal_gen.mean_reversion_signals(df)
target = int(signals['position'].iloc[-1])
current = current_positions.get(symbol, 0)
if target == current:
continue # No change needed
price = df['close'].iloc[-1]
atr = self.signal_gen._compute_atr(df).iloc[-1]
# Position size based on ATR risk
if target != 0:
risk_dollars = portfolio_value * self.risk
shares = int(risk_dollars / (2 * atr))
shares = shares * target # Apply direction
else:
shares = -current # Flatten
orders.append({
'symbol': symbol,
'action': 'BUY' if shares > 0 else 'SELL',
'quantity': abs(shares),
'order_type': 'MARKET',
'z_score': signals['z_score'].iloc[-1],
'reason': f"BB z={signals['z_score'].iloc[-1]:.2f}"
})
return orders
Conclusion
Python provides everything needed to build a complete Bollinger Band trading system: pandas handles the rolling calculations, numpy enables vectorized backtesting, and broker API libraries (ib_insync, alpaca-trade-api) connect signals to live markets. The key implementation details are: use the %B indicator for cross-sectional comparisons, shift all signals by one bar to prevent look-ahead bias, include realistic transaction costs, and limit portfolio concentration. Start with the mean-reversion strategy on a small universe, validate with out-of-sample backtesting, paper trade for 30+ days, then deploy with small capital.
Frequently Asked Questions
What Python libraries do I need for Bollinger Band trading?
Core: pandas, numpy, matplotlib. For backtesting: you can use the custom framework above or libraries like backtrader, zipline, or vectorbt. For live trading: ib_insync (Interactive Brokers), alpaca-trade-api (Alpaca), or ccxt (crypto). For data: yfinance (free), polygon-api-client (professional).
How do I handle missing data in the rolling calculations?
Pandas rolling functions handle NaN values by default -- the first (period-1) bars will return NaN. For real-time systems, ensure you have enough history loaded before generating signals. If a stock is halted or has missing bars, forward-fill the last known price for indicator computation but do not generate new signals on filled data.
Should I use SMA or EMA for Bollinger Bands?
John Bollinger's original specification uses SMA, and this is what most traders and platforms use. EMA-based bands are more responsive to recent price changes, which can be advantageous for shorter timeframes. Test both in your backtest -- the difference is typically small (less than 0.1 Sharpe). Use SMA for consistency with the broader trading community.
How do I vectorize the position tracking (avoid the for loop)?
The for-loop state machine is needed for complex entry/exit logic. For simple threshold-based signals, use numpy: positions = np.where(z_score < -2, 1, np.where(z_score > 0, 0, np.nan)) then pd.Series(positions).ffill(). This is 100x faster but cannot handle complex state-dependent logic like trailing stops.
What is a realistic Sharpe ratio for a Bollinger Band strategy?
On a single stock: 0.3-0.7. On a diversified portfolio of 20-50 stocks: 0.7-1.2. These numbers are after transaction costs. If your backtest shows a Sharpe above 2.0, you likely have look-ahead bias or insufficient transaction cost modeling. The strategy works best as one component of a multi-factor system rather than standalone.