如何编写bundle_indicators函数批量整合Pandas指标至DataFrame

Got it, let's walk through building this bundle_indicators function to batch add technical indicators (like your custom MA) to one or multiple DataFrames. Here's a complete, practical solution tailored to your needs:

First, let's define a flexible moving average function that works with any Series and window size—you can tweak the logic (e.g., switch to EMA) later without breaking the rest of the code:

import pandas as pd
from functools import partial

def custom_ma(series: pd.Series, window: int) -> pd.Series:
    """Custom simple moving average calculation (adjust logic as needed)"""
    return series.rolling(window=window).mean()

We'll use three parameters as you planned, each with clear responsibilities:

• data: A single DataFrame or list of DataFrames to process
• indicator_specs: A dictionary mapping indicator column names to pre-configured indicator functions (we'll use partial to bind parameters like window sizes)
• target_col: The name of the column in your DataFrame(s) to use as input for indicators (e.g., "close" for closing prices)

Here's the code:

def bundle_indicators(data, indicator_specs: dict, target_col: str) -> pd.DataFrame | list[pd.DataFrame]:
    """
    Batch generate technical indicators for one or multiple DataFrames.
    
    Args:
        data: Single pandas DataFrame or list of DataFrames
        indicator_specs: Dict where keys are output column names, values are functions that take a Series and return a Series
        target_col: Name of the column to use as input for all indicators
        
    Returns:
        DataFrame with added indicators (or list of DataFrames if input was a list)
    """
    # Handle single DataFrame input
    if isinstance(data, pd.DataFrame):
        df_copy = data.copy()
        for indicator_name, indicator_func in indicator_specs.items():
            df_copy[indicator_name] = indicator_func(df_copy[target_col])
        # Optional: Return only the indicator columns if you don't need the original data
        # return df_copy[[*indicator_specs.keys()]]
        return df_copy
    
    # Handle list of DataFrames input
    elif isinstance(data, list) and all(isinstance(df, pd.DataFrame) for df in data):
        processed_dfs = []
        for df in data:
            df_copy = df.copy()
            for indicator_name, indicator_func in indicator_specs.items():
                df_copy[indicator_name] = indicator_func(df_copy[target_col])
            processed_dfs.append(df_copy)
        return processed_dfs
    
    else:
        raise TypeError("Input 'data' must be a pandas DataFrame or list of DataFrames")

Let's test this with sample price data and multiple MA windows:

# Sample price DataFrame
df_price = pd.DataFrame({
    "close": [10, 12, 11, 13, 14, 16, 15, 17, 19, 20]
})

# Configure your indicators: map output column names to bound MA functions
indicator_config = {
    "MA_5": partial(custom_ma, window=5),
    "MA_10": partial(custom_ma, window=10),
    # Add more indicators here later (e.g., RSI, MACD) by defining their functions
}

# Generate the indicator DataFrame
df_indicators = bundle_indicators(df_price, indicator_config, target_col="close")

print(df_indicators)

This setup is fully reusable for other technical indicators. For example, adding an exponential moving average:

def custom_ema(series: pd.Series, window: int) -> pd.Series:
    return series.ewm(span=window, adjust=False).mean()

# Update the indicator config
indicator_config.update({
    "EMA_5": partial(custom_ema, window=5)
})

# Re-run to include the new EMA
df_indicators = bundle_indicators(df_price, indicator_config, target_col="close")

Key perks of this implementation:

• Reusability: Works for any number of indicators and DataFrames
• Flexibility: Easy to add new indicators by just defining their functions
• Safety: Operates on copies of your input data to avoid accidental modifications

内容的提问来源于stack exchange，提问作者Alex Niemi