tessl/pypi-hierarchicalforecast

Hierarchical Methods Time series forecasting

—

Pending

Overview

Eval results

Files

Reconciliation Methods

Name: tessl/pypi-hierarchicalforecast
Author: tessl

Hierarchical reconciliation methods that ensure coherent forecasts across all levels of a hierarchy. Each method implements different approaches to resolving inconsistencies between forecasts at different aggregation levels, with both dense and sparse matrix variants for scalability.

Capabilities

Base Reconciler Interface

All reconciliation methods inherit from the HReconciler base class, providing a consistent interface across different algorithms.

class HReconciler:
    """Abstract base class for all reconciliation methods."""
    
    def fit(self, *args, **kwargs):
        """Fit the reconciliation method to data."""
    
    def fit_predict(self, *args, **kwargs):
        """Fit and predict in one step."""
    
    def predict(self, S, y_hat, level=None):
        """Generate reconciled predictions."""
    
    def sample(self, num_samples: int):
        """Generate coherent samples from fitted model."""
    
    def __call__(self, *args, **kwargs):
        """Alias for fit_predict."""
    
    # Key attributes
    fitted: bool              # Whether model is fitted
    is_sparse_method: bool    # Whether method supports sparse matrices
    insample: bool           # Whether method requires historical data

Bottom-Up Reconciliation

Bottom-up reconciliation aggregates forecasts from the most disaggregated level upward, ensuring perfect coherence by construction.

class BottomUp:
    """
    Bottom-up reconciliation method.
    
    Aggregates forecasts from leaf nodes upward through the hierarchy.
    This is the simplest reconciliation method that maintains perfect coherence.
    """
    
    def __init__(self):
        """Initialize BottomUp reconciler."""
    
    def fit(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        idx_bottom: np.ndarray,
        y_insample: Optional[np.ndarray] = None,
        y_hat_insample: Optional[np.ndarray] = None,
        sigmah: Optional[np.ndarray] = None,
        intervals_method: Optional[str] = None,
        num_samples: Optional[int] = None,
        seed: Optional[int] = None,
        tags: Optional[dict[str, np.ndarray]] = None
    ):
        """
        Fit the bottom-up reconciliation method.
        
        Parameters:
        - S: summing matrix DataFrame
        - y_hat: base forecasts array
        - idx_bottom: indices of bottom-level series
        - y_insample: historical data (not used for BottomUp)
        - y_hat_insample: in-sample forecasts (not used)
        - sigmah: forecast standard deviations
        - intervals_method: method for generating intervals
        - num_samples: number of samples for probabilistic reconciliation
        - seed: random seed
        - tags: hierarchy tags dictionary
        """
    
    def fit_predict(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        idx_bottom: np.ndarray,
        **kwargs
    ) -> np.ndarray:
        """
        Fit and generate bottom-up reconciled forecasts.
        
        Returns:
        Reconciled forecasts array with coherent predictions
        """

class BottomUpSparse:
    """Sparse matrix version of BottomUp for large hierarchies."""
    # Same interface as BottomUp but optimized for sparse matrices

Top-Down Reconciliation

Top-down reconciliation starts from the top level and disaggregates forecasts downward using historical proportions.

class TopDown:
    """
    Top-down reconciliation method.
    
    Disaggregates forecasts from the root level downward using proportions
    derived from historical data or base forecasts.
    """
    
    def __init__(self, method: str = 'forecast_proportions'):
        """
        Initialize TopDown reconciler.
        
        Parameters:
        - method: str, disaggregation method
            'forecast_proportions': Use base forecast proportions
            'average_proportions': Use average historical proportions  
            'proportion_averages': Use proportions of historical averages
        """
    
    def fit(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        y_insample: np.ndarray,
        y_hat_insample: Optional[np.ndarray] = None,
        sigmah: Optional[np.ndarray] = None,
        intervals_method: Optional[str] = None,
        num_samples: Optional[int] = None,
        seed: Optional[int] = None,
        tags: Optional[dict[str, np.ndarray]] = None,
        idx_bottom: Optional[np.ndarray] = None
    ):
        """
        Fit the top-down reconciliation method.
        
        Parameters:
        - S: summing matrix DataFrame
        - y_hat: base forecasts array
        - y_insample: historical data (required for proportion calculation)
        - y_hat_insample: in-sample forecasts
        - sigmah: forecast standard deviations
        - intervals_method: method for generating intervals
        - num_samples: number of samples
        - seed: random seed
        - tags: hierarchy tags dictionary
        - idx_bottom: bottom-level series indices
        """
    
    def fit_predict(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        tags: dict[str, np.ndarray],
        idx_bottom: Optional[np.ndarray] = None,
        y_insample: Optional[np.ndarray] = None,
        **kwargs
    ) -> np.ndarray:
        """
        Fit and generate top-down reconciled forecasts.
        
        Returns:
        Reconciled forecasts array using top-down approach
        """

class TopDownSparse:
    """Sparse matrix version of TopDown for large hierarchies."""
    # Same interface as TopDown

Middle-Out Reconciliation

Middle-out reconciliation anchors at a specified middle level, applying bottom-up above and top-down below that level.

class MiddleOut:
    """
    Middle-out reconciliation method.
    
    Anchors reconciliation at a middle level of the hierarchy, applying
    bottom-up reconciliation above and top-down reconciliation below.
    """
    
    def __init__(
        self,
        middle_level: str,
        top_down_method: str = 'forecast_proportions'
    ):
        """
        Initialize MiddleOut reconciler.
        
        Parameters:
        - middle_level: str, name of the middle level to anchor at
        - top_down_method: str, method for top-down portion
        """
    
    def fit_predict(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        tags: dict[str, np.ndarray],
        y_insample: Optional[np.ndarray] = None,
        level: Optional[list[int]] = None,
        intervals_method: Optional[str] = None,
        **kwargs
    ) -> np.ndarray:
        """
        Generate middle-out reconciled forecasts.
        
        Returns:
        Reconciled forecasts using middle-out approach
        """

class MiddleOutSparse:
    """Sparse matrix version of MiddleOut for large hierarchies."""
    # Same interface as MiddleOut

Minimum Trace Reconciliation

MinTrace uses generalized least squares to find reconciled forecasts that minimize the trace of the covariance matrix.

class MinTrace:
    """
    Minimum trace reconciliation using generalized least squares.
    
    Finds reconciled forecasts that minimize the trace of the forecast
    error covariance matrix subject to aggregation constraints.
    """
    
    def __init__(
        self,
        method: str = 'ols',
        nonnegative: bool = False,
        mint_shr_ridge: float = 2e-8,
        num_threads: int = 1
    ):
        """
        Initialize MinTrace reconciler.
        
        Parameters:
        - method: str, estimation method
            'ols': Ordinary least squares (identity covariance)
            'wls_struct': Weighted least squares with structural scaling
            'wls_var': Weighted least squares with variance scaling
            'mint_cov': Use sample covariance matrix
            'mint_shrink': Use shrinkage covariance estimator
        - nonnegative: bool, enforce non-negative constraints
        - mint_shr_ridge: float, ridge parameter for shrinkage estimator
        - num_threads: int, number of threads for optimization
        """
    
    def fit(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        y_insample: Optional[np.ndarray] = None,
        y_hat_insample: Optional[np.ndarray] = None,
        sigmah: Optional[np.ndarray] = None,
        intervals_method: Optional[str] = None,
        num_samples: Optional[int] = None,
        seed: Optional[int] = None,
        tags: Optional[dict[str, np.ndarray]] = None,
        idx_bottom: Optional[np.ndarray] = None
    ):
        """Fit the MinTrace reconciliation method."""
    
    def fit_predict(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        idx_bottom: Optional[np.ndarray] = None,
        y_insample: Optional[np.ndarray] = None,
        y_hat_insample: Optional[np.ndarray] = None,
        **kwargs
    ) -> np.ndarray:
        """
        Generate MinTrace reconciled forecasts.
        
        Returns:
        Reconciled forecasts using minimum trace approach
        """

class MinTraceSparse:
    """Sparse matrix version of MinTrace for large hierarchies."""
    # Same interface as MinTrace

Optimal Combination

OptimalCombination is equivalent to specific MinTrace variants, providing an alternative interface for classic optimal reconciliation.

class OptimalCombination:
    """
    Optimal combination method (equivalent to specific MinTrace variants).
    
    Provides optimal linear combination of base forecasts to achieve coherence
    while minimizing forecast error variance.
    """
    
    def __init__(
        self,
        method: str = 'ols',
        nonnegative: bool = False,
        num_threads: int = 1
    ):
        """
        Initialize OptimalCombination reconciler.
        
        Parameters:
        - method: str, combination method ('ols', 'wls_struct')
        - nonnegative: bool, enforce non-negative constraints
        - num_threads: int, number of threads for optimization
        """
    
    # Same interface as MinTrace for fit/fit_predict methods

Empirical Risk Minimization

ERM reconciliation minimizes empirical risk using in-sample errors, with options for regularization.

class ERM:
    """
    Empirical Risk Minimization reconciliation.
    
    Minimizes in-sample reconciliation errors using various optimization
    approaches including closed-form solutions and regularized methods.
    """
    
    def __init__(
        self,
        method: str = 'closed',
        lambda_reg: float = 1e-2
    ):
        """
        Initialize ERM reconciler.
        
        Parameters:
        - method: str, optimization method
            'closed': Closed-form solution
            'reg': Regularized solution with lasso penalty
            'reg_bu': Regularized bottom-up solution
        - lambda_reg: float, regularization parameter for lasso methods
        """
    
    def fit(
        self,
        S: pd.DataFrame,
        y_hat: np.ndarray,
        y_insample: pd.DataFrame,
        y_hat_insample: np.ndarray,
        sigmah: np.ndarray = None,
        intervals_method: str = None,
        num_samples: int = None,
        seed: int = None,
        tags: dict = None,
        idx_bottom: np.ndarray = None
    ):
        """
        Fit the ERM reconciliation method.
        
        Note: ERM requires both historical data and in-sample forecasts.
        """
    
    def fit_predict(
        self,
        S: np.ndarray,
        y_hat: np.ndarray,
        idx_bottom: Optional[np.ndarray] = None,
        y_insample: Optional[np.ndarray] = None,
        y_hat_insample: Optional[np.ndarray] = None,
        **kwargs
    ) -> np.ndarray:
        """
        Generate ERM reconciled forecasts.
        
        Returns:
        Reconciled forecasts using empirical risk minimization
        """

Usage Examples

Basic Method Usage

from hierarchicalforecast.methods import BottomUp, TopDown, MinTrace

# Simple bottom-up reconciliation
bottom_up = BottomUp()
reconciled = bottom_up.fit_predict(S=summing_matrix, y_hat=forecasts, idx_bottom=bottom_indices)

# Top-down with forecast proportions
top_down = TopDown(method='forecast_proportions')
reconciled = top_down.fit_predict(
    S=summing_matrix, 
    y_hat=forecasts, 
    tags=hierarchy_tags,
    y_insample=historical_data
)

# MinTrace with shrinkage covariance
min_trace = MinTrace(method='mint_shrink', nonnegative=True)
reconciled = min_trace.fit_predict(
    S=summing_matrix,
    y_hat=forecasts,
    y_insample=historical_data
)

Using Sparse Methods for Large Hierarchies

from hierarchicalforecast.methods import BottomUpSparse, MinTraceSparse

# For very large hierarchies, use sparse variants
bottom_up_sparse = BottomUpSparse()
min_trace_sparse = MinTraceSparse(method='ols')

# Same interface, but optimized for sparse matrices
reconciled = bottom_up_sparse.fit_predict(S=sparse_summing_matrix, y_hat=forecasts, idx_bottom=bottom_indices)

Method Comparison

# Compare multiple methods
methods = {
    'BottomUp': BottomUp(),
    'TopDown_FP': TopDown(method='forecast_proportions'),
    'TopDown_AP': TopDown(method='average_proportions'),
    'MinTrace_OLS': MinTrace(method='ols'),
    'MinTrace_WLS': MinTrace(method='wls_struct'),
    'ERM_Closed': ERM(method='closed')
}

results = {}
for name, method in methods.items():
    results[name] = method.fit_predict(
        S=summing_matrix,
        y_hat=forecasts,
        y_insample=historical_data,
        y_hat_insample=insample_forecasts
    )

Install with Tessl CLI

npx tessl i tessl/pypi-hierarchicalforecast

docs

core-reconciliation.md

data-utilities.md

evaluation.md

index.md

probabilistic-methods.md

reconciliation-methods.md

visualization.md

tile.json

tessl/pypi-hierarchicalforecast