fusionlab.nn.anomaly_detection.SequenceAnomalyScoreLayer¶

class fusionlab.nn.anomaly_detection.SequenceAnomalyScoreLayer[source]¶

Bases: Layer, NNLearner

Computes an anomaly score from input features using a Multi-Layer: Perceptron (MLP).

This layer processes input features, typically representing learned embeddings or aggregated sequence information from upstream layers, through a configurable MLP to produce a scalar anomaly score for each input sample.

It provides flexibility in defining the depth and width of the MLP, activation functions, normalization, and dropout for regularization. The output score reflects the model’s learned assessment of how anomalous the input features are.

Parameters:

hidden_units (int or list of int) –
Specifies the structure of the hidden layers in the MLP. * If int: A single hidden layer with that many units is used. * If list[int]: Creates multiple hidden layers, where each

integer in the list defines the number of units for the corresponding layer.
activation (str, default 'relu') – Activation function applied after each hidden dense layer (but before normalization or dropout). Common choices include ‘relu’, ‘elu’, ‘gelu’, ‘tanh’.
dropout_rate (float, default 0.1) – Dropout rate applied after activation (and normalization, if used) in each hidden layer. Value between 0 and 1.
use_norm (bool or str, default False) –
Specifies whether to apply normalization after the activation in hidden layers. * False: No normalization. * True or ‘layer’: Use Layer Normalization. * ‘batch’: Use Batch Normalization. Note that Batch Normalization

behaves differently during training and inference.
final_activation (str, default 'linear') –
Activation function applied to the final output neuron that produces the scalar anomaly score. * ‘linear’: Produces an unbounded score. * ‘sigmoid’: Produces a score between 0 and 1, interpretable

as a probability or normalized score.
- Other activations like ‘softplus’ can also be used to ensure non-negative scores.
**kwargs – Additional keyword arguments passed to the parent Keras Layer.

Notes

This layer typically expects input features with shape (Batch, Features). If your input is sequential (Batch, TimeSteps, Features), you might need to flatten or pool it before feeding it to this layer.

Use Case and Importance

This layer is designed to be a part of a larger model, acting as a dedicated “scoring head” that learns to map complex internal features to an anomaly score. It’s useful when you want the model to learn what constitutes an anomaly based on learned representations, rather than relying solely on reconstruction error or predefined rules. This approach aligns well with the concept of feature-based anomaly detection within models like XTFT. Training this layer effectively requires integrating it into a larger network and defining a suitable loss function that utilizes its output score, potentially combining it with the primary task’s loss (e.g., forecasting loss) or using anomaly labels if available (supervised training).

Mathematical Formulation

The layer implements a standard Multi-Layer Perceptron (MLP). For an input feature vector $\mathbf{h}$ and $L$ hidden layers:

Let $\mathbf{h}^{(0)} = \mathbf{h}$. For each hidden layer $i = 1 dots L$:

\[\begin{split}\mathbf{a}^{(i)} = \text{Dense}_i(\mathbf{h}^{(i-1)}) \\ \mathbf{n}^{(i)} = \text{Activation}(\mathbf{a}^{(i)}) \\ \mathbf{o}^{(i)} = \text{Normalization}(\mathbf{n}^{(i)}) \quad (\text{if use_norm=True}) \\ \mathbf{h}^{(i)} = \text{Dropout}(\mathbf{o}^{(i)} \text{ or } \mathbf{n}^{(i)})\end{split}\]

The final score is computed from the last hidden layer’s output $\mathbf{h}^{(L)}$:

\[\text{Score} = \text{FinalActivation}(\text{Dense}_{out}(\mathbf{h}^{(L)}))\]

where Dense includes weights, biases, and the specified activation or normalization steps.

call(inputs, training=False)[source]¶: Performs the forward pass to compute anomaly scores.

Examples

>>> from fusionlab.nn.anomaly_detection import SequenceAnomalyScoreLayer
>>> import tensorflow as tf
>>> B, F = 32, 64 # Batch, Features
>>> # Assume 'features' are output from another layer
>>> features = tf.random.normal((B, F))
>>> # Instantiate with multiple hidden layers and LayerNorm
>>> anomaly_scorer = SequenceAnomalyScoreLayer(
...     hidden_units=[64, 32], # Two hidden layers
...     activation='relu',
...     dropout_rate=0.2,
...     use_norm='layer', # Use Layer Normalization
...     final_activation='sigmoid' # Output score between 0 and 1
... )
>>> # Get anomaly scores
>>> scores = anomaly_scorer(features, training=True) # Pass training flag
>>> scores.shape
TensorShape([32, 1])

See also

tensorflow.keras.layers.Layer: Base class for Keras layers.
tensorflow.keras.layers.Dense: Fully-connected layer used internally.
tensorflow.keras.layers.Dropout: Dropout regularization layer.
tensorflow.keras.layers.LayerNormalization: Normalization layer option.
tensorflow.keras.layers.BatchNormalization: Normalization layer option.
fusionlab.nn.transformers.XTFT: Can incorporate feature-based anomaly detection potentially using layers like this.
fusionlab.nn.losses.AnomalyLoss: Loss component for anomaly scores.
LSTMAutoencoderAnomaly: Alternative reconstruction-based component.

References

__init__(hidden_units, activation='relu', dropout_rate=0.1, use_norm=False, final_activation='linear', kernel_regularizer=None, bias_regularizer=None, **kwargs)[source]¶

Parameters:

hidden_units (int | List[int])
activation (str)
dropout_rate (float)
use_norm (bool | str)
final_activation (str)

Methods

`__init__`(hidden_units[, activation, ...])
`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])	Forward pass: (Dense -> [Norm] -> Dropout) * N -> Dense Output.
`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__(hidden_units, activation='relu', dropout_rate=0.1, use_norm=False, final_activation='linear', kernel_regularizer=None, bias_regularizer=None, **kwargs)[source]¶

Parameters:

hidden_units (int | List[int])
activation (str)
dropout_rate (float)
use_norm (bool | str)
final_activation (str)

call(inputs, training=False)[source]¶: Forward pass: (Dense -> [Norm] -> Dropout) * N -> Dense Output. Expects inputs of shape (Batch, Features).

get_config()[source]¶: Returns the layer configuration.

classmethod from_config(config)[source]¶: Creates layer from its config.

help(**kwargs)¶

my_params = SequenceAnomalyScoreLayer( hidden_units, activation='relu', dropout_rate=0.1, use_norm=False, final_activation='linear', kernel_regularizer=None, bias_regularizer=None )¶