Abstract and Streaming APIs

class Function:
    """
    Abstract function wrapper for flexible indicator calculation
    
    Supports both direct instantiation and dictionary-based inputs.
    """
    def __init__(self, function_name, *args, **kwargs):
        """
        Initialize abstract function
        
        Parameters:
        - function_name: str, name of the TA-Lib function (case-insensitive)
        - args, kwargs: additional arguments passed to the function
        """

def Function(function_name, *args, **kwargs):
    """
    Factory function to create abstract function instances
    
    Parameters:
    - function_name: str, name of TA-Lib function
    - args, kwargs: function parameters
    
    Returns:
    Function: Abstract function instance
    """

# All TA-Lib functions are available as abstract functions
import talib.abstract as abstract

SMA = abstract.SMA
RSI = abstract.RSI  
MACD = abstract.MACD
BBANDS = abstract.BBANDS

# Dictionary with named arrays
inputs = {
    'open': np.array([...]),
    'high': np.array([...]),
    'low': np.array([...]), 
    'close': np.array([...]),
    'volume': np.array([...])
}

# pandas DataFrame with OHLCV columns
df = pd.DataFrame({
    'open': [...],
    'high': [...],
    'low': [...],
    'close': [...],
    'volume': [...]
})

# polars DataFrame
import polars as pl
df = pl.DataFrame({
    'open': [...],
    'high': [...],
    'low': [...],
    'close': [...],
    'volume': [...]
})

def _get_defaults_and_docs(func_name):
    """
    Get function parameter defaults and documentation
    
    Parameters:
    - func_name: str, function name
    
    Returns:
    dict: Function metadata including parameters and defaults
    """

# Regular function returns array
sma_array = talib.SMA(close_prices, timeperiod=20)

# Streaming function returns single float
latest_sma = talib.stream.SMA(close_prices, timeperiod=20)
# or
latest_sma = talib.stream_SMA(close_prices, timeperiod=20)

# Overlap Studies
def stream_SMA(real, timeperiod=30): ...
def stream_EMA(real, timeperiod=30): ...
def stream_BBANDS(real, timeperiod=5, nbdevup=2, nbdevdn=2, matype=MA_Type.SMA): ...

# Momentum Indicators  
def stream_RSI(real, timeperiod=14): ...
def stream_MACD(real, fastperiod=12, slowperiod=26, signalperiod=9): ...
def stream_STOCH(high, low, close, fastk_period=5, slowk_period=3, 
                 slowk_matype=MA_Type.SMA, slowd_period=3, slowd_matype=MA_Type.SMA): ...

# Volume Indicators
def stream_AD(high, low, close, volume): ...
def stream_OBV(close, volume): ...

# All other categories follow the same pattern...

import talib.abstract as abstract
import pandas as pd
import numpy as np

# Using dictionary inputs
inputs = {
    'open': np.random.random(100) * 100,
    'high': np.random.random(100) * 100 + 5,
    'low': np.random.random(100) * 100 - 5,
    'close': np.random.random(100) * 100,
    'volume': np.random.random(100) * 1000
}

# Calculate indicators using abstract API
sma = abstract.SMA(inputs, timeperiod=20)
rsi = abstract.RSI(inputs, timeperiod=14)
upper, middle, lower = abstract.BBANDS(inputs, timeperiod=20)

# Using pandas DataFrame
df = pd.DataFrame(inputs)
macd_line, signal_line, histogram = abstract.MACD(df)

# Advanced usage with custom price selection
sma_high = abstract.SMA(inputs, timeperiod=20, price='high')
sma_volume = abstract.SMA(inputs, timeperiod=20, price='volume')

# Multiple output handling with DataFrames
stoch_k, stoch_d = abstract.STOCH(df)
if isinstance(df, pd.DataFrame):
    # Returns pandas Series with matching index
    print(f"Latest %K: {stoch_k.iloc[-1]:.2f}")
    print(f"Latest %D: {stoch_d.iloc[-1]:.2f}")

import talib.stream as stream
import numpy as np

# Sample streaming data (in real application, this would be live data)
close_prices = np.array([100, 101, 102, 101, 103, 105, 104, 106])

# Calculate streaming indicators (returns only latest values)
latest_sma = stream.SMA(close_prices, timeperiod=5)
latest_rsi = stream.RSI(close_prices, timeperiod=14)
latest_ema = stream.EMA(close_prices, timeperiod=10)

print(f"Current SMA(5): {latest_sma:.2f}")
print(f"Current RSI(14): {latest_rsi:.2f}")
print(f"Current EMA(10): {latest_ema:.2f}")

# For indicators that return multiple values
macd, signal, histogram = stream.MACD(close_prices)
print(f"MACD: {macd:.4f}, Signal: {signal:.4f}, Histogram: {histogram:.4f}")

# Real-time trading system example
def process_new_price(new_price, price_history, max_history=100):
    # Add new price and maintain history size
    price_history.append(new_price)
    if len(price_history) > max_history:
        price_history.pop(0)
    
    # Calculate streaming indicators
    current_sma = stream.SMA(np.array(price_history), timeperiod=20)
    current_rsi = stream.RSI(np.array(price_history), timeperiod=14)
    
    # Trading logic
    if current_rsi < 30 and new_price > current_sma:
        return "BUY_SIGNAL"
    elif current_rsi > 70 and new_price < current_sma:
        return "SELL_SIGNAL"
    else:
        return "HOLD"

# Usage in streaming context
price_buffer = [100, 101, 102, 103, 104]
signal = process_new_price(105, price_buffer)
print(f"Trading signal: {signal}")

import talib
import talib.abstract as abstract
import talib.stream as stream
import numpy as np

close_prices = np.random.random(50) * 100

# Function API - returns full array
sma_array = talib.SMA(close_prices, timeperiod=10)
print(f"Function API - Array length: {len(sma_array)}")

# Abstract API - flexible input handling
inputs = {'close': close_prices}
sma_abstract = abstract.SMA(inputs, timeperiod=10) 
print(f"Abstract API - Array length: {len(sma_abstract)}")

# Streaming API - returns single value
sma_latest = stream.SMA(close_prices, timeperiod=10)
print(f"Streaming API - Single value: {sma_latest:.2f}")

# All should give the same latest value
print(f"Values match: {abs(sma_array[-1] - sma_latest) < 1e-10}")