Strategy Development with RustyBT¶

Learn how to build custom trading strategies from simple to advanced implementations.

This comprehensive guide covers:

1. Moving Average Crossover - Entry methods (market vs limit orders) & order management
2. Mean Reversion - Exit methods (stop loss, take profit, closing positions)
3. Momentum Strategy - Trailing stops & dynamic position sizing
4. Multi-Factor Strategy - TA-Lib indicators & advanced order types

Key Concepts Demonstrated:

• Entry Methods: Market orders, limit orders, stop orders, conditional entries
• Exit Methods: Market exits, stop-loss, take-profit, trailing stops, close all positions
• Order Management: Cancelling orders, replacing orders, checking open orders
• Position Management: Tracking position size, direction, value, and properties
• Indicators: Using both numpy-based and TA-Lib indicators with temporal isolation
• Order Types: Market, limit, stop, stop-limit, trailing stop orders

Estimated runtime: 5 minutes

---

📋 Notebook Information

• RustyBT Version: 0.1.2+
• Last Validated: 2025-11-07
• API Compatibility: Verified ✅
• Documentation: API Reference

---

In [ ]:

from rustybt.analytics import setup_notebook

setup_notebook()

import numpy as np

from rustybt import TradingAlgorithm
from rustybt.api import (
    cancel_order,
    get_open_orders,
    order,
    order_target,
    order_target_percent,
    record,
    symbol,
)

# Optional: Import TA-Lib for advanced indicators
try:
    import talib
    TALIB_AVAILABLE = True
except ImportError:
    print("TA-Lib not available. Install with: pip install TA-Lib")
    print("Strategies will use numpy-based indicators as fallback.")
    TALIB_AVAILABLE = False

from rustybt.analytics import setup_notebook setup_notebook() import numpy as np from rustybt import TradingAlgorithm from rustybt.api import ( cancel_order, get_open_orders, order, order_target, order_target_percent, record, symbol, ) # Optional: Import TA-Lib for advanced indicators try: import talib TALIB_AVAILABLE = True except ImportError: print("TA-Lib not available. Install with: pip install TA-Lib") print("Strategies will use numpy-based indicators as fallback.") TALIB_AVAILABLE = False

1. Moving Average Crossover - Market Entries & Order Management¶

Strategy Logic:

• Entry: Use market orders to enter positions immediately when crossover signals occur (capture momentum)
• Exit: Use limit orders to exit at favorable prices
• Demonstrates: Order management (cancelling stale limit orders, checking open orders)

Key Features:

• Fast market entries when fast MA crosses above slow MA (bullish signal)
• Limit exit orders to capture profits at target price
• Cancel old exit orders if not filled and replace with updated targets
• Track and display open orders

In [ ]:

class MovingAverageCrossover(TradingAlgorithm):
    """
    Moving Average Crossover with market entries and limit exits.
    
    Demonstrates:
    - Market orders for fast entry (capturing momentum)
    - Limit orders for exits (profit targets)
    - Order cancellation and replacement
    - Checking open orders
    """
    
    def initialize(self, context) -> None:
        context.asset = symbol("SPY")
        context.fast_period = 20
        context.slow_period = 50
        context.prices = []
        context.profit_target_pct = 0.05  # 5% profit target for limit exit
        
    def handle_data(self, context, data) -> None:
        price = data.current(context.asset, "close")
        context.prices.append(price)
        
        # Wait for enough data
        if len(context.prices) < context.slow_period:
            return
        
        # Calculate moving averages using numpy (temporal isolation enforced by context.prices)
        fast_ma = np.mean(context.prices[-context.fast_period:])
        slow_ma = np.mean(context.prices[-context.slow_period:])
        
        # Get current position
        current_position = context.portfolio.positions[context.asset].amount
        
        # Check for open orders
        open_orders = get_open_orders(context.asset)
        
        # Entry Logic: Market orders for fast execution
        if fast_ma > slow_ma and current_position == 0:
            # Bullish crossover: Enter with MARKET order
            order_id = order(context.asset, 100)  # Market order for 100 shares
            record(signal="buy", fast_ma=fast_ma, slow_ma=slow_ma)
            
        # Exit Logic: Limit orders for profit targets
        elif fast_ma < slow_ma and current_position > 0:
            # Bearish crossover: Exit with LIMIT order at profit target
            limit_price = price * (1 + context.profit_target_pct)
            
            # Cancel any existing exit orders first
            for open_order in open_orders:
                cancel_order(open_order.id)
            
            # Place new limit exit order
            order_id = order(context.asset, -int(current_position), limit_price=limit_price)
            record(signal="sell_limit", fast_ma=fast_ma, slow_ma=slow_ma, limit_price=limit_price)
        
        # Monitor: If we have open orders but no clear signal, track them
        elif len(open_orders) > 0:
            record(signal="waiting", fast_ma=fast_ma, slow_ma=slow_ma, open_orders=len(open_orders))
        else:
            record(signal="hold", fast_ma=fast_ma, slow_ma=slow_ma)

class MovingAverageCrossover(TradingAlgorithm): """ Moving Average Crossover with market entries and limit exits. Demonstrates: - Market orders for fast entry (capturing momentum) - Limit orders for exits (profit targets) - Order cancellation and replacement - Checking open orders """ def initialize(self, context) -> None: context.asset = symbol("SPY") context.fast_period = 20 context.slow_period = 50 context.prices = [] context.profit_target_pct = 0.05 # 5% profit target for limit exit def handle_data(self, context, data) -> None: price = data.current(context.asset, "close") context.prices.append(price) # Wait for enough data if len(context.prices) < context.slow_period: return # Calculate moving averages using numpy (temporal isolation enforced by context.prices) fast_ma = np.mean(context.prices[-context.fast_period:]) slow_ma = np.mean(context.prices[-context.slow_period:]) # Get current position current_position = context.portfolio.positions[context.asset].amount # Check for open orders open_orders = get_open_orders(context.asset) # Entry Logic: Market orders for fast execution if fast_ma > slow_ma and current_position == 0: # Bullish crossover: Enter with MARKET order order_id = order(context.asset, 100) # Market order for 100 shares record(signal="buy", fast_ma=fast_ma, slow_ma=slow_ma) # Exit Logic: Limit orders for profit targets elif fast_ma < slow_ma and current_position > 0: # Bearish crossover: Exit with LIMIT order at profit target limit_price = price * (1 + context.profit_target_pct) # Cancel any existing exit orders first for open_order in open_orders: cancel_order(open_order.id) # Place new limit exit order order_id = order(context.asset, -int(current_position), limit_price=limit_price) record(signal="sell_limit", fast_ma=fast_ma, slow_ma=slow_ma, limit_price=limit_price) # Monitor: If we have open orders but no clear signal, track them elif len(open_orders) > 0: record(signal="waiting", fast_ma=fast_ma, slow_ma=slow_ma, open_orders=len(open_orders)) else: record(signal="hold", fast_ma=fast_ma, slow_ma=slow_ma)

2. Mean Reversion - Limit Entries & Stop Loss/Take Profit¶

Strategy Logic:

• Entry: Use limit orders to enter positions at favorable prices (buy low, sell high)
• Exit: Use stop-loss and take-profit orders for risk management
• Demonstrates: Limit entry orders, stop orders, position closing, z-score calculations

Key Features:

• Calculate z-score to identify overbought/oversold conditions
• Place limit buy orders when oversold (z-score < -2)
• Place limit sell orders when overbought (z-score > +2)
• Set stop-loss orders to limit downside risk
• Set take-profit orders to lock in gains
• Close all positions using order_target_percent

In [ ]:

class MeanReversion(TradingAlgorithm):
    """
    Mean Reversion with limit entries and stop-loss/take-profit exits.
    
    Demonstrates:
    - Limit orders for entry (buy low, sell high)
    - Stop-loss orders for risk management
    - Take-profit orders for profit targets
    - Position closing with order_target_percent
    """
    
    def initialize(self, context) -> None:
        context.asset = symbol("SPY")
        context.lookback = 20
        context.z_entry = 2.0  # Enter when z-score exceeds ±2
        context.z_exit = 0.5   # Exit when z-score returns to ±0.5
        context.stop_loss_pct = 0.03  # 3% stop loss
        context.take_profit_pct = 0.06  # 6% take profit
        context.prices = []
        context.entry_price = None
        
    def handle_data(self, context, data) -> None:
        price = data.current(context.asset, "close")
        context.prices.append(price)
        
        # Wait for enough data
        if len(context.prices) < context.lookback:
            return
        
        # Calculate z-score (temporal isolation: only uses past data from context.prices)
        recent_prices = context.prices[-context.lookback:]
        mean = np.mean(recent_prices)
        std = np.std(recent_prices)
        z_score = (price - mean) / std if std > 0 else 0
        
        # Get current position
        position = context.portfolio.positions[context.asset]
        current_position = position.amount
        
        # Entry Logic: Limit orders at better prices
        if current_position == 0:
            # Oversold: Place limit BUY order below current price
            if z_score < -context.z_entry:
                limit_buy_price = price * 0.995  # Limit order 0.5% below current price
                order_id = order(context.asset, 100, limit_price=limit_buy_price)
                record(signal="buy_limit", z_score=z_score, limit_price=limit_buy_price)
            
            # Overbought: Place limit SELL order above current price (short)
            elif z_score > context.z_entry:
                limit_sell_price = price * 1.005  # Limit order 0.5% above current price
                order_id = order(context.asset, -100, limit_price=limit_sell_price)
                record(signal="sell_limit", z_score=z_score, limit_price=limit_sell_price)
            else:
                record(signal="neutral", z_score=z_score)
        
        # Exit Logic: Stop-loss, Take-profit, or Mean Reversion
        else:
            # Track entry price
            if context.entry_price is None:
                context.entry_price = position.cost_basis
            
            # Calculate profit/loss percentage
            pnl_pct = (price - context.entry_price) / context.entry_price
            
            # LONG position exits
            if current_position > 0:
                # Stop-loss: Exit immediately if loss exceeds threshold
                if pnl_pct <= -context.stop_loss_pct:
                    order_target_percent(context.asset, 0.0)  # Close all positions
                    context.entry_price = None
                    record(signal="stop_loss", z_score=z_score, pnl_pct=pnl_pct)
                
                # Take-profit: Exit if profit target reached
                elif pnl_pct >= context.take_profit_pct:
                    order_target_percent(context.asset, 0.0)
                    context.entry_price = None
                    record(signal="take_profit", z_score=z_score, pnl_pct=pnl_pct)
                
                # Mean reversion exit: Exit when z-score normalizes
                elif z_score > -context.z_exit:
                    order_target_percent(context.asset, 0.0)
                    context.entry_price = None
                    record(signal="mean_revert_exit", z_score=z_score, pnl_pct=pnl_pct)
                else:
                    record(signal="hold_long", z_score=z_score, pnl_pct=pnl_pct)
            
            # SHORT position exits
            elif current_position < 0:
                # Stop-loss for short: Exit if price rises too much
                if pnl_pct <= -context.stop_loss_pct:  # For short, rising price is loss
                    order_target_percent(context.asset, 0.0)
                    context.entry_price = None
                    record(signal="stop_loss_short", z_score=z_score, pnl_pct=pnl_pct)
                
                # Take-profit for short
                elif pnl_pct >= context.take_profit_pct:
                    order_target_percent(context.asset, 0.0)
                    context.entry_price = None
                    record(signal="take_profit_short", z_score=z_score, pnl_pct=pnl_pct)
                
                # Mean reversion exit for short
                elif z_score < context.z_exit:
                    order_target_percent(context.asset, 0.0)
                    context.entry_price = None
                    record(signal="mean_revert_exit_short", z_score=z_score, pnl_pct=pnl_pct)
                else:
                    record(signal="hold_short", z_score=z_score, pnl_pct=pnl_pct)

class MeanReversion(TradingAlgorithm): """ Mean Reversion with limit entries and stop-loss/take-profit exits. Demonstrates: - Limit orders for entry (buy low, sell high) - Stop-loss orders for risk management - Take-profit orders for profit targets - Position closing with order_target_percent """ def initialize(self, context) -> None: context.asset = symbol("SPY") context.lookback = 20 context.z_entry = 2.0 # Enter when z-score exceeds ±2 context.z_exit = 0.5 # Exit when z-score returns to ±0.5 context.stop_loss_pct = 0.03 # 3% stop loss context.take_profit_pct = 0.06 # 6% take profit context.prices = [] context.entry_price = None def handle_data(self, context, data) -> None: price = data.current(context.asset, "close") context.prices.append(price) # Wait for enough data if len(context.prices) < context.lookback: return # Calculate z-score (temporal isolation: only uses past data from context.prices) recent_prices = context.prices[-context.lookback:] mean = np.mean(recent_prices) std = np.std(recent_prices) z_score = (price - mean) / std if std > 0 else 0 # Get current position position = context.portfolio.positions[context.asset] current_position = position.amount # Entry Logic: Limit orders at better prices if current_position == 0: # Oversold: Place limit BUY order below current price if z_score < -context.z_entry: limit_buy_price = price * 0.995 # Limit order 0.5% below current price order_id = order(context.asset, 100, limit_price=limit_buy_price) record(signal="buy_limit", z_score=z_score, limit_price=limit_buy_price) # Overbought: Place limit SELL order above current price (short) elif z_score > context.z_entry: limit_sell_price = price * 1.005 # Limit order 0.5% above current price order_id = order(context.asset, -100, limit_price=limit_sell_price) record(signal="sell_limit", z_score=z_score, limit_price=limit_sell_price) else: record(signal="neutral", z_score=z_score) # Exit Logic: Stop-loss, Take-profit, or Mean Reversion else: # Track entry price if context.entry_price is None: context.entry_price = position.cost_basis # Calculate profit/loss percentage pnl_pct = (price - context.entry_price) / context.entry_price # LONG position exits if current_position > 0: # Stop-loss: Exit immediately if loss exceeds threshold if pnl_pct <= -context.stop_loss_pct: order_target_percent(context.asset, 0.0) # Close all positions context.entry_price = None record(signal="stop_loss", z_score=z_score, pnl_pct=pnl_pct) # Take-profit: Exit if profit target reached elif pnl_pct >= context.take_profit_pct: order_target_percent(context.asset, 0.0) context.entry_price = None record(signal="take_profit", z_score=z_score, pnl_pct=pnl_pct) # Mean reversion exit: Exit when z-score normalizes elif z_score > -context.z_exit: order_target_percent(context.asset, 0.0) context.entry_price = None record(signal="mean_revert_exit", z_score=z_score, pnl_pct=pnl_pct) else: record(signal="hold_long", z_score=z_score, pnl_pct=pnl_pct) # SHORT position exits elif current_position < 0: # Stop-loss for short: Exit if price rises too much if pnl_pct <= -context.stop_loss_pct: # For short, rising price is loss order_target_percent(context.asset, 0.0) context.entry_price = None record(signal="stop_loss_short", z_score=z_score, pnl_pct=pnl_pct) # Take-profit for short elif pnl_pct >= context.take_profit_pct: order_target_percent(context.asset, 0.0) context.entry_price = None record(signal="take_profit_short", z_score=z_score, pnl_pct=pnl_pct) # Mean reversion exit for short elif z_score < context.z_exit: order_target_percent(context.asset, 0.0) context.entry_price = None record(signal="mean_revert_exit_short", z_score=z_score, pnl_pct=pnl_pct) else: record(signal="hold_short", z_score=z_score, pnl_pct=pnl_pct)

3. Momentum Strategy - Trailing Stops & Position Sizing¶

Strategy Logic:

• Entry: Market orders when momentum is strong
• Exit: Trailing stop orders to protect profits while letting winners run
• Demonstrates: Position sizing based on portfolio value, trailing stops, RSI momentum indicator

Key Features:

• Calculate RSI (Relative Strength Index) for momentum
• Dynamic position sizing based on available capital
• Trailing stop orders that move with price
• Track position properties (size, value, direction)
• Demonstrates both order_target and percentage-based sizing

In [ ]:

class MomentumStrategy(TradingAlgorithm):
    """
    Momentum strategy with trailing stops and dynamic position sizing.
    
    Demonstrates:
    - RSI momentum indicator
    - Position sizing based on portfolio value
    - Trailing stop orders
    - Position property tracking (size, value, direction)
    """
    
    def initialize(self, context) -> None:
        context.asset = symbol("SPY")
        context.rsi_period = 14
        context.rsi_oversold = 30
        context.rsi_overbought = 70
        context.prices = []
        context.position_pct = 0.20  # Use 20% of portfolio per position
        context.trailing_stop_pct = 0.05  # 5% trailing stop
        context.highest_price = None
        
    def calculate_rsi(self, prices, period=14):
        """Calculate RSI indicator (temporal isolation: only uses provided prices)."""
        if len(prices) < period + 1:
            return 50  # Neutral RSI
        
        # Calculate price changes
        deltas = np.diff(prices)
        gains = np.where(deltas > 0, deltas, 0)
        losses = np.where(deltas < 0, -deltas, 0)
        
        # Calculate average gains and losses
        avg_gain = np.mean(gains[-period:])
        avg_loss = np.mean(losses[-period:])
        
        if avg_loss == 0:
            return 100
        
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        return rsi
    
    def handle_data(self, context, data) -> None:
        price = data.current(context.asset, "close")
        context.prices.append(price)
        
        # Wait for enough data
        if len(context.prices) < context.rsi_period + 1:
            return
        
        # Calculate RSI momentum indicator
        rsi = self.calculate_rsi(context.prices, context.rsi_period)
        
        # Get current position information
        position = context.portfolio.positions[context.asset]
        current_position = position.amount
        position_value = position.amount * price if current_position != 0 else 0
        portfolio_value = context.portfolio.portfolio_value
        
        # Entry Logic: Buy when RSI shows strong momentum
        if current_position == 0 and rsi > context.rsi_overbought:
            # Calculate position size based on portfolio value
            target_value = portfolio_value * context.position_pct
            shares_to_buy = int(target_value / price)
            
            if shares_to_buy > 0:
                # Enter with market order
                order_id = order(context.asset, shares_to_buy)
                context.highest_price = price  # Initialize trailing stop tracker
                record(
                    signal="buy_momentum",
                    rsi=rsi,
                    shares=shares_to_buy,
                    position_pct=context.position_pct
                )
        
        # Exit Logic: Trailing stop to protect profits
        elif current_position > 0:
            # Update highest price for trailing stop
            if context.highest_price is None or price > context.highest_price:
                context.highest_price = price
            
            # Calculate trailing stop price
            trailing_stop_price = context.highest_price * (1 - context.trailing_stop_pct)
            
            # Exit if price drops below trailing stop
            if price <= trailing_stop_price:
                # Close entire position
                order_target(context.asset, 0)
                record(
                    signal="trailing_stop_exit",
                    rsi=rsi,
                    exit_price=price,
                    highest_price=context.highest_price,
                    pnl_pct=(price - position.cost_basis) / position.cost_basis
                )
                context.highest_price = None
            
            # Also exit if RSI shows momentum reversal
            elif rsi < context.rsi_oversold:
                order_target(context.asset, 0)
                record(
                    signal="rsi_exit",
                    rsi=rsi,
                    exit_price=price,
                    pnl_pct=(price - position.cost_basis) / position.cost_basis
                )
                context.highest_price = None
            
            # Hold and track position
            else:
                record(
                    signal="hold_momentum",
                    rsi=rsi,
                    position_size=current_position,
                    position_value=position_value,
                    trailing_stop=trailing_stop_price,
                    unrealized_pnl_pct=(price - position.cost_basis) / position.cost_basis
                )
        else:
            record(signal="no_position", rsi=rsi)

class MomentumStrategy(TradingAlgorithm): """ Momentum strategy with trailing stops and dynamic position sizing. Demonstrates: - RSI momentum indicator - Position sizing based on portfolio value - Trailing stop orders - Position property tracking (size, value, direction) """ def initialize(self, context) -> None: context.asset = symbol("SPY") context.rsi_period = 14 context.rsi_oversold = 30 context.rsi_overbought = 70 context.prices = [] context.position_pct = 0.20 # Use 20% of portfolio per position context.trailing_stop_pct = 0.05 # 5% trailing stop context.highest_price = None def calculate_rsi(self, prices, period=14): """Calculate RSI indicator (temporal isolation: only uses provided prices).""" if len(prices) < period + 1: return 50 # Neutral RSI # Calculate price changes deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) # Calculate average gains and losses avg_gain = np.mean(gains[-period:]) avg_loss = np.mean(losses[-period:]) if avg_loss == 0: return 100 rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return rsi def handle_data(self, context, data) -> None: price = data.current(context.asset, "close") context.prices.append(price) # Wait for enough data if len(context.prices) < context.rsi_period + 1: return # Calculate RSI momentum indicator rsi = self.calculate_rsi(context.prices, context.rsi_period) # Get current position information position = context.portfolio.positions[context.asset] current_position = position.amount position_value = position.amount * price if current_position != 0 else 0 portfolio_value = context.portfolio.portfolio_value # Entry Logic: Buy when RSI shows strong momentum if current_position == 0 and rsi > context.rsi_overbought: # Calculate position size based on portfolio value target_value = portfolio_value * context.position_pct shares_to_buy = int(target_value / price) if shares_to_buy > 0: # Enter with market order order_id = order(context.asset, shares_to_buy) context.highest_price = price # Initialize trailing stop tracker record( signal="buy_momentum", rsi=rsi, shares=shares_to_buy, position_pct=context.position_pct ) # Exit Logic: Trailing stop to protect profits elif current_position > 0: # Update highest price for trailing stop if context.highest_price is None or price > context.highest_price: context.highest_price = price # Calculate trailing stop price trailing_stop_price = context.highest_price * (1 - context.trailing_stop_pct) # Exit if price drops below trailing stop if price <= trailing_stop_price: # Close entire position order_target(context.asset, 0) record( signal="trailing_stop_exit", rsi=rsi, exit_price=price, highest_price=context.highest_price, pnl_pct=(price - position.cost_basis) / position.cost_basis ) context.highest_price = None # Also exit if RSI shows momentum reversal elif rsi < context.rsi_oversold: order_target(context.asset, 0) record( signal="rsi_exit", rsi=rsi, exit_price=price, pnl_pct=(price - position.cost_basis) / position.cost_basis ) context.highest_price = None # Hold and track position else: record( signal="hold_momentum", rsi=rsi, position_size=current_position, position_value=position_value, trailing_stop=trailing_stop_price, unrealized_pnl_pct=(price - position.cost_basis) / position.cost_basis ) else: record(signal="no_position", rsi=rsi)

4. Multi-Factor Strategy - TA-Lib Indicators & Advanced Features¶

Strategy Logic:

• Indicators: Use TA-Lib for professional-grade indicators (EMA, RSI, MACD)
• Entry: Combine multiple factors for high-confidence signals
• Exit: Sophisticated exit logic combining multiple conditions
• Demonstrates: TA-Lib integration, data.history() for temporal isolation, multi-condition logic

Key Features:

• EMA (Exponential Moving Average) using TA-Lib
• RSI (Relative Strength Index) using TA-Lib
• MACD (Moving Average Convergence Divergence) using TA-Lib
• Fallback to numpy-based indicators if TA-Lib not available
• Demonstrates data.history() for temporal data access
• Complex entry/exit logic combining multiple indicators

In [ ]:

class MultiFactorStrategy(TradingAlgorithm):
    """
    Multi-factor strategy using TA-Lib indicators.
    
    Demonstrates:
    - TA-Lib indicator integration (EMA, RSI, MACD)
    - data.history() for temporal data access
    - Multi-condition entry/exit logic
    - Fallback to numpy when TA-Lib unavailable
    """
    
    def initialize(self, context) -> None:
        context.asset = symbol("SPY")
        context.fast_ema_period = 12
        context.slow_ema_period = 26
        context.rsi_period = 14
        context.lookback = 50  # Days of history to fetch
        
    def handle_data(self, context, data) -> None:
        # Use data.history() for temporal isolation - guaranteed no lookahead bias
        # This fetches past price data automatically respecting simulation time
        price_history = data.history(context.asset, "close", context.lookback, "1d")
        
        # Check if we have enough data
        if price_history.isnull().values.any() or len(price_history) < context.lookback:
            return
        
        prices = price_history.values
        current_price = prices[-1]
        
        # Calculate indicators using TA-Lib (or numpy fallback)
        if TALIB_AVAILABLE:
            # TA-Lib provides professional-grade indicators
            fast_ema = talib.EMA(prices, timeperiod=context.fast_ema_period)[-1]
            slow_ema = talib.EMA(prices, timeperiod=context.slow_ema_period)[-1]
            rsi = talib.RSI(prices, timeperiod=context.rsi_period)[-1]
            macd, macd_signal, macd_hist = talib.MACD(prices)
            macd = macd[-1]
            macd_signal = macd_signal[-1]
            macd_hist = macd_hist[-1]
        else:
            # Fallback to numpy-based calculations
            fast_ema = np.mean(prices[-context.fast_ema_period:])
            slow_ema = np.mean(prices[-context.slow_ema_period:])
            
            # Simple RSI calculation
            deltas = np.diff(prices[-context.rsi_period-1:])
            gains = np.where(deltas > 0, deltas, 0)
            losses = np.where(deltas < 0, -deltas, 0)
            avg_gain = np.mean(gains)
            avg_loss = np.mean(losses)
            rs = avg_gain / avg_loss if avg_loss != 0 else 100
            rsi = 100 - (100 / (1 + rs))
            
            # Simple MACD approximation
            macd = fast_ema - slow_ema
            macd_signal = np.mean([macd] * 9)  # Simplified
            macd_hist = macd - macd_signal
        
        # Get current position
        position = context.portfolio.positions[context.asset]
        current_position = position.amount
        
        # Multi-Factor Entry Logic: All conditions must align
        if current_position == 0:
            # Bullish conditions:
            # 1. Fast EMA above Slow EMA (trend)
            # 2. RSI between 40-60 (not overbought/oversold)
            # 3. MACD histogram positive (momentum)
            bullish = (
                fast_ema > slow_ema and
                40 < rsi < 60 and
                macd_hist > 0
            )
            
            if bullish:
                # Enter with 50% of portfolio
                order_target_percent(context.asset, 0.5)
                record(
                    signal="buy_multi_factor",
                    fast_ema=fast_ema,
                    slow_ema=slow_ema,
                    rsi=rsi,
                    macd_hist=macd_hist
                )
            else:
                record(
                    signal="waiting",
                    fast_ema=fast_ema,
                    slow_ema=slow_ema,
                    rsi=rsi,
                    macd_hist=macd_hist
                )
        
        # Multi-Factor Exit Logic: Exit if any bearish condition
        else:
            # Bearish conditions:
            # 1. Fast EMA crosses below Slow EMA (trend reversal)
            # 2. RSI becomes overbought (>70) or oversold (<30)
            # 3. MACD histogram turns negative (momentum loss)
            bearish = (
                fast_ema < slow_ema or
                rsi > 70 or
                rsi < 30 or
                macd_hist < 0
            )
            
            if bearish:
                # Exit position
                order_target_percent(context.asset, 0.0)
                pnl_pct = (current_price - position.cost_basis) / position.cost_basis
                record(
                    signal="sell_multi_factor",
                    fast_ema=fast_ema,
                    slow_ema=slow_ema,
                    rsi=rsi,
                    macd_hist=macd_hist,
                    pnl_pct=pnl_pct
                )
            else:
                # Hold and track
                pnl_pct = (current_price - position.cost_basis) / position.cost_basis
                record(
                    signal="hold_multi_factor",
                    fast_ema=fast_ema,
                    slow_ema=slow_ema,
                    rsi=rsi,
                    macd_hist=macd_hist,
                    position_size=current_position,
                    pnl_pct=pnl_pct
                )

class MultiFactorStrategy(TradingAlgorithm): """ Multi-factor strategy using TA-Lib indicators. Demonstrates: - TA-Lib indicator integration (EMA, RSI, MACD) - data.history() for temporal data access - Multi-condition entry/exit logic - Fallback to numpy when TA-Lib unavailable """ def initialize(self, context) -> None: context.asset = symbol("SPY") context.fast_ema_period = 12 context.slow_ema_period = 26 context.rsi_period = 14 context.lookback = 50 # Days of history to fetch def handle_data(self, context, data) -> None: # Use data.history() for temporal isolation - guaranteed no lookahead bias # This fetches past price data automatically respecting simulation time price_history = data.history(context.asset, "close", context.lookback, "1d") # Check if we have enough data if price_history.isnull().values.any() or len(price_history) < context.lookback: return prices = price_history.values current_price = prices[-1] # Calculate indicators using TA-Lib (or numpy fallback) if TALIB_AVAILABLE: # TA-Lib provides professional-grade indicators fast_ema = talib.EMA(prices, timeperiod=context.fast_ema_period)[-1] slow_ema = talib.EMA(prices, timeperiod=context.slow_ema_period)[-1] rsi = talib.RSI(prices, timeperiod=context.rsi_period)[-1] macd, macd_signal, macd_hist = talib.MACD(prices) macd = macd[-1] macd_signal = macd_signal[-1] macd_hist = macd_hist[-1] else: # Fallback to numpy-based calculations fast_ema = np.mean(prices[-context.fast_ema_period:]) slow_ema = np.mean(prices[-context.slow_ema_period:]) # Simple RSI calculation deltas = np.diff(prices[-context.rsi_period-1:]) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = np.mean(gains) avg_loss = np.mean(losses) rs = avg_gain / avg_loss if avg_loss != 0 else 100 rsi = 100 - (100 / (1 + rs)) # Simple MACD approximation macd = fast_ema - slow_ema macd_signal = np.mean([macd] * 9) # Simplified macd_hist = macd - macd_signal # Get current position position = context.portfolio.positions[context.asset] current_position = position.amount # Multi-Factor Entry Logic: All conditions must align if current_position == 0: # Bullish conditions: # 1. Fast EMA above Slow EMA (trend) # 2. RSI between 40-60 (not overbought/oversold) # 3. MACD histogram positive (momentum) bullish = ( fast_ema > slow_ema and 40 < rsi < 60 and macd_hist > 0 ) if bullish: # Enter with 50% of portfolio order_target_percent(context.asset, 0.5) record( signal="buy_multi_factor", fast_ema=fast_ema, slow_ema=slow_ema, rsi=rsi, macd_hist=macd_hist ) else: record( signal="waiting", fast_ema=fast_ema, slow_ema=slow_ema, rsi=rsi, macd_hist=macd_hist ) # Multi-Factor Exit Logic: Exit if any bearish condition else: # Bearish conditions: # 1. Fast EMA crosses below Slow EMA (trend reversal) # 2. RSI becomes overbought (>70) or oversold (<30) # 3. MACD histogram turns negative (momentum loss) bearish = ( fast_ema < slow_ema or rsi > 70 or rsi < 30 or macd_hist < 0 ) if bearish: # Exit position order_target_percent(context.asset, 0.0) pnl_pct = (current_price - position.cost_basis) / position.cost_basis record( signal="sell_multi_factor", fast_ema=fast_ema, slow_ema=slow_ema, rsi=rsi, macd_hist=macd_hist, pnl_pct=pnl_pct ) else: # Hold and track pnl_pct = (current_price - position.cost_basis) / position.cost_basis record( signal="hold_multi_factor", fast_ema=fast_ema, slow_ema=slow_ema, rsi=rsi, macd_hist=macd_hist, position_size=current_position, pnl_pct=pnl_pct )

Summary of Concepts Covered¶

Entry Methods Demonstrated¶

1. Market Orders (order(asset, amount)) - Fast execution for momentum strategies
2. Limit Orders (order(asset, amount, limit_price=price)) - Enter at specific price levels
3. Conditional Entries - Using indicators to trigger entries

Exit Methods Demonstrated¶

1. Market Exits (order_target_percent(asset, 0.0)) - Close positions immediately
2. Limit Exits - Exit at profit target prices
3. Stop-Loss Exits - Risk management with percentage-based stops
4. Trailing Stops - Protect profits while letting winners run
5. Indicator-Based Exits - Exit when technical conditions change

Order Management¶

1. Checking Open Orders - get_open_orders(asset) to track pending orders
2. Cancelling Orders - cancel_order(order_id) to remove unfilled orders
3. Replacing Orders - Cancel old orders and place new ones with updated parameters

Position Management¶

1. Position Size - Access via context.portfolio.positions[asset].amount
2. Position Value - Calculate as position.amount * current_price
3. Cost Basis - Track entry price via position.cost_basis
4. P&L Tracking - Calculate unrealized profit/loss percentages

Indicators & Temporal Isolation¶

1. Numpy-based - Using context.prices list ensures temporal isolation
2. TA-Lib Integration - Professional indicators with talib.EMA(), talib.RSI(), talib.MACD()
3. data.history() - Automatic temporal isolation for historical data access
4. Custom Calculations - Building custom indicators with guaranteed no lookahead bias

Order Types Reference¶

• order(asset, amount) - Market order
• order(asset, amount, limit_price=X) - Limit order
• order(asset, amount, stop_price=X) - Stop order
• order(asset, amount, limit_price=X, stop_price=Y) - Stop-limit order
• order_target(asset, shares) - Target specific share count
• order_target_percent(asset, pct) - Target portfolio percentage
• order_target_value(asset, value) - Target dollar amount

Next Steps¶

1. Test with Real Data - Ingest data using 02_data_ingestion.ipynb
2. Optimize Parameters - Use 05_optimization.ipynb for parameter tuning
3. Analyze Performance - Evaluate results with 04_performance_analysis.ipynb
4. Risk Analysis - Assess risk metrics with 07_risk_analytics.ipynb
5. Live Trading - Deploy strategies with 09_live_paper_trading.ipynb

Additional Resources¶

• API Documentation: See rustybt.api for complete function signatures
• Order Execution: Review rustybt.finance.execution for execution styles
• Indicators: Install TA-Lib with pip install TA-Lib for professional indicators
• Examples: Check examples/ directory for more strategy implementations