- 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.
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
- Strategy-Controlled Timeframes: Each strategy specifies its required timeframes
- Automatic Resampling: Framework resamples 1-minute data to strategy needs
- Precision Execution: All strategies maintain 1-minute data for accurate stop-loss execution
- 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
-
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] -
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] -
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
- Dynamic Timeframe Switching: Strategies adapt timeframes based on market conditions
- Timeframe Confidence Weighting: Different confidence levels per timeframe
- Cross-Timeframe Signal Validation: Automatic signal confirmation across timeframes
- 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