Utilities

class IndicatorMixin:
    """
    Utility mixin indicator class - Base class for all indicators.
    
    Provides common functionality for handling NaN values and calculating
    true range for volatility-based indicators.
    """
    
    def _check_fillna(self, series, value=0):
        """
        Check if fillna flag is True and fill NaN values.
        
        Parameters:
        - series (Series): Input pandas Series
        - value (numeric): Value to fill NaN with (default: 0)
        
        Returns:
        Series: Series with NaN values filled if fillna=True
        """
    
    @staticmethod
    def _true_range(high, low, prev_close):
        """
        Calculate true range for volatility indicators.
        
        Parameters:
        - high (Series): High prices
        - low (Series): Low prices  
        - prev_close (Series): Previous close prices
        
        Returns:
        Series: True range values
        """

def dropna(df):
    """
    Drop rows with "NaN" values from DataFrame.
    
    Parameters:
    - df (DataFrame): DataFrame to clean
    
    Returns:
    DataFrame: DataFrame with NaN values removed
    """

def _sma(series, periods, fillna=False):
    """
    Calculate Simple Moving Average.
    
    Parameters:
    - series (Series): Price series
    - periods (int): Number of periods for calculation
    - fillna (bool): If True, fill NaN values (default: False)
    
    Returns:
    Series: Simple moving average values
    """

def _ema(series, periods, fillna=False):
    """
    Calculate Exponential Moving Average.
    
    Parameters:
    - series (Series): Price series
    - periods (int): Number of periods for calculation  
    - fillna (bool): If True, fill NaN values (default: False)
    
    Returns:
    Series: Exponential moving average values
    """

def _get_min_max(series1, series2, function="min"):
    """
    Find min or max value between two lists for each index.
    
    Parameters:
    - series1 (Series): First series
    - series2 (Series): Second series
    - function (str): "min" or "max" operation (default: "min")
    
    Returns:
    Series: Series with min/max values for each index
    """

from ta.utils import IndicatorMixin
import pandas as pd

class CustomIndicator(IndicatorMixin):
    def __init__(self, close, window=14, fillna=False):
        self._close = close
        self._window = window
        self._fillna = fillna
        self._run()
    
    def _run(self):
        # Custom calculation logic here
        self._custom_value = self._close.rolling(self._window).mean()
        
        # Use inherited fillna functionality
        if self._fillna:
            self._custom_value = self._check_fillna(self._custom_value, 0)
    
    def custom_indicator(self):
        return self._custom_value

from ta.utils import dropna
import pandas as pd

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

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

# Clean the data
df_clean = dropna(df)

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

# Now safe to use with indicators
import ta
df_with_indicators = ta.add_all_ta_features(
    df_clean, 
    open='Open', 
    high='High', 
    low='Low', 
    close='Close', 
    volume='Volume'
)

from ta.utils import _sma, _ema
import pandas as pd

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

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

print("Simple Moving Average (5-period):")
print(sma_5)
print("\nExponential Moving Average (5-period):")
print(ema_5)

# Compare the smoothing characteristics
import matplotlib.pyplot as plt

plt.figure(figsize=(10, 6))
plt.plot(prices.index, prices, 'b-', label='Price', linewidth=2)
plt.plot(sma_5.index, sma_5, 'r--', label='SMA(5)', linewidth=1.5)
plt.plot(ema_5.index, ema_5, 'g--', label='EMA(5)', linewidth=1.5)
plt.legend()
plt.title('Price vs Moving Averages')
plt.show()

from ta.utils import _get_min_max
import pandas as pd

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

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

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

print("Resistance levels:")
print(resistance_levels.dropna())
print("\nSupport levels:")
print(support_levels.dropna())

from ta.utils import IndicatorMixin, _sma
import pandas as pd
import numpy as np

class PriceVelocityIndicator(IndicatorMixin):
    """
    Custom indicator measuring price velocity (rate of change acceleration).
    """
    
    def __init__(self, close, window=14, fillna=False):
        self._close = close
        self._window = window
        self._fillna = fillna
        self._run()
    
    def _run(self):
        # Calculate price changes
        price_change = self._close.diff()
        
        # Calculate velocity (change in price change)
        velocity = price_change.diff()
        
        # Smooth with moving average
        self._velocity = _sma(velocity, self._window)
        
        # Handle NaN values using inherited method
        if self._fillna:
            self._velocity = self._check_fillna(self._velocity, 0)
    
    def price_velocity(self):
        """Returns: Series with price velocity values"""
        return self._velocity

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

velocity_indicator = PriceVelocityIndicator(
    close=df['Close'], 
    window=5, 
    fillna=True
)

df['Price_Velocity'] = velocity_indicator.price_velocity()
print(df)

from ta.utils import dropna, IndicatorMixin
import pandas as pd
import numpy as np

def safe_indicator_calculation(df, indicator_func, **kwargs):
    """
    Safely calculate indicators with proper error handling.
    """
    try:
        # Clean data first
        df_clean = dropna(df)
        
        if df_clean.empty:
            raise ValueError("No data remaining after cleaning NaN values")
        
        # Calculate indicator
        result = indicator_func(df_clean, **kwargs)
        
        return result
        
    except Exception as e:
        print(f"Error calculating indicator: {e}")
        return pd.Series(dtype=float)

# Example usage with error handling
import ta

try:
    df_with_rsi = safe_indicator_calculation(
        df,
        ta.add_momentum_ta,
        high='High',
        low='Low', 
        close='Close',
        volume='Volume'
    )
    print("Successfully calculated momentum indicators")
except Exception as e:
    print(f"Failed to calculate indicators: {e}")