def get_crypto_rsi_thresholds(volatility_regime='normal'):
    """
    Adjust RSI thresholds based on market volatility
    """
    thresholds = {
        'low_volatility': {
            'overbought': 75,
            'oversold': 25,
            'description': 'Bitcoin under $30k volatility'
        },
        'normal': {
            'overbought': 70,
            'oversold': 30,
            'description': 'Typical market conditions'
        },
        'high_volatility': {
            'overbought': 65,
            'oversold': 35,
            'description': 'Crypto bear market 2022-style'
        },
        'extreme': {
            'overbought': 60,
            'oversold': 40,
            'description': 'March 2020, November 2021 crash'
        }
    }
return thresholds[volatility_regime]
Example
thresholds = get_crypto_rsi_thresholds('normal')
print(f"Overbought: {thresholds['overbought']}")
print(f"Oversold: {thresholds['oversold']}")

def optimize_rsi_period_crypto(prices, lookback_periods=[7, 9, 14, 21]):
    """
    Test different RSI periods, find optimal
    Crypto typically performs better with 7-14 period
    """
    results = {}
for period in lookback_periods:
        rsi = calculate_rsi(prices, period)
# Count oversold/overbought occurrences
        overbought_count = (rsi > 70).sum()
        oversold_count = (rsi < 30).sum()
        signal_frequency = overbought_count + oversold_count
results[period] = {
            'signal_frequency': signal_frequency,
            'overbought_count': overbought_count,
            'oversold_count': oversold_count,
            'avg_rsi': rsi.mean()
        }
return results
Example with Bitcoin prices
btc_prices = pd.Series([...])  # Bitcoin OHLC prices
optimal = optimize_rsi_period_crypto(btc_prices, lookback_periods=[7, 9, 14, 21])
for period, stats in optimal.items():
    print(f"Period {period}: {stats['signal_frequency']} signals/year")

class CryptoRSIBacktest:
    """RSI backtester optimized for 24/7 cryptocurrency trading"""
def __init__(
        self,
        prices,
        volumes,
        rsi_period=9,
        overbought=70,
        oversold=30,
        initial_capital=10000,
        position_size_pct=0.05,  # More aggressive for crypto
        leverage=1.0
    ):
        self.prices = prices
        self.volumes = volumes
        self.rsi_period = rsi_period
        self.overbought = overbought
        self.oversold = oversold
        self.initial_capital = initial_capital
        self.position_size_pct = position_size_pct
        self.leverage = leverage
self.rsi = self.calculate_rsi()
        self.trades = []
        self.equity_curve = [initial_capital]
        self.capital = initial_capital
def calculate_rsi(self):
        """Calculate RSI for crypto"""
        delta = self.prices.diff()
        gains = delta.clip(lower=0)
        losses = abs(delta.clip(upper=0))
avg_gain = gains.rolling(self.rsi_period).mean()
        avg_loss = losses.rolling(self.rsi_period).mean()
rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        return rsi.fillna(50)
def check_volume_confirmation(self, idx, min_volume_percentile=40):
        """
        Require volume confirmation for signals
        Crypto often spikes on low volume; filter them out
        """
        recent_volumes = self.volumes[max(0, idx-20):idx]
        volume_threshold = np.percentile(recent_volumes, min_volume_percentile)
return self.volumes[idx] >= volume_threshold
def calculate_dynamic_position_size(self, current_price, atr_value):
        """
        Position size scales with volatility
        High volatility (ATR) = smaller position
        """
        base_position = self.capital * self.position_size_pct / current_price
        historical_atr = np.mean(self.prices[max(0, len(self.prices)-50):].pct_change() * current_price)
        volatility_factor = historical_atr / atr_value if atr_value > 0 else 1.0
return int(base_position  volatility_factor  self.leverage)
def run(self, max_hold_hours=48, use_volume_filter=True):
        """Execute crypto RSI backtest"""
        position = None
for i in range(len(self.prices)):
            price = self.prices.iloc[i]
            rsi = self.rsi.iloc[i]
# Volume filter for entries
            has_volume = self.check_volume_confirmation(i) if use_volume_filter else True
# Exit position
            if position:
                hours_held = (i - position['entry_bar']) / 24  # Assume hourly data
                if rsi > self.overbought or hours_held >= max_hold_hours:
                    pnl = (price - position['entry_price']) * position['shares']
                    self.capital += pnl
                    self.equity_curve.append(self.capital)
self.trades.append({
                        'entry': position['entry_price'],
                        'exit': price,
                        'shares': position['shares'],
                        'pnl': pnl,
                        'hours_held': hours_held
                    })
position = None
# Entry signal
            if not position and rsi < self.oversold and has_volume:
                # Dynamic sizing based on ATR
                atr = np.std(self.prices[max(0, i-14):i])
                shares = self.calculate_dynamic_position_size(price, atr)
if shares > 0:
                    position = {
                        'entry_price': price,
                        'entry_bar': i,
                        'shares': shares,
                        'entry_rsi': rsi
                    }
return {
            'trades': self.trades,
            'final_capital': self.capital,
            'total_return': (self.capital - self.initial_capital) / self.initial_capital
        }
def metrics(self):
        """Calculate crypto-specific metrics"""
        if not self.trades:
            return {}
pnl_values = np.array([t['pnl'] for t in self.trades])
        returns = pnl_values / self.capital
return {
            'total_return': (self.capital - self.initial_capital) / self.initial_capital,
            'num_trades': len(self.trades),
            'win_rate': (pnl_values > 0).sum() / len(pnl_values),
            'profit_factor': np.sum(pnl_values[pnl_values > 0]) / abs(np.sum(pnl_values[pnl_values < 0])),
            'sharpe_ratio': np.mean(returns) / np.std(returns) * np.sqrt(365) if np.std(returns) > 0 else 0,
            'max_drawdown': self._calculate_max_drawdown(),
            'avg_hold_hours': np.mean([t['hours_held'] for t in self.trades]),
            'avg_pnl': np.mean(pnl_values)
        }
def _calculate_max_drawdown(self):
        """Calculate maximum drawdown"""
        cumulative = np.cumprod(1 + np.array([t['pnl'] / self.capital for t in self.trades]))
        running_max = np.maximum.accumulate(cumulative)
        return np.min((cumulative - running_max) / running_max) if len(cumulative) > 0 else 0

# Load Bitcoin hourly data (2024-2026)
btc_prices = pd.Series([...])  # Bitcoin hourly close prices
btc_volumes = pd.Series([...])  # Bitcoin hourly volumes
Test RSI 9 period (optimal for crypto)
backtest = CryptoRSIBacktest(
    prices=btc_prices,
    volumes=btc_volumes,
    rsi_period=9,
    overbought=70,
    oversold=30,
    initial_capital=10000,
    position_size_pct=0.05,
    leverage=1.0
)
results = backtest.run(max_hold_hours=48, use_volume_filter=True)
metrics = backtest.metrics()
print(f"Total Return: {metrics['total_return']:.1%}")
print(f"Win Rate: {metrics['win_rate']:.1%}")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f"Max Drawdown: {metrics['max_drawdown']:.2%}")
print(f"Avg Trade PnL: ${metrics['avg_pnl']:,.0f}")
print(f"Avg Hold Time: {metrics['avg_hold_hours']:.1f} hours")

def multi_timeframe_crypto_signal(rsi_1h, rsi_4h, oversold_1h=30, oversold_4h=40):
    """
    Generate signal only if both timeframes confirm
    Buy if: 1H RSI < 30 AND 4H RSI < 40
    This prevents catching falling knives
    """
    if rsi_1h < oversold_1h and rsi_4h < oversold_4h:
        return 1  # Strong buy signal
    elif rsi_1h > 70 and rsi_4h > 60:
        return -1  # Strong sell signal
    else:
        return 0  # No signal