|
|
| import json
|
| from pathlib import Path
|
| import numpy as np
|
| import pandas as pd
|
| import joblib
|
|
|
| def load_forecaster(model_dir="."):
|
| path = Path(model_dir)
|
| config = json.loads((path / "config.json").read_text())
|
| models = joblib.load(path / "models.joblib")
|
| features = config["feature_names"]
|
| q_correction = config["calibrator"]["q_correction"] if config.get("calibrator") else 0.0
|
|
|
| def predict(df_features, apply_calibration=True):
|
| X = df_features[features]
|
| p10 = models[0.1].predict(X)
|
| p50 = models[0.5].predict(X)
|
| p90 = models[0.9].predict(X)
|
| stacked = np.sort(np.vstack([p10, p50, p90]), axis=0)
|
| p10, p50, p90 = stacked[0], stacked[1], stacked[2]
|
| if apply_calibration:
|
| p10 -= q_correction
|
| p90 += q_correction
|
| stacked_cal = np.sort(np.vstack([p10, p50, p90]), axis=0)
|
| p10, p50, p90 = stacked_cal[0], stacked_cal[1], stacked_cal[2]
|
| return pd.DataFrame({"p10": p10, "p50_point": p50, "p90": p90}, index=df_features.index)
|
|
|
| return predict
|
|
|
