Multi-Timeframe Strategies¶

Runtime: ~8 minutes
Level: Advanced

This notebook demonstrates how to build strategies that analyze and trade across multiple timeframes:

1. Higher Timeframe for Trend - Use daily/weekly data for overall direction
2. Lower Timeframe for Entry - Use hourly/minute data for precise entries
3. Timeframe Alignment - Ensure signals across timeframes agree
4. Adaptive Timeframes - Switch timeframes based on market conditions

Why Multi-Timeframe?¶

• Better context - See the big picture and details simultaneously
• Improved entries - Find optimal entry points within confirmed trends
• Reduced false signals - Filter out noise by requiring multi-timeframe confirmation
• Risk management - Use higher timeframes for stops, lower for entries

---

📋 Notebook Information

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

---

In [ ]:

# Setup
from rustybt.analytics import setup_notebook
setup_notebook()

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

from rustybt import TradingAlgorithm
from rustybt.api import (
    order_target_percent,
    record,
    symbol,
    schedule_function,
)
from rustybt.pipeline import Pipeline
from rustybt.pipeline.factors import (
    SimpleMovingAverage,
    RSI,
    ATR,
    Returns,
)

print("✓ Imports successful")

# Setup from rustybt.analytics import setup_notebook setup_notebook() import pandas as pd import numpy as np from datetime import datetime, timedelta from rustybt import TradingAlgorithm from rustybt.api import ( order_target_percent, record, symbol, schedule_function, ) from rustybt.pipeline import Pipeline from rustybt.pipeline.factors import ( SimpleMovingAverage, RSI, ATR, Returns, ) print("✓ Imports successful")

1. Basic Multi-Timeframe Strategy¶

Classic approach: Higher timeframe for trend, lower for entries

Example: Daily trend + hourly entries

In [ ]:

class DailyTrendHourlyEntry(TradingAlgorithm):
    """
    Trade with daily trend, enter on hourly pullbacks.
    
    Strategy:
    - Daily: Use 50/200 MA crossover for trend
    - Hourly: Enter on RSI oversold in uptrend (or overbought in downtrend)
    """
    
    def initialize(self, context):
        context.asset = self.symbol('AAPL')
        
        # Timeframe parameters
        context.daily_fast_ma = 50
        context.daily_slow_ma = 200
        context.hourly_rsi_period = 14
        context.hourly_rsi_oversold = 30
        context.hourly_rsi_overbought = 70
        
        # State
        context.daily_trend = None  # 'up', 'down', or None
        
        # Schedule daily trend check
        self.schedule_function(
            self.check_daily_trend,
            date_rules=self.date_rules.every_day(),
            time_rules=self.time_rules.market_open()
        )
        
        # Schedule hourly entry checks (every hour during market hours)
        for hour in range(9, 16):  # 9 AM to 4 PM
            self.schedule_function(
                self.check_hourly_entry,
                date_rules=self.date_rules.every_day(),
                time_rules=self.time_rules.market_open(hours=hour-9)
            )
    
    def check_daily_trend(self, context, data):
        """Check daily timeframe for overall trend"""
        # Get daily data
        daily_prices = data.history(
            context.asset,
            'close',
            context.daily_slow_ma,
            '1d'
        )
        
        if len(daily_prices) < context.daily_slow_ma:
            return
        
        # Calculate moving averages
        fast_ma = daily_prices[-context.daily_fast_ma:].mean()
        slow_ma = daily_prices.mean()
        
        # Determine trend
        if fast_ma > slow_ma:
            context.daily_trend = 'up'
        elif fast_ma < slow_ma:
            context.daily_trend = 'down'
        else:
            context.daily_trend = None
        
        self.log.info(f"Daily trend: {context.daily_trend}")
    
    def check_hourly_entry(self, context, data):
        """Check hourly timeframe for entry signals"""
        if context.daily_trend is None:
            return
        
        # Get hourly data
        hourly_prices = data.history(
            context.asset,
            'close',
            context.hourly_rsi_period + 1,
            '1h'
        )
        
        if len(hourly_prices) < context.hourly_rsi_period + 1:
            return
        
        # Calculate hourly RSI
        rsi = self.calculate_rsi(hourly_prices, context.hourly_rsi_period)
        
        position = context.portfolio.positions[context.asset]
        
        # Entry logic
        if context.daily_trend == 'up' and position.amount == 0:
            # In uptrend, buy on hourly pullback (RSI oversold)
            if rsi < context.hourly_rsi_oversold:
                self.order_target_percent(context.asset, 0.95)
                self.log.info(f"Long entry: Daily uptrend + Hourly RSI={rsi:.1f}")
        
        elif context.daily_trend == 'down' and position.amount == 0:
            # In downtrend, short on hourly rally (RSI overbought)
            if rsi > context.hourly_rsi_overbought:
                self.order_target_percent(context.asset, -0.95)
                self.log.info(f"Short entry: Daily downtrend + Hourly RSI={rsi:.1f}")
        
        # Exit logic
        elif position.amount > 0 and context.daily_trend != 'up':
            # Exit long if daily trend changes
            self.order_target_percent(context.asset, 0)
            self.log.info("Exiting long: Daily trend changed")
        
        elif position.amount < 0 and context.daily_trend != 'down':
            # Exit short if daily trend changes
            self.order_target_percent(context.asset, 0)
            self.log.info("Exiting short: Daily trend changed")
        
        # Record
        self.record(
            daily_trend=1 if context.daily_trend == 'up' else (-1 if context.daily_trend == 'down' else 0),
            hourly_rsi=rsi,
            position=position.amount,
        )
    
    def calculate_rsi(self, prices, period):
        """Calculate RSI"""
        deltas = np.diff(prices)
        gains = np.where(deltas > 0, deltas, 0)
        losses = np.where(deltas < 0, -deltas, 0)
        
        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

print("✓ Basic multi-timeframe strategy defined")

class DailyTrendHourlyEntry(TradingAlgorithm): """ Trade with daily trend, enter on hourly pullbacks. Strategy: - Daily: Use 50/200 MA crossover for trend - Hourly: Enter on RSI oversold in uptrend (or overbought in downtrend) """ def initialize(self, context): context.asset = self.symbol('AAPL') # Timeframe parameters context.daily_fast_ma = 50 context.daily_slow_ma = 200 context.hourly_rsi_period = 14 context.hourly_rsi_oversold = 30 context.hourly_rsi_overbought = 70 # State context.daily_trend = None # 'up', 'down', or None # Schedule daily trend check self.schedule_function( self.check_daily_trend, date_rules=self.date_rules.every_day(), time_rules=self.time_rules.market_open() ) # Schedule hourly entry checks (every hour during market hours) for hour in range(9, 16): # 9 AM to 4 PM self.schedule_function( self.check_hourly_entry, date_rules=self.date_rules.every_day(), time_rules=self.time_rules.market_open(hours=hour-9) ) def check_daily_trend(self, context, data): """Check daily timeframe for overall trend""" # Get daily data daily_prices = data.history( context.asset, 'close', context.daily_slow_ma, '1d' ) if len(daily_prices) < context.daily_slow_ma: return # Calculate moving averages fast_ma = daily_prices[-context.daily_fast_ma:].mean() slow_ma = daily_prices.mean() # Determine trend if fast_ma > slow_ma: context.daily_trend = 'up' elif fast_ma < slow_ma: context.daily_trend = 'down' else: context.daily_trend = None self.log.info(f"Daily trend: {context.daily_trend}") def check_hourly_entry(self, context, data): """Check hourly timeframe for entry signals""" if context.daily_trend is None: return # Get hourly data hourly_prices = data.history( context.asset, 'close', context.hourly_rsi_period + 1, '1h' ) if len(hourly_prices) < context.hourly_rsi_period + 1: return # Calculate hourly RSI rsi = self.calculate_rsi(hourly_prices, context.hourly_rsi_period) position = context.portfolio.positions[context.asset] # Entry logic if context.daily_trend == 'up' and position.amount == 0: # In uptrend, buy on hourly pullback (RSI oversold) if rsi < context.hourly_rsi_oversold: self.order_target_percent(context.asset, 0.95) self.log.info(f"Long entry: Daily uptrend + Hourly RSI={rsi:.1f}") elif context.daily_trend == 'down' and position.amount == 0: # In downtrend, short on hourly rally (RSI overbought) if rsi > context.hourly_rsi_overbought: self.order_target_percent(context.asset, -0.95) self.log.info(f"Short entry: Daily downtrend + Hourly RSI={rsi:.1f}") # Exit logic elif position.amount > 0 and context.daily_trend != 'up': # Exit long if daily trend changes self.order_target_percent(context.asset, 0) self.log.info("Exiting long: Daily trend changed") elif position.amount < 0 and context.daily_trend != 'down': # Exit short if daily trend changes self.order_target_percent(context.asset, 0) self.log.info("Exiting short: Daily trend changed") # Record self.record( daily_trend=1 if context.daily_trend == 'up' else (-1 if context.daily_trend == 'down' else 0), hourly_rsi=rsi, position=position.amount, ) def calculate_rsi(self, prices, period): """Calculate RSI""" deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) 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 print("✓ Basic multi-timeframe strategy defined")

2. Three-Timeframe Alignment¶

Require agreement across three timeframes for stronger signals.

Example: Weekly trend + Daily setup + Hourly trigger

In [ ]:

class ThreeTimeframeAlignment(TradingAlgorithm):
    """
    Trade only when all three timeframes align.
    
    Weekly: Overall trend (MA crossover)
    Daily: Setup condition (pullback to support)
    Hourly: Entry trigger (momentum confirmation)
    """
    
    def initialize(self, context):
        context.asset = self.symbol('AAPL')
        
        # Timeframe states
        context.weekly_trend = None
        context.daily_setup = False
        context.hourly_trigger = False
        
        # Schedule checks
        self.schedule_function(
            self.check_weekly_trend,
            date_rules=self.date_rules.week_start(),
            time_rules=self.time_rules.market_open()
        )
        
        self.schedule_function(
            self.check_daily_setup,
            date_rules=self.date_rules.every_day(),
            time_rules=self.time_rules.market_open()
        )
        
        self.schedule_function(
            self.check_hourly_trigger,
            date_rules=self.date_rules.every_day(),
            time_rules=self.time_rules.market_open(hours=1)
        )
    
    def check_weekly_trend(self, context, data):
        """Weekly: Determine overall trend"""
        weekly_prices = data.history(context.asset, 'close', 52, '1w')  # 1 year of weekly data
        
        if len(weekly_prices) < 52:
            return
        
        # 13-week vs 26-week MA
        fast_ma = weekly_prices[-13:].mean()
        slow_ma = weekly_prices[-26:].mean()
        
        if fast_ma > slow_ma * 1.01:  # 1% buffer
            context.weekly_trend = 'up'
        elif fast_ma < slow_ma * 0.99:
            context.weekly_trend = 'down'
        else:
            context.weekly_trend = None
        
        self.log.info(f"✓ Weekly trend: {context.weekly_trend}")
    
    def check_daily_setup(self, context, data):
        """Daily: Look for pullback to MA"""
        if context.weekly_trend is None:
            context.daily_setup = False
            return
        
        daily_prices = data.history(context.asset, 'close', 20, '1d')
        
        if len(daily_prices) < 20:
            context.daily_setup = False
            return
        
        current_price = daily_prices[-1]
        ma_20 = daily_prices.mean()
        
        # Setup: Price near 20-day MA (within 2%)
        distance_from_ma = abs(current_price - ma_20) / ma_20
        
        if distance_from_ma < 0.02:  # Within 2%
            if context.weekly_trend == 'up' and current_price < ma_20:
                context.daily_setup = True
                self.log.info("✓ Daily setup: Bullish pullback to MA")
            elif context.weekly_trend == 'down' and current_price > ma_20:
                context.daily_setup = True
                self.log.info("✓ Daily setup: Bearish rally to MA")
            else:
                context.daily_setup = False
        else:
            context.daily_setup = False
    
    def check_hourly_trigger(self, context, data):
        """Hourly: Look for momentum confirmation"""
        if not context.daily_setup:
            return
        
        # Get last 4 hours of data
        hourly_prices = data.history(context.asset, 'close', 4, '1h')
        
        if len(hourly_prices) < 4:
            return
        
        # Trigger: Price moving in trend direction for 2+ consecutive hours
        hourly_changes = np.diff(hourly_prices)
        
        if context.weekly_trend == 'up':
            # Need 2+ consecutive positive hours
            positive_hours = np.sum(hourly_changes[-3:] > 0)
            if positive_hours >= 2:
                context.hourly_trigger = True
                self.execute_trade(context, 'long')
        
        elif context.weekly_trend == 'down':
            # Need 2+ consecutive negative hours
            negative_hours = np.sum(hourly_changes[-3:] < 0)
            if negative_hours >= 2:
                context.hourly_trigger = True
                self.execute_trade(context, 'short')
    
    def execute_trade(self, context, direction):
        """Execute trade when all timeframes align"""
        position = context.portfolio.positions[context.asset]
        
        if direction == 'long' and position.amount == 0:
            self.order_target_percent(context.asset, 0.95)
            self.log.info(
                "🚀 LONG ENTRY: Weekly UP + Daily pullback + Hourly momentum"
            )
        
        elif direction == 'short' and position.amount == 0:
            self.order_target_percent(context.asset, -0.95)
            self.log.info(
                "🚀 SHORT ENTRY: Weekly DOWN + Daily rally + Hourly momentum"
            )
        
        # Reset triggers
        context.daily_setup = False
        context.hourly_trigger = False
    
    def handle_data(self, context, data):
        """Exit management"""
        position = context.portfolio.positions[context.asset]
        
        # Exit if weekly trend changes
        if position.amount > 0 and context.weekly_trend != 'up':
            self.order_target_percent(context.asset, 0)
            self.log.info("Exit long: Weekly trend changed")
        
        elif position.amount < 0 and context.weekly_trend != 'down':
            self.order_target_percent(context.asset, 0)
            self.log.info("Exit short: Weekly trend changed")

print("✓ Three-timeframe alignment strategy defined")

class ThreeTimeframeAlignment(TradingAlgorithm): """ Trade only when all three timeframes align. Weekly: Overall trend (MA crossover) Daily: Setup condition (pullback to support) Hourly: Entry trigger (momentum confirmation) """ def initialize(self, context): context.asset = self.symbol('AAPL') # Timeframe states context.weekly_trend = None context.daily_setup = False context.hourly_trigger = False # Schedule checks self.schedule_function( self.check_weekly_trend, date_rules=self.date_rules.week_start(), time_rules=self.time_rules.market_open() ) self.schedule_function( self.check_daily_setup, date_rules=self.date_rules.every_day(), time_rules=self.time_rules.market_open() ) self.schedule_function( self.check_hourly_trigger, date_rules=self.date_rules.every_day(), time_rules=self.time_rules.market_open(hours=1) ) def check_weekly_trend(self, context, data): """Weekly: Determine overall trend""" weekly_prices = data.history(context.asset, 'close', 52, '1w') # 1 year of weekly data if len(weekly_prices) < 52: return # 13-week vs 26-week MA fast_ma = weekly_prices[-13:].mean() slow_ma = weekly_prices[-26:].mean() if fast_ma > slow_ma * 1.01: # 1% buffer context.weekly_trend = 'up' elif fast_ma < slow_ma * 0.99: context.weekly_trend = 'down' else: context.weekly_trend = None self.log.info(f"✓ Weekly trend: {context.weekly_trend}") def check_daily_setup(self, context, data): """Daily: Look for pullback to MA""" if context.weekly_trend is None: context.daily_setup = False return daily_prices = data.history(context.asset, 'close', 20, '1d') if len(daily_prices) < 20: context.daily_setup = False return current_price = daily_prices[-1] ma_20 = daily_prices.mean() # Setup: Price near 20-day MA (within 2%) distance_from_ma = abs(current_price - ma_20) / ma_20 if distance_from_ma < 0.02: # Within 2% if context.weekly_trend == 'up' and current_price < ma_20: context.daily_setup = True self.log.info("✓ Daily setup: Bullish pullback to MA") elif context.weekly_trend == 'down' and current_price > ma_20: context.daily_setup = True self.log.info("✓ Daily setup: Bearish rally to MA") else: context.daily_setup = False else: context.daily_setup = False def check_hourly_trigger(self, context, data): """Hourly: Look for momentum confirmation""" if not context.daily_setup: return # Get last 4 hours of data hourly_prices = data.history(context.asset, 'close', 4, '1h') if len(hourly_prices) < 4: return # Trigger: Price moving in trend direction for 2+ consecutive hours hourly_changes = np.diff(hourly_prices) if context.weekly_trend == 'up': # Need 2+ consecutive positive hours positive_hours = np.sum(hourly_changes[-3:] > 0) if positive_hours >= 2: context.hourly_trigger = True self.execute_trade(context, 'long') elif context.weekly_trend == 'down': # Need 2+ consecutive negative hours negative_hours = np.sum(hourly_changes[-3:] < 0) if negative_hours >= 2: context.hourly_trigger = True self.execute_trade(context, 'short') def execute_trade(self, context, direction): """Execute trade when all timeframes align""" position = context.portfolio.positions[context.asset] if direction == 'long' and position.amount == 0: self.order_target_percent(context.asset, 0.95) self.log.info( "🚀 LONG ENTRY: Weekly UP + Daily pullback + Hourly momentum" ) elif direction == 'short' and position.amount == 0: self.order_target_percent(context.asset, -0.95) self.log.info( "🚀 SHORT ENTRY: Weekly DOWN + Daily rally + Hourly momentum" ) # Reset triggers context.daily_setup = False context.hourly_trigger = False def handle_data(self, context, data): """Exit management""" position = context.portfolio.positions[context.asset] # Exit if weekly trend changes if position.amount > 0 and context.weekly_trend != 'up': self.order_target_percent(context.asset, 0) self.log.info("Exit long: Weekly trend changed") elif position.amount < 0 and context.weekly_trend != 'down': self.order_target_percent(context.asset, 0) self.log.info("Exit short: Weekly trend changed") print("✓ Three-timeframe alignment strategy defined")

3. Adaptive Timeframe Selection¶

Switch timeframes based on market volatility.

High volatility → Use longer timeframes (less noise)
Low volatility → Use shorter timeframes (more opportunities)

In [ ]:

class AdaptiveTimeframeStrategy(TradingAlgorithm):
    """
    Adapt timeframes based on market volatility.
    """
    
    def initialize(self, context):
        context.asset = self.symbol('SPY')
        
        # Volatility thresholds
        context.high_vol_threshold = 0.02  # 2% daily volatility
        context.low_vol_threshold = 0.01   # 1% daily volatility
        
        # Current regime
        context.vol_regime = 'medium'  # 'low', 'medium', 'high'
        context.current_timeframe = '1h'  # Active trading timeframe
        
        # Schedule volatility assessment
        self.schedule_function(
            self.assess_volatility,
            date_rules=self.date_rules.every_day(),
            time_rules=self.time_rules.market_open()
        )
        
        # Schedule trading
        self.schedule_function(
            self.trade_adaptive,
            date_rules=self.date_rules.every_day(),
            time_rules=self.time_rules.market_open(hours=1)
        )
    
    def assess_volatility(self, context, data):
        """Assess market volatility and adapt timeframe"""
        # Calculate 20-day historical volatility
        daily_prices = data.history(context.asset, 'close', 21, '1d')
        
        if len(daily_prices) < 21:
            return
        
        daily_returns = np.diff(daily_prices) / daily_prices[:-1]
        volatility = np.std(daily_returns)
        
        # Determine regime
        if volatility > context.high_vol_threshold:
            context.vol_regime = 'high'
            context.current_timeframe = '4h'  # Longer timeframe for high vol
            context.ma_period = 10
        elif volatility < context.low_vol_threshold:
            context.vol_regime = 'low'
            context.current_timeframe = '15m'  # Shorter timeframe for low vol
            context.ma_period = 20
        else:
            context.vol_regime = 'medium'
            context.current_timeframe = '1h'  # Standard timeframe
            context.ma_period = 15
        
        self.log.info(
            f"Volatility: {volatility:.4f} | "
            f"Regime: {context.vol_regime} | "
            f"Timeframe: {context.current_timeframe}"
        )
        
        self.record(
            volatility=volatility,
            regime={
                'low': 0,
                'medium': 1,
                'high': 2
            }[context.vol_regime]
        )
    
    def trade_adaptive(self, context, data):
        """Trade using adaptive timeframe"""
        # Get data for current timeframe
        prices = data.history(
            context.asset,
            'close',
            context.ma_period * 2,
            context.current_timeframe
        )
        
        if len(prices) < context.ma_period * 2:
            return
        
        # Simple MA crossover on adaptive timeframe
        fast_ma = prices[-context.ma_period:].mean()
        slow_ma = prices.mean()
        current_price = prices[-1]
        
        position = context.portfolio.positions[context.asset]
        
        # Entry/exit logic
        if fast_ma > slow_ma and position.amount == 0:
            # Bullish crossover
            self.order_target_percent(context.asset, 0.95)
            self.log.info(
                f"LONG entry on {context.current_timeframe} timeframe "
                f"(Vol regime: {context.vol_regime})"
            )
        
        elif fast_ma < slow_ma and position.amount > 0:
            # Bearish crossover
            self.order_target_percent(context.asset, 0)
            self.log.info(
                f"Exit long on {context.current_timeframe} timeframe"
            )

print("✓ Adaptive timeframe strategy defined")

class AdaptiveTimeframeStrategy(TradingAlgorithm): """ Adapt timeframes based on market volatility. """ def initialize(self, context): context.asset = self.symbol('SPY') # Volatility thresholds context.high_vol_threshold = 0.02 # 2% daily volatility context.low_vol_threshold = 0.01 # 1% daily volatility # Current regime context.vol_regime = 'medium' # 'low', 'medium', 'high' context.current_timeframe = '1h' # Active trading timeframe # Schedule volatility assessment self.schedule_function( self.assess_volatility, date_rules=self.date_rules.every_day(), time_rules=self.time_rules.market_open() ) # Schedule trading self.schedule_function( self.trade_adaptive, date_rules=self.date_rules.every_day(), time_rules=self.time_rules.market_open(hours=1) ) def assess_volatility(self, context, data): """Assess market volatility and adapt timeframe""" # Calculate 20-day historical volatility daily_prices = data.history(context.asset, 'close', 21, '1d') if len(daily_prices) < 21: return daily_returns = np.diff(daily_prices) / daily_prices[:-1] volatility = np.std(daily_returns) # Determine regime if volatility > context.high_vol_threshold: context.vol_regime = 'high' context.current_timeframe = '4h' # Longer timeframe for high vol context.ma_period = 10 elif volatility < context.low_vol_threshold: context.vol_regime = 'low' context.current_timeframe = '15m' # Shorter timeframe for low vol context.ma_period = 20 else: context.vol_regime = 'medium' context.current_timeframe = '1h' # Standard timeframe context.ma_period = 15 self.log.info( f"Volatility: {volatility:.4f} | " f"Regime: {context.vol_regime} | " f"Timeframe: {context.current_timeframe}" ) self.record( volatility=volatility, regime={ 'low': 0, 'medium': 1, 'high': 2 }[context.vol_regime] ) def trade_adaptive(self, context, data): """Trade using adaptive timeframe""" # Get data for current timeframe prices = data.history( context.asset, 'close', context.ma_period * 2, context.current_timeframe ) if len(prices) < context.ma_period * 2: return # Simple MA crossover on adaptive timeframe fast_ma = prices[-context.ma_period:].mean() slow_ma = prices.mean() current_price = prices[-1] position = context.portfolio.positions[context.asset] # Entry/exit logic if fast_ma > slow_ma and position.amount == 0: # Bullish crossover self.order_target_percent(context.asset, 0.95) self.log.info( f"LONG entry on {context.current_timeframe} timeframe " f"(Vol regime: {context.vol_regime})" ) elif fast_ma < slow_ma and position.amount > 0: # Bearish crossover self.order_target_percent(context.asset, 0) self.log.info( f"Exit long on {context.current_timeframe} timeframe" ) print("✓ Adaptive timeframe strategy defined")

4. Higher Timeframe Stops¶

Use higher timeframe structure for stop placement.

In [ ]:

class HigherTimeframeStops(TradingAlgorithm):
    """
    Enter on 1-hour signals, but place stops based on daily structure.
    """
    
    def initialize(self, context):
        context.asset = self.symbol('AAPL')
        context.stop_order_id = None
        context.entry_price = None
        
        self.schedule_function(
            self.check_entry,
            date_rules=self.date_rules.every_day(),
            time_rules=self.time_rules.market_open(hours=1)
        )
    
    def check_entry(self, context, data):
        """Check for entry on hourly timeframe"""
        position = context.portfolio.positions[context.asset]
        
        if position.amount != 0:
            return  # Already in position
        
        # Hourly entry signal (simple RSI oversold)
        hourly_prices = data.history(context.asset, 'close', 15, '1h')
        
        if len(hourly_prices) < 15:
            return
        
        rsi = self.calculate_rsi(hourly_prices, 14)
        
        if rsi < 30:  # Oversold
            # Get daily support level for stop
            daily_low = self.get_daily_support(context, data)
            
            if daily_low is None:
                return
            
            # Enter long
            shares = 100
            self.order(context.asset, shares)
            context.entry_price = hourly_prices[-1]
            
            # Place stop below daily support
            stop_price = daily_low * 0.98  # 2% below support
            
            from rustybt.finance import StopOrder
            context.stop_order_id = self.order(
                context.asset,
                -shares,
                style=StopOrder(stop_price)
            )
            
            self.log.info(
                f"LONG entry: Hourly RSI={rsi:.1f} | "
                f"Entry=${context.entry_price:.2f} | "
                f"Daily support=${daily_low:.2f} | "
                f"Stop=${stop_price:.2f}"
            )
            
            self.record(
                entry_price=context.entry_price,
                stop_price=stop_price,
                daily_support=daily_low,
            )
    
    def get_daily_support(self, context, data):
        """Find support on daily timeframe (20-day low)"""
        daily_lows = data.history(context.asset, 'low', 20, '1d')
        
        if len(daily_lows) < 20:
            return None
        
        return daily_lows.min()
    
    def calculate_rsi(self, prices, period):
        """Calculate RSI"""
        deltas = np.diff(prices)
        gains = np.where(deltas > 0, deltas, 0)
        losses = np.where(deltas < 0, -deltas, 0)
        
        avg_gain = np.mean(gains[-period:])
        avg_loss = np.mean(losses[-period:])
        
        if avg_loss == 0:
            return 100
        
        rs = avg_gain / avg_loss
        return 100 - (100 / (1 + rs))

print("✓ Higher timeframe stops strategy defined")

class HigherTimeframeStops(TradingAlgorithm): """ Enter on 1-hour signals, but place stops based on daily structure. """ def initialize(self, context): context.asset = self.symbol('AAPL') context.stop_order_id = None context.entry_price = None self.schedule_function( self.check_entry, date_rules=self.date_rules.every_day(), time_rules=self.time_rules.market_open(hours=1) ) def check_entry(self, context, data): """Check for entry on hourly timeframe""" position = context.portfolio.positions[context.asset] if position.amount != 0: return # Already in position # Hourly entry signal (simple RSI oversold) hourly_prices = data.history(context.asset, 'close', 15, '1h') if len(hourly_prices) < 15: return rsi = self.calculate_rsi(hourly_prices, 14) if rsi < 30: # Oversold # Get daily support level for stop daily_low = self.get_daily_support(context, data) if daily_low is None: return # Enter long shares = 100 self.order(context.asset, shares) context.entry_price = hourly_prices[-1] # Place stop below daily support stop_price = daily_low * 0.98 # 2% below support from rustybt.finance import StopOrder context.stop_order_id = self.order( context.asset, -shares, style=StopOrder(stop_price) ) self.log.info( f"LONG entry: Hourly RSI={rsi:.1f} | " f"Entry=${context.entry_price:.2f} | " f"Daily support=${daily_low:.2f} | " f"Stop=${stop_price:.2f}" ) self.record( entry_price=context.entry_price, stop_price=stop_price, daily_support=daily_low, ) def get_daily_support(self, context, data): """Find support on daily timeframe (20-day low)""" daily_lows = data.history(context.asset, 'low', 20, '1d') if len(daily_lows) < 20: return None return daily_lows.min() def calculate_rsi(self, prices, period): """Calculate RSI""" deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = np.mean(gains[-period:]) avg_loss = np.mean(losses[-period:]) if avg_loss == 0: return 100 rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) print("✓ Higher timeframe stops strategy defined")

Summary¶

Key Concepts¶

1. Timeframe Hierarchy - Higher timeframes provide context, lower provide timing
2. Alignment - Require multiple timeframes to agree for stronger signals
3. Adaptation - Adjust timeframes based on market conditions
4. Structure - Use higher timeframe levels for stops and targets

Common Timeframe Combinations¶

Strategy Type	Higher TF	Mid TF	Lower TF
Swing Trading	Weekly	Daily	4-hour
Day Trading	Daily	1-hour	15-min
Scalping	1-hour	15-min	5-min
Position Trading	Monthly	Weekly	Daily

Best Practices¶

✅ Use 3-4x ratio between timeframes (e.g., daily → 4-hour → 1-hour)
✅ Higher TF for trend - Always know the bigger picture
✅ Lower TF for entries - Fine-tune entry timing
✅ Higher TF for stops - Avoid getting stopped out by noise
✅ Require alignment - Don't fight higher timeframe trends
✅ Adapt to volatility - Use longer timeframes when markets are choppy
✅ Be patient - Wait for all timeframes to align

Common Mistakes to Avoid¶

❌ Trading against higher timeframe trend
❌ Using stops based on lower timeframe noise
❌ Ignoring timeframe alignment
❌ Using timeframes that are too close together
❌ Overcomplicating with too many timeframes

Next Steps¶

• Experiment with different timeframe combinations
• Backtest performance across various market conditions
• Combine with other techniques (Pipeline, optimization)
• Test in paper trading before going live

Resources¶

• Multi-Timeframe Analysis Guide
• Data History Reference
• Timeframe Best Practices