Strategy Development Guide¶
This guide explains how to develop, test, and deploy trading strategies in the DSTA backtesting framework.
Overview¶
DSTA uses an event-driven architecture for strategy development, ensuring strategies can be backtested without lookahead bias and deployed to live trading with minimal changes.
Key Concepts¶
- Strategy: Class that implements trading logic and generates signals
- Indicators: Technical indicators (SMA, RSI, MACD, etc.) for decision making
- Signals: Buy/sell/exit signals generated by strategies
- Event-Driven: Strategies respond to market events (new bars) in real-time
Strategy Interface¶
Base Strategy Class¶
All strategies inherit from BaseStrategy:
from backtesting.strategy import BaseStrategy
from backtesting.events import SignalType, Event
class MyStrategy(BaseStrategy):
"""
Your custom strategy.
"""
def __init__(self, bars, events_queue, symbol_list, **params):
super().__init__(
bars=bars,
events_queue=events_queue,
symbol_list=symbol_list,
strategy_id="MyStrategy"
)
# Store strategy parameters
self.param1 = params.get('param1', default_value)
def calculate_signals(self, event: Event) -> None:
"""
Generate trading signals based on market data.
Args:
event: MarketEvent with new bar data
"""
# Implement your trading logic here
pass
Required Methods¶
calculate_signals(event)
- Called when new market data arrives
- Analyzes indicators and generates signals
- Emits SignalEvent via
self.emit_signal()
Available Methods¶
Indicator Calculation:
# Moving Averages
sma = self.calculate_sma(symbol, period=20, price_type='close')
ema = self.calculate_ema(symbol, period=12, price_type='close')
# Oscillators
rsi = self.calculate_rsi(symbol, period=14)
stoch = self.calculate_stochastic(symbol, period=14)
# Trend Indicators
macd = self.calculate_macd(symbol, fast=12, slow=26, signal=9)
adx = self.calculate_adx(symbol, period=14)
# Volatility
atr = self.calculate_atr(symbol, period=14)
bb = self.calculate_bollinger_bands(symbol, period=20, num_std=2.0)
Signal Emission:
# Generate a buy signal
self.emit_signal(
symbol='BTCUSDT',
signal_type=SignalType.LONG,
strength=0.8, # 0.0 to 1.0
# Optional metadata
entry_reason='golden_cross',
stop_loss=40000,
take_profit=50000
)
# Generate a sell signal
self.emit_signal(
symbol='BTCUSDT',
signal_type=SignalType.EXIT,
strength=1.0,
exit_reason='death_cross'
)
Data Access:
# Get latest bar
latest = self.bars.get_latest_bar(symbol)
close = latest['close']
high = latest['high']
# Get multiple bars
bars = self.bars.get_latest_bars(symbol, N=20)
closes = [bar['close'] for bar in bars]
Example Strategies¶
1. SMA Crossover Strategy¶
Classic trend-following strategy using moving average crossovers.
Logic: - BUY when fast SMA crosses above slow SMA (golden cross) - SELL when fast SMA crosses below slow SMA (death cross)
Implementation:
"""
Simple Moving Average Crossover Strategy.
"""
from backtesting.strategy import BaseStrategy
from backtesting.events import SignalType, EventType
class SMACrossoverStrategy(BaseStrategy):
"""
SMA Crossover Strategy.
Parameters:
fast_period: Fast SMA period (default: 50)
slow_period: Slow SMA period (default: 200)
"""
def __init__(self, bars, events_queue, symbol_list, fast_period=50, slow_period=200):
super().__init__(
bars=bars,
events_queue=events_queue,
symbol_list=symbol_list,
strategy_id=f"SMA_Crossover_{fast_period}_{slow_period}"
)
self.fast_period = fast_period
self.slow_period = slow_period
# Track previous values for crossover detection
self.prev_fast_sma = {symbol: None for symbol in symbol_list}
self.prev_slow_sma = {symbol: None for symbol in symbol_list}
def calculate_signals(self, event):
"""Generate signals on SMA crossover."""
if event.event_type != EventType.MARKET:
return
symbol = event.symbol
# Calculate current SMAs
fast_sma = self.calculate_sma(symbol, self.fast_period)
slow_sma = self.calculate_sma(symbol, self.slow_period)
if fast_sma is None or slow_sma is None:
return # Not enough data
# Get previous values
prev_fast = self.prev_fast_sma[symbol]
prev_slow = self.prev_slow_sma[symbol]
# Detect crossovers
if prev_fast is not None and prev_slow is not None:
# Golden cross: fast crosses above slow
if prev_fast <= prev_slow and fast_sma > slow_sma:
if not self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.LONG,
strength=1.0,
fast_sma=float(fast_sma),
slow_sma=float(slow_sma)
)
self.bought[symbol] = True
# Death cross: fast crosses below slow
elif prev_fast >= prev_slow and fast_sma < slow_sma:
if self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.EXIT,
strength=1.0
)
self.bought[symbol] = False
# Update previous values
self.prev_fast_sma[symbol] = fast_sma
self.prev_slow_sma[symbol] = slow_sma
2. RSI Mean Reversion Strategy¶
Mean reversion strategy using RSI oversold/overbought levels.
Logic: - BUY when RSI < 30 (oversold) - SELL when RSI > 70 (overbought)
Implementation:
"""
RSI Mean Reversion Strategy.
"""
from backtesting.strategy import BaseStrategy
from backtesting.events import SignalType, EventType
class RSIMeanReversionStrategy(BaseStrategy):
"""
RSI Mean Reversion Strategy.
Parameters:
rsi_period: RSI calculation period (default: 14)
oversold_level: RSI oversold threshold (default: 30)
overbought_level: RSI overbought threshold (default: 70)
"""
def __init__(
self,
bars,
events_queue,
symbol_list,
rsi_period=14,
oversold_level=30,
overbought_level=70
):
super().__init__(
bars=bars,
events_queue=events_queue,
symbol_list=symbol_list,
strategy_id=f"RSI_MeanReversion_{rsi_period}"
)
self.rsi_period = rsi_period
self.oversold_level = oversold_level
self.overbought_level = overbought_level
def calculate_signals(self, event):
"""Generate signals on RSI extremes."""
if event.event_type != EventType.MARKET:
return
symbol = event.symbol
# Calculate RSI
rsi = self.calculate_rsi(symbol, self.rsi_period)
if rsi is None:
return # Not enough data
# Store in cache
self.indicators[symbol]['rsi'] = rsi
# Generate signals
if float(rsi) < self.oversold_level:
if not self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.LONG,
strength=1.0,
rsi=float(rsi),
entry_reason='oversold'
)
self.bought[symbol] = True
elif float(rsi) > self.overbought_level:
if self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.EXIT,
strength=1.0,
rsi=float(rsi),
exit_reason='overbought'
)
self.bought[symbol] = False
3. MACD Trend Following Strategy¶
Trend-following strategy using MACD crossovers and histogram.
Logic: - BUY when MACD line crosses above signal line - SELL when MACD line crosses below signal line - Optional: Filter with histogram strength
Implementation:
"""
MACD Trend Following Strategy.
"""
from backtesting.strategy import BaseStrategy
from backtesting.events import SignalType, EventType
class MACDStrategy(BaseStrategy):
"""
MACD Trend Following Strategy.
Parameters:
fast_period: Fast EMA period (default: 12)
slow_period: Slow EMA period (default: 26)
signal_period: Signal line period (default: 9)
use_histogram: Filter signals by histogram (default: True)
"""
def __init__(
self,
bars,
events_queue,
symbol_list,
fast_period=12,
slow_period=26,
signal_period=9,
use_histogram=True
):
super().__init__(
bars=bars,
events_queue=events_queue,
symbol_list=symbol_list,
strategy_id="MACD_Trend"
)
self.fast_period = fast_period
self.slow_period = slow_period
self.signal_period = signal_period
self.use_histogram = use_histogram
# Track previous MACD values
self.prev_macd = {symbol: None for symbol in symbol_list}
self.prev_signal = {symbol: None for symbol in symbol_list}
def calculate_signals(self, event):
"""Generate signals on MACD crossover."""
if event.event_type != EventType.MARKET:
return
symbol = event.symbol
# Calculate MACD
macd_data = self.calculate_macd(
symbol,
self.fast_period,
self.slow_period,
self.signal_period
)
if macd_data is None:
return
macd = macd_data['macd']
signal = macd_data['signal']
histogram = macd_data['histogram']
# Get previous values
prev_macd = self.prev_macd[symbol]
prev_signal = self.prev_signal[symbol]
# Detect crossovers
if prev_macd is not None and prev_signal is not None:
# Bullish crossover: MACD crosses above signal
if prev_macd <= prev_signal and macd > signal:
# Optional: Confirm with histogram
if not self.use_histogram or histogram > 0:
if not self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.LONG,
strength=abs(float(histogram)),
macd=float(macd),
signal=float(signal),
histogram=float(histogram)
)
self.bought[symbol] = True
# Bearish crossover: MACD crosses below signal
elif prev_macd >= prev_signal and macd < signal:
if self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.EXIT,
strength=abs(float(histogram)),
macd=float(macd),
signal=float(signal)
)
self.bought[symbol] = False
# Update previous values
self.prev_macd[symbol] = macd
self.prev_signal[symbol] = signal
4. Bollinger Bands Breakout Strategy¶
Volatility breakout strategy using Bollinger Bands.
Logic: - BUY when price breaks above upper band (breakout) - SELL when price breaks below lower band or crosses back to middle
Implementation:
"""
Bollinger Bands Breakout Strategy.
"""
from backtesting.strategy import BaseStrategy
from backtesting.events import SignalType, EventType
class BollingerBandsStrategy(BaseStrategy):
"""
Bollinger Bands Breakout Strategy.
Parameters:
period: BB period (default: 20)
num_std: Number of standard deviations (default: 2.0)
"""
def __init__(self, bars, events_queue, symbol_list, period=20, num_std=2.0):
super().__init__(
bars=bars,
events_queue=events_queue,
symbol_list=symbol_list,
strategy_id=f"BB_Breakout_{period}_{num_std}"
)
self.period = period
self.num_std = num_std
def calculate_signals(self, event):
"""Generate signals on Bollinger Band breakouts."""
if event.event_type != EventType.MARKET:
return
symbol = event.symbol
# Calculate Bollinger Bands
bb = self.calculate_bollinger_bands(symbol, self.period, self.num_std)
if bb is None:
return
# Get current price
latest = self.bars.get_latest_bar(symbol)
close = latest['close']
upper = bb['upper']
middle = bb['middle']
lower = bb['lower']
# Breakout above upper band
if close > upper:
if not self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.LONG,
strength=1.0,
close=float(close),
upper_band=float(upper),
entry_reason='upper_breakout'
)
self.bought[symbol] = True
# Exit on cross back to middle or lower band
elif close < middle or close < lower:
if self.bought[symbol]:
exit_reason = 'lower_band' if close < lower else 'mean_reversion'
self.emit_signal(
symbol=symbol,
signal_type=SignalType.EXIT,
strength=1.0,
close=float(close),
exit_reason=exit_reason
)
self.bought[symbol] = False
5. Multi-Indicator Confluence Strategy¶
Advanced strategy combining multiple indicators for higher confidence signals.
Logic: - BUY when multiple conditions align (e.g., RSI oversold + MACD bullish + price above SMA) - SELL when conditions reverse
Implementation:
"""
Multi-Indicator Confluence Strategy.
"""
from backtesting.strategy import BaseStrategy
from backtesting.events import SignalType, EventType
class ConfluenceStrategy(BaseStrategy):
"""
Multi-Indicator Confluence Strategy.
Combines SMA, RSI, and MACD for high-confidence signals.
"""
def __init__(self, bars, events_queue, symbol_list):
super().__init__(
bars=bars,
events_queue=events_queue,
symbol_list=symbol_list,
strategy_id="Confluence"
)
# Strategy parameters
self.sma_period = 200
self.rsi_period = 14
self.macd_fast = 12
self.macd_slow = 26
self.macd_signal = 9
# Thresholds
self.rsi_oversold = 30
self.rsi_overbought = 70
# Track previous MACD for crossover
self.prev_macd = {symbol: None for symbol in symbol_list}
self.prev_macd_signal = {symbol: None for symbol in symbol_list}
def calculate_signals(self, event):
"""Generate signals when multiple indicators align."""
if event.event_type != EventType.MARKET:
return
symbol = event.symbol
# Calculate all indicators
sma = self.calculate_sma(symbol, self.sma_period)
rsi = self.calculate_rsi(symbol, self.rsi_period)
macd_data = self.calculate_macd(
symbol,
self.macd_fast,
self.macd_slow,
self.macd_signal
)
if sma is None or rsi is None or macd_data is None:
return
# Get current price
latest = self.bars.get_latest_bar(symbol)
close = latest['close']
macd = macd_data['macd']
macd_signal = macd_data['signal']
# Get previous MACD values
prev_macd = self.prev_macd[symbol]
prev_signal = self.prev_macd_signal[symbol]
# Check bullish confluence
bullish_conditions = 0
# Condition 1: Price above SMA
if close > sma:
bullish_conditions += 1
# Condition 2: RSI oversold (potential bounce)
if float(rsi) < self.rsi_oversold:
bullish_conditions += 1
# Condition 3: MACD bullish crossover
if prev_macd is not None and prev_signal is not None:
if prev_macd <= prev_signal and macd > macd_signal:
bullish_conditions += 1
# Generate BUY signal if 2+ conditions met
if bullish_conditions >= 2 and not self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.LONG,
strength=bullish_conditions / 3.0,
conditions_met=bullish_conditions,
rsi=float(rsi),
price_vs_sma='above' if close > sma else 'below',
macd_trend='bullish'
)
self.bought[symbol] = True
# Check bearish confluence for exit
bearish_conditions = 0
# Condition 1: Price below SMA
if close < sma:
bearish_conditions += 1
# Condition 2: RSI overbought
if float(rsi) > self.rsi_overbought:
bearish_conditions += 1
# Condition 3: MACD bearish crossover
if prev_macd is not None and prev_signal is not None:
if prev_macd >= prev_signal and macd < macd_signal:
bearish_conditions += 1
# Generate EXIT signal if 2+ conditions met
if bearish_conditions >= 2 and self.bought[symbol]:
self.emit_signal(
symbol=symbol,
signal_type=SignalType.EXIT,
strength=bearish_conditions / 3.0,
conditions_met=bearish_conditions
)
self.bought[symbol] = False
# Update previous MACD values
self.prev_macd[symbol] = macd
self.prev_macd_signal[symbol] = macd_signal
Testing Strategies¶
Unit Testing¶
Test individual strategy components:
"""
Test strategy signal generation.
"""
import pytest
from decimal import Decimal
from datetime import datetime, timezone
from backtesting.strategies.sma_crossover import SMACrossoverStrategy
from backtesting.data_handler import HistoricCSVDataHandler
from queue import Queue
def test_sma_crossover_golden_cross():
"""Test golden cross signal generation."""
# Setup
symbol_list = ['BTCUSDT']
events_queue = Queue()
# Create data handler with test data
bars = create_test_data_handler(
symbol='BTCUSDT',
prices=[100, 102, 105, 108, 110] # Rising prices
)
# Create strategy
strategy = SMACrossoverStrategy(
bars=bars,
events_queue=events_queue,
symbol_list=symbol_list,
fast_period=2,
slow_period=3
)
# Feed market events
for _ in range(5):
bars.update_bars()
market_event = events_queue.get()
strategy.calculate_signals(market_event)
# Check signal generated
signal = events_queue.get()
assert signal.signal_type == SignalType.LONG
assert signal.symbol == 'BTCUSDT'
Backtesting¶
Test strategy on historical data:
"""
Backtest SMA crossover strategy.
"""
from backtesting.backtest import Backtest
from backtesting.strategies.sma_crossover import SMACrossoverStrategy
from datetime import datetime
# Configure backtest
backtest = Backtest(
symbol_list=['BTCUSDT'],
initial_capital=100000,
start_date=datetime(2023, 1, 1),
end_date=datetime(2023, 12, 31),
data_handler='DatabaseDataHandler',
strategy=SMACrossoverStrategy,
strategy_params={'fast_period': 50, 'slow_period': 200}
)
# Run backtest
results = backtest.run()
# Print results
print(f"Total Return: {results['total_return']:.2f}%")
print(f"Sharpe Ratio: {results['sharpe_ratio']:.2f}")
print(f"Max Drawdown: {results['max_drawdown']:.2f}%")
print(f"Win Rate: {results['win_rate']:.2f}%")
Best Practices¶
1. Avoid Lookahead Bias¶
# ❌ Bad - Uses future data
def calculate_signals(self, event):
# This looks ahead at future data!
all_bars = self.bars.get_all_bars(symbol)
future_high = all_bars[-1]['high'] # This is the future!
# ✅ Good - Only uses past data
def calculate_signals(self, event):
# Only access data up to current timestamp
past_bars = self.bars.get_latest_bars(symbol, N=20)
current_bar = self.bars.get_latest_bar(symbol)
2. Handle Missing Data¶
# ✅ Always check for None
def calculate_signals(self, event):
rsi = self.calculate_rsi(symbol, 14)
if rsi is None:
return # Not enough data yet
# Safe to use rsi now
if float(rsi) < 30:
self.emit_signal(...)
3. Use Proper Position Tracking¶
# ✅ Track position state
def calculate_signals(self, event):
if buy_condition:
if not self.bought[symbol]: # Check not already in position
self.emit_signal(signal_type=SignalType.LONG)
self.bought[symbol] = True
if sell_condition:
if self.bought[symbol]: # Check we have position to exit
self.emit_signal(signal_type=SignalType.EXIT)
self.bought[symbol] = False
4. Add Metadata to Signals¶
# ✅ Include useful metadata
self.emit_signal(
symbol=symbol,
signal_type=SignalType.LONG,
strength=0.8,
# Metadata for analysis
entry_reason='golden_cross',
fast_sma=float(fast_sma),
slow_sma=float(slow_sma),
rsi=float(rsi),
atr=float(atr),
# Risk management
suggested_stop_loss=float(stop_loss),
suggested_take_profit=float(take_profit)
)
5. Parameter Validation¶
def __init__(self, bars, events_queue, symbol_list, fast_period=50, slow_period=200):
# Validate parameters
if fast_period >= slow_period:
raise ValueError(f"Fast period ({fast_period}) must be < slow period ({slow_period})")
if fast_period < 2:
raise ValueError(f"Fast period must be >= 2")
super().__init__(...)
6. Logging and Debugging¶
import logging
logger = logging.getLogger(__name__)
def calculate_signals(self, event):
# Log important events
logger.debug(f"Processing {symbol} at {event.timestamp}")
rsi = self.calculate_rsi(symbol, 14)
logger.debug(f"{symbol} RSI: {rsi}")
if buy_signal:
logger.info(f"BUY signal: {symbol} RSI={rsi}")
self.emit_signal(...)
Integration with Backtesting¶
Strategy Registration¶
Register your strategy for backtesting:
# In backtesting/strategies/__init__.py
from .sma_crossover import SMACrossoverStrategy
from .rsi_mean_reversion import RSIMeanReversionStrategy
from .your_strategy import YourStrategy
AVAILABLE_STRATEGIES = {
'sma_crossover': SMACrossoverStrategy,
'rsi_mean_reversion': RSIMeanReversionStrategy,
'your_strategy': YourStrategy,
}
Running Your Strategy¶
from backtesting.backtest import Backtest
from backtesting.strategies import AVAILABLE_STRATEGIES
# Get your strategy
StrategyClass = AVAILABLE_STRATEGIES['your_strategy']
# Configure and run
backtest = Backtest(
symbol_list=['BTCUSDT', 'ETHUSDT'],
initial_capital=100000,
start_date=datetime(2023, 1, 1),
end_date=datetime(2023, 12, 31),
strategy=StrategyClass,
strategy_params={
'param1': value1,
'param2': value2
}
)
results = backtest.run()
Advanced Topics¶
Custom Indicators¶
Create custom indicators not in the built-in library:
def calculate_custom_indicator(self, symbol, period):
"""Calculate your custom indicator."""
bars = self.bars.get_latest_bars(symbol, period)
if len(bars) < period:
return None
# Your calculation logic
values = [bar['close'] for bar in bars]
result = your_calculation(values)
return Decimal(str(result))
Multi-Timeframe Strategies¶
Use multiple timeframes for confirmation:
def calculate_signals(self, event):
# Higher timeframe trend
daily_sma = self.calculate_sma(symbol, 200) # Daily data
# Lower timeframe entry
hourly_rsi = self.calculate_rsi(symbol, 14) # Hourly data
# Only buy if daily trend is up
if close > daily_sma and hourly_rsi < 30:
self.emit_signal(...)
Portfolio-Level Strategies¶
Strategies that consider portfolio composition:
def calculate_signals(self, event):
# Get portfolio state
portfolio_value = self.get_portfolio_value()
current_positions = self.get_positions()
# Limit number of concurrent positions
if len(current_positions) >= 5:
return # Portfolio full
# Size position based on portfolio
position_size = portfolio_value * 0.2 # 20% per position
Resources¶
- Backtesting Guide: See
docs/BACKTESTING_GUIDE.md - Risk Management: See
docs/RISK_MANAGEMENT.md - Example Strategies: Check
src/backtesting/strategies/ - Indicator Library: See
src/backtesting/indicators.py - TA-Lib Wrapper: See
src/backtesting/talib_wrapper.py
Support¶
For questions about strategy development:
- Review existing strategy implementations
- Check the backtesting architecture docs
- Run unit tests to understand behavior
- Open an issue with the
strategylabel