Fix/lag transformer

.gitignore

@@ -155,3 +155,5 @@ main[1-9]*.py
 *.jsonl
 docs/node_modules/
 tinkering
+
+examples/raw_02_example_reproduction.py

epftoolbox2/data/transformers/lag.py

@@ -1,6 +1,8 @@
 import pandas as pd
 from .base import Transformer
 
+_UNIT_THRESHOLDS = [(86400, "d"), (3600, "h"), (60, "min"), (1, "s")]
+
 
 class LagTransformer(Transformer):
     """Transform columns by shifting them to create lagged features.
@@ -15,20 +17,15 @@ class LagTransformer(Transformer):
         >>> result = transformer.transform(df)
     """
 
-    _FREQ_MAPPING = {
-        "day": "1D",
-        "days": "1D",
-        "d": "1D",
-        "hour": "1h",
-        "hours": "1h",
-        "h": "1h",
-        "minute": "1min",
-        "minutes": "1min",
-        "min": "1min",
-        "second": "1s",
-        "seconds": "1s",
-        "s": "1s",
-    }
+    _FREQ_MAPPING: dict[str, tuple[str, str]] = {}
+    for _aliases, _td, _unit in [
+        (("day", "days", "d", "1d"), "1D", "d"),
+        (("hour", "hours", "h", "1h"), "1h", "h"),
+        (("minute", "minutes", "min", "1min"), "1min", "min"),
+        (("second", "seconds", "s", "1s"), "1s", "s"),
+    ]:
+        for _alias in _aliases:
+            _FREQ_MAPPING[_alias] = (_td, _unit)
 
     def __init__(
         self,
@@ -39,37 +36,29 @@ def __init__(
         self.columns = [columns] if isinstance(columns, str) else columns
         self.lags = [lags] if isinstance(lags, int) else list(lags)
         self.freq = freq
-        freq_normalized = self._FREQ_MAPPING.get(freq.lower(), freq)
-        self._freq = pd.Timedelta(freq_normalized)
-        self._validate()
 
-    def _validate(self) -> None:
         if not self.lags:
             raise ValueError("At least one lag value must be provided")
-
         if not self.columns:
             raise ValueError("At least one column must be provided")
 
-    def _get_timedelta(self, lag: int) -> pd.Timedelta:
-        return self._freq * lag
+        mapping = self._FREQ_MAPPING.get(freq.lower())
+        if mapping:
+            freq_normalized, self._freq_unit = mapping
+        else:
+            freq_normalized, self._freq_unit = freq, None
+        self._freq = pd.Timedelta(freq_normalized)
 
     def _format_lag_name(self, column: str, lag: int) -> str:
-        total_td = self._freq * abs(lag)
-        total_seconds = int(total_td.total_seconds())
-
-        if total_seconds % 86400 == 0:
-            value = total_seconds // 86400
-            unit = "d"
-        elif total_seconds % 3600 == 0:
-            value = total_seconds // 3600
-            unit = "h"
-        elif total_seconds % 60 == 0:
-            value = total_seconds // 60
-            unit = "min"
+        if self._freq_unit:
+            unit, value = self._freq_unit, abs(lag)
         else:
-            value = total_seconds
-            unit = "s"
-
+            total_seconds = int((self._freq * abs(lag)).total_seconds())
+            value, unit = next(
+                (total_seconds // d, u)
+                for d, u in _UNIT_THRESHOLDS
+                if total_seconds % d == 0
+            )
         sign = "-" if lag >= 0 else "+"
         return f"{column}_{unit}{sign}{value}"
 
@@ -79,9 +68,7 @@ def transform(self, df: pd.DataFrame) -> pd.DataFrame:
             series = df[column]
             for lag in self.lags:
                 name = self._format_lag_name(column, lag)
-                timedelta = self._get_timedelta(lag)
-                shifted_index = df.index + timedelta
-                shifted_series = pd.Series(series.values, index=shifted_index)
-                lagged_data[name] = shifted_series.reindex(df.index)
+                shifted = pd.Series(series.values, index=df.index + self._freq * lag)
+                lagged_data[name] = shifted.reindex(df.index)
 
         return pd.concat([df, pd.DataFrame(lagged_data, index=df.index)], axis=1)

epftoolbox2/models/_worker.py

@@ -44,7 +44,6 @@ def _fit_one_numpy(hour: int, hz: int, day_in_test: int) -> Dict:
     run_date        = datetime.date.fromisoformat(cfg["test_start"]) + datetime.timedelta(days=day_in_test)
     run_date_str    = run_date.isoformat()
     target_date_str = (run_date + datetime.timedelta(days=hz)).isoformat()
-    run_weekday     = run_date.weekday()
 
     scaler = StandardScaler()
     train_x, train_y, test_x = scaler.fit_transform(
@@ -56,7 +55,7 @@ def _fit_one_numpy(hour: int, hz: int, day_in_test: int) -> Dict:
     return {
         "run_date":     run_date_str,
         "target_date":  target_date_str,
-        "run_weekday":  run_weekday,
+        "run_weekday":  run_date.weekday(),
         "hour":         hour,
         "horizon":      hz,
         "day_in_test":  day_in_test,

epftoolbox2/models/base.py

@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
 from concurrent.futures import ProcessPoolExecutor, as_completed
 from typing import List, Dict, Callable, Union, Tuple, Optional
-from datetime import date
+
 import gc
 import inspect
 import multiprocessing
@@ -34,7 +34,6 @@ def __init__(
         self._hour_days: Dict[int, np.ndarray] = {}
         self._offset: int = 0
         self._target: str = ""
-        self._run_date: str = ""
 
     @property
     def _model_kwargs(self) -> Dict:
@@ -69,7 +68,6 @@ def run(
             test_end   = resolve_date(test_end)
 
         self._target = target
-        self._run_date = date.today().isoformat()
 
         self._data = self._preprocess(data, horizon, target)
 
@@ -126,7 +124,6 @@ def run(
 
         in_memory = None if save_to else []
         mp_context = multiprocessing.get_context("fork" if sys.platform != "win32" else "spawn")
-
         with ProcessPoolExecutor(
             max_workers=n_processes,
             mp_context=mp_context,
@@ -192,8 +189,16 @@ def _cleanup(self) -> None:
 
     def _preprocess(self, data: pd.DataFrame, horizon: int, target: str) -> pd.DataFrame:
         data = data.drop(columns=["hour", "day"], errors="ignore")
-        new_cols = {f"{target}_d+{h}": data[target].shift(-24 * h) for h in range(1, horizon + 1)}
-        new_cols["day"] = pd.Series(np.arange(len(data)) // 24, index=data.index)
+
+        data = data[~data.index.duplicated(keep="first")]
+        new_cols = {}
+        for h in range(1, horizon + 1):
+            future_idx = data.index + pd.Timedelta(hours=24 * h)
+            aligned = data[target].reindex(future_idx)
+            aligned.index = data.index
+            new_cols[f"{target}_d+{h}"] = aligned
+        midnight = data.index.normalize()
+        new_cols["day"] = pd.Series((midnight - midnight[0]).days, index=data.index)
         new_cols["hour"] = pd.Series(data.index.hour, index=data.index)
         return pd.concat([data, pd.DataFrame(new_cols)], axis=1)
 
@@ -222,10 +227,10 @@ def _precompute_scalable_masks(
         for hour in range(24):
             days = self._hour_days[hour]
             day = self._offset
-            day_min = day - self.training_window - 1
-            mask = (days >= day_min) & (days <= day)
             hour_df = self._hour_data[hour]
             for hz in range(1, horizon + 1):
+                day_min = day - self.training_window - hz
+                mask = (days >= day_min) & (days <= day)
                 preds = self._expand_predictors(hz, hour=hour)
                 sample_x = (
                     hour_df[mask][preds]

examples/01_data_pipeline_only.py

@@ -40,6 +40,7 @@
         )
     )
     .add_transformer(TimezoneTransformer(target_tz="Europe/Warsaw"))
+    .add_transformer(ResampleTransformer(freq="1h", columns=["load_forecast_daily_min","load_forecast_daily_max"], method="ffill"))
     .add_transformer(ResampleTransformer(freq="1h"))
     .add_transformer(
         LagTransformer(

examples/02_model_pipeline_only.py

@@ -17,8 +17,8 @@
 from epftoolbox2.exporters import ExcelExporter, TerminalExporter
 
 if __name__ == "__main__":
-    df = pd.read_csv("data_output.csv", index_col=0, parse_dates=True)
-    df.index = pd.to_datetime(df.index, utc=True).tz_convert("Europe/Warsaw").tz_localize(None)
+    df = pd.read_csv("data_output.csv", index_col=0)
+    df.index = pd.to_datetime(df.index, utc=True).tz_convert("Europe/Warsaw")
     print(f"Loaded {len(df)} rows")
 
     seasonal_indicators = [
@@ -38,6 +38,8 @@
         *seasonal_indicators,
         "load_actual_d-1",
         "load_actual_d-7",
+        "price_d-1",
+        "price_d-7",
         "warsaw_temperature_2m_d+{horizon}",
     ]
 
@@ -59,7 +61,7 @@
             )
         )
         .add_evaluator(MAEEvaluator())
-        .add_exporter(TerminalExporter())
+        .add_exporter(TerminalExporter(['horizon']))
         .add_exporter(ExcelExporter("model_results.xlsx"))
     )
 
@@ -77,6 +79,3 @@
 
     print("\n=== Results by Hour ===")
     print(report.by_hour())
-
-    print("\n=== Results by Horizon ===")
-    print(report.by_horizon())

examples/03_full_workflow.py

@@ -21,11 +21,11 @@
 
 ENTSOE_API_KEY = os.environ.get("ENTSOE_API_KEY", "YOUR_API_KEY_HERE")
 
-DATA_START = "2023-05-01"
-DATA_END = "2024-08-01"
+DATA_START = "2023-01-01"
+DATA_END = "2024-04-01"
 
-TEST_START = "2024-06-01"
-TEST_END = "2024-07-01"
+TEST_START = "2024-02-01"
+TEST_END = "2024-03-01"
 if __name__ == "__main__":
 
     df = (
@@ -40,8 +40,6 @@
         .add_validator(NullCheckValidator(columns=["load_actual", "price"]))
         .run(start=DATA_START, end=DATA_END, cache=True)
     )
-
-
     seasonal_indicators = [
         "is_monday_d+{horizon}",
         "is_tuesday_d+{horizon}",
@@ -58,9 +56,9 @@
         "load_actual",
         *seasonal_indicators,
         "load_actual_d-1",
-        "load_actual_d-2",
+        "load_actual_d-7",
         "price_d-1",
-        "price_d-2",
+        "price_d-7",
         lambda h: f"warsaw_temperature_2m_d+{h}",
     ]
 
@@ -81,6 +79,3 @@
         horizon=7,
         save_dir="results",
     )
-
-    print(report.summary())
-    print(report.by_horizon())