HierarchicalForecast 包含层次协调方法的纯 Python 实现，以及一个 core.HierarchicalReconciliation 包装类，通过包含层次时间序列和基础预测的 pandas DataFrame，可以轻松地与这些方法进行交互。

core.HierarchicalReconciliation 协调类使用层次时间序列 pd.DataFrame Y_df、基础预测 pd.DataFrame Y_hat_df 和聚合约束矩阵 S 进行操作。有关聚合约束矩阵创建的更多信息，请参阅 utils 聚合方法。

​
HierarchicalReconciliation

源

​
HierarchicalReconciliation

 HierarchicalReconciliation
                             (reconcilers:list[hierarchicalforecast.method
                             s.HReconciler])

*层次协调类。

core.HierarchicalReconciliation 类允许您高效地为存储在 pandas DataFrame 中的时间序列集合和基础预测拟合多种 HierarchicaForecast 方法。Y_df DataFrame 使用 unique_id 和 ds 列标识时间序列和日期戳，而 y 列表示目标时间序列变量。Y_h DataFrame 存储基础预测，例如 (AutoARIMA, ETS 等)。

参数
reconcilers：[协调方法](https://nixtla.github.io/hierarchicalforecast/methods.html)模块中实例化类的列表。

参考资料
Rob J. Hyndman and George Athanasopoulos (2018)。“预测原理与实践，层次和分组时间序列”。*

源

​
reconcile

 reconcile (Y_hat_df:Union[ForwardRef('DataFrame[Any]'),ForwardRef('LazyFr
            ame[Any]')], S:Union[ForwardRef('DataFrame[Any]'),ForwardRef('
            LazyFrame[Any]')], tags:dict[str,numpy.ndarray], Y_df:Union[Fo
            rwardRef('DataFrame[Any]'),ForwardRef('LazyFrame[Any]'),NoneTy
            pe]=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)

*层次协调方法。

reconcile 方法类似于 SKLearn 的 fit_predict 方法，它应用在 reconcilers 列表中实例化的不同协调技术。

大多数协调方法可以用以下简便的线性代数符号表示

y~[a,b],τ=S[a,b][b]P[b][a,b]y^[a,b],τ\tilde{\mathbf{y}}_{[a,b],\tau} = \mathbf{S}_{[a,b][b]} \mathbf{P}_{[b][a,b]} \hat{\mathbf{y}}_{[a,b],\tau}S[a,b][b]​P[b][a,b]​y^​[a,b],τ​

其中 a,ba, b 表示聚合层和底层，S[a,b][b]\mathbf{S}_{[a,b][b]} 包含层次聚合约束，而 P[b][a,b]\mathbf{P}_{[b][a,b]} 因协调方法而异。协调后的预测是 y~[a,b],τ\tilde{\mathbf{y}}_{[a,b],\tau}，基础预测是 y^[a,b],τ\hat{\mathbf{y}}_{[a,b],\tau}。

参数
Y_hat_df：DataFrame，包含 [‘unique_id’, ‘ds’] 列和待协调模型的原始预测。
Y_df：DataFrame，包含 ['unique_id', 'ds', 'y'] 列的基础时间序列训练集。
如果 self.reconciles 的某个类接收 y_hat_insample，则 Y_df 必须将其作为列包含进来。
S：大小为 (base, bottom) 的求和矩阵 DataFrame，参见[聚合方法](https://nixtla.github.io/hierarchicalforecast/utils.html#aggregate)。
tags：每个键表示一个级别，其值包含与该级别相关的标签。
level：正浮点数列表 [0,100)，用于预测区间的置信水平。
intervals_method：字符串，用于计算预测区间的方法，可选值为 normality、bootstrap、permbu 之一。
num_samples：整数，-1 表示不返回样本；如果为正，则返回指定数量的概率相干样本。 seed：整数，0 表示默认值；numpy 生成器的随机种子，用于重现性。
is_balanced：布尔值，默认为 False；表示 Y_df 是否均衡，如果 Y_df 是均衡的，设置为 True 可以加快速度。
id_col：字符串，默认为 ‘unique_id’；标识每个序列的列。
time_col：字符串，默认为 ‘ds’；标识每个时间步的列，其值可以是时间戳或整数。
target_col：字符串，默认为 ‘y’；包含目标的列。

返回值
Y_tilde_df：DataFrame，包含协调后的预测结果。*

源

​
bootstrap_reconcile

 bootstrap_reconcile (Y_hat_df:Union[ForwardRef('DataFrame[Any]'),ForwardR
                      ef('LazyFrame[Any]')], S_df:Union[ForwardRef('DataFr
                      ame[Any]'),ForwardRef('LazyFrame[Any]')],
                      tags:dict[str,numpy.ndarray], Y_df:Union[ForwardRef(
                      'DataFrame[Any]'),ForwardRef('LazyFrame[Any]'),NoneT
                      ype]=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')

*基于 Bootstrap 的层次协调方法。

基于不同的随机种子，对 reconcilers 列表中实例化的不同协调技术重复应用 reconcile 方法 N 次。

参数
Y_hat_df：DataFrame，包含 [‘unique_id’, ‘ds’] 列和待协调模型的原始预测。
Y_df：DataFrame，包含 ['unique_id', 'ds', 'y'] 列的基础时间序列训练集。
如果 self.reconciles 的某个类接收 y_hat_insample，则 Y_df 必须将其作为列包含进来。
S：大小为 (base, bottom) 的求和矩阵 DataFrame，参见[聚合方法](https://nixtla.github.io/hierarchicalforecast/utils.html#aggregate)。
tags：每个键表示一个级别，其值包含与该级别相关的标签。
level：正浮点数列表 [0,100)，用于预测区间的置信水平。
intervals_method：字符串，用于计算预测区间的方法，可选值为 normality、bootstrap、permbu 之一。
num_samples：整数，-1 表示不返回样本；如果为正，则返回指定数量的概率相干样本。 num_seeds：整数，默认为 1；numpy 生成器的随机种子，用于重现性。
id_col：字符串，默认为 ‘unique_id’；标识每个序列的列。
time_col：字符串，默认为 ‘ds’；标识每个时间步的列，其值可以是时间戳或整数。
target_col：字符串，默认为 ‘y’；包含目标的列。

返回值
Y_bootstrap_df：DataFrame，包含基于 Bootstrap 的协调后预测结果。*

​
示例

import pandas as pd

from hierarchicalforecast.core import HierarchicalReconciliation
from hierarchicalforecast.methods import BottomUp, MinTrace
from hierarchicalforecast.utils import aggregate
from hierarchicalforecast.evaluation import evaluate
from statsforecast.core import StatsForecast
from statsforecast.models import AutoETS
from utilsforecast.losses import mase, rmse
from functools import partial

# Load TourismSmall dataset
df = pd.read_csv('https://raw.githubusercontent.com/Nixtla/transfer-learning-time-series/main/datasets/tourism.csv')
df = df.rename({'Trips': 'y', 'Quarter': 'ds'}, axis=1)
df.insert(0, 'Country', 'Australia')
qs = df['ds'].str.replace(r'(\d+) (Q\d)', r'\1-\2', regex=True)
df['ds'] = pd.PeriodIndex(qs, freq='Q').to_timestamp()

# Create hierarchical seires based on geographic levels and purpose
# And Convert quarterly ds string to pd.datetime format
hierarchy_levels = [['Country'],
                    ['Country', 'State'], 
                    ['Country', 'Purpose'], 
                    ['Country', 'State', 'Region'], 
                    ['Country', 'State', 'Purpose'], 
                    ['Country', 'State', 'Region', 'Purpose']]

Y_df, S_df, tags = aggregate(df=df, spec=hierarchy_levels)

# Split train/test sets
Y_test_df  = Y_df.groupby('unique_id').tail(8)
Y_train_df = Y_df.drop(Y_test_df.index)

# Compute base auto-ETS predictions
# Careful identifying correct data freq, this data quarterly 'Q'
fcst = StatsForecast(models=[AutoETS(season_length=4, model='ZZA')], freq='QS', n_jobs=-1)
Y_hat_df = fcst.forecast(df=Y_train_df, h=8, fitted=True)
Y_fitted_df = fcst.forecast_fitted_values()

reconcilers = [
                BottomUp(),
                MinTrace(method='ols'),
                MinTrace(method='mint_shrink'),
               ]
hrec = HierarchicalReconciliation(reconcilers=reconcilers)
Y_rec_df = hrec.reconcile(Y_hat_df=Y_hat_df, 
                          Y_df=Y_fitted_df,
                          S=S_df, tags=tags)

# Evaluate
eval_tags = {}
eval_tags['Total'] = tags['Country']
eval_tags['Purpose'] = tags['Country/Purpose']
eval_tags['State'] = tags['Country/State']
eval_tags['Regions'] = tags['Country/State/Region']
eval_tags['Bottom'] = tags['Country/State/Region/Purpose']

Y_rec_df_with_y = Y_rec_df.merge(Y_test_df, on=['unique_id', 'ds'], how='left')
mase_p = partial(mase, seasonality=4)

evaluation = evaluate(Y_rec_df_with_y, 
         metrics=[mase_p, rmse], 
         tags=eval_tags, 
         train_df=Y_train_df)

numeric_cols = evaluation.select_dtypes(include="number").columns
evaluation[numeric_cols] = evaluation[numeric_cols].map('{:.2f}'.format)