fusionlab.nn.anomaly_detection.PredictionErrorAnomalyScore

class fusionlab.nn.anomaly_detection.PredictionErrorAnomalyScore

Bases: Layer, NNLearner

Calculates an anomaly score based on prediction error between

true and predicted sequences.

This layer quantifies the discrepancy between ground truth (y_true) and model predictions (y_pred) for time series, aggregating the error across time and features to produce a single anomaly score per sequence.

It provides a direct way to measure how well a model’s predictions match the actual outcomes, with larger errors typically indicating more anomalous or unexpected behavior.

Parameters:

• error_metric ({'mae', 'mse'}, default 'mae') –

  The metric used to calculate the element-wise error between y_true and y_pred at each time step and feature. * 'mae': Mean Absolute Error, $|y_{true} - y_{pred}|$. Less

  sensitive to large outliers.

  • 'mse': Mean Squared Error, $(y_{true} - y_{pred})^2$. Penalizes larger errors more heavily.

• aggregation ({'mean', 'max'}, default 'mean') –

  The method used to aggregate the per-step errors (which are already averaged across features) into a single score for the entire sequence. * 'mean': Computes the average error across all time steps. * 'max': Takes the maximum error encountered across all time

  steps. More sensitive to single large deviations.

• **kwargs – Additional keyword arguments passed to the parent Keras Layer.

Notes

This layer expects input as a list or tuple containing two tensors: [y_true, y_pred], both with the shape (Batch, TimeSteps, Features).

Use Case and Importance

This component directly implements the core logic behind prediction-based anomaly detection. It assumes that anomalies manifest as poor predictions by a model trained on normal patterns. It’s particularly useful when integrated into a multi-task learning setup where a forecasting model generates y_pred. The output score from this layer can then be fed into a loss function (like AnomalyLoss or used within prediction_based_loss()) to penalize the model for large prediction errors, implicitly guiding it to recognize or adapt to anomalous points. This approach links anomaly detection directly to the model’s predictive performance.

Mathematical Formulation

1. Element-wise Error: Calculate the error term \(e_{t,f}\) at each time step \(t\) and feature \(f\).

   \[e_{t,f} = y_{true; t,f} - y_{pred; t,f}\]

2. Step Error Score: Apply the chosen metric (mae or mse) and average across features ($F$) to get a score for each time step \(t\).

   \[\text{Error}_t = \frac{1}{F} \sum_{f=1}^F \text{metric}(e_{t,f})\]

   where \(\text{metric}(e) = |e|\) for MAE, and \(\text{metric}(e) = e^2\) for MSE.

3. Sequence Aggregation: Aggregate the step errors \(\{\text{Error}_t\}_{t=1}^T\) across time ($T$) using the chosen aggregation method (mean or max).

   \[\text{Score}_{seq} = \text{Aggregation}_{t=1}^T (\text{Error}_t)\]

call(inputs, training=False)

Calculates the anomaly score based on input [y_true, y_pred].

Examples

>>> from fusionlab.nn.anomaly_detection import PredictionErrorAnomalyScore
>>> import tensorflow as tf
>>> B, T, F = 32, 20, 3 # Batch, TimeSteps, Features
>>> # Assume y_true and y_pred come from your model/data
>>> y_true = tf.random.normal((B, T, F))
>>> y_pred = y_true + tf.random.normal((B, T, F), stddev=0.5) # Add noise
>>> # Instantiate the layer using Mean Absolute Error and Max aggregation
>>> error_scorer = PredictionErrorAnomalyScore(
...     error_metric='mae',
...     aggregation='max'
... )
>>> # Calculate scores
>>> anomaly_scores = error_scorer([y_true, y_pred])
>>> anomaly_scores.shape
TensorShape([32, 1])

See also

tensorflow.keras.layers.Layer

Base class for Keras layers.

fusionlab.nn.losses.prediction_based_loss

Loss function factory using a similar error-based anomaly concept.

fusionlab.nn.components.AnomalyLoss

Loss component that can take scores from this or other layers.

LSTMAutoencoderAnomaly

Reconstruction-based anomaly detection.

SequenceAnomalyScoreLayer

Feature-based anomaly scoring layer.

References

[1]

Chandola, V., Banerjee, A., & Kumar, V. (2009). Anomaly detection: A survey. ACM computing surveys (CSUR), 41(3), 1-58. (General survey covering deviation-based methods).

__init__(error_metric='mae', aggregation='mean', **kwargs)

Initialize layer.

Parameters:

• error_metric (str) – Metric for step-wise error (‘mae’ or ‘mse’). Default is ‘mae’.

• aggregation (str) – How to aggregate step-wise errors (‘mean’ or ‘max’). Default is ‘mean’.

Methods

__init__([error_metric, aggregation])	Initialize layer.
add_loss(loss)	Can be called inside of the call() method to add a scalar loss.
add_metric(*args, **kwargs)
add_variable(shape, initializer[, dtype, ...])	Add a weight variable to the layer.
add_weight([shape, initializer, dtype, ...])	Add a weight variable to the layer.
build(input_shape)
build_from_config(config)	Builds the layer's states with the supplied config dict.
call(inputs[, training])	Calculate anomaly score from prediction error.
compute_mask(inputs, previous_mask)
compute_output_shape(*args, **kwargs)
compute_output_spec(*args, **kwargs)
count_params()	Count the total number of scalars composing the weights.
from_config(config)	Creates layer from its config.
get_build_config()	Returns a dictionary with the layer's input shape.
get_config()	Returns the layer configuration.
get_params([deep])	Get the parameters for this learner.
get_weights()	Return the values of layer.weights as a list of NumPy arrays.
help(**kwargs)
load(file_path[, format])	Load the learner's state from a specified file in the desired format.
load_own_variables(store)	Loads the state of the layer.
quantize(mode[, type_check, config])
quantized_build(input_shape, mode)
quantized_call(*args, **kwargs)
rematerialized_call(layer_call, *args, **kwargs)	Enable rematerialization dynamically for layer's call method.
save([file_path, format, overwrite, ...])	Save the learner's state to a specified file in the desired format.
save_own_variables(store)	Saves the state of the layer.
set_params(**params)	Set the parameters of this learner.
set_weights(weights)	Sets the values of layer.weights from a list of NumPy arrays.
stateless_call(trainable_variables, ...[, ...])	Call the layer without any side effects.
summary()	Provide a summary of the learner's parameters.
symbolic_call(*args, **kwargs)

Attributes

compute_dtype	The dtype of the computations performed by the layer.
dtype	Alias of layer.variable_dtype.
dtype_policy
input	Retrieves the input tensor(s) of a symbolic operation.
input_dtype	The dtype layer inputs should be converted to.
input_spec
losses	List of scalar losses from add_loss, regularizers and sublayers.
metrics	List of all metrics.
metrics_variables	List of all metric variables.
my_params
non_trainable_variables	List of all non-trainable layer state.
non_trainable_weights	List of all non-trainable weight variables of the layer.
output	Retrieves the output tensor(s) of a layer.
path	The path of the layer.
quantization_mode	The quantization mode of this layer, None if not quantized.
supports_masking	Whether this layer supports computing a mask using compute_mask.
trainable	Settable boolean, whether this layer should be trainable or not.
trainable_variables	List of all trainable layer state.
trainable_weights	List of all trainable weight variables of the layer.
variable_dtype	The dtype of the state (weights) of the layer.
variables	List of all layer state, including random seeds.
weights	List of all weight variables of the layer.

__init__(error_metric='mae', aggregation='mean', **kwargs)

Initialize layer.

Parameters:

• error_metric (str) – Metric for step-wise error (‘mae’ or ‘mse’). Default is ‘mae’.

• aggregation (str) – How to aggregate step-wise errors (‘mean’ or ‘max’). Default is ‘mean’.

help(**kwargs)

my_params = PredictionErrorAnomalyScore(error_metric='mae', aggregation='mean')

call(inputs, training=False)

Calculate anomaly score from prediction error.

Parameters:

• inputs (list[Tensor]) – List containing [y_true, y_pred]. Both tensors should have shape (Batch, TimeSteps, Features).

• training (bool) – Ignored.

Returns:

Anomaly scores, shape (Batch, 1).

Return type:

Tensor

get_config()

Returns the layer configuration.

classmethod from_config(config)

Creates layer from its config.