code: def pipeline_fitted(series, X_df, horizon):
      models = [SeasonalNaive(season_length=7)]
      pd_series = fa.as_pandas(series)
      pd_X = None if X_df is None else fa.as_pandas(X_df)
      sf = StatsForecast(models=models, freq="D")
      sf.forecast(
          df=pd_series,
          h=horizon,
          X_df=pd_X,
          level=[80, 90],
          fitted=True,
      )
      fitted = sf.forecast_fitted_values()
      sf.forecast(
          df=series, h=horizon, X_df=X_df, level=[80, 90], fitted=True
      )
      distributed_fitted = (
          fa.as_pandas(sf.forecast_fitted_values())
          .sort_values(["unique_id", "ds"])
          .reset_index(drop=True)[fitted.columns]
          .astype(
              fitted.dtypes
          )  # fugue returns nullable and pyarrow dtypes
      )
      pd.testing.assert_frame_equal(
          fitted, distributed_fitted, atol=1e-5
      )
  
imports: import fugue.api
  import pandas
  from statsforecast.core import StatsForecast
  from statsforecast.models import ADIDA, AutoARIMA, ARIMA, CrostonClassic, CrostonOptimized, CrostonSBA, AutoETS, HistoricAverage, IMAPA, Naive, RandomWalkWithDrift, SeasonalExponentialSmoothing, SeasonalNaive, SeasonalWindowAverage, SimpleExponentialSmoothing, TSB, WindowAverage
label_desc: Use image node_id 3 for calling a global function with example usage: pipeline_fitted(series, X_df, horizon) without return types