crypto_trader/docs/developer/creating_strategies.md

Creating Custom Strategies

This guide explains how to create custom trading strategies for Crypto Trader.

Strategy Framework

All strategies inherit from BaseStrategy and implement a standardized interface. This allows the trading engine to execute any strategy uniformly.

Basic Strategy Structure

from src.strategies.base import BaseStrategy, StrategyRegistry
from src.data.indicators import get_indicators
import pandas as pd
from typing import Dict, Any

class MyCustomStrategy(BaseStrategy):
    """My custom trading strategy."""
    
    def __init__(self, strategy_id: int, name: str, symbol: str, timeframe: str, parameters: Dict[str, Any]):
        super().__init__(strategy_id, name, symbol, timeframe, parameters)
        # Initialize strategy-specific parameters
        self.my_param = parameters.get("my_param", 10)
    
    async def on_data(self, new_data: pd.DataFrame):
        """Called when new data is available."""
        # Update internal data
        self.current_data = pd.concat([self.current_data, new_data])
        
        # Generate signal if enough data
        if len(self.current_data) >= self.my_param:
            signal = await self.generate_signal()
            if signal["signal"] != "hold":
                # Signal generated - will be handled by trading engine
                pass
    
    async def generate_signal(self) -> Dict[str, Any]:
        """Generate trading signal."""
        # Calculate indicators
        indicators = get_indicators()
        rsi = indicators.rsi(self.current_data['close'], period=14)
        
        # Generate signal based on strategy logic
        current_rsi = rsi.iloc[-1]
        current_price = self.current_data['close'].iloc[-1]
        
        if current_rsi < 30:
            return {"signal": "buy", "price": current_price}
        elif current_rsi > 70:
            return {"signal": "sell", "price": current_price}
        else:
            return {"signal": "hold", "price": current_price}
    
    async def calculate_position_size(self, capital: float, risk_percentage: float) -> float:
        """Calculate position size."""
        return capital * risk_percentage

# Register strategy
StrategyRegistry.register_strategy("my_custom", MyCustomStrategy)

Required Methods

on_data(new_data: pd.DataFrame)

Called when new OHLCV data is available for the strategy's timeframe.

Responsibilities:

• Update internal data storage
• Calculate indicators
• Generate signals when conditions are met

generate_signal() -> Dict[str, Any]

Analyzes current data and generates a trading signal.

Returns:

{
    "signal": "buy" | "sell" | "hold",
    "price": float,  # Optional, current price
    "confidence": float,  # Optional, 0.0 to 1.0
}

calculate_position_size(capital: float, risk_percentage: float) -> float

Calculates the appropriate position size based on capital and risk.

Parameters:

• capital: Available trading capital
• risk_percentage: Percentage of capital to risk (0.0 to 1.0)

Returns: Position size in base currency

Using Technical Indicators

Access indicators through the indicators library:

from src.data.indicators import get_indicators

indicators = get_indicators()

# Calculate indicators
sma = indicators.sma(data['close'], period=20)
ema = indicators.ema(data['close'], period=20)
rsi = indicators.rsi(data['close'], period=14)
macd_result = indicators.macd(data['close'], fast=12, slow=26, signal=9)
bbands = indicators.bollinger_bands(data['close'], period=20, std_dev=2)

Multi-Timeframe Strategies

Strategies can use multiple timeframes:

from src.strategies.timeframe_manager import get_timeframe_manager

timeframe_manager = get_timeframe_manager()

# Get data from different timeframes
primary_data = timeframe_manager.get_data(self.symbol, self.timeframe)
higher_tf_data = timeframe_manager.get_data(self.symbol, "1d")  # Daily for trend

# Use higher timeframe for trend confirmation
if higher_tf_data is not None and len(higher_tf_data) > 0:
    daily_trend = higher_tf_data['close'].iloc[-1] > higher_tf_data['close'].iloc[-20]
    if daily_trend:
        # Only trade in direction of higher timeframe trend
        pass

Strategy Parameters

Define configurable parameters:

def __init__(self, strategy_id: int, name: str, symbol: str, timeframe: str, parameters: Dict[str, Any]):
    super().__init__(strategy_id, name, symbol, timeframe, parameters)
    
    # Required parameters with defaults
    self.rsi_period = parameters.get("rsi_period", 14)
    self.overbought = parameters.get("overbought", 70)
    self.oversold = parameters.get("oversold", 30)
    
    # Validate parameters
    if not (0 < self.oversold < self.overbought < 100):
        raise ValueError("Invalid RSI thresholds")

Strategy Registration

Register your strategy so it can be used:

from src.strategies.base import StrategyRegistry

StrategyRegistry.register_strategy("my_strategy", MyCustomStrategy)

Strategy Lifecycle

1. Initialization: Strategy is created with parameters
2. Start: start() is called when strategy is activated
3. Data Updates: on_data() is called with new candles
4. Signal Generation: generate_signal() is called when conditions are met
5. Order Execution: Trading engine executes signals
6. Stop: stop() is called when strategy is deactivated

Best Practices

1. Parameter Validation: Validate all parameters in __init__
2. Error Handling: Handle missing data and calculation errors
3. Logging: Use strategy logger for important events
4. Testing: Write unit tests for your strategy
5. Documentation: Document strategy logic and parameters
6. Backtesting: Always backtest before live trading

Example: Complete Strategy

See src/strategies/technical/rsi_strategy.py for a complete example.

Available Strategy Examples

The codebase includes several advanced strategy implementations that serve as examples:

Multi-Indicator Confirmation

See src/strategies/technical/confirmed_strategy.py for an example of combining multiple indicators.

Divergence Detection

See src/strategies/technical/divergence_strategy.py for an example of divergence detection using the indicators API.

Ensemble Methods

See src/strategies/ensemble/consensus_strategy.py for an example of combining multiple strategies with weighted voting.

Trend Filtering

All strategies can use the optional trend filter method from BaseStrategy:

signal = strategy.apply_trend_filter(signal, ohlcv_data, adx_period=14, min_adx=25.0)

Testing Your Strategy

import pytest
from src.strategies.my_strategy import MyCustomStrategy

def test_strategy_signal_generation():
    """Test strategy signal generation."""
    strategy = MyCustomStrategy(
        strategy_id=1,
        name="Test Strategy",
        symbol="BTC/USD",
        timeframe="1h",
        parameters={"rsi_period": 14}
    )
    
    # Create test data
    test_data = pd.DataFrame({
        'close': [100, 101, 102, 103, 104]
    })
    
    # Test signal generation
    signal = await strategy.generate_signal()
    assert signal["signal"] in ["buy", "sell", "hold"]