寻求适用于实时股票指标迭代计算的动态数据计算设计模式

Great question! For real-time stock indicator calculations that rely on iterative, history-dependent logic (like EMA or SMA), you need a design that balances performance (avoiding full recalculations) and architectural consistency. Here's a tailored approach using established patterns and practical implementations:

Core Design Principles #

Your key needs are:

• Incremental calculation: Avoid full recalculations when new data arrives
• State caching: Maintain necessary historical state/values for iteration
• Unified architecture: All indicators follow a consistent interface via a base class

1. Template Method Pattern - The Foundation for Consistency #

The Template Method Pattern is perfect here. It lets you define a universal workflow for all indicators in a base class, while letting specific indicator subclasses implement their unique calculation logic.

Base Class Implementation (Pseudocode)

from abc import ABC, abstractmethod

class StockIndicator(ABC):
    def __init__(self, period: int):
        self.period = period
        self._history = []  # Stores recent price data (or relevant state)
        self._current_value = None  # Caches the latest calculated value

    def update(self, new_price: float) -> float:
        """Standard workflow for updating with new price data"""
        self._preprocess_new_data(new_price)
        if not self._is_ready():
            self._initialize_indicator()
        else:
            self._calculate_incremental(new_price)
        return self._current_value

    def _preprocess_new_data(self, new_price: float):
        """Common logic to maintain a rolling window of price data"""
        self._history.append(new_price)
        if len(self._history) > self.period:
            self._history.pop(0)

    @abstractmethod
    def _is_ready(self) -> bool:
        """Subclasses define when they have enough data to start calculating"""
        pass

    @abstractmethod
    def _initialize_indicator(self):
        """Subclasses implement initial calculation (e.g., SMA for first EMA)"""
        pass

    @abstractmethod
    def _calculate_incremental(self, new_price: float):
        """Subclasses implement iterative calculation logic"""
        pass

    @property
    def current_value(self):
        return self._current_value

2. State Caching for Performance #

Each indicator subclass maintains only the state it needs to avoid full recalculations. For example:

• EMA needs the previous period's EMA value
• SMA can cache the sum of recent prices instead of recalculating the sum every time

EMA Indicator Subclass

class EMAIndicator(StockIndicator):
    def __init__(self, period: int):
        super().__init__(period)
        self._previous_ema = None
        self.multiplier = 2 / (period + 1)

    def _is_ready(self) -> bool:
        # EMA needs a full period of data to compute the initial SMA
        return len(self._history) == self.period and self._previous_ema is not None

    def _initialize_indicator(self):
        # Initial EMA uses SMA of the first period's data
        initial_sma = sum(self._history) / self.period
        self._current_value = initial_sma
        self._previous_ema = initial_sma

    def _calculate_incremental(self, new_price: float):
        # Use the iterative EMA formula with cached previous value
        self._current_value = (new_price - self._previous_ema) * self.multiplier + self._previous_ema
        self._previous_ema = self._current_value

SMA Indicator Subclass

class SMAIndicator(StockIndicator):
    def __init__(self, period: int):
        super().__init__(period)
        self._rolling_sum = 0.0  # Cache sum to avoid recalculating every time

    def _preprocess_new_data(self, new_price: float):
        # Override to maintain rolling sum incrementally
        old_price = self._history[0] if len(self._history) == self.period else None
        super()._preprocess_new_data(new_price)
        if old_price:
            self._rolling_sum = self._rolling_sum - old_price + new_price
        else:
            self._rolling_sum += new_price

    def _is_ready(self) -> bool:
        return len(self._history) == self.period

    def _initialize_indicator(self):
        self._current_value = self._rolling_sum / self.period

    def _calculate_incremental(self, new_price: float):
        # Use cached rolling sum for instant calculation
        self._current_value = self._rolling_sum / self.period

3. Additional Optimizations #

• Thread Safety: If your real-time data comes from multiple threads, add locks to the update method to prevent race conditions on shared state.
• State Persistence: Serialize and store indicator state (e.g., _previous_ema for EMA) if you need to resume calculations after a restart.
• Composite Indicators: For indicators that depend on others (like MACD, which uses two EMAs), you can inject other StockIndicator instances into the subclass and use their current_value property.

This design ensures all indicators follow a consistent interface, uses incremental calculation to avoid performance hits from full recalculations, and is easy to extend with new indicators (just implement the abstract methods in the base class).

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