Trend Trading with Moving Average Slope A Dual-Directional Strategy with Trailing Stops and Rolling Backtesting

import backtrader as bt
import numpy as np

class Slope(bt.Indicator):
    lines = ('slope',)
    params = dict(period=14)

    plotinfo = dict(subplot=True) # Plot in separate panel below price
    plotlines = dict(slope=dict(_name='MA Slope')) # Label for plot legend

    def __init__(self):
        # We need at least 'period' data points for slope calculation.
        # This implementation calculates the simple change over 'period' bars.
        # A more robust implementation might use numpy.polyfit for linear regression.
        
        # self.data(-N) gets value N bars ago. self.data(0) is current.
        data_prev = self.data(-self.p.period)
        delta_y = self.data(0) - data_prev

        if self.p.period > 0:
            # Simple average rate of change over the period
            self.lines.slope = delta_y / self.p.period
        else:
            # Handle period=0 case to avoid division by zero
            self.lines.slope = self.data * 0.0 # Assign zeros

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import dateutil.relativedelta as rd
import matplotlib.pyplot as plt
import seaborn as sns

# Make sure the Slope indicator class is defined or imported above this
# from slope_indicator import Slope

# Define the MA Slope Strategy (Long & Short)
class MaSlopeStrategy(bt.Strategy):
    params = (
        ('ma_period', 30),          # Period for the EMA
        ('slope_period', 10),       # Period for calculating the slope of the EMA
        ('ma_type', 'EMA'),         # Type of MA ('EMA' or 'SMA')
        ('slope_long_entry_threshold', 0.01),  # Slope must cross ABOVE this to enter long
        ('slope_short_entry_threshold', -0.01), # Slope must cross BELOW this to enter short
        ('min_prior_slope_long', -0.01),   # Slope should have been below this prior to long entry
        ('max_prior_slope_short', 0.01),   # Slope should have been above this prior to short entry
        ('trailing_stop_pct', 0.05), # 5% trailing stop
        ('order_percentage', 0.95), # Percentage of cash to use for orders
        ('ticker', 'BTC-USD'),      # Default ticker
        ('printlog', True),         # Enable/disable logging
    )

    def __init__(self):
        self.data_close = self.datas[0].close

        # Select MA type based on params
        ma_indicator = bt.indicators.EMA if self.params.ma_type == 'EMA' else bt.indicators.SMA
        self.ma = ma_indicator(self.data_close, period=self.params.ma_period)

        # Calculate the slope of the MA using our custom Slope indicator
        self.ma_slope = Slope(self.ma, period=self.params.slope_period)

        # Order tracking
        self.order = None
        
        # Trailing stop variables
        self.highest_price = None   # For long positions: tracks highest price reached in trade
        self.lowest_price = None    # For short positions: tracks lowest price reached in trade

        if self.params.printlog:
            print(f"Strategy Parameters: MA Period={self.p.ma_period}, Slope Period={self.p.slope_period}, "
                  f"MA Type={self.p.ma_type}, Long Entry Threshold={self.p.slope_long_entry_threshold}, "
                  f"Short Entry Threshold={self.p.slope_short_entry_threshold}, "
                  f"Min Prior Long={self.p.min_prior_slope_long}, Max Prior Short={self.p.max_prior_slope_short}, "
                  f"Trailing Stop Pct={self.p.trailing_stop_pct * 100:.2f}%")

    def log(self, txt, dt=None, doprint=False):
        ''' Logging function '''
        if self.params.printlog or doprint:
            dt = dt or self.datas[0].datetime.date(0)
            print(f'{dt.isoformat()} - {txt}')

    def notify_order(self, order):
        # Ignore submitted/accepted orders, wait for completion or failure
        if order.status in [order.Submitted, order.Accepted]: 
            return

        # Handle completed orders
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
                # Initialize highest price for trailing stop for new long position
                if self.position.size > 0: # Ensure it's an actual entry, not cover
                    self.highest_price = order.executed.price
                    self.lowest_price = None # Clear for long positions
            elif order.issell():
                self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
                # Initialize lowest price for trailing stop for new short position
                if self.position.size < 0: # Ensure it's an actual entry, not exit long
                    self.lowest_price = order.executed.price
                    self.highest_price = None # Clear for short positions
            self.bar_executed = len(self) # Record the bar number when the order was executed
        
        # Handle failed orders
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log(f'Order Canceled/Margin/Rejected: Status {order.getstatusname()}')
        
        # Clear order reference regardless of outcome to allow new orders
        self.order = None

    def notify_trade(self, trade):
        # Report profit/loss when a trade is closed
        if not trade.isclosed: 
            return
        self.log(f'TRADE CLOSED - GROSS: {trade.pnl:.2f}, NET: {trade.pnlcomm:.2f}')

    def next(self):
        # Check if indicators have enough data to produce valid outputs
        # min_data_needed covers MA and Slope calculation periods
        min_data_needed = max(self.p.ma_period, self.p.slope_period) 
        if len(self.data_close) < min_data_needed + 1: # +1 for previous slope comparison
            return

        # Prevent new orders if an order is already pending
        if self.order:
            return

        # Get current and previous slope values
        # [0] refers to the current bar's value, [-1] refers to the previous bar's value
        current_slope = self.ma_slope[0]
        previous_slope = self.ma_slope[-1]
        current_price = self.data_close[0]

        # No position - look for entry signals
        if not self.position:
            # --- LONG Entry Logic ---
            # Condition 1: Slope was recently flat or negative, setting up for a reversal/uptrend
            was_flat_or_down = previous_slope < self.params.min_prior_slope_long
            # Condition 2: Slope just crossed above the positive entry threshold
            crossed_up = (previous_slope <= self.params.slope_long_entry_threshold and 
                          current_slope > self.params.slope_long_entry_threshold)

            if was_flat_or_down and crossed_up:
                self.log(f'LONG SIGNAL (Slope Turn Up): Close={current_price:.2f}, Slope={current_slope:.3f}, Prev={previous_slope:.3f}')
                cash = self.broker.get_cash()
                price = current_price
                if price > 0: # Avoid division by zero
                    size = (cash * self.params.order_percentage) / price
                    self.log(f'BUY Size: {size:.6f}')
                    self.order = self.buy(size=size)
                else:
                    self.log("Cannot place BUY order: current price is zero or negative.", doprint=True)

            # --- SHORT Entry Logic ---
            # Condition 1: Slope was recently flat or positive, setting up for a reversal/downtrend
            was_flat_or_up = previous_slope > self.params.max_prior_slope_short
            # Condition 2: Slope just crossed below the negative entry threshold
            crossed_down = (previous_slope >= self.params.slope_short_entry_threshold and 
                            current_slope < self.params.slope_short_entry_threshold)

            if was_flat_or_up and crossed_down:
                self.log(f'SHORT SIGNAL (Slope Turn Down): Close={current_price:.2f}, Slope={current_slope:.3f}, Prev={previous_slope:.3f}')
                cash = self.broker.get_cash()
                price = current_price
                if price > 0: # Avoid division by zero
                    size = (cash * self.params.order_percentage) / price
                    self.log(f'SELL Size: {size:.6f}')
                    self.order = self.sell(size=size)
                else:
                    self.log("Cannot place SELL order: current price is zero or negative.", doprint=True)

        # Currently LONG - check for trailing stop exit
        elif self.position.size > 0:
            # Update highest price for trailing stop if current price is higher
            if self.highest_price is None or current_price > self.highest_price:
                self.highest_price = current_price
                
            # Calculate trailing stop price
            trailing_stop_price = self.highest_price * (1 - self.params.trailing_stop_pct)
            
            # Check if current price falls below the trailing stop
            if current_price <= trailing_stop_price:
                self.log(f'LONG TRAILING STOP HIT: Price={current_price:.2f} <= Stop={trailing_stop_price:.2f} (High={self.highest_price:.2f})')
                self.order = self.sell(size=self.position.size) # Exit long position

        # Currently SHORT - check for trailing stop exit
        elif self.position.size < 0:
            # Update lowest price for trailing stop if current price is lower
            if self.lowest_price is None or current_price < self.lowest_price:
                self.lowest_price = current_price
                
            # Calculate trailing stop price
            trailing_stop_price = self.lowest_price * (1 + self.params.trailing_stop_pct)
            
            # Check if current price rises above the trailing stop
            if current_price >= trailing_stop_price:
                self.log(f'SHORT TRAILING STOP HIT: Price={current_price:.2f} >= Stop={trailing_stop_price:.2f} (Low={self.lowest_price:.2f})')
                self.order = self.buy(size=abs(self.position.size)) # Exit short position

# ... (imports and strategy/Slope definitions as above) ...

def run_rolling_backtest(
    ticker="BTC-USD",
    start="2018-01-01",
    end="2025-12-31",
    window_months=6,
    strategy_params=None
):
    strategy_params = strategy_params or {}
    all_results = []
    start_dt = pd.to_datetime(start)
    end_dt = pd.to_datetime(end)
    current_start = start_dt

    while True:
        current_end = current_start + rd.relativedelta(months=window_months)
        # Adjust the end of the current window if it exceeds the overall end date
        if current_end > end_dt:
            current_end = end_dt
            if current_start >= current_end: # If no valid period left, break
                break

        print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")

        # Data download using yfinance, respecting the user's preference for auto_adjust=False and droplevel
        # Using the saved preference: yfinance download with auto_adjust=False and droplevel(axis=1, level=1)
        data = yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
        
        # Apply droplevel if data is a MultiIndex, as per user's preference
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, axis=1)

        # Check for sufficient data after droplevel
        # Calculate minimum data needed based on strategy parameters
        ma_period = strategy_params.get('ma_period', 30)
        slope_period = strategy_params.get('slope_period', 10)
        min_data_for_indicators = max(ma_period, ma_period + slope_period) + 1 # +1 for previous value access

        if data.empty or len(data) < min_data_for_indicators:
            print(f"Not enough data for period {current_start.date()} to {current_end.date()}. Skipping.")
            # Move to the next window. If moving to the next window makes us pass overall end, break.
            if current_end == end_dt: # If the current window already reached overall end_dt
                break
            current_start = current_end # Move to the end of the current (insufficient) period
            continue

        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(MaSlopeStrategy, **strategy_params) # Use MaSlopeStrategy
        cerebro.adddata(feed)
        cerebro.broker.setcash(100000)
        cerebro.broker.setcommission(commission=0.001)
        cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

        start_val = cerebro.broker.getvalue()
        cerebro.run()
        final_val = cerebro.broker.getvalue()
        ret = (final_val - start_val) / start_val * 100

        all_results.append({
            'start': current_start.date(),
            'end': current_end.date(),
            'return_pct': ret,
            'final_value': final_val,
        })

        print(f"Return: {ret:.2f}% | Final Value: {final_val:.2f}")
        
        # Move to the next window. If current_end already reached overall end_dt, then break.
        if current_end == end_dt:
            break
        current_start = current_end # For non-overlapping windows, next start is current end

    return pd.DataFrame(all_results)

# ... (all code as above) ...

def report_stats(df):
    returns = df['return_pct']
    stats = {
        'Mean Return %': np.mean(returns),
        'Median Return %': np.median(returns),
        'Std Dev %': np.std(returns),
        'Min Return %': np.min(returns),
        'Max Return %': np.max(returns),
        'Sharpe Ratio': np.mean(returns) / np.std(returns) if np.std(returns) > 0 else np.nan
    }
    print("\n=== ROLLING BACKTEST STATISTICS ===")
    for k, v in stats.items():
        print(f"{k}: {v:.2f}")
    return stats

def plot_return_distribution(df):
    sns.set(style="whitegrid")
    plt.figure(figsize=(10, 5))
    sns.histplot(df['return_pct'], bins=20, kde=True, color='dodgerblue')
    plt.axvline(df['return_pct'].mean(), color='black', linestyle='--', label='Mean')
    plt.title('Rolling Backtest Return Distribution')
    plt.xlabel('Return %')
    plt.ylabel('Frequency')
    plt.legend()
    plt.tight_layout()
    plt.show()

def plot_four_charts(df, rolling_sharpe_window=4):
    """
    Generates four analytical plots for rolling backtest results.
    """
    fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(12, 8)) # Adjusted figsize for clarity

    # Calculate period numbers (0, 1, 2, 3, ...)
    periods = list(range(len(df)))
    returns = df['return_pct']

    # 1. Period Returns (Top Left)
    colors = ['green' if r >= 0 else 'red' for r in returns]
    ax1.bar(periods, returns, color=colors, alpha=0.7)
    ax1.set_title('Period Returns', fontsize=14, fontweight='bold')
    ax1.set_xlabel('Period')
    ax1.set_ylabel('Return %')
    ax1.axhline(y=0, color='black', linestyle='-', alpha=0.3)
    ax1.grid(True, alpha=0.3)

    # 2. Cumulative Returns (Top Right)
    cumulative_returns = (1 + returns / 100).cumprod() * 100 - 100
    ax2.plot(periods, cumulative_returns, marker='o', linewidth=2, markersize=4, color='blue') # Smaller markers
    ax2.set_title('Cumulative Returns', fontsize=14, fontweight='bold')
    ax2.set_xlabel('Period')
    ax2.set_ylabel('Cumulative Return %')
    ax2.grid(True, alpha=0.3)

    # 3. Rolling Sharpe Ratio (Bottom Left)
    rolling_sharpe = returns.rolling(window=rolling_sharpe_window).apply(
        lambda x: x.mean() / x.std() if x.std() > 0 else np.nan, raw=False # Added raw=False for lambda
    )
    # Only plot where we have valid rolling calculations
    valid_mask = ~rolling_sharpe.isna()
    valid_periods = [i for i, valid in enumerate(valid_mask) if valid]
    valid_sharpe = rolling_sharpe[valid_mask]

    ax3.plot(valid_periods, valid_sharpe, marker='o', linewidth=2, markersize=4, color='orange') # Smaller markers
    ax3.axhline(y=0, color='red', linestyle='--', alpha=0.5)
    ax3.set_title(f'Rolling Sharpe Ratio ({rolling_sharpe_window}-period)', fontsize=14, fontweight='bold')
    ax3.set_xlabel('Period')
    ax3.set_ylabel('Sharpe Ratio')
    ax3.grid(True, alpha=0.3)

    # 4. Return Distribution (Bottom Right)
    bins = min(15, max(5, len(returns)//2))
    ax4.hist(returns, bins=bins, alpha=0.7, color='steelblue', edgecolor='black')
    mean_return = returns.mean()
    ax4.axvline(mean_return, color='red', linestyle='--', linewidth=2,
                label=f'Mean: {mean_return:.2f}%')
    ax4.set_title('Return Distribution', fontsize=14, fontweight='bold')
    ax4.set_xlabel('Return %')
    ax4.set_ylabel('Frequency')
    ax4.legend()
    ax4.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

if __name__ == '__main__':
    # Running with default parameters (BTC-USD, 6-month windows)
    # The end date is set to the current date for a more up-to-date backtest.
    df = run_rolling_backtest(end="2025-06-20") 

    print("\n=== ROLLING BACKTEST RESULTS ===")
    print(df)

    stats = report_stats(df)
    plot_four_charts(df)