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(losses, **kwargs)	Add loss tensor(s), potentially dependent on layer inputs.
add_metric(value[, name])	Adds metric tensor to the layer.
add_update(updates)	Add update op(s), potentially dependent on layer inputs.
add_variable(*args, **kwargs)	Deprecated, do NOT use! Alias for add_weight.
add_weight([name, shape, dtype, ...])	Adds a new variable to the layer.
build(input_shape)	Creates the variables of the layer (for subclass implementers).
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[, mask])	Computes an output mask tensor.
compute_output_shape(input_shape)	Computes the output shape of the layer.
compute_output_signature(input_signature)	Compute the output tensor signature of the layer based on the inputs.
count_params()	Count the total number of scalars composing the weights.
finalize_state()	Finalizes the layers state after updating layer 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_input_at(node_index)	Retrieves the input tensor(s) of a layer at a given node.
get_input_mask_at(node_index)	Retrieves the input mask tensor(s) of a layer at a given node.
get_input_shape_at(node_index)	Retrieves the input shape(s) of a layer at a given node.
get_output_at(node_index)	Retrieves the output tensor(s) of a layer at a given node.
get_output_mask_at(node_index)	Retrieves the output mask tensor(s) of a layer at a given node.
get_output_shape_at(node_index)	Retrieves the output shape(s) of a layer at a given node.
get_params([deep])	Get the parameters for this learner.
get_weights()	Returns the current weights of the layer, as 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.
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 weights of the layer, from NumPy arrays.
summary()	Provide a summary of the learner's parameters.
with_name_scope(method)	Decorator to automatically enter the module name scope.

Attributes

activity_regularizer	Optional regularizer function for the output of this layer.
compute_dtype	The dtype of the layer's computations.
dtype	The dtype of the layer weights.
dtype_policy	The dtype policy associated with this layer.
dynamic	Whether the layer is dynamic (eager-only); set in the constructor.
inbound_nodes	Return Functional API nodes upstream of this layer.
input	Retrieves the input tensor(s) of a layer.
input_mask	Retrieves the input mask tensor(s) of a layer.
input_shape	Retrieves the input shape(s) of a layer.
input_spec	InputSpec instance(s) describing the input format for this layer.
losses	List of losses added using the add_loss() API.
metrics	List of metrics attached to the layer.
my_params
name	Name of the layer (string), set in the constructor.
name_scope	Returns a tf.name_scope instance for this class.
non_trainable_variables	Sequence of non-trainable variables owned by this module and its submodules.
non_trainable_weights	List of all non-trainable weights tracked by this layer.
outbound_nodes	Return Functional API nodes downstream of this layer.
output	Retrieves the output tensor(s) of a layer.
output_mask	Retrieves the output mask tensor(s) of a layer.
output_shape	Retrieves the output shape(s) of a layer.
stateful
submodules	Sequence of all sub-modules.
supports_masking	Whether this layer supports computing a mask using compute_mask.
trainable
trainable_variables	Sequence of trainable variables owned by this module and its submodules.
trainable_weights	List of all trainable weights tracked by this layer.
updates
variable_dtype	Alias of Layer.dtype, the dtype of the weights.
variables	Returns the list of all layer variables/weights.
weights	Returns the list of all layer variables/weights.

__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.