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    try:
 94        # Convert Excel serial dates to datetime if needed
 95        if pd.api.types.is_numeric_dtype(df["date"]):
 96            df["date"] = df["date"].apply(excel_date_to_datetime)
 97
 98        with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as tmp:
 99            df.to_csv(tmp.name, index=False)
100            result = api.forecasting(
101                file_path=tmp.name,
102                base_model=base_model,
103                n_hidden_features=n_hidden_features,
104                lags=lags,
105                type_pi=type_pi,
106                replications=replications,
107                h=h,
108            )
109        output_dates = result["date"]
110        if df.shape[1] == 2:  # univariate time series
111            res_df = pd.DataFrame(result.pop("sims"))
112            if return_sims:
113                res2_df = pd.DataFrame([])
114                res2_df["date"] = output_dates
115                sims_df = res_df.transpose()
116                sims_df.columns = [(i + 1) for i in range(sims_df.shape[1])]
117                return pd.concat([res2_df, sims_df], axis=1)
118            return pd.DataFrame(result)
119        else:  # multivariate time series
120            n_series = len(result["mean"][0])
121            series_names = [col for col in df.columns if col != "date"]
122            if len(series_names) != n_series:
123                series_names = [f"series_{i+1}" for i in range(n_series)]
124
125            # If series_choice is provided and valid, extract its index and return as univariate
126            if series_choice is not None and series_choice in series_names:
127                idx = series_names.index(series_choice)
128                # Extract only the selected series from each stat
129                summary_df = pd.DataFrame(
130                    {
131                        "date": output_dates,
132                        "mean": [x[idx] for x in result["mean"]],
133                        "lower": [x[idx] for x in result["lower"]],
134                        "upper": [x[idx] for x in result["upper"]],
135                    }
136                )
137                if return_sims:
138                    # sims: shape (replications, horizon, n_series)
139                    # list of replications, each is list of horizon, each is list of n_series
140                    sims = result["sims"]
141                    # For each replication, extract the selected series only
142                    sims_selected = [
143                        [h[idx] for h in rep] for rep in sims
144                    ]  # shape: (replications, horizon)
145                    sims_df = pd.DataFrame(sims_selected).transpose()
146                    sims_df.columns = [
147                        f"sim_{i+1}_{series_choice}"
148                        for i in range(sims_df.shape[1])
149                    ]
150                    res2_df = pd.DataFrame({"date": output_dates})
151                    return pd.concat([res2_df, sims_df], axis=1)
152                return summary_df
153
154            # Otherwise, return the full multivariate summary as before
155            summary_data = {"date": output_dates}
156            for stat in ["mean", "lower", "upper"]:
157                for s in range(n_series):
158                    summary_data[f"{stat}_{series_names[s]}"] = [
159                        x[s] for x in result[stat]
160                    ]
161            summary_df = pd.DataFrame(summary_data)
162            if return_sims:
163                # sims: shape (replications, horizon, n_series)
164                sims = result["sims"]
165                flat = []
166                for rep in sims:
167                    for s in range(n_series):
168                        flat.append([h[s] for h in rep])
169                sims_df = pd.DataFrame(flat).transpose()
170                colnames = []
171                for r in range(len(sims)):
172                    for s in range(n_series):
173                        colnames.append(f"sim_{r+1}_{series_names[s]}")
174                sims_df.columns = colnames
175                sims_df.insert(0, "date", output_dates)
176                return sims_df
177            return summary_df
178    except Exception as e:
179        return pd.DataFrame({"error": [str(e)]})