import backtrader as bt import pandas as pd class SmaCross(bt.Strategy): """ Simple moving average crossover strategy. Buy when fast SMA crosses above slow SMA. Sell when fast SMA crosses below slow SMA. User prompt: "Go long when the 10-period SMA crosses above the 100-period SMA, and exit when the 10-period SMA crosses below the 100-period SMA." Call: cerebro.addstrategy(SmaCross, pfast=10, pslow=100) """ params = dict(pfast=10, pslow=100) def __init__(self): self.sma_fast = bt.ind.SMA(period=self.p.pfast) self.sma_slow = bt.ind.SMA(period=self.p.pslow) self.crossover = bt.ind.CrossOver(self.sma_fast, self.sma_slow) def next(self): if not self.position: if self.crossover > 0: # Golden cross self.buy() elif self.crossover < 0: # Death cross self.close() class TrendMomentumLongStrategy(bt.Strategy): """ Multi-indicator strategy with trend following and momentum Long entries """ params = dict( sma_fast=5,# 20 sma_slow=30,# 50 rsi_period=14, rsi_upper=90, # 70 rsi_lower=30, atr_period=14, atr_multiplier=2.0 ) def __init__(self): # Trend indicators self.sma_fast = bt.ind.SMA(period=self.p.sma_fast) self.sma_slow = bt.ind.SMA(period=self.p.sma_slow) self.trend = self.sma_fast - self.sma_slow # Momentum indicator self.rsi = bt.ind.RSI(period=self.p.rsi_period) # Volatility for position sizing self.atr = bt.ind.ATR(period=self.p.atr_period) # Crossovers focall_liner entry signals self.crossover = bt.ind.CrossOver(self.sma_fast, self.sma_slow) def next(self): # Calculate position size based on volatility (1% risk per trade) if self.atr[0] > 0: risk_amount = self.broker.getvalue() * 0.01 size = risk_amount / (self.atr[0] * self.p.atr_multiplier) size = int(size) else: size = 100 # Default size # Entry conditions: Golden cross + RSI not overbought if not self.position: if (self.crossover > 0 and self.rsi < self.p.rsi_upper and self.trend > 0): self.buy(size=size) # Exit conditions: Death cross OR RSI overbought elif self.position: if (self.crossover < 0 or self.rsi > self.p.rsi_upper or self.trend < 0): self.close() class TrendMomentumShortStrategy(bt.Strategy): """ Multi-indicator strategy with trend following and momentum Short entries only """ params = dict( sma_fast=5, # Fast SMA period sma_slow=30, # Slow SMA period rsi_period=14, # RSI period rsi_upper=70, # Overbought threshold rsi_lower=10, # Oversold threshold atr_period=14, # ATR period for volatility-based sizing atr_multiplier=2.0 ) def __init__(self): # Trend indicators self.sma_fast = bt.ind.SMA(period=self.p.sma_fast) self.sma_slow = bt.ind.SMA(period=self.p.sma_slow) self.trend = self.sma_fast - self.sma_slow # Momentum indicator self.rsi = bt.ind.RSI(period=self.p.rsi_period) # Volatility indicator self.atr = bt.ind.ATR(period=self.p.atr_period) # Crossover signal self.crossover = bt.ind.CrossOver(self.sma_fast, self.sma_slow) def next(self): # Volatility-based position sizing (1% risk per trade) if self.atr[0] > 0: risk_amount = self.broker.getvalue() * 0.01 size = risk_amount / (self.atr[0] * self.p.atr_multiplier) size = int(size) else: size = 100 # Fallback default # Entry conditions: Death cross + RSI not oversold if not self.position: if (self.crossover < 0 and self.rsi > self.p.rsi_lower and self.trend < 0): self.sell(size=size) # Exit conditions: Golden cross OR RSI oversold elif self.position: if (self.crossover > 0 or self.rsi < self.p.rsi_lower or self.trend > 0): self.close() # Go long when the 10-period SMA crosses above the 100-period SMA, and exit when the 10-period SMA crosses below the 100-period SMA. # With logging class SmaCrossExtended(bt.Strategy): ''' Follow two moving average lines on a stock chart to decide when to buy and sell. ''' params = dict(pfast=10, pslow=100) def __init__(self): self.sma1 = bt.ind.SMA(period=self.p.pfast) self.sma2 = bt.ind.SMA(period=self.p.pslow) self.crossover = bt.ind.CrossOver(self.sma1, self.sma2) # Initialize order tracking self.order = None def log(self, txt, dt=None): """Logging function for this strategy""" dt = dt or self.datas[0].datetime.date(0) print(f'{dt.isoformat()}: {txt}') def notify_order(self, order): """Called when order status changes""" if order.status in [order.Submitted, order.Accepted]: # Order submitted/accepted - nothing to do return # Order completed if order.status in [order.Completed]: if order.isbuy(): self.log(f'BUY EXECUTED - Price: {order.executed.price:.2f}, ' f'Cost: {order.executed.value:.2f}, ' f'Comm: {order.executed.comm:.2f}, ' f'Size: {order.executed.size}') else: self.log(f'SELL EXECUTED - Price: {order.executed.price:.2f}, ' f'Cost: {order.executed.value:.2f}, ' f'Comm: {order.executed.comm:.2f}, ' f'Size: {order.executed.size}') elif order.status in [order.Canceled, order.Margin, order.Rejected]: self.log('Order Canceled/Margin/Rejected') # Reset order self.order = None def notify_trade(self, trade): """Called when a trade is closed""" if not trade.isclosed: return self.log(f'TRADE CLOSED - PnL: {trade.pnl:.2f}, PnL Net: {trade.pnlcomm:.2f}') def next(self): # Check if we have a pending order if self.order: return if not self.position: # not in market if self.crossover > 0: # Golden cross self.log('BUY SIGNAL DETECTED') self.order = self.buy() else: # in market if self.crossover < 0: # Death cross self.log('SELL SIGNAL DETECTED') self.order = self.close() import backtrader as bt class TrendMomentumLongStrategyTS(bt.Strategy): """ Multi-indicator strategy with trend following and momentum Long entries with take profit and stop loss """ params = dict( sma_fast=5, sma_slow=30, rsi_period=14, rsi_upper=90, rsi_lower=30, atr_period=14, atr_multiplier=2.0, stop_loss_pct=0.05, # 5% stop loss take_profit_pct=0.10 # 10% take profit ) def __init__(self): # Trend indicators self.sma_fast = bt.ind.SMA(period=self.p.sma_fast) self.sma_slow = bt.ind.SMA(period=self.p.sma_slow) self.trend = self.sma_fast - self.sma_slow # Momentum indicator self.rsi = bt.ind.RSI(period=self.p.rsi_period) # Volatility for position sizing self.atr = bt.ind.ATR(period=self.p.atr_period) # Crossovers for entry signals self.crossover = bt.ind.CrossOver(self.sma_fast, self.sma_slow) # Track entry price for stop loss and take profit self.entry_price = None def next(self): # Calculate position size based on volatility (1% risk per trade) if self.atr[0] > 0: risk_amount = self.broker.getvalue() * 0.01 size = risk_amount / (self.atr[0] * self.p.atr_multiplier) size = int(size) else: size = 100 # Default size # Entry conditions: Golden cross + RSI not overbought if not self.position: if (self.crossover > 0 and self.rsi < self.p.rsi_upper and self.trend > 0): self.buy(size=size) self.entry_price = self.data.close[0] # Track entry price # Exit conditions: Death cross OR RSI overbought OR stop loss/take profit elif self.position: current_price = self.data.close[0] # Calculate stop loss and take profit levels stop_loss_price = self.entry_price * (1 - self.p.stop_loss_pct) take_profit_price = self.entry_price * (1 + self.p.take_profit_pct) # Check exit conditions if (self.crossover < 0 or self.rsi > self.p.rsi_upper or self.trend < 0 or current_price <= stop_loss_price or current_price >= take_profit_price): self.close() self.entry_price = None # Reset entry price class TrendMomentumShortStrategyTS(bt.Strategy): """ Multi-indicator strategy with trend following and momentum Short entries only with take profit and stop loss """ params = dict( sma_fast=5, # Fast SMA period sma_slow=30, # Slow SMA period rsi_period=14, # RSI period rsi_upper=70, # Overbought threshold rsi_lower=10, # Oversold threshold atr_period=14, # ATR period for volatility-based sizing atr_multiplier=2.0, stop_loss_pct=0.05, # 5% stop loss take_profit_pct=0.10 # 10% take profit ) def __init__(self): # Trend indicators self.sma_fast = bt.ind.SMA(period=self.p.sma_fast) self.sma_slow = bt.ind.SMA(period=self.p.sma_slow) self.trend = self.sma_fast - self.sma_slow # Momentum indicator self.rsi = bt.ind.RSI(period=self.p.rsi_period) # Volatility indicator self.atr = bt.ind.ATR(period=self.p.atr_period) # Crossover signal self.crossover = bt.ind.CrossOver(self.sma_fast, self.sma_slow) # Entry price for stop loss and take profit self.entry_price = None def next(self): # Volatility-based position sizing (1% risk per trade) if self.atr[0] > 0: risk_amount = self.broker.getvalue() * 0.01 size = risk_amount / (self.atr[0] * self.p.atr_multiplier) size = int(size) else: size = 100 # Fallback default # Entry conditions: Death cross + RSI not oversold if not self.position: if (self.crossover < 0 and self.rsi > self.p.rsi_lower and self.trend < 0): self.sell(size=size) self.entry_price = self.data.close[0] # Track entry price # Exit conditions: Golden cross OR RSI oversold OR stop loss/take profit elif self.position: current_price = self.data.close[0] # Calculate stop loss and take profit levels # For short positions: # - Stop loss triggers when price goes UP (price > entry * (1 + stop_loss_pct)) # - Take profit triggers when price goes DOWN (price < entry * (1 - take_profit_pct)) stop_loss_price = self.entry_price * (1 + self.p.stop_loss_pct) # Stop loss above entry take_profit_price = self.entry_price * (1 - self.p.take_profit_pct) # Take profit below entry if (self.crossover > 0 or # Golden cross self.rsi < self.p.rsi_lower or # RSI oversold self.trend > 0 or # Trend turned positive current_price >= stop_loss_price or # Stop loss hit (price went up) current_price <= take_profit_price): # Take profit hit (price went down) self.close() self.entry_price = None # Reset entry price class ScalpingBB(bt.Strategy): """ Scalping strategy using Bollinger Bands with RSI during high volatility periods. Only trades during specific time windows for 1-minute. """ params = dict( bb_period=20, bb_dev=2.0, rsi_period=14, rsi_oversold=30, rsi_overbought=70, start_hour=9, # 9:00 AM end_hour=16, # 4:00 PM position_size=100 ) def __init__(self): # Bollinger Bands indicator self.bb = bt.ind.BollingerBands( period=self.p.bb_period, devfactor=self.p.bb_dev ) # RSI indicator for confirmation self.rsi = bt.ind.RSI( period=self.p.rsi_period ) # Track current time self.current_time = None def is_trading_hours(self): """Check if current time is within trading hours""" if self.current_time is None: return False hour = self.current_time.hour minute = self.current_time.minute # Check if within 9:00 AM to 4:00 PM if hour < self.p.start_hour or hour >= self.p.end_hour: return False return True def next(self): # Get current datetime self.current_time = self.data.datetime.datetime() # Only trade during specified hours if not self.is_trading_hours(): if self.position: self.close() return # Check for buy signal (price touches lower band, RSI oversold) if self.data.close[0] <= self.bb.lines.bot[0] and self.rsi[0] <= self.p.rsi_oversold: if not self.position: self.buy(size=self.p.position_size) # Check for sell signal (price touches upper band, RSI overbought) elif self.data.close[0] >= self.bb.lines.top[0] and self.rsi[0] >= self.p.rsi_overbought: if self.position: self.sell(size=self.p.position_size) # Exit if price returns to middle band elif self.position: if abs(self.data.close[0] - self.bb.lines.mid[0]) < (self.bb.lines.top[0] - self.bb.lines.mid[0]) * 0.3: self.close() class TripleMACross(bt.Strategy): """ Triple moving average crossover strategy with 10% position sizing. - Entry: Fast SMA (5) crosses above Medium SMA (10) and Medium SMA is above Slow SMA (20) - Exit: Fast SMA crosses below Medium SMA OR Medium SMA crosses below Slow SMA - Position sizing: 10% of portfolio per trade """ params = ( ('fast_period', 5), ('medium_period', 10), ('slow_period', 20), ) def __init__(self): # Three moving averages self.sma_fast = bt.indicators.SMA(period=self.p.fast_period) self.sma_medium = bt.indicators.SMA(period=self.p.medium_period) self.sma_slow = bt.indicators.SMA(period=self.p.slow_period) # Crossover indicators self.cross_fast_medium = bt.indicators.CrossOver(self.sma_fast, self.sma_medium) self.cross_medium_slow = bt.indicators.CrossOver(self.sma_medium, self.sma_slow) # Track position for conditional logic self.position_open = False def next(self): # Entry condition: Fast crosses above Medium AND Medium > Slow (no existing position) if not self.position: if self.cross_fast_medium > 0 and self.sma_medium[0] > self.sma_slow[0]: self.buy(size=self.get_target_size()) # Use dynamic sizing self.position_open = True # Exit conditions: Fast crosses below Medium OR Medium crosses below Slow elif self.position_open: if self.cross_fast_medium < 0 or self.cross_medium_slow < 0: self.close() self.position_open = False def get_target_size(self): """Calculate 10% of current portfolio value""" return int((self.broker.getvalue() * 0.90) / self.data.close[0])