Cycles/docs/timeframe_system.md
Vasily.onl 9629d3090b Enhance README and documentation for Cycles framework
- Expanded the README.md to provide a comprehensive overview of the Cycles framework, including features, quick start instructions, and configuration examples.
- Updated strategies documentation to detail the architecture, available strategies, and their configurations, emphasizing the new multi-timeframe capabilities.
- Added a new timeframe system documentation to explain the strategy-controlled timeframe management and automatic data resampling.
- Improved the strategy manager documentation to clarify its role in orchestrating multiple strategies and combining signals effectively.
- Adjusted configuration examples to reflect recent changes in strategy parameters and usage.
2025-05-23 17:06:35 +08:00

13 KiB

Timeframe System Documentation

Overview

The Cycles framework features a sophisticated timeframe management system that allows strategies to operate on their preferred timeframes while maintaining precise execution control. This system supports both single-timeframe and multi-timeframe strategies with automatic data resampling and intelligent signal mapping.

Architecture

Core Concepts

  1. Strategy-Controlled Timeframes: Each strategy specifies its required timeframes
  2. Automatic Resampling: Framework resamples 1-minute data to strategy needs
  3. Precision Execution: All strategies maintain 1-minute data for accurate stop-loss execution
  4. Signal Mapping: Intelligent mapping between different timeframe resolutions

Data Flow

Original 1min Data
        ↓
Strategy.get_timeframes() → ["15min", "1h"]
        ↓
Automatic Resampling
        ↓
Strategy Logic (15min + 1h analysis)
        ↓
Signal Generation
        ↓
Map to Working Timeframe
        ↓
Backtesting Engine

Strategy Timeframe Interface

StrategyBase Methods

All strategies inherit timeframe capabilities from StrategyBase:

class MyStrategy(StrategyBase):
    def get_timeframes(self) -> List[str]:
        """Specify required timeframes for this strategy"""
        return ["15min", "1h"]  # Strategy needs both timeframes
    
    def initialize(self, backtester) -> None:
        # Automatic resampling happens here
        self._resample_data(backtester.original_df)
        
        # Access resampled data
        data_15m = self.get_data_for_timeframe("15min")
        data_1h = self.get_data_for_timeframe("1h")
        
        # Calculate indicators on each timeframe
        self.indicators_15m = self._calculate_indicators(data_15m)
        self.indicators_1h = self._calculate_indicators(data_1h)
        
        self.initialized = True

Data Access Methods

# Get data for specific timeframe
data_15m = strategy.get_data_for_timeframe("15min")

# Get primary timeframe data (first in list)
primary_data = strategy.get_primary_timeframe_data()

# Check available timeframes
timeframes = strategy.get_timeframes()

Supported Timeframes

Standard Timeframes

  • "1min": 1-minute bars (original resolution)
  • "5min": 5-minute bars
  • "15min": 15-minute bars
  • "30min": 30-minute bars
  • "1h": 1-hour bars
  • "4h": 4-hour bars
  • "1d": Daily bars

Custom Timeframes

Any pandas-compatible frequency string is supported:

  • "2min": 2-minute bars
  • "10min": 10-minute bars
  • "2h": 2-hour bars
  • "12h": 12-hour bars

Strategy Examples

Single Timeframe Strategy

class SingleTimeframeStrategy(StrategyBase):
    def get_timeframes(self):
        return ["15min"]  # Only needs 15-minute data
    
    def initialize(self, backtester):
        self._resample_data(backtester.original_df)
        
        # Work with 15-minute data
        data = self.get_primary_timeframe_data()
        self.indicators = self._calculate_indicators(data)
        self.initialized = True
    
    def get_entry_signal(self, backtester, df_index):
        # df_index refers to 15-minute data
        if self.indicators['signal'][df_index]:
            return StrategySignal("ENTRY", confidence=0.8)
        return StrategySignal("HOLD", confidence=0.0)

Multi-Timeframe Strategy

class MultiTimeframeStrategy(StrategyBase):
    def get_timeframes(self):
        return ["15min", "1h", "4h"]  # Multiple timeframes
    
    def initialize(self, backtester):
        self._resample_data(backtester.original_df)
        
        # Access different timeframes
        self.data_15m = self.get_data_for_timeframe("15min")
        self.data_1h = self.get_data_for_timeframe("1h")
        self.data_4h = self.get_data_for_timeframe("4h")
        
        # Calculate indicators on each timeframe
        self.trend_4h = self._calculate_trend(self.data_4h)
        self.momentum_1h = self._calculate_momentum(self.data_1h)
        self.entry_signals_15m = self._calculate_entries(self.data_15m)
        
        self.initialized = True
    
    def get_entry_signal(self, backtester, df_index):
        # Primary timeframe is 15min (first in list)
        # Map df_index to other timeframes for confirmation
        
        # Get current 15min timestamp
        current_time = self.data_15m.index[df_index]
        
        # Find corresponding indices in other timeframes
        h1_idx = self.data_1h.index.get_indexer([current_time], method='ffill')[0]
        h4_idx = self.data_4h.index.get_indexer([current_time], method='ffill')[0]
        
        # Multi-timeframe confirmation
        trend_ok = self.trend_4h[h4_idx] > 0
        momentum_ok = self.momentum_1h[h1_idx] > 0.5
        entry_signal = self.entry_signals_15m[df_index]
        
        if trend_ok and momentum_ok and entry_signal:
            confidence = 0.9  # High confidence with all timeframes aligned
            return StrategySignal("ENTRY", confidence=confidence)
        
        return StrategySignal("HOLD", confidence=0.0)

Configurable Timeframe Strategy

class ConfigurableStrategy(StrategyBase):
    def get_timeframes(self):
        # Strategy timeframe configurable via parameters
        primary_tf = self.params.get("timeframe", "15min")
        return [primary_tf, "1min"]  # Primary + 1min for stop-loss
    
    def initialize(self, backtester):
        self._resample_data(backtester.original_df)
        
        primary_tf = self.get_timeframes()[0]
        self.data = self.get_data_for_timeframe(primary_tf)
        
        # Indicator parameters can also be timeframe-dependent
        if primary_tf == "5min":
            self.ma_period = 20
        elif primary_tf == "15min":
            self.ma_period = 14
        else:
            self.ma_period = 10
            
        self.indicators = self._calculate_indicators(self.data)
        self.initialized = True

Built-in Strategy Timeframe Behavior

Default Strategy

Timeframes: Configurable primary + 1min for stop-loss

# Configuration
{
    "name": "default",
    "params": {
        "timeframe": "5min"  # Configurable timeframe
    }
}

# Resulting timeframes: ["5min", "1min"]

Features:

  • Supertrend analysis on configured timeframe
  • 1-minute precision for stop-loss execution
  • Optimized for 15-minute default, but works on any timeframe

BBRS Strategy

Timeframes: 1min input (internal resampling)

# Configuration
{
    "name": "bbrs",
    "params": {
        "strategy_name": "MarketRegimeStrategy"
    }
}

# Resulting timeframes: ["1min"]

Features:

  • Uses 1-minute data as input
  • Internal resampling to 15min/1h by Strategy class
  • Signals mapped back to 1-minute resolution
  • No double-resampling issues

Advanced Features

Timeframe Synchronization

When working with multiple timeframes, synchronization is crucial:

def _get_synchronized_signals(self, df_index, primary_timeframe="15min"):
    """Get signals synchronized across timeframes"""
    
    # Get timestamp from primary timeframe
    primary_data = self.get_data_for_timeframe(primary_timeframe)
    current_time = primary_data.index[df_index]
    
    signals = {}
    for tf in self.get_timeframes():
        if tf == primary_timeframe:
            signals[tf] = df_index
        else:
            # Find corresponding index in other timeframe
            tf_data = self.get_data_for_timeframe(tf)
            tf_idx = tf_data.index.get_indexer([current_time], method='ffill')[0]
            signals[tf] = tf_idx
    
    return signals

Dynamic Timeframe Selection

Strategies can adapt timeframes based on market conditions:

class AdaptiveStrategy(StrategyBase):
    def get_timeframes(self):
        # Fixed set of timeframes strategy might need
        return ["5min", "15min", "1h"]
    
    def _select_active_timeframe(self, market_volatility):
        """Select timeframe based on market conditions"""
        if market_volatility > 0.8:
            return "5min"   # High volatility -> shorter timeframe
        elif market_volatility > 0.4:
            return "15min"  # Medium volatility -> medium timeframe
        else:
            return "1h"     # Low volatility -> longer timeframe
    
    def get_entry_signal(self, backtester, df_index):
        # Calculate market volatility
        volatility = self._calculate_volatility(df_index)
        
        # Select appropriate timeframe
        active_tf = self._select_active_timeframe(volatility)
        
        # Generate signal on selected timeframe
        return self._generate_signal_for_timeframe(active_tf, df_index)

Configuration Examples

Single Timeframe Configuration

{
    "strategies": [
        {
            "name": "default",
            "weight": 1.0,
            "params": {
                "timeframe": "15min",
                "stop_loss_pct": 0.03
            }
        }
    ]
}

Multi-Timeframe Configuration

{
    "strategies": [
        {
            "name": "multi_timeframe_strategy",
            "weight": 1.0,
            "params": {
                "primary_timeframe": "15min",
                "confirmation_timeframes": ["1h", "4h"],
                "signal_timeframe": "5min"
            }
        }
    ]
}

Mixed Strategy Configuration

{
    "strategies": [
        {
            "name": "default",
            "weight": 0.6,
            "params": {
                "timeframe": "15min"
            }
        },
        {
            "name": "bbrs",
            "weight": 0.4,
            "params": {
                "strategy_name": "MarketRegimeStrategy"
            }
        }
    ]
}

Performance Considerations

Memory Usage

  • Only required timeframes are resampled and stored
  • Original 1-minute data shared across all strategies
  • Efficient pandas resampling with minimal memory overhead

Processing Speed

  • Resampling happens once during initialization
  • No repeated resampling during backtesting
  • Vectorized operations on pre-computed timeframes

Data Alignment

  • All timeframes aligned to original 1-minute timestamps
  • Forward-fill resampling ensures data availability
  • Intelligent handling of missing data points

Best Practices

1. Minimize Timeframe Requirements

# Good - minimal timeframes
def get_timeframes(self):
    return ["15min"]

# Less optimal - unnecessary timeframes
def get_timeframes(self):
    return ["1min", "5min", "15min", "1h", "4h", "1d"]

2. Use Appropriate Timeframes for Strategy Logic

# Good - timeframe matches strategy logic
class TrendStrategy(StrategyBase):
    def get_timeframes(self):
        return ["1h"]  # Trend analysis works well on hourly data

class ScalpingStrategy(StrategyBase):
    def get_timeframes(self):
        return ["1min", "5min"]  # Scalping needs fine-grained data

3. Include 1min for Stop-Loss Precision

def get_timeframes(self):
    primary_tf = self.params.get("timeframe", "15min")
    timeframes = [primary_tf]
    
    # Always include 1min for precise stop-loss
    if "1min" not in timeframes:
        timeframes.append("1min")
    
    return timeframes

4. Handle Timeframe Edge Cases

def get_entry_signal(self, backtester, df_index):
    # Check bounds for all timeframes
    if df_index >= len(self.get_primary_timeframe_data()):
        return StrategySignal("HOLD", confidence=0.0)
    
    # Robust timeframe indexing
    try:
        signal = self._calculate_signal(df_index)
        return signal
    except IndexError:
        return StrategySignal("HOLD", confidence=0.0)

Troubleshooting

Common Issues

  1. Index Out of Bounds

    # Problem: Different timeframes have different lengths
    # Solution: Always check bounds
    if df_index < len(self.data_1h):
        signal = self.data_1h[df_index]
    
  2. Timeframe Misalignment

    # Problem: Assuming same index across timeframes
    # Solution: Use timestamp-based alignment
    current_time = primary_data.index[df_index]
    h1_idx = hourly_data.index.get_indexer([current_time], method='ffill')[0]
    
  3. Memory Issues with Large Datasets

    # Solution: Only include necessary timeframes
    def get_timeframes(self):
        # Return minimal set
        return ["15min"]  # Not ["1min", "5min", "15min", "1h"]
    

Debugging Tips

# Log timeframe information
def initialize(self, backtester):
    self._resample_data(backtester.original_df)
    
    for tf in self.get_timeframes():
        data = self.get_data_for_timeframe(tf)
        print(f"Timeframe {tf}: {len(data)} bars, "
              f"from {data.index[0]} to {data.index[-1]}")
    
    self.initialized = True

Future Enhancements

Planned Features

  1. Dynamic Timeframe Switching: Strategies adapt timeframes based on market conditions
  2. Timeframe Confidence Weighting: Different confidence levels per timeframe
  3. Cross-Timeframe Signal Validation: Automatic signal confirmation across timeframes
  4. Optimized Memory Management: Lazy loading and caching for large datasets

Extension Points

The timeframe system is designed for easy extension:

  • Custom resampling methods
  • Alternative timeframe synchronization strategies
  • Market-specific timeframe preferences
  • Real-time timeframe adaptation