Python Scripting

1import pandas as pd2import numpy as np34def initialize(bars_data):5    """6    Called once when the strategy starts.7    bars_data is a list of dicts with keys: time, open, high, low, close, volume8    """9    df = pd.DataFrame(bars_data)10    df['time'] = pd.to_datetime(df['time'], unit='s')11    df.set_index('time', inplace=True)12    13    # Add technical indicators14    df['sma_20'] = df['close'].rolling(20).mean()15    df['sma_50'] = df['close'].rolling(50).mean()16    df['rsi'] = compute_rsi(df['close'], 14)17    df['atr'] = compute_atr(df, 14)18    19    return df

def sma(series: pd.Series, period: int) -> pd.Series:
    """Simple Moving Average"""
    return series.rolling(window=period).mean()

def ema(series: pd.Series, period: int) -> pd.Series:
    """Exponential Moving Average"""
    return series.ewm(span=period, adjust=False).mean()

1def compute_rsi(close: pd.Series, period: int = 14) -> pd.Series:2    """RSI using Wilder's smoothing method"""3    delta = close.diff()4    gain = delta.where(delta > 0, 0.0)5    loss = -delta.where(delta < 0, 0.0)6    7    # Wilder's smoothing (equivalent to EMA with alpha=1/period)8    avg_gain = gain.ewm(alpha=1/period, min_periods=period).mean()9    avg_loss = loss.ewm(alpha=1/period, min_periods=period).mean()10    11    rs = avg_gain / avg_loss12    rsi = 100 - (100 / (1 + rs))13    return rsi

1def compute_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:2    """Average True Range for volatility measurement"""3    high = df['high']4    low = df['low']5    close = df['close']6    7    tr1 = high - low                       # Current range8    tr2 = (high - close.shift(1)).abs()    # Gap up9    tr3 = (low - close.shift(1)).abs()     # Gap down10    11    true_range = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)12    return true_range.ewm(alpha=1/period, min_periods=period).mean()

1def compute_macd(close: pd.Series, fast=12, slow=26, signal=9):2    """MACD: Moving Average Convergence Divergence"""3    ema_fast = close.ewm(span=fast, adjust=False).mean()4    ema_slow = close.ewm(span=slow, adjust=False).mean()5    6    macd_line = ema_fast - ema_slow7    signal_line = macd_line.ewm(span=signal, adjust=False).mean()8    histogram = macd_line - signal_line9    10    return macd_line, signal_line, histogram

1def compute_bollinger(close: pd.Series, period=20, std_dev=2.0):2    """Bollinger Bands"""3    middle = close.rolling(period).mean()4    std = close.rolling(period).std()5    6    upper = middle + std * std_dev7    lower = middle - std * std_dev8    9    # %B: Where price is relative to bands (0=lower, 1=upper)10    percent_b = (close - lower) / (upper - lower)11    12    # Bandwidth: How wide the bands are (volatility measure)13    bandwidth = (upper - lower) / middle14    15    return upper, middle, lower, percent_b, bandwidth

1def crossover(fast: pd.Series, slow: pd.Series) -> pd.Series:2    """Returns True where fast crosses above slow"""3    return (fast > slow) & (fast.shift(1) <= slow.shift(1))45def crossunder(fast: pd.Series, slow: pd.Series) -> pd.Series:6    """Returns True where fast crosses below slow"""7    return (fast < slow) & (fast.shift(1) >= slow.shift(1))89# Usage:10buy_signals = crossover(df['sma_20'], df['sma_50'])11sell_signals = crossunder(df['sma_20'], df['sma_50'])

1def crosses_above(series: pd.Series, threshold: float) -> pd.Series:2    """Returns True where series crosses above a threshold"""3    return (series > threshold) & (series.shift(1) <= threshold)45def crosses_below(series: pd.Series, threshold: float) -> pd.Series:6    """Returns True where series crosses below a threshold"""7    return (series < threshold) & (series.shift(1) >= threshold)89# Usage:10rsi_oversold = crosses_above(df['rsi'], 30)   # RSI crosses above 3011rsi_overbought = crosses_below(df['rsi'], 70)  # RSI crosses below 70

1def detect_engulfing(df: pd.DataFrame) -> pd.Series:2    """3    Detect bullish and bearish engulfing candles.4    Returns: 1 for bullish engulfing, -1 for bearish, 0 for none.5    """6    prev_body = df['close'].shift(1) - df['open'].shift(1)7    curr_body = df['close'] - df['open']8    9    bullish = (10        (prev_body < 0) &           # Previous candle is bearish11        (curr_body > 0) &            # Current candle is bullish12        (df['open'] <= df['close'].shift(1)) &   # Opens at or below prev close13        (df['close'] >= df['open'].shift(1))      # Closes at or above prev open14    )15    16    bearish = (17        (prev_body > 0) &           # Previous candle is bullish18        (curr_body < 0) &            # Current candle is bearish19        (df['open'] >= df['close'].shift(1)) &   # Opens at or above prev close20        (df['close'] <= df['open'].shift(1))      # Closes at or below prev open21    )22    23    result = pd.Series(0, index=df.index)24    result[bullish] = 125    result[bearish] = -126    return result

1class Backtester:2    """Simple backtesting engine for strategy validation."""3    4    def __init__(self, df, config):5        self.df = df6        self.config = config7        self.capital = config['initial_capital']8        self.position = None     # {'side': 'long'/'short', 'entry': price, 'size': qty, 'stop': price}9        self.trades = []10        self.equity_curve = []11    12    def run(self, signal_func):13        """Execute backtest with the given signal function."""14        for i in range(1, len(self.df)):15            bar = self.df.iloc[i]16            prev = self.df.iloc[i-1]17            18            # Check stop loss / take profit19            if self.position:20                self._check_exit(bar)21            22            # Generate signal23            signal = signal_func(self.df.iloc[:i+1], self.position, self.config)24            25            if signal['signal'] == Signal.BUY and not self.position:26                self._open_position('long', bar['close'], i,27                    stop=signal.get('stop_loss'),28                    tp=signal.get('take_profit'))29            30            elif signal['signal'] == Signal.SELL and not self.position:31                self._open_position('short', bar['close'], i,32                    stop=signal.get('stop_loss'),33                    tp=signal.get('take_profit'))34            35            elif signal['signal'] == Signal.CLOSE and self.position:36                self._close_position(bar['close'], i, 'Signal close')37            38            # Track equity39            equity = self.capital40            if self.position:41                pnl = self._unrealized_pnl(bar['close'])42                equity += pnl43            self.equity_curve.append({44                'time': bar.name if hasattr(bar.name, 'timestamp') else i,45                'equity': equity,46            })47        48        return self._calculate_metrics()49    50    def _open_position(self, side, price, bar_idx, stop=None, tp=None):51        risk_pct = self.config['position_size_pct'] / 10052        size = (self.capital * risk_pct) / price53        commission = price * size * (self.config['commission_pct'] / 100)54        55        self.position = {56            'side': side,57            'entry': price,58            'size': size,59            'stop': stop,60            'take_profit': tp,61            'entry_bar': bar_idx,62        }63        self.capital -= commission64    65    def _close_position(self, price, bar_idx, reason):66        pnl = self._unrealized_pnl(price)67        commission = price * self.position['size'] * (self.config['commission_pct'] / 100)68        69        self.capital += pnl - commission70        self.trades.append({71            'side': self.position['side'],72            'entry': self.position['entry'],73            'exit': price,74            'size': self.position['size'],75            'pnl': pnl - commission,76            'bars_held': bar_idx - self.position['entry_bar'],77            'reason': reason,78        })79        self.position = None80    81    def _unrealized_pnl(self, current_price):82        if self.position['side'] == 'long':83            return (current_price - self.position['entry']) * self.position['size']84        else:85            return (self.position['entry'] - current_price) * self.position['size']86    87    def _check_exit(self, bar):88        if self.position['stop']:89            if self.position['side'] == 'long' and bar['low'] <= self.position['stop']:90                self._close_position(self.position['stop'], -1, 'Stop loss')91            elif self.position['side'] == 'short' and bar['high'] >= self.position['stop']:92                self._close_position(self.position['stop'], -1, 'Stop loss')93        94        if self.position and self.position.get('take_profit'):95            if self.position['side'] == 'long' and bar['high'] >= self.position['take_profit']:96                self._close_position(self.position['take_profit'], -1, 'Take profit')97            elif self.position['side'] == 'short' and bar['low'] <= self.position['take_profit']:98                self._close_position(self.position['take_profit'], -1, 'Take profit')99    100    def _calculate_metrics(self):101        if not self.trades:102            return {'total_trades': 0, 'message': 'No trades generated'}103        104        pnls = [t['pnl'] for t in self.trades]105        wins = [p for p in pnls if p > 0]106        losses = [p for p in pnls if p <= 0]107        108        return {109            'total_trades': len(self.trades),110            'winning_trades': len(wins),111            'losing_trades': len(losses),112            'win_rate': len(wins) / len(self.trades) * 100,113            'total_pnl': sum(pnls),114            'avg_win': np.mean(wins) if wins else 0,115            'avg_loss': np.mean(losses) if losses else 0,116            'profit_factor': abs(sum(wins) / sum(losses)) if losses else float('inf'),117            'max_drawdown': self._max_drawdown(),118            'sharpe_ratio': self._sharpe_ratio(),119        }120    121    def _max_drawdown(self):122        equity = [e['equity'] for e in self.equity_curve]123        peak = equity[0]124        max_dd = 0125        for e in equity:126            peak = max(peak, e)127            dd = (peak - e) / peak * 100128            max_dd = max(max_dd, dd)129        return max_dd130    131    def _sharpe_ratio(self, risk_free=0.02):132        if len(self.equity_curve) < 2:133            return 0134        equity = [e['equity'] for e in self.equity_curve]135        returns = pd.Series(equity).pct_change().dropna()136        if returns.std() == 0:137            return 0138        return (returns.mean() - risk_free/252) / returns.std() * np.sqrt(252)