Custom Trading Indicators

@abstractmethod def getIndicatorSpec(self) -> IndicatorSpec: # Return IndicatorDefinition with name, description, required params, and valuetype # See IndicatorSpec abstract class for more details pass @abstractmethod def computeSeries(self, params: Dict[str, Any], bars: List[Bar]) -> List[DatedValue]: # This is the main function which must compute and return value. This function must return within 1 sec # or it will be killed pass

from typing import Dict, List, Any from datetime import datetime import numpy as np class MACrossoverSignal(Indicator): def getIndicatorSpec(self) -> IndicatorSpec: spec = IndicatorSpec() spec.name = "MA Crossover Signal" spec.description = "Generates buy signal when fast MA crosses above slow MA" # Fast MA period parameter fastParam = IntegerParameterDefinition() fastParam.name = "fastPeriod" fastParam.description = "Fast moving average period" fastParam.defaultValue = 5 fastParam.minValue = 2 fastParam.maxValue = 50 spec.parameters.append(fastParam) # Slow MA period parameter slowParam = IntegerParameterDefinition() slowParam.name = "slowPeriod" slowParam.description = "Slow moving average period" slowParam.defaultValue = 20 slowParam.minValue = 5 slowParam.maxValue = 200 spec.parameters.append(slowParam) # Return boolean values (true for buy signal, false for no signal) spec.valueType = ValueType.BOOLEAN return spec def computeSeries(self, params: Dict[str, Any], bars: List[Bar]) -> List[DatedValue]: # Extract parameters fast_period = params["fastPeriod"] slow_period = params["slowPeriod"] # Extract closing prices from bars closes = [bar.close for bar in bars] # Calculate moving averages fast_ma = self._calculate_sma(closes, fast_period) slow_ma = self._calculate_sma(closes, slow_period) # Generate signals when fast MA crosses above slow MA result = [] for i in range(len(bars)): if i < slow_period: # Not enough data for both MAs signal = False elif i == slow_period: # First point where we have both MAs, no previous point to compare signal = False else: # Check for crossover: fast MA was below slow MA yesterday and is above today yesterday_fast_above_slow = fast_ma[i-1] > slow_ma[i-1] today_fast_above_slow = fast_ma[i] > slow_ma[i] signal = not yesterday_fast_above_slow and today_fast_above_slow # Add result with date from bar result.append(DatedValue(bars[i].date, signal)) return result def _calculate_sma(self, values, period): # Using numpy for efficient calculation result = np.zeros(len(values)) for i in range(len(values)): if i < period - 1: result[i]=float('nan') # Not enough data else: result[i] = sum(values[i-(period-1):i+1]) / period return result