Keltner Channel Breakout Strategy with Optimization and Rolling Backtest Analysis

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import dateutil.relativedelta as rd
import warnings

warnings.filterwarnings("ignore")

# ------------------------------------------------------------------
# 1. Keltner Channel Components
# ------------------------------------------------------------------
class KeltnerChannel(bt.Indicator):
    """Keltner Channel indicator with EMA centerline and ATR-based bands"""
    lines = ('mid', 'top', 'bot')
    params = (
        ('ema_period', 30),
        ('atr_period', 14), 
        ('atr_multiplier', 1.0),
    )

    def __init__(self):
        self.lines.mid = bt.indicators.EMA(self.data.close, period=self.params.ema_period)
        atr = bt.indicators.ATR(self.data, period=self.params.atr_period)
        self.lines.top = self.lines.mid + (atr * self.params.atr_multiplier)
        self.lines.bot = self.lines.mid - (atr * self.params.atr_multiplier)

class KeltnerBreakoutStrategy(bt.Strategy):
    """Keltner Channel Breakout Strategy"""
    params = (
        ('ema_period', 30),
        ('atr_period', 7),
        ('atr_multiplier', 1.0),
        ('printlog', False),
    )

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

    def __init__(self):
        self.dataclose = self.datas[0].close
        self.dataopen = self.datas[0].open
        
        self.keltner = KeltnerChannel(
            self.data,
            ema_period=self.params.ema_period,
            atr_period=self.params.atr_period,
            atr_multiplier=self.params.atr_multiplier
        )
        
        self.order = None

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'BUY EXECUTED: {order.executed.price:.2f}')
            elif order.issell():
                self.log(f'SELL EXECUTED: {order.executed.price:.2f}')
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('Order Canceled/Margin/Rejected')
        self.order = None

    def notify_trade(self, trade):
        if not trade.isclosed:
            return
        self.log(f'TRADE PROFIT: {trade.pnl:.2f}')

    def next(self):
        if len(self.data) < max(self.params.ema_period, self.params.atr_period):
            return
        if self.order:
            return
        if len(self.data) < 2:
            return
            
        prev_close = self.dataclose[-1]
        prev_upper = self.keltner.top[-1]
        prev_lower = self.keltner.bot[-1]
        current_ema = self.keltner.mid[0]
        current_close = self.dataclose[0]

        if not self.position:
            # Long entry: Previous close > Previous upper band
            if prev_close > prev_upper:
                self.log(f'BUY CREATE: Breakout above {prev_upper:.2f}')
                self.order = self.buy()
            # Short entry: Previous close < Previous lower band  
            elif prev_close < prev_lower:
                self.log(f'SELL CREATE: Breakout below {prev_lower:.2f}')
                self.order = self.sell()
        else:
            # Exit conditions based on EMA
            if self.position.size > 0:  # Long position
                if current_close < current_ema:
                    self.log(f'CLOSE LONG: {current_close:.2f} < EMA {current_ema:.2f}')
                    self.order = self.close()
            elif self.position.size < 0:  # Short position
                if current_close > current_ema:
                    self.log(f'CLOSE SHORT: {current_close:.2f} > EMA {current_ema:.2f}')
                    self.order = self.close()

# ------------------------------------------------------------------
# 2. Simple Optimization Function
# ------------------------------------------------------------------
def optimize_keltner_parameters():
    """Run optimization to find best parameters"""
    print("="*60)
    print("KELTNER CHANNEL STRATEGY OPTIMIZATION")
    print("="*60)
    
    # Fetch data for optimization
    print("Fetching data for optimization...")
    df = yf.download('BTC-USD', start='2020-01-01', end='2025-01-01', auto_adjust=False, progress=False)
    if isinstance(df.columns, pd.MultiIndex):
        df = df.droplevel(1, axis=1)

    # Set up optimization
    cerebro = bt.Cerebro()
    data = bt.feeds.PandasData(dataname=df)
    cerebro.adddata(data)
    
    start_cash = 10000.0
    cerebro.broker.setcash(start_cash)
    cerebro.broker.setcommission(commission=0.001)
    
    # Add analyzers
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', 
                       timeframe=bt.TimeFrame.Days, annualize=True, riskfreerate=0.0)
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')

    print("Testing parameter combinations...")
    # Parameter optimization ranges
    cerebro.optstrategy(
        KeltnerBreakoutStrategy,
        ema_period=[20, 30, 40, 50],                    # 4 values
        atr_period=[7, 14, 21],                         # 3 values  
        atr_multiplier=[0.5, 1.0, 1.5, 2.0, 2.5]       # 5 values
    )
    # Total: 60 combinations

    stratruns = cerebro.run()
    print("Optimization complete!")

    # Collect and analyze results
    results = []
    for run in stratruns:
        strategy = run[0]
        sharpe_analysis = strategy.analyzers.sharpe.get_analysis()
        returns_analysis = strategy.analyzers.returns.get_analysis()
        
        rtot = returns_analysis.get('rtot', 0.0)
        final_value = start_cash * (1 + rtot)
        sharpe_ratio = sharpe_analysis.get('sharperatio', None)
        
        if sharpe_ratio is None or np.isnan(sharpe_ratio):
            sharpe_ratio = -999.0
        
        results.append({
            'sharpe_ratio': sharpe_ratio,
            'final_value': final_value,
            'return_pct': rtot * 100,
            'ema_period': strategy.p.ema_period,
            'atr_period': strategy.p.atr_period,
            'atr_multiplier': strategy.p.atr_multiplier,
        })

    # Filter valid results and sort by Sharpe ratio
    valid_results = [r for r in results if r['sharpe_ratio'] != -999.0]
    if not valid_results:
        print("No valid results found!")
        return None
        
    results_sorted = sorted(valid_results, key=lambda x: x['sharpe_ratio'], reverse=True)
    
    print(f"\n{'='*80}")
    print("TOP 10 PARAMETER COMBINATIONS BY SHARPE RATIO")
    print(f"{'='*80}")
    print("Rank | Sharpe | Return% |   Value   | EMA | ATR | Multiplier")
    print("-" * 80)
    
    for i, result in enumerate(results_sorted[:10]):
        print(f"{i+1:4d} | {result['sharpe_ratio']:5.2f} | {result['return_pct']:6.1f}% | "
              f"${result['final_value']:8,.0f} | {result['ema_period']:3d} | {result['atr_period']:3d} | "
              f"{result['atr_multiplier']:10.1f}")
    
    best_params = results_sorted[0]
    print(f"\n{'='*60}")
    print("BEST PARAMETERS FOUND:")
    print(f"{'='*60}")
    print(f"EMA Period: {best_params['ema_period']}")
    print(f"ATR Period: {best_params['atr_period']}")
    print(f"ATR Multiplier: {best_params['atr_multiplier']:.1f}")
    print(f"Sharpe Ratio: {best_params['sharpe_ratio']:.3f}")
    print(f"Total Return: {best_params['return_pct']:.1f}%")
    print(f"Final Value: ${best_params['final_value']:,.2f}")
    
    return best_params

# ------------------------------------------------------------------
# 3. Your Rolling Backtest Function (slightly modified)
# ------------------------------------------------------------------
def run_rolling_backtest(ticker, start, end, window_months, strategy_params=None):
    """Your clean rolling backtest function"""
    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)
        if current_end > end_dt:
            break
            
        print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")
        
        # Fetch data using yfinance
        data = yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
        
        if data.empty or len(data) < 90:
            print("Not enough data for this period.")
            current_start += rd.relativedelta(months=window_months)
            continue
            
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, 1)
        
        # Calculate Buy & Hold return for the period
        start_price = data['Close'].iloc[0]
        end_price = data['Close'].iloc[-1]
        benchmark_ret = (end_price - start_price) / start_price * 100
        
        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        
        cerebro.addstrategy(KeltnerBreakoutStrategy, **strategy_params)
        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()
        strategy_ret = (final_val - start_val) / start_val * 100
        
        all_results.append({
            'start': current_start.date(),
            'end': current_end.date(),
            'return_pct': strategy_ret,
            'benchmark_pct': benchmark_ret,
            'final_value': final_val,
        })
        
        print(f"Strategy Return: {strategy_ret:.2f}% | Buy & Hold Return: {benchmark_ret:.2f}%")
        current_start += rd.relativedelta(months=window_months)
    
    return pd.DataFrame(all_results)