techtonique_apis.techto_forecast

  1import tempfile
  2import datetime as dt
  3import numpy as np
  4import pandas as pd
  5
  6# import seaborn as sns
  7from xlwings import func, arg, ret
  8from .techtonique_apis import TechtoniqueAPI
  9
 10api = TechtoniqueAPI()
 11
 12
 13def excel_date_to_datetime(excel_serial):
 14    # Excel's day 0 is 1899-12-30
 15    return pd.to_datetime("1899-12-30") + pd.to_timedelta(
 16        excel_serial, unit="D"
 17    )
 18
 19
 20@func
 21@arg(
 22    "df",
 23    index=False,
 24    doc="Excel range with columns 'date' and one or more series.",
 25)
 26@arg("base_model", doc='Forecasting model (default: "RidgeCV")')
 27@arg("n_hidden_features", doc="Number of hidden features (default: 5)")
 28@arg("lags", doc="Number of lags (default: 25)")
 29@arg("type_pi", doc='Prediction interval type (default: "kde")')
 30@arg("replications", doc="Number of simulation replications (default: 10)")
 31@arg("h", doc="Forecast horizon (default: 5)")
 32@arg(
 33    "return_sims",
 34    doc="If TRUE, return simulation matrix; else, return forecast summary (default: FALSE)",
 35)
 36@arg(
 37    "series_choice",
 38    doc="(Optional) Name of the series to forecast if multiple are present",
 39)
 40@ret(index=False, doc="Forecast or simulation results as a table for Excel")
 41def techto_forecast(
 42    df: pd.DataFrame,
 43    base_model: str = "RidgeCV",
 44    n_hidden_features: int = 5,
 45    lags: int = 25,
 46    type_pi: str = "kde",
 47    replications: int = 10,
 48    h: int = 5,
 49    return_sims: bool = False,
 50    series_choice: str = None,
 51) -> pd.DataFrame:
 52    """Forecasting: pass a time series as a DataFrame from Excel, return forecast.
 53
 54    Excel/xlwings custom function: Forecast a time series passed as a DataFrame from Excel, using the Techtonique API.
 55
 56    Parameters
 57    ----------
 58
 59    df : pd.DataFrame
 60        The input time series data as a DataFrame (from Excel range).
 61
 62    base_model : str, default "RidgeCV"
 63        The base model to use for forecasting.
 64
 65    n_hidden_features : int, default 5
 66        Number of hidden features for the model.
 67
 68    lags : int, default 25
 69        Number of lags to use in the model.
 70
 71    type_pi : str, default "kde"
 72        Type of prediction interval ("kde" or other supported types).
 73
 74    replications : int, default 10
 75        Number of simulation replications.
 76
 77    h : int, default 5
 78        Forecast horizon (number of periods ahead to forecast).
 79
 80    return_sims : bool, default False
 81        If True, return the simulation matrix; otherwise, return the forecast summary bounds.
 82
 83    series_choice : str, optional
 84        If provided, specifies which series to forecast from the DataFrame.
 85
 86    Returns
 87    -------
 88
 89    pd.DataFrame
 90        The forecast results or simulation matrix as a DataFrame for Excel.
 91
 92    """
 93    # Convert Excel serial dates to datetime if needed
 94    if pd.api.types.is_numeric_dtype(df["date"]):
 95        df["date"] = df["date"].apply(excel_date_to_datetime)
 96
 97    with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as tmp:
 98        df.to_csv(tmp.name, index=False)
 99        result = api.forecasting(
100            file_path=tmp.name,
101            base_model=base_model,
102            n_hidden_features=n_hidden_features,
103            lags=lags,
104            type_pi=type_pi,
105            replications=replications,
106            h=h,
107        )
108    output_dates = result["date"]
109    if df.shape[1] == 2:  # univariate time series
110        res_df = pd.DataFrame(result.pop("sims"))
111        if return_sims:
112            res2_df = pd.DataFrame([])
113            res2_df["date"] = output_dates
114            sims_df = res_df.transpose()
115            sims_df.columns = [(i + 1) for i in range(sims_df.shape[1])]
116            return pd.concat([res2_df, sims_df], axis=1)
117        return pd.DataFrame(result)
118    else:  # multivariate time series
119        n_series = len(result["mean"][0])
120        series_names = [col for col in df.columns if col != "date"]
121        if len(series_names) != n_series:
122            series_names = [f"series_{i+1}" for i in range(n_series)]
123
124        # If series_choice is provided and valid, extract its index and return as univariate
125        if series_choice is not None and series_choice in series_names:
126            idx = series_names.index(series_choice)
127            # Extract only the selected series from each stat
128            summary_df = pd.DataFrame(
129                {
130                    "date": output_dates,
131                    "mean": [x[idx] for x in result["mean"]],
132                    "lower": [x[idx] for x in result["lower"]],
133                    "upper": [x[idx] for x in result["upper"]],
134                }
135            )
136            if return_sims:
137                # sims: shape (replications, horizon, n_series)
138                # list of replications, each is list of horizon, each is list of n_series
139                sims = result["sims"]
140                # For each replication, extract the selected series only
141                sims_selected = [
142                    [h[idx] for h in rep]
143                    # shape: (replications, horizon)
144                    for rep in sims
145                ]
146                sims_df = pd.DataFrame(sims_selected).transpose()
147                sims_df.columns = [
148                    f"sim_{i+1}_{series_choice}"
149                    for i in range(sims_df.shape[1])
150                ]
151                res2_df = pd.DataFrame({"date": output_dates})
152                return pd.concat([res2_df, sims_df], axis=1)
153            return summary_df
154
155        # Otherwise, return the full multivariate summary as before
156        summary_data = {"date": output_dates}
157        for stat in ["mean", "lower", "upper"]:
158            for s in range(n_series):
159                summary_data[f"{stat}_{series_names[s]}"] = [
160                    x[s] for x in result[stat]
161                ]
162        summary_df = pd.DataFrame(summary_data)
163        if return_sims:
164            # sims: shape (replications, horizon, n_series)
165            sims = result["sims"]
166            flat = []
167            for rep in sims:
168                for s in range(n_series):
169                    flat.append([h[s] for h in rep])
170            sims_df = pd.DataFrame(flat).transpose()
171            colnames = []
172            for r in range(len(sims)):
173                for s in range(n_series):
174                    colnames.append(f"sim_{r+1}_{series_names[s]}")
175            sims_df.columns = colnames
176            sims_df.insert(0, "date", output_dates)
177            return sims_df
178        return summary_df