delta-barth/delta-barth-py
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 3 Releases 3 Wiki Activity

src/delta_barth/analysis/forecast.py aktualisiert

Browse Source
This commit is contained in:
frasu 2025-04-10 17:10:56 +00:00
parent 4ef8fc5e9d
commit 2934326258
1 changed files with 27 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
src/delta_barth/analysis/forecast.py
View File

@ -1,6 +1,8 @@
from __future__ import annotations
import datetime
# --- new: for calculating timedelta
from dateutil.relativedelta import relativedelta
import math
from collections.abc import Mapping, Set
from dataclasses import asdict
@ -215,8 +217,16 @@ def _process_sales(
features = ["jahr", "monat"]
target = SALES_FEAT
current_year = datetime.datetime.now().year
first_year = cast(int, df_cust["jahr"].min())
# --- new: not necessary anymore
#current_year = datetime.datetime.now().year
#first_year = cast(int, df_cust["jahr"].min())
# --- new: dates und forecast
#last_date = pd.to_datetime(monthly_sum.index[-1], format="%m.%Y")
last_date = pd.to_datetime(datetime.now().strftime("%m.%Y"), format="%m.%Y")
future_dates = pd.date_range(start=last_date + pd.DateOffset(months=1), periods=6, freq="MS")
forecast = pd.DataFrame({"datum": future_dates.strftime("%m.%Y")}).set_index("datum")
# Randomized Search
kfold = KFold(n_splits=5, shuffle=True)
@ -231,8 +241,9 @@ def _process_sales(
"early_stopping_rounds": [20, 50],
}
# --- new: best_estimator
best_estimator: BestEstimatorXGBRegressor | None = None
# --- new: best_estimator (internal usage)
best_estimator = None
best_params: BestParametersXGBRegressor | None = None
best_score_mae: float | None = float("inf")
best_score_r2: float | None = None
@ -240,20 +251,19 @@ def _process_sales(
too_few_month_points: bool = True
forecast: pd.DataFrame | None = None
# --- new: dates und forecast
#last_date = pd.to_datetime(monthly_sum.index[-1], format="%m.%Y")
last_date = pd.to_datetime(datetime.now().strftime("%m.%Y"), format="%m.%Y")
future_dates = pd.date_range(start=last_date + pd.DateOffset(months=1), periods=anzahl, freq="MS")
forecast = pd.DataFrame({"Datum": future_dates.strftime("%m.%Y")}).set_index("Datum")
dates = monthly_sum.index
for index, i in enumerate(range(len(dates)-36, -1, -12)):
current_date = dates[i]
split_date = dates[-anzahl]
# --- new: use monthly basis for time windows
starting_date = datetime.now() - relativedelta(months=36)
#starting_date = dates.max() - relativedelta(months=36)
start_index = next((i for i, date in enumerate(dates) if date >= starting_date), len(dates) - 1)
for index, i in enumerate(range(start_index, -1, -12)):
start_date = dates[i]
split_date = dates[-6]
train = cast(
pd.DataFrame,
monthly_sum.loc[current_date:split_date].copy(), # type: ignore
monthly_sum.loc[start_date:split_date].copy(), # type: ignore
)
test = cast(
pd.DataFrame,
@ -287,12 +297,13 @@ def _process_sales(
best_params = cast(BestParametersXGBRegressor, rand.best_params_)
best_score_mae = error
best_score_r2 = cast(float, r2_score(y_test, y_pred))
best_start_year = start_year
# --- new: use store start_date in best_start_year
best_start_year = start_date
# --- new: use best_estimator to calculate future values and store them in forecast
if best_estimator is not None:
X_future = pd.DataFrame({"jahr": future_dates.year, "monat": future_dates.month}, index=future_dates)
y_future = rand.best_estimator_.predict(X_future)
y_future = best_estimator.predict(X_future)
forecast["vorhersage"] = y_future
best_score_mae = best_score_mae if not math.isinf(best_score_mae) else None