Hierarchical Methods Time series forecasting
npx @tessl/cli install tessl/pypi-hierarchicalforecast@1.2.0A comprehensive Python library for hierarchical time series forecasting that provides cross-sectional and temporal reconciliation methods including BottomUp, TopDown, MiddleOut, MinTrace, and ERM, as well as probabilistic coherent prediction methods such as Normality, Bootstrap, and PERMBU. The library addresses the challenge of maintaining coherent forecasts across different levels of hierarchical data structures (like geographical, categorical, or temporal aggregations) which is essential for consistent decision-making and planning in business and research contexts.
pip install hierarchicalforecastfrom hierarchicalforecast.core import HierarchicalReconciliationCommon imports for reconciliation methods:
from hierarchicalforecast.methods import (
BottomUp, BottomUpSparse, TopDown, TopDownSparse,
MiddleOut, MiddleOutSparse, MinTrace, MinTraceSparse,
OptimalCombination, ERM
)For evaluation:
from hierarchicalforecast.evaluation import evaluateFor probabilistic methods:
from hierarchicalforecast.probabilistic_methods import Normality, Bootstrap, PERMBUFor utilities:
from hierarchicalforecast.utils import aggregate, aggregate_temporal, HierarchicalPlotimport pandas as pd
import numpy as np
from hierarchicalforecast.core import HierarchicalReconciliation
from hierarchicalforecast.methods import BottomUp, TopDown, MinTrace
from hierarchicalforecast.utils import aggregate
# Sample hierarchical data preparation
df = pd.DataFrame({
'unique_id': ['A', 'A', 'B', 'B', 'C', 'C'],
'ds': pd.date_range('2020-01-01', periods=2, freq='D').tolist() * 3,
'y': [100, 110, 200, 220, 150, 160]
})
# Create hierarchical structure
spec = [['A', 'B', 'C']] # Group bottom level series
Y_df, S_df, tags = aggregate(df, spec)
# Generate base forecasts (example)
forecasts = pd.DataFrame({
'unique_id': ['A', 'A', 'B', 'B', 'C', 'C', 'A/B/C', 'A/B/C'],
'ds': pd.date_range('2020-01-03', periods=2, freq='D').tolist() * 4,
'AutoARIMA': [115, 120, 230, 240, 170, 180, 515, 540]
})
# Initialize reconciliation methods
reconcilers = [
BottomUp(),
TopDown(method='forecast_proportions'),
MinTrace(method='ols')
]
# Create reconciliation object
hrec = HierarchicalReconciliation(reconcilers=reconcilers)
# Perform reconciliation
reconciled_forecasts = hrec.reconcile(
Y_hat_df=forecasts,
S=S_df,
tags=tags,
Y_df=Y_df
)
print(reconciled_forecasts)HierarchicalForecast is built around a modular architecture:
Main orchestration functionality for applying multiple hierarchical reconciliation methods to base forecasts, ensuring coherence across hierarchy levels.
class HierarchicalReconciliation:
def __init__(self, reconcilers: list[HReconciler]): ...
def reconcile(
self,
Y_hat_df: Frame,
S: Frame,
tags: dict[str, np.ndarray],
Y_df: Optional[Frame] = None,
level: Optional[list[int]] = None,
intervals_method: str = "normality",
num_samples: int = -1,
seed: int = 0,
is_balanced: bool = False,
id_col: str = "unique_id",
time_col: str = "ds",
target_col: str = "y",
id_time_col: str = "temporal_id",
temporal: bool = False,
) -> FrameT: ...
def bootstrap_reconcile(
self,
Y_hat_df: Frame,
S_df: Frame,
tags: dict[str, np.ndarray],
Y_df: Optional[Frame] = None,
level: Optional[list[int]] = None,
intervals_method: str = "normality",
num_samples: int = -1,
num_seeds: int = 1,
id_col: str = "unique_id",
time_col: str = "ds",
target_col: str = "y",
) -> FrameT: ...Various reconciliation algorithms including bottom-up, top-down, middle-out, minimum trace, optimal combination, and empirical risk minimization approaches, each with sparse variants for scalability.
class BottomUp:
def fit_predict(self, S, y_hat, idx_bottom, **kwargs): ...
class TopDown:
def __init__(self, method='forecast_proportions'): ...
def fit_predict(self, S, y_hat, tags, y_insample=None, **kwargs): ...
class MinTrace:
def __init__(self, method='ols', nonnegative=False, mint_shr_ridge=2e-8, num_threads=1): ...
def fit_predict(self, S, y_hat, y_insample=None, **kwargs): ...Comprehensive evaluation framework for measuring forecast accuracy across different hierarchy levels using standard metrics from utilsforecast.losses.
def evaluate(df, metrics, tags, models=None, train_df=None, level=None, id_col='unique_id', time_col='ds', target_col='y', agg_fn='mean', benchmark=None): ...Probabilistic reconciliation methods for generating prediction intervals and samples from hierarchically coherent distributions, supporting normality assumptions, bootstrap sampling, and empirical copula approaches.
class Normality:
def __init__(self, S: np.ndarray, P: np.ndarray, y_hat: np.ndarray, sigmah: np.ndarray, W: np.ndarray, seed: int = 0): ...
def get_samples(self, num_samples: int) -> np.ndarray: ...
def get_prediction_quantiles(self, results: dict, quantiles: np.ndarray) -> dict: ...
class Bootstrap:
def __init__(self, S: np.ndarray, P: np.ndarray, y_hat: np.ndarray, y_insample: np.ndarray, y_hat_insample: np.ndarray, num_samples: int, seed: int = 0): ...
def get_samples(self, num_samples: int) -> np.ndarray: ...
def get_prediction_quantiles(self, results: dict, quantiles: np.ndarray) -> dict: ...
class PERMBU:
def __init__(self, S: np.ndarray, P: np.ndarray, y_hat: np.ndarray, tags: dict, y_insample: np.ndarray, y_hat_insample: np.ndarray, sigmah: np.ndarray, num_samples: int, seed: int = 0): ...
def get_samples(self, num_samples: int) -> np.ndarray: ...
def get_prediction_quantiles(self, results: dict, quantiles: np.ndarray) -> dict: ...Utilities for creating hierarchical data structures from bottom-level time series, including cross-sectional and temporal aggregation capabilities.
def aggregate(df, spec, exog_vars=None, sparse_s=False, id_col='unique_id', time_col='ds', id_time_col=None, target_cols=('y',)): ...
def aggregate_temporal(df, spec, exog_vars=None, sparse_s=False, id_col='unique_id', time_col='ds', id_time_col='temporal_id', target_cols=('y',), aggregation_type='local'): ...
def make_future_dataframe(df, freq, h, id_col='unique_id', time_col='ds'): ...Plotting utilities for visualizing hierarchical time series data, reconciliation results, and hierarchy structures.
class HierarchicalPlot:
def __init__(self, S, tags, S_id_col='unique_id'): ...
def plot_summing_matrix(self): ...
def plot_series(self, series, Y_df, models=None, level=None, **kwargs): ...
def plot_hierarchically_linked_series(self, bottom_series, Y_df, models=None, level=None, **kwargs): ...
def plot_hierarchical_predictions_gap(self, Y_df, models=None, **kwargs): ...# Import necessary types
from typing import Optional
from narwhals.typing import Frame, FrameT
import numpy as np
import scipy.sparse
# DataFrame types expected by the library
Frame = Frame # Narwhals Frame for input DataFrames
FrameT = FrameT # Narwhals FrameT for output DataFrames
# Reconciler base class
class HReconciler:
"""Base class for all reconciliation methods."""
fitted: bool
is_sparse_method: bool
insample: bool
# Common column specifications
id_col: str = 'unique_id' # Series identifier column
time_col: str = 'ds' # Timestamp column
target_col: str = 'y' # Target variable column
id_time_col: str = 'temporal_id' # Temporal hierarchy identifier
# Aggregation specification
spec: list[list[str]] # Cross-sectional hierarchy specification
spec: dict[str, int] # Temporal aggregation specification
# Tags dictionary mapping hierarchy levels to series indices
tags: dict[str, np.ndarray]
# Summing matrix formats
S: Frame # Dense summing matrix as Frame
S: scipy.sparse.csr_matrix # Sparse summing matrix for large hierarchies