Source code for py_ssa_lib.weights_strategies
from typing import Any, Hashable, Sequence
from numpy.typing import NDArray
from abc import ABC, abstractmethod
import numpy as np
[docs]
class WeightingStrategy(ABC):
"""
Base class for weighting of spectral channels.
"""
[docs]
@abstractmethod
def compute_weights(
self,
X: NDArray[np.float64],
**kwargs: Any
) -> NDArray[np.float64]:
"""
Parameters
----------
X : ndarray
Time-series matrix with shape (N, S).
**kwargs : Any
Strategy-specific parameters.
Returns
-------
ndarray
Weight vector with shape (S,).
"""
pass
[docs]
class PrecomputedWeights(WeightingStrategy):
"""
Pass precomputed weights for the spectral channels in SSA and MSSA. Weights must be strictly positive!
"""
[docs]
def compute_weights(
self,
X: NDArray[np.float64],
weights: Sequence[float],
**kwargs: Any
) -> NDArray[np.float64]:
"""
Parameters
----------
X : ndarray
Time-series matrix.
weights : Sequence[float]
User-provided weight for each series.
Returns
-------
ndarray
Precomputed weights as a NumPy array.
"""
return np.asarray(weights, dtype=float)
[docs]
class DistanceWeights(WeightingStrategy):
"""
Distance-based implementation for weighting of spectral channels in SSA and MSSA.
"""
[docs]
def compute_weights(
self,
X: NDArray[np.float64],
distances: Sequence[float],
bandwidth: float = 1.0,
**kwargs: Any
) -> NDArray[np.float64]:
"""
Parameters
----------
X : ndarray
Time-series matrix.
distances : Sequence[float]
Distance from the focal series to each series.
bandwidth : float
Scale controlling distance-weight decay.
Returns
-------
ndarray
Distance-based weights.
"""
distances = np.asarray(distances, dtype=float)
weights = np.exp(-0.5 * (distances / bandwidth)**2 )
return weights
[docs]
class ClusterAwareWeights(WeightingStrategy):
"""
Distance-based with cluster membership awareness implementation for weighting of spectral channels in SSA and MSSA.
"""
[docs]
def compute_weights(
self,
X: NDArray[np.float64],
distances: Sequence[float],
focal_cluster: Hashable,
neighbour_clusters: Sequence[Hashable],
bandwidth: float = 1.0,
cluster_penalty: float = 0.5,
**kwargs: Any
) -> NDArray[np.float64]:
"""
Parameters
----------
X : ndarray
Time-series matrix.
distances : Sequence[float]
Distance from the focal series to each neighbour.
focal_cluster : Hashable
Cluster label of the focal series.
neighbour_clusters : Sequence[Hashable]
Cluster labels of neighbouring series.
bandwidth : float
Scale controlling distance-weight decay.
cluster_penalty : float
Weight multiplier applied to different-cluster neighbours.
Returns
-------
ndarray
Combined spatial and cluster-aware weights.
"""
distances = np.asarray(distances, dtype=float)
neighbour_clusters = np.asarray(neighbour_clusters)
spatial = np.exp( -0.5 * (distances / bandwidth)**2 )
cluster = np.where(
neighbour_clusters == focal_cluster,
1.0,
cluster_penalty
)
return spatial * cluster