bt_strategies.py

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])