gitea_admin/sientia-dataops-model-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1430: Implement advanced model training capabilities and enhanced data preprocessing. This includes support for Polynomial Regression with configurable degree and interaction terms, flexible per-variable lag configurations, and new data filtering options by date range and removed intervals. Comprehensive business validations are now enforced for all parameters, and MLflow logging has been extended to capture these detailed configurations. Additionally, Reduced Coulomb Energy (RCE) metrics are added for drift detection, with a new changelog documenting all pipeline parameter updates.

Browse Source
This commit is contained in:
Bruno Domingues
2025-12-17 21:39:43 -03:00
parent 4d6674758c
commit 6e8f87b2a3
10 changed files with 796 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
model_manager/sientia/metrics.py
View File

@@ -26,3 +26,123 @@ def r2(real_data: pd.Series, predictions: pd.Series) -> float:
Calculates the R2 score between the real data and the predictions.
"""
return round(r2_score(real_data.astype(np.float64), predictions.astype(np.float64)), 2)
def silverman_radius(data: np.ndarray) -> float:
"""
Calculate the Silverman bandwidth (radius) for a given dataset.
Args:
data (np.ndarray): Input data (1D array)
Returns:
float: Silverman bandwidth (radius)
"""
n = len(data)
sigma = np.std(data)
iqr = np.percentile(data, 75) - np.percentile(data, 25)
radius = 0.9 * min(sigma, iqr / 1.34) * n ** (-1 / 5)
return radius
def rce_train(training_set: pd.DataFrame, radius: float) -> pd.DataFrame:
"""
Get the Reduced Coulomb Energy (RCE) prototypes.
Args:
training_set (pd.DataFrame): The training set
radius (float): The radius of the RCE prototypes
Returns:
pd.DataFrame: The RCE prototypes
"""
train_vectors = training_set.values
# Vectorized distance computation for the radius calculation
diff_vectors = train_vectors[:, np.newaxis] - train_vectors[np.newaxis, :]
distances = np.linalg.norm(diff_vectors, axis=-1)
# Non-parametric radius: Silverman Radius
radius = silverman_radius(distances.flatten())
# Initialize prototypes with the first vector
prototypes = [train_vectors[0]]
for vector in train_vectors[1:]:
# Vectorized distance check between current vector and all prototypes
distances_to_prototypes = np.linalg.norm(prototypes - vector, axis=1)
# If no prototype is close, add the current vector as a new prototype
if np.all(distances_to_prototypes > radius):
prototypes.append(vector)
return pd.DataFrame(prototypes)
def rce_test(test_set: pd.DataFrame, prototypes: pd.DataFrame) -> pd.Series:
"""
Get the signed Reduced Coulomb Energy (RCE) predictions.
Args:
test_set (pd.DataFrame): The test set
prototypes (pd.DataFrame): The RCE prototypes
Returns:
pd.Series: The signed distances to the closest prototype for each test vector
"""
test_vectors = test_set.values
prototype_vectors = prototypes.values
# Vectorized computation of distances between test vectors and all prototypes
diff_vectors = test_vectors[:, np.newaxis] - prototype_vectors[np.newaxis, :]
distances = np.linalg.norm(diff_vectors, axis=-1)
# Find the closest prototype for each test vector
min_distances = np.min(distances, axis=1)
closest_prototypes = prototype_vectors[np.argmin(distances, axis=1)]
# Compute the signed distance for each test vector
signed_distances = np.sqrt(min_distances**2) * np.sign(
np.mean(test_vectors - closest_prototypes, axis=1)
)
return pd.Series(signed_distances)
def rce_drift(reference_data: pd.DataFrame, real_data: pd.DataFrame, column: str) -> pd.Series:
"""
Detect drift using the Reduced Coulomb Energy (RCE) method.
Args:
reference_data (pd.DataFrame): The reference data
real_data (pd.DataFrame): The real data
column (str): The target column to be analyzed. 'target' or 'prediction'
Returns:
pd.Series: Normalized drift distances
"""
common_columns = list(set(reference_data.columns).intersection(real_data.columns))
reference_data = reference_data[common_columns]
real_data = real_data[common_columns]
# Get prototypes
if column == 'target':
prototypes = rce_train(reference_data.drop(columns=['prediction']), 0.1)
else:
prototypes = rce_train(reference_data.drop(columns=['target']), 0.1)
# Distances to prototypes
if column == 'target':
distances_train = rce_test(reference_data.drop(columns=['prediction']), prototypes)
distances_test = rce_test(real_data.drop(columns=['prediction']), prototypes)
else:
distances_train = rce_test(reference_data.drop(columns=['target']), prototypes)
distances_test = rce_test(real_data.drop(columns=['target']), prototypes)
# Find the maximum absolute distance in the training set
max_abs_distance = max(abs(distances_train.max()), abs(distances_train.min()))
# Normalize while preserving sign
distances = distances_test / max_abs_distance
return distances

246
model_manager/sientia/models.py
View File

@@ -6,19 +6,24 @@ from sientia_do.operations.df_preprocessor import create_features, limit_dataset
from sientia_do.timeseries.analyzer import TimeSeriesDiscontinuityAnalyzer
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import PolynomialFeatures, StandardScaler
DISCONTINUITY_TREATMENT = 'Discontinuity Treatment'
LAG_SELECTION = 'Lag Selection'
RANGE_SELECTION = 'Range Selection & Data Removal'
STATIC_WINDOW_REMOVAL = 'Static Window Removal'
DEFINE_VARIABLES_LIMITS = 'Define Variables Limits'
NORMALIZATION = 'Normalization'
FEATURE_CREATION = 'Feature Creation'
LAG_CREATION = 'Lag Creation'
class LinearRegressionModel(BaseEstimator, TransformerMixin):
"""
Linear Regression Model for Time Series Analysis.
Supports both simple linear regression and polynomial regression.
Thread-safety: This class is NOT thread-safe during fit() operations.
Do not call fit() on the same instance from multiple threads simultaneously.
After fitting, predict() is thread-safe for read-only operations.
@@ -36,6 +41,9 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
model_params: dict[str, Any] | None = None,
clipping: dict[str, float] | None = None,
weights: dict[str, float] | None = None,
degree: int = 1,
interaction_only: bool = False,
verbose: bool = False,
):
"""
Linear Regression Model for Time Series Analysis
@@ -48,6 +56,9 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
*Format: {'min': min_value, 'max': max_value}*
weights (dict): The weights for the Linear Regression model \\
*Format: {'variable_name': weight}*
degree (int): The degree of the polynomial features (1 = linear, >1 = polynomial)
interaction_only (bool): If True, only interaction features are produced
verbose (bool): If True, print verbose output during fitting
Returns:
LinearRegressionModel: The prediction model object
@@ -60,6 +71,44 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
self.q1_target: float | None = None
self.q3_target: float | None = None
self.weights: dict[str, float] | None = weights
self.degree: int = degree
self.interaction_only: bool = interaction_only
self.verbose: bool = verbose
self.poly: PolynomialFeatures | None = None
self.poly_feature_names: list[str] | None = None
def create_poly_features(self, input_data: pd.DataFrame, fit: bool = False) -> pd.DataFrame:
"""
Create polynomial features from input data.
Args:
input_data (pd.DataFrame): Input data with feature columns
fit (bool): If True, fit the PolynomialFeatures transformer
Returns:
pd.DataFrame: DataFrame with polynomial features
"""
if self.degree <= 1:
return input_data
if fit:
self.poly = PolynomialFeatures(
degree=self.degree,
interaction_only=self.interaction_only,
include_bias=False,
)
poly_features = self.poly.fit_transform(input_data)
self.poly_feature_names = list(self.poly.get_feature_names_out(input_data.columns))
else:
if self.poly is None:
raise ValueError('PolynomialFeatures not fitted. Call fit() first.')
poly_features = self.poly.transform(input_data)
return pd.DataFrame(
poly_features,
columns=self.poly_feature_names,
index=input_data.index,
)
def fit(self, input_data: pd.DataFrame) -> 'LinearRegressionModel':
"""
@@ -71,13 +120,52 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
Returns:
LinearRegressionModel: The prediction model object
"""
assert self.variable_columns is not None, 'variable_columns must be set before fitting'
X_train = input_data[self.variable_columns]
y_train = input_data[self.target_variable]
if not self.target_variable:
raise ValueError('target_variable must be set before fitting')
# Infer variable_columns if not provided
if self.variable_columns is None:
self.variable_columns = [
col for col in input_data.columns if col != self.target_variable
]
# Validate columns exist
missing_cols = [col for col in self.variable_columns if col not in input_data.columns]
if missing_cols:
raise ValueError(f'Columns not found in input data: {missing_cols}')
if self.target_variable not in input_data.columns:
raise ValueError(f'Target variable {self.target_variable} not found in input data')
X_train = input_data[self.variable_columns].copy()
y_train = input_data[self.target_variable].copy()
# Handle infinite values
X_train = X_train.replace([np.inf, -np.inf], np.nan)
y_train = y_train.replace([np.inf, -np.inf], np.nan)
# Remove rows with NaN
valid_mask = ~(X_train.isna().any(axis=1) | y_train.isna())
X_train = X_train[valid_mask]
y_train = y_train[valid_mask]
# Remove columns with all NaN values
cols_to_drop = X_train.columns[X_train.isna().all()].tolist()
if cols_to_drop:
if self.verbose:
print(f'Dropping columns with all NaN values: {cols_to_drop}')
X_train = X_train.drop(columns=cols_to_drop)
self.variable_columns = [c for c in self.variable_columns if c not in cols_to_drop]
self.q1_target = y_train.quantile(0.25)
self.q3_target = y_train.quantile(0.75)
# Apply polynomial features if degree > 1
if self.degree > 1:
X_train = self.create_poly_features(X_train, fit=True)
if self.verbose and self.poly_feature_names is not None:
print(f'Created {len(self.poly_feature_names)} polynomial features')
# Fit the model
self.regr.fit(X_train, y_train)
@@ -86,11 +174,16 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
round_intercept = np.round(self.regr.intercept_, 3)
# Save the weights
weights = dict(zip(self.variable_columns, [float(c) for c in round_coef], strict=True))
feature_names = self.poly_feature_names if self.degree > 1 else self.variable_columns
assert feature_names is not None, 'feature_names should be set at this point'
weights = dict(zip(feature_names, [float(c) for c in round_coef], strict=True))
weights = dict(sorted(weights.items(), key=lambda item: abs(item[1]), reverse=True))
weights = {'Bias': float(round_intercept), **weights}
self.weights = weights
if self.verbose and feature_names is not None:
print(f'Model fitted with {len(feature_names)} features')
return self
def predict(self, input_data: pd.DataFrame) -> np.ndarray:
@@ -104,7 +197,16 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
Returns:
numpy.ndarray: The predicted target variable
"""
X_test = input_data[self.variable_columns]
assert self.variable_columns is not None, 'variable_columns must be set before predict'
X_test: pd.DataFrame = input_data[self.variable_columns].copy()
# Handle infinite values
X_test = X_test.replace([np.inf, -np.inf], np.nan)
# Apply polynomial features if degree > 1
if self.degree > 1:
X_test = self.create_poly_features(X_test, fit=False)
y_pred = self.regr.predict(X_test)
if self.clipping:
@@ -116,6 +218,15 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
return y_pred
def get_regressor(self) -> LinearRegression:
"""
Get the underlying LinearRegression model.
Returns:
LinearRegression: The sklearn LinearRegression model
"""
return self.regr
class DataPreprocessor(BaseEstimator, TransformerMixin):
"""
@@ -141,6 +252,9 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
nan_treatment: str | None = None,
lag_train: dict[str, int] | None = None,
lag_transform: dict[str, int] | None = None,
start_date: str | None = None,
end_date: str | None = None,
removed_intervals: list[tuple[str, str]] | None = None,
static_threshold: int | None = None,
low_lim: dict[str, float] | None = None,
upp_lim: dict[str, float] | None = None,
@@ -152,6 +266,7 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
cross_operations: list[str] | None = None,
created_lags: dict[str, int] | None = None,
steps_order: list[str] | None = None,
verbose: bool = False,
):
"""
Data Preprocessor for Time Series Analysis
@@ -161,11 +276,15 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
target_variable (str): The target variable name
input_columns (list): The input columns names in a list
nan_treatment (str): The treatment for missing values \\
*Options: 'drop', 'fill linear'*
*Options: 'drop', 'fill linear', 'linear interpolation'*
lag_train (dict): The lags for each variable to be applyed during training \\
*Format: {'variable_name': lag}*
lag_transform (dict): The lags for each variable to be applyed during transformation \\
*Format: {'variable_name': lag}*
start_date (str): The start date for filtering data (format: 'YYYY-MM-DD HH:MM:SS')
end_date (str): The end date for filtering data (format: 'YYYY-MM-DD HH:MM:SS')
removed_intervals (list): List of tuples with intervals to remove from data \\
*Format: [('start_date', 'end_date'), ...]*
static_threshold (int): The number of repeated values to be considered as static
low_lim (dict): The lower limits for each variable \\
*Format: {'variable_name': limit}*
@@ -188,11 +307,13 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
steps_order (list): The order of the steps to be executed in the pipeline \\
*Options for list: 'Discontinuity Treatment',
'Lag Selection',
'Range Selection & Data Removal',
'Static Window Removal',
'Define Variables Limits',
'Normalization',
'Feature Creation',
'Lag Creation'*
verbose (bool): If True, print verbose output during preprocessing
Returns:
DataPreprocessor: The data preprocessor object
@@ -203,6 +324,9 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
self.nan_treatment = nan_treatment
self.lag_train = lag_train if lag_train else {}
self.lag_transform = lag_transform if lag_transform else {}
self.start_date = start_date
self.end_date = end_date
self.removed_intervals = removed_intervals if removed_intervals else []
self.ar_var = ar_var
self.self_operations = self_operations
self.cross_operations = cross_operations
@@ -214,6 +338,8 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
self.scaler_name = scaler_name
self.scaler_params = scaler_params
self.feature_names_order: list[str] = [] # Initialize to avoid AttributeError
self.verbose = verbose
self._fitted_feature_order: list[str] | None = None # Track feature order after fit
if self.scaler_name == 'Standard Scaler':
self.scaler = StandardScaler()
@@ -226,11 +352,12 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
possible_steps = [
DISCONTINUITY_TREATMENT,
LAG_SELECTION,
RANGE_SELECTION,
STATIC_WINDOW_REMOVAL,
DEFINE_VARIABLES_LIMITS,
NORMALIZATION,
'Feature Creation',
'Lag Creation',
FEATURE_CREATION,
LAG_CREATION,
]
self.steps_order = steps_order or possible_steps
for step in possible_steps:
@@ -321,7 +448,61 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
pandas.DataFrame: The treated data
"""
if self.nan_treatment:
input_data = treat_nan(input_data, self.nan_treatment)
# Map 'linear interpolation' to 'fill linear' for compatibility
treatment = self.nan_treatment
if treatment == 'linear interpolation':
treatment = 'fill linear'
input_data = treat_nan(input_data, treatment)
if self.verbose:
print(f'Applied NaN treatment: {self.nan_treatment}')
return input_data
def range_selection(self, input_data: pd.DataFrame) -> pd.DataFrame:
"""
Filter data by date range and remove specified intervals.
Args:
input_data (pandas.DataFrame): The input data with datetime index
Returns:
pandas.DataFrame: The filtered data
"""
# Filter by start_date and end_date
if self.start_date:
try:
start = pd.to_datetime(self.start_date)
input_data = input_data[input_data.index >= start]
if self.verbose:
print(f'Filtered data from start_date: {self.start_date}')
except (ValueError, TypeError):
pass # Invalid date format, skip filtering
if self.end_date:
try:
end = pd.to_datetime(self.end_date)
input_data = input_data[input_data.index <= end]
if self.verbose:
print(f'Filtered data to end_date: {self.end_date}')
except (ValueError, TypeError):
pass # Invalid date format, skip filtering
# Remove specified intervals
if self.removed_intervals:
for interval in self.removed_intervals:
if len(interval) >= 2:
try:
interval_start = pd.to_datetime(interval[0])
interval_end = pd.to_datetime(interval[1])
mask = ~(
(input_data.index >= interval_start)
& (input_data.index <= interval_end)
)
input_data = input_data[mask]
if self.verbose:
print(f'Removed interval: {interval[0]} to {interval[1]}')
except (ValueError, TypeError):
pass # Invalid date format, skip this interval
return input_data
def lag_selection(self, input_data: pd.DataFrame, lag_dict: dict) -> pd.DataFrame:
@@ -469,6 +650,10 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
if step == LAG_SELECTION:
data_treat = self.lag_selection(data_treat, self.lag_train)
# Range Selection & Data Removal
if step == RANGE_SELECTION:
data_treat = self.range_selection(data_treat)
# Static Window Treatment
if step == STATIC_WINDOW_REMOVAL:
data_treat = self.treat_static_windows(data_treat)
@@ -493,6 +678,9 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
'variance': round(variance, 3),
}
# Store fitted feature order for predict method
self._fitted_feature_order = list(existing_columns)
return self
def transform(self, x: pd.DataFrame) -> pd.DataFrame:
@@ -525,6 +713,10 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
if step == LAG_SELECTION:
data_treat = self.lag_selection(data_treat, self.lag_transform)
# Range Selection & Data Removal (typically skipped in transform)
if step == RANGE_SELECTION:
data_treat = self.range_selection(data_treat)
# Static Window Treatment
if step == STATIC_WINDOW_REMOVAL:
data_treat = self.treat_static_windows(data_treat)
@@ -540,11 +732,11 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
data_treat[feature_cols] = self.scaler.transform(data_treat[feature_cols])
# Feature Creation
if step == 'Feature Creation':
if step == FEATURE_CREATION:
data_treat = self.create_features(data_treat)
# Lag Creation
if step == 'Lag Creation':
if step == LAG_CREATION:
# Autoregressive Variable
if self.input_columns is not None and self.ar_var in self.input_columns:
data_treat = self.create_ar(data_treat)
@@ -553,3 +745,33 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
data_treat = self.create_lags(data_treat)
return data_treat
def predict(self, x: pd.DataFrame) -> pd.DataFrame:
"""
Transform data for prediction (removes target variable).
This method is a wrapper around transform() that:
1. Transforms the input data
2. Removes the target variable column
3. Ensures features are in the same order as during fit
Args:
x (pandas.DataFrame): The input data
Returns:
pandas.DataFrame: The transformed data without target variable,
with features in the same order as during fit
"""
data_treat = self.transform(x)
# Remove target variable if present
if self.target_variable in data_treat.columns:
data_treat = data_treat.drop(columns=self.target_variable)
# Ensure features are in the same order as during fit
if self._fitted_feature_order is not None:
# Filter to only include columns that exist in both
available_cols = [c for c in self._fitted_feature_order if c in data_treat.columns]
data_treat = data_treat[available_cols]
return data_treat