import numpy as np
import pandas as pd
def calculate_crypto_position_size(
    account_size,
    entry_price,
    stop_loss_price,
    current_volatility,
    benchmark_volatility=0.02,
    base_risk=0.02
):
    """
    Adjust position sizing for crypto volatility
    Lower volatility relative to benchmark = larger position
    Higher volatility = smaller position
    """
    # Volatility adjustment factor
    vol_ratio = current_volatility / benchmark_volatility
# Scale risk inversely to volatility
    adjusted_risk = base_risk / vol_ratio
# Cap risk to prevent over-leveraging
    adjusted_risk = min(adjusted_risk, 0.05)  # Never exceed 5%
risk_amount = account_size * adjusted_risk
    stop_distance = abs(entry_price - stop_loss_price)
position_size = risk_amount / stop_distance
return {
        'position_size': position_size,
        'position_value': position_size * entry_price,
        'adjusted_risk': adjusted_risk,
        'volatility_factor': vol_ratio
    }
Example: Bitcoin with high volatility
btc_price = 45000
entry = 45000
stop = 42750  # 6% stop loss (wider for crypto)
vol_30day = 0.035  # 3.5% annualized = 0.35% daily
benchmark = 0.02
size = calculate_crypto_position_size(
    account_size=100000,
    entry_price=entry,
    stop_loss_price=stop,
    current_volatility=vol_30day,
    benchmark_volatility=benchmark,
    base_risk=0.02
)
print(f"Volatility ratio: {size['volatility_factor']:.2f}x")
print(f"Adjusted risk: {size['adjusted_risk']:.2%}")
print(f"Position size: {size['position_size']:.2f} BTC")
print(f"Position value: ${size['position_value']:,.0f}")

def crypto_position_with_leverage(
    account_size,
    entry_price,
    stop_loss_price,
    leverage=1.0,
    max_loss_percent=0.05
):
    """
    Position size with leverage constraints
    Ensures max loss doesn't exceed account * max_loss_percent
    """
    # Maximum dollar loss
    max_dollar_loss = account_size * max_loss_percent
# Stop distance
    stop_distance = abs(entry_price - stop_loss_price)
# Base position size from max loss
    base_position = max_dollar_loss / stop_distance
# Apply leverage cap: total notional exposure ≤ account × (1 + leverage)
    max_notional = account_size * (1 + leverage)
# Final position size
    position = min(base_position * leverage, max_notional / entry_price)
return {
        'position_size': position,
        'notional_exposure': position * entry_price,
        'leverage_used': (position * entry_price) / account_size,
        'max_loss': max_dollar_loss
    }
With 2x leverage on crypto
result = crypto_position_with_leverage(
    account_size=100000,
    entry_price=45000,
    stop_loss_price=40000,
    leverage=2.0,
    max_loss_percent=0.05
)
print(f"Position size: {result['position_size']:.2f}")
print(f"Notional exposure: ${result['notional_exposure']:,.0f}")
print(f"Actual leverage: {result['leverage_used']:.2f}x")

class CryptoBacktest:
    """Specialized backtesting engine for cryptocurrency strategies"""
def __init__(
        self,
        prices,
        volumes,
        signals,
        initial_capital=100000,
        leverage=1.0,
        maker_fee=0.001,
        taker_fee=0.0015,
        slippage_bps=5
    ):
        self.prices = prices
        self.volumes = volumes
        self.signals = signals
        self.capital = initial_capital
        self.leverage = leverage
        self.maker_fee = maker_fee
        self.taker_fee = taker_fee
        self.slippage_bps = slippage_bps
self.trades = []
        self.equity_curve = [initial_capital]
        self.position = None
def calculate_volatility(self, idx, lookback=20):
        """Calculate rolling volatility (percent change std dev)"""
        if idx < lookback:
            return np.std(np.diff(self.prices[:idx]) / self.prices[:idx-1])
prices_subset = self.prices[idx-lookback:idx]
        returns = np.diff(prices_subset) / prices_subset[:-1]
        return np.std(returns)
def calculate_position_size(self, entry_price, stop_price, vol_idx):
        """Crypto-specific position sizing"""
        volatility = self.calculate_volatility(vol_idx)
        benchmark_vol = 0.02
# Volatility scaling
        vol_factor = benchmark_vol / volatility
        adjusted_risk = min(0.02 * vol_factor, 0.05)
# Maximum dollar loss
        max_loss = self.capital * adjusted_risk
        stop_distance = abs(entry_price - stop_price)
return max_loss / stop_distance
def apply_slippage_and_fees(self, price, is_buy):
        """Apply realistic execution costs"""
        slippage = price * self.slippage_bps / 10000
        fee = self.taker_fee
if is_buy:
            return price + slippage, 1 + fee
        else:
            return price - slippage, 1 - fee
def run_backtest(self):
        """Execute backtest with crypto adjustments"""
        for i in range(1, len(self.signals)):
            signal = self.signals[i]
            price = self.prices[i]
# Check volume filter (ensure liquidity)
            if self.volumes[i] < np.percentile(self.volumes[max(0, i-20):i], 10):
                continue
# Exit position if signal reverses
            if self.position and signal != self.position['signal']:
                self._close_position(price, i)
                self.position = None
# Open new position
            if signal != 0 and not self.position:
                self._open_position(signal, price, i)
# Close final position
        if self.position:
            self._close_position(self.prices[-1], len(self.prices) - 1)
return self.equity_curve, self.trades
def _open_position(self, signal, price, idx):
        """Open trade with crypto-adjusted stops"""
        stop_loss = price  0.94 if signal == 1 else price  1.06  # 6% stop
position_size = self.calculate_position_size(price, stop_loss, idx)
# Apply slippage
        entry_slipped, entry_fee = self.apply_slippage_and_fees(price, signal == 1)
self.position = {
            'signal': signal,
            'entry': entry_slipped,
            'entry_fee': entry_fee,
            'stop': stop_loss,
            'size': position_size,
            'value': entry_slipped * position_size,
            'idx': idx
        }
def _close_position(self, exit_price, idx):
        """Close position with fees and slippage"""
        exit_slipped, exit_fee = self.apply_slippage_and_fees(exit_price, self.position['signal'] == 1)
# Calculate PnL
        entry_value = self.position['value'] * self.position['entry_fee']
        exit_value = exit_slipped  self.position['size']  exit_fee
if self.position['signal'] == 1:
            pnl = exit_value - entry_value
        else:
            pnl = entry_value - exit_value
self.capital += pnl
self.trades.append({
            'entry': self.position['entry'],
            'exit': exit_slipped,
            'size': self.position['size'],
            'pnl': pnl,
            'return': pnl / self.capital,
            'bars_held': idx - self.position['idx']
        })
self.equity_curve.append(self.capital)
def metrics(self):
        """Calculate crypto-specific metrics"""
        if not self.trades:
            return {}
returns = np.array([t['return'] for t in self.trades])
# Calmar ratio: return / max drawdown
        cumulative = np.cumprod(1 + returns)
        running_max = np.maximum.accumulate(cumulative)
        drawdown = (cumulative - running_max) / running_max
        max_dd = np.abs(np.min(drawdown))
total_return = (self.capital - 100000) / 100000
return {
            'total_return': total_return,
            'sharpe_ratio': np.mean(returns) / np.std(returns) * np.sqrt(365),
            'calmar_ratio': total_return / max_dd if max_dd > 0 else np.inf,
            'max_drawdown': max_dd,
            'win_rate': (returns > 0).sum() / len(returns),
            'num_trades': len(self.trades),
            'avg_holding_hours': np.mean([t['bars_held'] for t in self.trades])  # Assume 4h bars
        }