feat: configure MVLR source-data filtering

Author: Molier

openenergyid/mvlr/__init__.py

@@ -7,6 +7,7 @@
     MultiVariableRegressionInput,
     MultiVariableRegressionResult,
     OutlierFilteringDiagnostics,
+    SourceDataFilteringParameters,
     ValidationParameters,
 )
 from .source_data_filtering import clean_regression_frame
@@ -18,6 +19,7 @@
     "MultiVariableRegressionInput",
     "MultiVariableRegressionResult",
     "OutlierFilteringDiagnostics",
+    "SourceDataFilteringParameters",
     "ValidationParameters",
     "IndependentVariableResult",
 ]

openenergyid/mvlr/main.py

@@ -14,7 +14,11 @@ def find_best_mvlr(
     best_filtering = None
     for granularity in data.granularities:
         frame = data.data_frame()
-        frame, filtering = clean_regression_frame(frame, data.dependent_variable)
+        frame, filtering = clean_regression_frame(
+            frame,
+            data.dependent_variable,
+            data.source_data_filtering,
+        )
         best_filtering = filtering
         frame = resample_input_data(data=frame, granularity=granularity)
         mvlr = MultiVariableLinearRegression(

openenergyid/mvlr/models.py

@@ -34,6 +34,56 @@ class ValidationParameters(BaseModel):
     )
 
 
+class SourceDataFilteringParameters(BaseModel):
+    """Parameters for source-data filtering before regression fitting."""
+
+    enabled: bool = True
+    minimum_retained_fraction: float = Field(
+        0.50,
+        ge=0,
+        le=1,
+        alias="minimumRetainedFraction",
+        description="Minimum fraction of original observations that must remain after filtering.",
+    )
+    minimum_retained_rows: int = Field(
+        30,
+        ge=0,
+        alias="minimumRetainedRows",
+        description="Minimum number of observations that must remain after filtering.",
+    )
+    solar_reference_names: tuple[str, ...] = Field(
+        ("solarPowerGeneration", "solarRadiation"),
+        alias="solarReferenceNames",
+        description="Preferred source column names used as solar production/radiation references.",
+    )
+    positive_reference_median_fraction: float = Field(
+        0.10,
+        ge=0,
+        alias="positiveReferenceMedianFraction",
+        description="Fraction of the positive solar-reference median used as meaningful-solar threshold.",
+    )
+    minimum_positive_reference: float = Field(
+        0.05,
+        ge=0,
+        alias="minimumPositiveReference",
+        description="Minimum meaningful-solar threshold for the solar reference column.",
+    )
+    ratio_iqr_multiplier: float = Field(
+        3.0,
+        gt=0,
+        alias="ratioIqrMultiplier",
+        description="IQR multiplier used when MAD-based ratio filtering is unavailable.",
+    )
+    ratio_robust_z_threshold: float = Field(
+        4.5,
+        gt=0,
+        alias="ratioRobustZThreshold",
+        description="Robust z-score threshold for production-to-reference ratio outliers.",
+    )
+
+    model_config = ConfigDict(populate_by_name=True)
+
+
 class IndependentVariableInput(BaseModel):
     """
     Independent variable.
@@ -71,7 +121,11 @@ class MultiVariableRegressionInput(BaseModel):
     granularities: list[Granularity]
     allow_negative_predictions: bool = Field(alias="allowNegativePredictions", default=False)
     validation_parameters: ValidationParameters = Field(
-        alias="validationParameters", default=ValidationParameters()
+        alias="validationParameters", default_factory=ValidationParameters
+    )
+    source_data_filtering: SourceDataFilteringParameters = Field(
+        alias="sourceDataFiltering",
+        default_factory=SourceDataFilteringParameters,
     )
     single_use_exog_prefixes: list[str] | None = Field(
         # default=["HDD", "CDD", "FDD"],

openenergyid/mvlr/source_data_filtering.py

@@ -3,15 +3,14 @@
 import numpy as np
 import pandas as pd
 
-from .models import OutlierFilteringDiagnostics
+from .models import OutlierFilteringDiagnostics, SourceDataFilteringParameters
 
-MINIMUM_RETAINED_FRACTION = 0.50
-MINIMUM_RETAINED_ROWS = 30
-SOLAR_REFERENCE_NAMES = ("solarPowerGeneration", "solarRadiation")
 
-
-def _solar_reference_column(frame: pd.DataFrame) -> str | None:
-    for name in SOLAR_REFERENCE_NAMES:
+def _solar_reference_column(
+    frame: pd.DataFrame,
+    parameters: SourceDataFilteringParameters,
+) -> str | None:
+    for name in parameters.solar_reference_names:
         if name in frame.columns:
             return name
 
@@ -23,24 +22,37 @@ def _solar_reference_column(frame: pd.DataFrame) -> str | None:
     return None
 
 
-def _is_solar_production_model(dependent_variable: str, frame: pd.DataFrame) -> bool:
+def _is_solar_production_model(
+    dependent_variable: str,
+    frame: pd.DataFrame,
+    parameters: SourceDataFilteringParameters,
+) -> bool:
     dependent = dependent_variable.lower()
     if "solarphotovoltaic" in dependent:
         return True
     if "solar" in dependent and "production" in dependent:
         return True
-    return "production" in dependent and _solar_reference_column(frame) is not None
+    return "production" in dependent and _solar_reference_column(frame, parameters) is not None
 
 
-def _positive_reference_threshold(series: pd.Series) -> float:
+def _positive_reference_threshold(
+    series: pd.Series,
+    parameters: SourceDataFilteringParameters,
+) -> float:
     positive = series[series > 0]
     if positive.empty:
         return 0.0
-    return max(float(positive.median()) * 0.10, 0.05)
+    return max(
+        float(positive.median()) * parameters.positive_reference_median_fraction,
+        parameters.minimum_positive_reference,
+    )
 
 
-def _robust_ratio_outlier_mask(ratio: pd.Series) -> pd.Series:
-    if len(ratio) < MINIMUM_RETAINED_ROWS:
+def _robust_ratio_outlier_mask(
+    ratio: pd.Series,
+    parameters: SourceDataFilteringParameters,
+) -> pd.Series:
+    if len(ratio) < parameters.minimum_retained_rows:
         return pd.Series(False, index=ratio.index)
 
     median = float(ratio.median())
@@ -51,26 +63,35 @@ def _robust_ratio_outlier_mask(ratio: pd.Series) -> pd.Series:
         iqr = q3 - q1
         if not np.isfinite(iqr) or iqr <= 0:
             return pd.Series(False, index=ratio.index)
-        return (ratio < q1 - 3.0 * iqr) | (ratio > q3 + 3.0 * iqr)
+        return (ratio < q1 - parameters.ratio_iqr_multiplier * iqr) | (
+            ratio > q3 + parameters.ratio_iqr_multiplier * iqr
+        )
 
     robust_z = 0.6745 * (ratio - median).abs() / mad
-    return robust_z > 4.5
+    return robust_z > parameters.ratio_robust_z_threshold
 
 
 def clean_regression_frame(
     frame: pd.DataFrame,
     dependent_variable: str,
+    parameters: SourceDataFilteringParameters | None = None,
 ) -> tuple[pd.DataFrame, OutlierFilteringDiagnostics]:
     """Remove obvious bad source observations before fitting a regression model."""
 
+    parameters = parameters or SourceDataFilteringParameters()
     original_count = len(frame)
     diagnostics = OutlierFilteringDiagnostics(
+        enabled=parameters.enabled,
         originalObservationCount=original_count,
         retainedObservationCount=original_count,
         removedObservationCount=0,
         applied=False,
     )
 
+    if not parameters.enabled:
+        diagnostics.reason = "source-data filtering disabled"
+        return frame, diagnostics
+
     if original_count == 0 or dependent_variable not in frame.columns:
         diagnostics.reason = "empty frame or missing dependent variable"
         return frame, diagnostics
@@ -82,7 +103,7 @@ def clean_regression_frame(
     diagnostics.removed_non_finite_count = int((keep & ~finite_mask).sum())
     keep &= finite_mask
 
-    if not _is_solar_production_model(dependent_variable, numeric_frame):
+    if not _is_solar_production_model(dependent_variable, numeric_frame, parameters):
         cleaned = numeric_frame.loc[keep].copy()
         diagnostics.retained_observation_count = len(cleaned)
         diagnostics.removed_observation_count = original_count - len(cleaned)
@@ -95,30 +116,30 @@ def clean_regression_frame(
     diagnostics.removed_negative_count = int((keep & negative_mask).sum())
     keep &= ~negative_mask
 
-    solar_column = _solar_reference_column(numeric_frame)
+    solar_column = _solar_reference_column(numeric_frame, parameters)
     if solar_column is not None:
         solar_reference = numeric_frame[solar_column]
-        solar_threshold = _positive_reference_threshold(solar_reference[keep])
+        solar_threshold = _positive_reference_threshold(solar_reference[keep], parameters)
 
         zero_with_solar_mask = (y <= 0) & (solar_reference > solar_threshold)
         diagnostics.removed_zero_with_solar_count = int((keep & zero_with_solar_mask).sum())
         keep &= ~zero_with_solar_mask
 
         ratio_candidates = keep & (y > 0) & (solar_reference > solar_threshold)
         ratios = y[ratio_candidates] / solar_reference[ratio_candidates]
-        ratio_outliers = _robust_ratio_outlier_mask(ratios)
+        ratio_outliers = _robust_ratio_outlier_mask(ratios, parameters)
         diagnostics.removed_ratio_outlier_count = int(ratio_outliers.sum())
         keep.loc[ratio_outliers[ratio_outliers].index] = False
 
     cleaned = numeric_frame.loc[keep].copy()
     retained_count = len(cleaned)
     removed_count = original_count - retained_count
 
-    if retained_count < MINIMUM_RETAINED_ROWS:
+    if retained_count < parameters.minimum_retained_rows:
         diagnostics.reason = "too few observations retained after filtering"
         return numeric_frame, diagnostics
 
-    if retained_count / original_count < MINIMUM_RETAINED_FRACTION:
+    if retained_count / original_count < parameters.minimum_retained_fraction:
         diagnostics.reason = "too much source data would be removed"
         return numeric_frame, diagnostics
 

tests/mvlr/test_source_data_filtering.py

@@ -7,6 +7,7 @@
 from openenergyid.models import TimeDataFrame
 from openenergyid.mvlr import (
     MultiVariableRegressionInput,
+    SourceDataFilteringParameters,
     clean_regression_frame,
     find_best_mvlr,
 )
@@ -30,7 +31,9 @@ def _solar_regression_input(
 
     if zero_slice:
         production.iloc[zero_slice] = 0.0
-    for idx, value in (spikes or {45: 70.0, 60: 75.0, 75: 80.0}).items():
+    if spikes is None:
+        spikes = {45: 70.0, 60: 75.0, 75: 80.0}
+    for idx, value in spikes.items():
         production.iloc[idx] = value
 
     frame = pd.DataFrame(
@@ -104,6 +107,62 @@ def test_clean_regression_frame_keeps_original_data_when_filtering_too_much() ->
     assert len(cleaned) == 90
 
 
+def test_clean_regression_frame_uses_filtering_parameters() -> None:
+    """Retained-data guardrails should be caller-configurable."""
+    data = _solar_regression_input(zero_slice=slice(0, 55), spikes={})
+    frame = data.data_frame()
+
+    cleaned, diagnostics = clean_regression_frame(
+        frame,
+        DEPENDENT,
+        SourceDataFilteringParameters(
+            minimum_retained_fraction=0.30,
+            ratio_robust_z_threshold=999.0,
+        ),
+    )
+
+    assert diagnostics.applied
+    assert diagnostics.removed_zero_with_solar_count == 55
+    assert diagnostics.removed_observation_count == 55
+    assert len(cleaned) == 35
+
+
+def test_source_data_filtering_parameters_support_json_aliases() -> None:
+    """Filtering parameters should be usable from API-shaped input."""
+    parameters = SourceDataFilteringParameters.model_validate(
+        {
+            "enabled": False,
+            "minimumRetainedRows": 12,
+            "minimumRetainedFraction": 0.25,
+            "solarReferenceNames": ["customSolarReference"],
+            "ratioRobustZThreshold": 8.0,
+        }
+    )
+
+    assert not parameters.enabled
+    assert parameters.minimum_retained_rows == 12
+    assert parameters.minimum_retained_fraction == 0.25
+    assert parameters.solar_reference_names == ("customSolarReference",)
+    assert parameters.model_dump(by_alias=True)["ratioRobustZThreshold"] == 8.0
+
+
+def test_clean_regression_frame_can_be_disabled() -> None:
+    """Filtering can be disabled without changing the source frame."""
+    data = _solar_regression_input()
+    frame = data.data_frame()
+
+    cleaned, diagnostics = clean_regression_frame(
+        frame,
+        DEPENDENT,
+        SourceDataFilteringParameters(enabled=False),
+    )
+
+    assert not diagnostics.enabled
+    assert not diagnostics.applied
+    assert diagnostics.reason == "source-data filtering disabled"
+    assert cleaned is frame
+
+
 def test_clean_regression_frame_removes_non_finite_rows_for_non_solar_models() -> None:
     """Generic MVLR cleaning should drop non-finite observations."""
     index = pd.date_range("2025-04-01", periods=40, freq="D", tz="Europe/Brussels")