0
# Utilities
1

2
Core utility functions and base classes used throughout the TA library for data handling, indicator calculations, and common operations. These utilities provide the foundation for all technical indicators and enable consistent behavior across the library.
3

4
## Capabilities
5

6
### IndicatorMixin Base Class
7

8
Base class for all technical indicators, providing common functionality and ensuring consistent behavior across all indicator implementations.
9

10
```python { .api }
11
class IndicatorMixin:
12
    """
13
    Utility mixin indicator class - Base class for all indicators.
14
    
15
    Provides common functionality for handling NaN values and calculating
16
    true range for volatility-based indicators.
17
    """
18
    
19
    def _check_fillna(self, series, value=0):
20
        """
21
        Check if fillna flag is True and fill NaN values.
22
        
23
        Parameters:
24
        - series (Series): Input pandas Series
25
        - value (numeric): Value to fill NaN with (default: 0)
26
        
27
        Returns:
28
        Series: Series with NaN values filled if fillna=True
29
        """
30
    
31
    @staticmethod
32
    def _true_range(high, low, prev_close):
33
        """
34
        Calculate true range for volatility indicators.
35
        
36
        Parameters:
37
        - high (Series): High prices
38
        - low (Series): Low prices  
39
        - prev_close (Series): Previous close prices
40
        
41
        Returns:
42
        Series: True range values
43
        """
44
```
45

46
### Data Cleaning
47

48
Utility function for cleaning DataFrames by removing rows with NaN values.
49

50
```python { .api }
51
def dropna(df):
52
    """
53
    Drop rows with "NaN" values from DataFrame.
54
    
55
    Parameters:
56
    - df (DataFrame): DataFrame to clean
57
    
58
    Returns:
59
    DataFrame: DataFrame with NaN values removed
60
    """
61
```
62

63
### Moving Average Calculations
64

65
Internal utility functions for calculating different types of moving averages used by indicators.
66

67
```python { .api }
68
def _sma(series, periods, fillna=False):
69
    """
70
    Calculate Simple Moving Average.
71
    
72
    Parameters:
73
    - series (Series): Price series
74
    - periods (int): Number of periods for calculation
75
    - fillna (bool): If True, fill NaN values (default: False)
76
    
77
    Returns:
78
    Series: Simple moving average values
79
    """
80

81
def _ema(series, periods, fillna=False):
82
    """
83
    Calculate Exponential Moving Average.
84
    
85
    Parameters:
86
    - series (Series): Price series
87
    - periods (int): Number of periods for calculation  
88
    - fillna (bool): If True, fill NaN values (default: False)
89
    
90
    Returns:
91
    Series: Exponential moving average values
92
    """
93
```
94

95
### Comparison Utilities
96

97
Internal utility function for finding minimum or maximum values between two series.
98

99
```python { .api }
100
def _get_min_max(series1, series2, function="min"):
101
    """
102
    Find min or max value between two lists for each index.
103
    
104
    Parameters:
105
    - series1 (Series): First series
106
    - series2 (Series): Second series
107
    - function (str): "min" or "max" operation (default: "min")
108
    
109
    Returns:
110
    Series: Series with min/max values for each index
111
    """
112
```
113

114
## Usage Examples
115

116
### Using IndicatorMixin in Custom Indicators
117

118
```python
119
from ta.utils import IndicatorMixin
120
import pandas as pd
121

122
class CustomIndicator(IndicatorMixin):
123
    def __init__(self, close, window=14, fillna=False):
124
        self._close = close
125
        self._window = window
126
        self._fillna = fillna
127
        self._run()
128
    
129
    def _run(self):
130
        # Custom calculation logic here
131
        self._custom_value = self._close.rolling(self._window).mean()
132
        
133
        # Use inherited fillna functionality
134
        if self._fillna:
135
            self._custom_value = self._check_fillna(self._custom_value, 0)
136
    
137
    def custom_indicator(self):
138
        return self._custom_value
139
```
140

141
### Data Cleaning Workflow
142

143
```python
144
from ta.utils import dropna
145
import pandas as pd
146

147
# Load data that might contain NaN values
148
df = pd.read_csv('financial_data.csv')
149

150
print(f"Original shape: {df.shape}")
151
print(f"NaN values: {df.isnull().sum().sum()}")
152

153
# Clean the data
154
df_clean = dropna(df)
155

156
print(f"Cleaned shape: {df_clean.shape}")
157
print(f"NaN values after cleaning: {df_clean.isnull().sum().sum()}")
158

159
# Now safe to use with indicators
160
import ta
161
df_with_indicators = ta.add_all_ta_features(
162
    df_clean, 
163
    open='Open', 
164
    high='High', 
165
    low='Low', 
166
    close='Close', 
167
    volume='Volume'
168
)
169
```
170

171
### Working with Moving Averages
172

173
```python
174
from ta.utils import _sma, _ema
175
import pandas as pd
176

177
# Example price data
178
prices = pd.Series([100, 102, 101, 103, 105, 104, 106, 108, 107, 109])
179

180
# Calculate different moving averages
181
sma_5 = _sma(prices, periods=5, fillna=True)
182
ema_5 = _ema(prices, periods=5, fillna=True)
183

184
print("Simple Moving Average (5-period):")
185
print(sma_5)
186
print("\nExponential Moving Average (5-period):")
187
print(ema_5)
188

189
# Compare the smoothing characteristics
190
import matplotlib.pyplot as plt
191

192
plt.figure(figsize=(10, 6))
193
plt.plot(prices.index, prices, 'b-', label='Price', linewidth=2)
194
plt.plot(sma_5.index, sma_5, 'r--', label='SMA(5)', linewidth=1.5)
195
plt.plot(ema_5.index, ema_5, 'g--', label='EMA(5)', linewidth=1.5)
196
plt.legend()
197
plt.title('Price vs Moving Averages')
198
plt.show()
199
```
200

201
### Comparative Analysis Utilities
202

203
```python
204
from ta.utils import _get_min_max
205
import pandas as pd
206

207
# Example: Finding support and resistance levels
208
highs = pd.Series([105, 107, 106, 108, 110, 109, 111, 113, 112, 114])
209
lows = pd.Series([98, 100, 99, 101, 103, 102, 104, 106, 105, 107])
210

211
# Find the maximum between consecutive highs (resistance)
212
resistance_levels = _get_min_max(highs, highs.shift(1), function="max")
213

214
# Find the minimum between consecutive lows (support)  
215
support_levels = _get_min_max(lows, lows.shift(1), function="min")
216

217
print("Resistance levels:")
218
print(resistance_levels.dropna())
219
print("\nSupport levels:")
220
print(support_levels.dropna())
221
```
222

223
### Custom Indicator Development
224

225
```python
226
from ta.utils import IndicatorMixin, _sma
227
import pandas as pd
228
import numpy as np
229

230
class PriceVelocityIndicator(IndicatorMixin):
231
    """
232
    Custom indicator measuring price velocity (rate of change acceleration).
233
    """
234
    
235
    def __init__(self, close, window=14, fillna=False):
236
        self._close = close
237
        self._window = window
238
        self._fillna = fillna
239
        self._run()
240
    
241
    def _run(self):
242
        # Calculate price changes
243
        price_change = self._close.diff()
244
        
245
        # Calculate velocity (change in price change)
246
        velocity = price_change.diff()
247
        
248
        # Smooth with moving average
249
        self._velocity = _sma(velocity, self._window)
250
        
251
        # Handle NaN values using inherited method
252
        if self._fillna:
253
            self._velocity = self._check_fillna(self._velocity, 0)
254
    
255
    def price_velocity(self):
256
        """Returns: Series with price velocity values"""
257
        return self._velocity
258

259
# Usage example
260
df = pd.DataFrame({
261
    'Close': [100, 102, 104, 101, 103, 106, 105, 108, 110, 107]
262
})
263

264
velocity_indicator = PriceVelocityIndicator(
265
    close=df['Close'], 
266
    window=5, 
267
    fillna=True
268
)
269

270
df['Price_Velocity'] = velocity_indicator.price_velocity()
271
print(df)
272
```
273

274
### Error Handling and Data Validation
275

276
```python
277
from ta.utils import dropna, IndicatorMixin
278
import pandas as pd
279
import numpy as np
280

281
def safe_indicator_calculation(df, indicator_func, **kwargs):
282
    """
283
    Safely calculate indicators with proper error handling.
284
    """
285
    try:
286
        # Clean data first
287
        df_clean = dropna(df)
288
        
289
        if df_clean.empty:
290
            raise ValueError("No data remaining after cleaning NaN values")
291
        
292
        # Calculate indicator
293
        result = indicator_func(df_clean, **kwargs)
294
        
295
        return result
296
        
297
    except Exception as e:
298
        print(f"Error calculating indicator: {e}")
299
        return pd.Series(dtype=float)
300

301
# Example usage with error handling
302
import ta
303

304
try:
305
    df_with_rsi = safe_indicator_calculation(
306
        df,
307
        ta.add_momentum_ta,
308
        high='High',
309
        low='Low', 
310
        close='Close',
311
        volume='Volume'
312
    )
313
    print("Successfully calculated momentum indicators")
314
except Exception as e:
315
    print(f"Failed to calculate indicators: {e}")
316
```