Time-Series Forecasting: Baselines, Backtesting, Calendar Features, Prediction Intervals and the Newsvendor Rule

"""Two-week sales forecasts for Crumb & Co's bakery. You write the functions marked TODO, one step at a time.

A forecaster is a function forecaster(history, future) -> numpy array: history is every row before the first
forecast day (date, loaves, bank_holiday); future holds the days to forecast, with date and bank_holiday only,
since the calendar is known in advance and the sales are not."""
import numpy as np
import pandas as pd
from sklearn.linear_model import Ridge

HORIZON = 14


def load(path="sales.csv"):
    df = pd.read_csv(path, parse_dates=["date"])
    return df


# ---------- Step 1: baselines ----------

def mae(actual, forecast):
    """Mean absolute error."""
    # TODO (Step 1): np.mean(np.abs(actual - forecast)) as a float.
    raise NotImplementedError("Step 1: write mae()")


def naive(history, future):
    """Every future day gets the last value seen."""
    # TODO (Step 1): np.full(len(future), the last loaves value).
    raise NotImplementedError("Step 1: write naive()")


def seasonal_naive(history, future):
    """Every future day gets the value of the same weekday in the last week of history (so day i of the future,
    counted from 0, gets history[-7 + i % 7])."""
    # TODO (Step 1): take the last 7 values of history["loaves"]; day i gets last_week[i % 7].
    raise NotImplementedError("Step 1: write seasonal_naive()")


# ---------- Step 2: backtesting ----------

def origins(df, first, last, step=7):
    """Forecast start dates from first to last (inclusive, pd.Timestamps or strings), every `step` days, keeping only
    those with HORIZON days of data from the start date on."""
    raise NotImplementedError("origins() arrives in Step 2")


def backtest(df, forecaster, starts, horizon=HORIZON):
    """For each start date: forecaster(history = rows before it, future = the next `horizon` rows without "loaves").
    A DataFrame with one row per forecast day: start, date, h (1..horizon), actual, forecast."""
    raise NotImplementedError("backtest() arrives in Step 2")


def score(bt):
    """MAE over the backtest days the shop was open (bank_holiday days have 0 sales and are forecast by rule)."""
    raise NotImplementedError("score() arrives in Step 2")


# ---------- Step 3: a model that knows the calendar ----------

LAGS = (14, 21, 28)   # days back; none shorter than HORIZON, or the value is not known when the forecast is made


def features(frame, sales):
    """Model inputs for the rows of frame (date, bank_holiday), a DataFrame with the columns
    dow_0 .. dow_6 (weekday one-hot, Monday = 0), sin and cos (of 2 pi day-of-year / 365.25), t (days since
    2023-01-01), bank_holiday, eve (the next day is a bank holiday, in frame or a zero-sales day in sales), and
    lag_14, lag_21, lag_28: sales on date - lag, looked up in `sales` (a Series of loaves indexed by date; a missing
    or zero value, as on a bank holiday, is replaced by the value a further 7 days back)."""
    raise NotImplementedError("features() arrives in Step 3")


def ridge_forecaster(history, future):
    """Ridge(alpha=1.0) on features() of the history's open days, predicting future; bank holidays forecast 0."""
    raise NotImplementedError("ridge_forecaster() arrives in Step 3")


# ---------- Step 4: how sure is the forecast ----------

def residual_quantiles(bt, qs):
    """{h: {q: quantile}} of actual - forecast over the open days of a backtest, per horizon step h."""
    raise NotImplementedError("residual_quantiles() arrives in Step 4")


def coverage(bt, quantiles, lo, hi):
    """Fraction of open backtest days whose actual lies within [forecast + quantiles[h][lo], forecast + quantiles[h][hi]]."""
    raise NotImplementedError("coverage() arrives in Step 4")


# ---------- Step 5: how many loaves to bake ----------

def critical_ratio(cost_leftover, cost_short):
    """The service level that minimises expected cost: cost_short / (cost_short + cost_leftover)."""
    raise NotImplementedError("critical_ratio() arrives in Step 5")


def bake_plan(forecast, h, quantiles, q):
    """Loaves to bake: forecast + the residual quantile q for each day's horizon step, rounded up, never below 0,
    and 0 on days the forecast is 0 (the shop is closed)."""
    raise NotImplementedError("bake_plan() arrives in Step 5")


def daily_cost(actual, baked, cost_leftover, cost_short):
    """Mean cost per day: leftover loaves x cost_leftover + loaves short x cost_short."""
    raise NotImplementedError("daily_cost() arrives in Step 5")