Quantile Channel Strategy with Optimization and Comprehensive Analysis

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import dateutil.relativedelta as rd
import warnings
# %matplotlib inline # This line is for Jupyter/IPython and should not be in the article code
warnings.filterwarnings("ignore")

# ------------------------------------------------------------------
# 1. Simplified Quantile Channel Strategy
# ------------------------------------------------------------------
class QuantileChannelStrategy(bt.Strategy):
    params = (
        ('lookback_period', 60),       # Lookback for channel estimation
        ('upper_quantile', 0.8),       # Upper channel quantile (80th percentile)
        ('lower_quantile', 0.2),       # Lower channel quantile (20th percentile)
        ('breakout_threshold', 1.02),  # Breakout confirmation (2% above/below)
        ('stop_loss_pct', 0.08),       # 8% stop loss
        ('rebalance_period', 5),       # Rebalance every 5 days (less frequent)
        ('min_channel_width', 0.02),   # Minimum 2% channel width
        ('order_percentage', 0.95),    # Position sizing
        ('printlog', False),
    )
    
    def __init__(self):
        # Use rolling windows for efficiency
        self.price_window = []
        
        # Pre-calculate quantile indices for faster lookup
        self.upper_idx = int(self.params.lookback_period * self.params.upper_quantile)
        self.lower_idx = int(self.params.lookback_period * self.params.lower_quantile)
        self.median_idx = int(self.params.lookback_period * 0.5)
        
        # Channel levels
        self.upper_channel = 0
        self.lower_channel = 0
        self.trend_line = 0
        
        # Trading variables
        self.rebalance_counter = 0
        self.stop_price = 0
        self.last_calculation_bar = -999
        self.order = None
        
        # Track strategy statistics
        self.trade_count = 0
        self.breakout_entries = 0
        self.stop_loss_exits = 0
        self.profitable_trades = 0
        self.upper_breakouts = 0
        self.lower_breakouts = 0
        self.mean_reversion_exits = 0

    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 calculate_channels_fast(self):
        """Fast channel calculation using numpy quantiles only"""
        if len(self.price_window) < self.params.lookback_period:
            return False
            
        # Simple quantile calculation - much faster than regression
        prices = np.array(self.price_window)
        
        # Calculate basic quantiles
        self.upper_channel = np.quantile(prices, self.params.upper_quantile)
        self.lower_channel = np.quantile(prices, self.params.lower_quantile)
        self.trend_line = np.quantile(prices, 0.5)  # Median
        
        # Ensure minimum channel width
        channel_width = (self.upper_channel - self.lower_channel) / self.trend_line
        if channel_width < self.params.min_channel_width:
            half_width = self.trend_line * self.params.min_channel_width / 2
            self.upper_channel = self.trend_line + half_width
            self.lower_channel = self.trend_line - half_width
            
        return True
    
    def detect_breakout_simple(self, current_price):
        """Simplified breakout detection"""
        if current_price > self.upper_channel * self.params.breakout_threshold:
            return 1  # Upper breakout
        elif current_price < self.lower_channel / self.params.breakout_threshold:
            return -1  # Lower breakout
        else:
            return 0  # No breakout

    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: Price {order.executed.price:.2f}')
                self.stop_price = order.executed.price * (1.0 - self.params.stop_loss_pct)
                self.trade_count += 1
                self.breakout_entries += 1
            elif order.issell():
                self.log(f'SELL EXECUTED: Price {order.executed.price:.2f}')
                self.stop_price = order.executed.price * (1.0 + self.params.stop_loss_pct)
                self.trade_count += 1
                self.breakout_entries += 1
                
        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
            
        if trade.pnl > 0:
            self.profitable_trades += 1
            
        self.log(f'TRADE CLOSED: PnL {trade.pnl:.2f}')
        self.stop_price = 0
    
    def next(self):
        current_price = self.data.close[0]
        current_bar = len(self.data)
        
        # Maintain rolling window
        self.price_window.append(current_price)
        if len(self.price_window) > self.params.lookback_period:
            self.price_window.pop(0)  # Remove oldest price
        
        # Check if we have an open order
        if self.order:
            return
        
        # Only recalculate channels periodically to save computation
        self.rebalance_counter += 1
        if (self.rebalance_counter >= self.params.rebalance_period or 
            current_bar - self.last_calculation_bar > self.params.rebalance_period):
            
            if not self.calculate_channels_fast():
                return  # Not enough data
                
            self.last_calculation_bar = current_bar
            self.rebalance_counter = 0
        
        # Check stop loss first (always active)
        if self.position.size > 0 and current_price <= self.stop_price:
            self.log(f'STOP LOSS LONG: Price {current_price:.2f} <= Stop {self.stop_price:.2f}')
            self.order = self.close()
            self.stop_loss_exits += 1
            return
        elif self.position.size < 0 and current_price >= self.stop_price:
            self.log(f'STOP LOSS SHORT: Price {current_price:.2f} >= Stop {self.stop_price:.2f}')
            self.order = self.close()
            self.stop_loss_exits += 1
            return
        
        # Only trade on rebalance periods
        if self.rebalance_counter != 0:
            return
            
        # Detect breakout
        breakout = self.detect_breakout_simple(current_price)
        
        # Current position
        current_pos = 0
        if self.position.size > 0:
            current_pos = 1
        elif self.position.size < 0:
            current_pos = -1
        
        # Trading logic
        if breakout != 0:
            # Close opposing position
            if current_pos != 0 and current_pos != breakout:
                self.order = self.close()
                return
            
            # Open new position
            if current_pos == 0:
                cash = self.broker.get_cash()
                size = (cash * self.params.order_percentage) / current_price
                
                if breakout == 1:  # Long breakout
                    self.log(f'UPPER BREAKOUT: Price {current_price:.2f} > Channel {self.upper_channel:.2f}')
                    self.order = self.buy(size=size)
                    self.upper_breakouts += 1
                elif breakout == -1:  # Short breakout
                    self.log(f'LOWER BREAKOUT: Price {current_price:.2f} < Channel {self.lower_channel:.2f}')
                    self.order = self.sell(size=size)
                    self.lower_breakouts += 1
        
        # Mean reversion exit
        elif self.position.size != 0:
            # Close if price returns to middle of channel
            if abs(current_price - self.trend_line) / self.trend_line < 0.01:  # Within 1% of median
                self.log(f'MEAN REVERSION EXIT: Price {current_price:.2f} near trend {self.trend_line:.2f}')
                self.order = self.close()
                self.mean_reversion_exits += 1

# ------------------------------------------------------------------
# 2. Optimization Function
# ------------------------------------------------------------------
def optimize_quantile_channel_parameters():
    """Run optimization to find best parameters with diagnostics"""
    print("="*60)
    print("QUANTILE 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)

    print(f"Data shape: {df.shape}")
    print(f"Date range: {df.index[0]} to {df.index[-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')
    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')

    print("Testing parameter combinations...")
    # Parameter ranges for quantile channel strategy
    cerebro.optstrategy(
        SimplifiedQuantileChannelStrategy,
        lookback_period=[30, 60, 90],                  # 3 values - Lookback period
        upper_quantile=[0.75, 0.8, 0.85, 0.9],         # 4 values - Upper quantile
        lower_quantile=[0.1, 0.15, 0.2, 0.25],         # 4 values - Lower quantile
        breakout_threshold=[1.015, 1.02, 1.03],        # 3 values - Breakout threshold
        stop_loss_pct=[0.05, 0.08, 0.10],              # 3 values - Stop loss %
        rebalance_period=[3, 5, 7]                     # 3 values - Rebalance frequency
    )
    # Total: 3 × 4 × 4 × 3 × 3 × 3 = 1,296 combinations

    stratruns = cerebro.run()
    print(f"Optimization complete! Tested {len(stratruns)} combinations.")

    # Collect and analyze results with detailed diagnostics
    results = []
    valid_count = 0
    no_trades_count = 0
    invalid_sharpe_count = 0
    
    for i, run in enumerate(stratruns):
        strategy = run[0]
        sharpe_analysis = strategy.analyzers.sharpe.get_analysis()
        returns_analysis = strategy.analyzers.returns.get_analysis()
        trades_analysis = strategy.analyzers.trades.get_analysis()
        
        rtot = returns_analysis.get('rtot', 0.0)
        final_value = start_cash * (1 + rtot)
        sharpe_ratio = sharpe_analysis.get('sharperatio', None)
        total_trades = trades_analysis.get('total', {}).get('total', 0)
        
        # Diagnostic information
        has_trades = total_trades > 0
        has_valid_sharpe = sharpe_ratio is not None and not np.isnan(sharpe_ratio)
        
        if not has_trades:
            no_trades_count += 1
        if not has_valid_sharpe:
            invalid_sharpe_count += 1
            sharpe_ratio = -999.0
        
        # Safe attribute access
        trade_count = getattr(strategy, 'trade_count', 0)
        breakout_entries = getattr(strategy, 'breakout_entries', 0)
        stop_loss_exits = getattr(strategy, 'stop_loss_exits', 0)
        profitable_trades = getattr(strategy, 'profitable_trades', 0)
        upper_breakouts = getattr(strategy, 'upper_breakouts', 0)
        lower_breakouts = getattr(strategy, 'lower_breakouts', 0)
        mean_reversion_exits = getattr(strategy, 'mean_reversion_exits', 0)
        
        result = {
            'combination_id': i + 1,
            'sharpe_ratio': sharpe_ratio,
            'final_value': final_value,
            'return_pct': rtot * 100,
            'total_analyzer_trades': total_trades,
            'lookback_period': strategy.p.lookback_period,
            'upper_quantile': strategy.p.upper_quantile,
            'lower_quantile': strategy.p.lower_quantile,
            'breakout_threshold': strategy.p.breakout_threshold,
            'stop_loss_pct': strategy.p.stop_loss_pct,
            'rebalance_period': strategy.p.rebalance_period,
            'trade_count': trade_count,
            'breakout_entries': breakout_entries,
            'stop_loss_exits': stop_loss_exits,
            'profitable_trades': profitable_trades,
            'upper_breakouts': upper_breakouts,
            'lower_breakouts': lower_breakouts,
            'mean_reversion_exits': mean_reversion_exits,
            'has_trades': has_trades,
            'has_valid_sharpe': has_valid_sharpe,
        }
        
        results.append(result)
        
        if has_trades and has_valid_sharpe:
            valid_count += 1

    # Print diagnostics
    print(f"\n{'='*60}")
    print("OPTIMIZATION DIAGNOSTICS")
    print(f"{'='*60}")
    print(f"Total combinations tested: {len(results)}")
    print(f"Combinations with trades: {len(results) - no_trades_count}")
    print(f"Combinations with no trades: {no_trades_count}")
    print(f"Combinations with invalid Sharpe: {invalid_sharpe_count}")
    print(f"Valid combinations: {valid_count}")
    
    # Show some examples of each category
    print(f"\n--- SAMPLE RESULTS ---")
    for result in results[:5]:  # Show first 5 combinations
        sharpe_display = f"{result['sharpe_ratio']:.3f}" if result['sharpe_ratio'] != -999.0 else "Invalid"
        print(f"Combination {result['combination_id']}: "
              f"Lookback({result['lookback_period']}) Upper({result['upper_quantile']:.2f}) "
              f"Lower({result['lower_quantile']:.2f}) Breakout({result['breakout_threshold']:.3f}) "
              f"Stop({result['stop_loss_pct']:.1%}) Rebal({result['rebalance_period']}) -> "
              f"Trades: {result['total_analyzer_trades']}, "
              f"Return: {result['return_pct']:.1f}%, "
              f"Sharpe: {sharpe_display}")
    
    # Try to find any valid results
    valid_results = [r for r in results if r['has_trades'] and r['has_valid_sharpe']]
    
    if not valid_results:
        print(f"\n{'='*60}")
        print("NO VALID RESULTS FOUND - RUNNING SINGLE STRATEGY TEST")
        print(f"{'='*60}")
        
        # Test a single strategy with logging enabled to see what's happening
        test_params = {
            'lookback_period': 60,
            'upper_quantile': 0.8,
            'lower_quantile': 0.2,
            'breakout_threshold': 1.02,
            'stop_loss_pct': 0.08,
            'rebalance_period': 5,
            'printlog': True  # Enable logging
        }
        
        print(f"Testing single strategy with parameters: {test_params}")
        
        cerebro_test = bt.Cerebro()
        cerebro_test.adddata(bt.feeds.PandasData(dataname=df))
        cerebro_test.addstrategy(SimplifiedQuantileChannelStrategy, **test_params)
        cerebro_test.broker.setcash(10000)
        cerebro_test.broker.setcommission(commission=0.001)
        cerebro_test.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
        
        print("Running test strategy...")
        test_result = cerebro_test.run()
        test_strategy = test_result[0]
        
        trades_test = test_strategy.analyzers.trades.get_analysis()
        total_trades_test = trades_test.get('total', {}).get('total', 0)
        
        print(f"Test strategy results:")
        print(f"  Total trades: {total_trades_test}")
        print(f"  Strategy trade count: {getattr(test_strategy, 'trade_count', 0)}")
        print(f"  Breakout entries: {getattr(test_strategy, 'breakout_entries', 0)}")
        print(f"  Upper breakouts: {getattr(test_strategy, 'upper_breakouts', 0)}")
        print(f"  Lower breakouts: {getattr(test_strategy, 'lower_breakouts', 0)}")
        print(f"  Final value: ${cerebro_test.broker.getvalue():.2f}")
        
        if total_trades_test == 0:
            print(f"\n*** STRATEGY IS NOT GENERATING ANY TRADES ***")
            print(f"Possible issues:")
            print(f"1. Lookback period too long - not enough data for setup")
            print(f"2. Quantile levels too extreme")
            print(f"3. Breakout threshold too high")
            print(f"4. Data period doesn't contain suitable breakout conditions")
            
            # Let's check the data characteristics for channel strategy
            print(f"\nDATA ANALYSIS:")
            print(f"Price range: ${df['Close'].min():.2f} - ${df['Close'].max():.2f}")
            print(f"Price change: {((df['Close'].iloc[-1] / df['Close'].iloc[0]) - 1) * 100:.1f}%")
            
            # Calculate sample quantiles to see channel behavior
            recent_prices = df['Close'].tail(60).values
            upper_80 = np.quantile(recent_prices, 0.8)
            lower_20 = np.quantile(recent_prices, 0.2)
            median = np.quantile(recent_prices, 0.5)
            current_price = df['Close'].iloc[-1]
            
            print(f"Recent 60-day quantiles:")
            print(f"  Upper 80%: ${upper_80:.2f}")
            print(f"  Median:    ${median:.2f}")
            print(f"  Lower 20%: ${lower_20:.2f}")
            print(f"  Current:   ${current_price:.2f}")
            print(f"  Channel width: {((upper_80 - lower_20) / median * 100):.1f}%")
            
            # Check how often breakouts occur
            upper_breakouts = (df['Close'] > upper_80 * 1.02).sum()
            lower_breakouts = (df['Close'] < lower_20 / 1.02).sum()
            
            print(f"Potential breakouts in data:")
            print(f"  Upper breakouts: {upper_breakouts}")
            print(f"  Lower breakouts: {lower_breakouts}")
            
            if upper_breakouts + lower_breakouts < 10:
                print("*** Very few potential breakouts - this may explain lack of trades ***")
            
        return None
        
    # If we have valid results, continue with normal processing
    results_sorted = sorted(valid_results, key=lambda x: x['sharpe_ratio'], reverse=True)
    
    print(f"\n{'='*140}")
    print("TOP 10 PARAMETER COMBINATIONS BY SHARPE RATIO")
    print(f"{'='*140}")
    print("Rank | Sharpe | Return% |    Value    | Lookback | Upper | Lower | Breakout | Stop% | Rebal | Trades | Upper BO | Lower BO | Win%")
    print("-" * 140)
    
    for i, result in enumerate(results_sorted[:10]):
        win_rate = (result['profitable_trades'] / max(1, result['trade_count'])) * 100
        print(f"{i+1:4d} | {result['sharpe_ratio']:5.2f} | {result['return_pct']:6.1f}% | "
              f"${result['final_value']:8,.0f} | {result['lookback_period']:8d} | {result['upper_quantile']:5.2f} | "
              f"{result['lower_quantile']:5.2f} | {result['breakout_threshold']:8.3f} | {result['stop_loss_pct']:4.1%} | "
              f"{result['rebalance_period']:5d} | {result['total_analyzer_trades']:6d} | {result['upper_breakouts']:8d} | "
              f"{result['lower_breakouts']:8d} | {win_rate:4.1f}%")
    
    best_params = results_sorted[0]
    print(f"\n{'='*60}")
    print("BEST PARAMETERS FOUND:")
    print(f"{'='*60}")
    print(f"Lookback Period: {best_params['lookback_period']}")
    print(f"Upper Quantile: {best_params['upper_quantile']:.2f}")
    print(f"Lower Quantile: {best_params['lower_quantile']:.2f}")
    print(f"Breakout Threshold: {best_params['breakout_threshold']:.3f}")
    print(f"Stop Loss: {best_params['stop_loss_pct']:.1%}")
    print(f"Rebalance Period: {best_params['rebalance_period']}")
    print(f"Sharpe Ratio: {best_params['sharpe_ratio']:.3f}")
    print(f"Total Return: {best_params['return_pct']:.1f}%")
    print(f"Total Trades: {best_params['total_analyzer_trades']}")
    print(f"Upper Breakouts: {best_params['upper_breakouts']}")
    print(f"Lower Breakouts: {best_params['lower_breakouts']}")
    print(f"Mean Reversion Exits: {best_params['mean_reversion_exits']}")
    win_rate = (best_params['profitable_trades'] / max(1, best_params['trade_count'])) * 100
    print(f"Win Rate: {win_rate:.1f}%")
    
    return best_params

# ------------------------------------------------------------------
# 3. Rolling Backtest Function
# ------------------------------------------------------------------
def run_rolling_backtest(ticker, start, end, window_months, strategy_params=None):
    """Rolling backtest function for Quantile Channel strategy"""
    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(SimplifiedQuantileChannelStrategy, **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()
        result = cerebro.run()
        final_val = cerebro.broker.getvalue()
        strategy_ret = (final_val - start_val) / start_val * 100
        
        # Get strategy statistics
        strategy_instance = result[0]
        trades = getattr(strategy_instance, 'trade_count', 0)
        breakouts = getattr(strategy_instance, 'breakout_entries', 0)
        stops = getattr(strategy_instance, 'stop_loss_exits', 0)
        profitable = getattr(strategy_instance, 'profitable_trades', 0)
        upper_breakouts = getattr(strategy_instance, 'upper_breakouts', 0)
        lower_breakouts = getattr(strategy_instance, 'lower_breakouts', 0)
        mean_reversions = getattr(strategy_instance, 'mean_reversion_exits', 0)
        
        all_results.append({
            'start': current_start.date(),
            'end': current_end.date(),
            'return_pct': strategy_ret,
            'benchmark_pct': benchmark_ret,
            'final_value': final_val,
            'trades': trades,
            'breakout_entries': breakouts,
            'stop_exits': stops,
            'profitable_trades': profitable,
            'upper_breakouts': upper_breakouts,
            'lower_breakouts': lower_breakouts,
            'mean_reversion_exits': mean_reversions,
        })
        
        print(f"Strategy Return: {strategy_ret:.2f}% | Buy & Hold Return: {benchmark_ret:.2f}% | Trades: {trades}")
        current_start += rd.relativedelta(months=window_months)
    
    return pd.DataFrame(all_results)