refactor: decouple solar source filtering from MVLR

Move source-data cleaning into a standalone openenergyid.cleaning module
and return MVLR to pure regression fitting.

- clean_solar_production_frame takes explicit production/reference
  columns instead of sniffing the dependent-variable name, and never
  mutates values: guardrail rejection returns the original frame.
- FilteringDiagnostics reports a three-state status (applied,
  rejected-guardrail, nothing-to-remove); identified counts survive
  guardrail rejection while removed/retained describe the actual result.
- find_best_mvlr no longer filters; sourceDataFiltering is removed from
  the MVLR input and outlierFiltering from the result. Callers run
  cleaning as an explicit pre-step and carry the diagnostics.

Author: Molier

openenergyid/cleaning/__init__.py

@@ -0,0 +1,20 @@
+"""Source-data cleaning module.
+
+Standalone data-quality filters that run as an explicit pre-step before
+analysis modules such as MVLR. Analysis modules stay pure; callers decide
+when to clean and carry the diagnostics.
+"""
+
+from .models import (
+    FilteringDiagnostics,
+    FilteringStatus,
+    SolarSourceFilteringParameters,
+)
+from .solar import clean_solar_production_frame
+
+__all__ = [
+    "FilteringDiagnostics",
+    "FilteringStatus",
+    "SolarSourceFilteringParameters",
+    "clean_solar_production_frame",
+]

openenergyid/cleaning/models.py

@@ -0,0 +1,76 @@
+"""Models for source-data cleaning."""
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+FilteringStatus = Literal["applied", "rejected-guardrail", "nothing-to-remove"]
+
+
+class SolarSourceFilteringParameters(BaseModel):
+    """Parameters for solar production source-data filtering."""
+
+    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 FilteringDiagnostics(BaseModel):
+    """Diagnostics for a source-data filtering run.
+
+    The `identified*` counts always describe what the filter identified as
+    invalid, even when guardrails rejected the filtering. The observation
+    counts describe what actually happened to the returned frame.
+    """
+
+    status: FilteringStatus
+    original_observation_count: int = Field(alias="originalObservationCount")
+    retained_observation_count: int = Field(alias="retainedObservationCount")
+    removed_observation_count: int = Field(alias="removedObservationCount")
+    identified_negative_count: int = Field(0, alias="identifiedNegativeCount")
+    identified_zero_with_solar_count: int = Field(0, alias="identifiedZeroWithSolarCount")
+    identified_ratio_outlier_count: int = Field(0, alias="identifiedRatioOutlierCount")
+    reason: str | None = None
+
+    model_config = ConfigDict(populate_by_name=True)

openenergyid/cleaning/solar.py

@@ -0,0 +1,160 @@
+"""Solar production source-data filtering.
+
+Removes physically implausible solar production observations from a source
+frame before analysis: negative production, zero production while a solar
+reference signal indicates meaningful production, and robust
+production-to-reference ratio outliers.
+
+The filter removes rows; it never mutates values. Run it on raw (pre-resample)
+data so aggregated energy totals are computed from the retained observations.
+"""
+
+import numpy as np
+import pandas as pd
+
+from .models import FilteringDiagnostics, SolarSourceFilteringParameters
+
+
+def _resolve_reference_column(
+    frame: pd.DataFrame,
+    reference_column: str | None,
+    parameters: SolarSourceFilteringParameters,
+) -> str | None:
+    if reference_column is not None:
+        if reference_column not in frame.columns:
+            raise ValueError(f"Reference column '{reference_column}' not found in frame.")
+        return reference_column
+
+    for name in parameters.solar_reference_names:
+        if name in frame.columns:
+            return name
+
+    for column in frame.columns:
+        lower = column.lower()
+        if "solar" in lower and ("generation" in lower or "radiation" in lower):
+            return column
+
+    return None
+
+
+def _positive_reference_threshold(
+    series: pd.Series,
+    parameters: SolarSourceFilteringParameters,
+) -> float:
+    positive = series[series > 0]
+    if positive.empty:
+        return 0.0
+    return max(
+        float(positive.median()) * parameters.positive_reference_median_fraction,
+        parameters.minimum_positive_reference,
+    )
+
+
+def _robust_ratio_outlier_mask(
+    ratio: pd.Series,
+    parameters: SolarSourceFilteringParameters,
+) -> pd.Series:
+    if len(ratio) < parameters.minimum_retained_rows:
+        return pd.Series(False, index=ratio.index)
+
+    median = float(ratio.median())
+    mad = float((ratio - median).abs().median())
+    if not np.isfinite(mad) or mad <= 0:
+        q1 = float(ratio.quantile(0.25))
+        q3 = float(ratio.quantile(0.75))
+        iqr = q3 - q1
+        if not np.isfinite(iqr) or iqr <= 0:
+            return pd.Series(False, index=ratio.index)
+        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 > parameters.ratio_robust_z_threshold
+
+
+def clean_solar_production_frame(
+    frame: pd.DataFrame,
+    production_column: str,
+    reference_column: str | None = None,
+    parameters: SolarSourceFilteringParameters | None = None,
+) -> tuple[pd.DataFrame, FilteringDiagnostics]:
+    """Remove invalid solar production observations from a source frame.
+
+    Returns the filtered frame and diagnostics. When guardrails reject the
+    filtering (too few or too small a fraction of observations would remain),
+    the original frame is returned untouched and the diagnostics keep the
+    identified counts so callers can see what would have been removed.
+    """
+
+    parameters = parameters or SolarSourceFilteringParameters()
+
+    if production_column not in frame.columns:
+        raise ValueError(f"Production column '{production_column}' not found in frame.")
+
+    original_count = len(frame)
+    diagnostics = FilteringDiagnostics(
+        status="nothing-to-remove",
+        originalObservationCount=original_count,
+        retainedObservationCount=original_count,
+        removedObservationCount=0,
+    )
+
+    if original_count == 0:
+        diagnostics.reason = "empty frame"
+        return frame, diagnostics
+
+    reference = _resolve_reference_column(frame, reference_column, parameters)
+
+    # Numeric view used for mask computation only; the returned frame keeps
+    # the original values.
+    numeric = frame.apply(pd.to_numeric, errors="coerce")
+    y = numeric[production_column]
+    keep = pd.Series(True, index=numeric.index)
+
+    negative_mask = y < 0
+    diagnostics.identified_negative_count = int(negative_mask.sum())
+    keep &= ~negative_mask
+
+    if reference is not None:
+        reference_series = numeric[reference]
+        reference_threshold = _positive_reference_threshold(reference_series[keep], parameters)
+
+        zero_with_solar_mask = (y <= 0) & (reference_series > reference_threshold)
+        diagnostics.identified_zero_with_solar_count = int((keep & zero_with_solar_mask).sum())
+        keep &= ~zero_with_solar_mask
+
+        ratio_candidates = (
+            keep
+            & np.isfinite(y)
+            & np.isfinite(reference_series)
+            & (y > 0)
+            & (reference_series > reference_threshold)
+        )
+        ratios = y[ratio_candidates] / reference_series[ratio_candidates]
+        ratio_outliers = _robust_ratio_outlier_mask(ratios, parameters)
+        diagnostics.identified_ratio_outlier_count = int(ratio_outliers.sum())
+        keep.loc[ratio_outliers[ratio_outliers].index] = False
+
+    retained_count = int(keep.sum())
+    identified_count = original_count - retained_count
+
+    if identified_count == 0:
+        diagnostics.reason = "no invalid solar observations identified"
+        return frame, diagnostics
+
+    if retained_count < parameters.minimum_retained_rows:
+        diagnostics.status = "rejected-guardrail"
+        diagnostics.reason = "too few observations would remain after filtering"
+        return frame, diagnostics
+
+    if retained_count / original_count < parameters.minimum_retained_fraction:
+        diagnostics.status = "rejected-guardrail"
+        diagnostics.reason = "too much source data would be removed"
+        return frame, diagnostics
+
+    diagnostics.status = "applied"
+    diagnostics.retained_observation_count = retained_count
+    diagnostics.removed_observation_count = identified_count
+    diagnostics.reason = "solar production source-data filtering applied"
+    return frame.loc[keep].copy(), diagnostics

openenergyid/mvlr/__init__.py

@@ -6,21 +6,14 @@
     IndependentVariableResult,
     MultiVariableRegressionInput,
     MultiVariableRegressionResult,
-    OutlierFilteringDiagnostics,
-    SourceDataFilteringParameters,
     ValidationParameters,
 )
-from .source_data_filtering import clean_regression_frame, clean_solar_source_frame
 
 __all__ = [
     "find_best_mvlr",
-    "clean_regression_frame",
-    "clean_solar_source_frame",
     "IndependentVariableInput",
     "MultiVariableRegressionInput",
     "MultiVariableRegressionResult",
-    "OutlierFilteringDiagnostics",
-    "SourceDataFilteringParameters",
     "ValidationParameters",
     "IndependentVariableResult",
 ]

openenergyid/mvlr/main.py

@@ -3,30 +3,16 @@
 from .helpers import resample_input_data
 from .models import MultiVariableRegressionInput, MultiVariableRegressionResult
 from .mvlr import MultiVariableLinearRegression
-from .source_data_filtering import clean_regression_frame, clean_solar_source_frame
 
 
 def find_best_mvlr(
     data: MultiVariableRegressionInput,
 ) -> MultiVariableRegressionResult:
     """Cycle through multiple granularities and return the best model."""
     best_rsquared = 0
-    best_filtering = None
     for granularity in data.granularities:
         frame = data.data_frame()
-        frame, solar_filtering = clean_solar_source_frame(
-            frame,
-            data.dependent_variable,
-            data.source_data_filtering,
-        )
         frame = resample_input_data(data=frame, granularity=granularity)
-        frame, finite_filtering = clean_regression_frame(
-            frame,
-            data.dependent_variable,
-            data.source_data_filtering,
-        )
-        filtering = solar_filtering if solar_filtering.applied else finite_filtering
-        best_filtering = filtering
         mvlr = MultiVariableLinearRegression(
             data=frame,
             y=data.dependent_variable,
@@ -41,17 +27,9 @@ def find_best_mvlr(
             max_f_pvalue=data.validation_parameters.f_pvalue,
             max_pvalues=data.validation_parameters.pvalues,
         ):
-            result = MultiVariableRegressionResult.from_mvlr(mvlr)
-            result.outlier_filtering = filtering
-            return result
+            return MultiVariableRegressionResult.from_mvlr(mvlr)
         best_rsquared = max(best_rsquared, mvlr.fit.rsquared_adj)
-    detail = (
+    raise ValueError(
         f"No valid model found. Best R²: {best_rsquared:.3f} "
         f"(need ≥{data.validation_parameters.rsquared})"
     )
-    if best_filtering and best_filtering.applied:
-        detail += (
-            f"; outlier filtering removed {best_filtering.removed_observation_count}/"
-            f"{best_filtering.original_observation_count} observations"
-        )
-    raise ValueError(detail)