Skip to content

Commit

Permalink
update metrics
Browse files Browse the repository at this point in the history
  • Loading branch information
AlexanderVNikitin committed Dec 14, 2023
1 parent 1576178 commit 6faf01b
Show file tree
Hide file tree
Showing 6 changed files with 117 additions and 5 deletions.
2 changes: 1 addition & 1 deletion docs/modules/root.rst
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ Augmentations

Metrics
--------------
.. automodule:: tsgm.models.metrics
.. automodule:: tsgm.metrics.metrics
:members:
:undoc-members:

Expand Down
1 change: 1 addition & 0 deletions requirements/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ tensorflow_privacy==0.8.5
tensorflow==2.9.1
tensorflow_probability==0.17.0
scipy>=1.7.3
antropy==0.1.6
numpy>=1.21.6
statsmodels
dtaidistance==2.3.10
Expand Down
7 changes: 7 additions & 0 deletions tests/test_metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,3 +205,10 @@ def test_discriminative_metric():
discr_metric = tsgm.metrics.DiscriminativeMetric()
assert discr_metric(d_hist=D1, d_syn=D2, model=model, test_size=0.2, random_seed=42, n_epochs=10) == 1.0
assert discr_metric(d_hist=D1, d_syn=D2, model=model, metric=sklearn.metrics.precision_score, test_size=0.2, random_seed=42, n_epochs=10) == 1.0


def test_entropy_metric():
ts = np.array([[[0, 2], [11, -11], [1, 2]], [[10, 21], [1, -1], [6, 8]]]).astype(np.float32)
D1 = tsgm.dataset.Dataset(ts, y=None)
spec_entropy_metric = tsgm.metrics.EntropyMetric()
assert spec_entropy_metric(D1) == 2.6402430161833763
2 changes: 1 addition & 1 deletion tsgm/metrics/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,5 +2,5 @@
from tsgm.metrics.metrics import (
DistanceMetric, ConsistencyMetric, BaseDownstreamEvaluator,
DownstreamPerformanceMetric, PrivacyMembershipInferenceMetric,
MMDMetric, DiscriminativeMetric
MMDMetric, DiscriminativeMetric, EntropyMetric
)
108 changes: 106 additions & 2 deletions tsgm/metrics/metrics.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
import abc
import antropy
import typing as T
import logging
import numpy as np
import itertools
import sklearn
import antropy
from tqdm import tqdm
from tensorflow.python.types.core import TensorLike

import tsgm

Expand Down Expand Up @@ -199,8 +202,19 @@ def __call__(self, d_tr: tsgm.dataset.Dataset, d_syn: tsgm.dataset.Dataset, d_te
class MMDMetric(Metric):
"""
This metric calculated MMD between real and synthetic samples
"""
Args:
d (tsgm.dataset.DatasetOrTensor): The input dataset or tensor.
Returns:
float: The computed spectral entropy.
Example:
>>> metric = MMDMetric(kernel)
>>> dataset, synth_dataset = tsgm.dataset.Dataset(...), tsgm.dataset.Dataset(...)
>>> result = metric(dataset)
>>> print(result)
"""
def __init__(self, kernel: T.Callable = tsgm.utils.mmd.exp_quad_kernel) -> None:
self.kernel = kernel

Expand All @@ -213,7 +227,49 @@ def __call__(self, D1: tsgm.dataset.DatasetOrTensor, D2: tsgm.dataset.DatasetOrT

class DiscriminativeMetric(Metric):
"""
The discriminative metric measures how accurately a discriminative model can separate synthetic and real data.
The DiscriminativeMetric measures the discriminative performance of a model in distinguishing
between synthetic and real datasets.
This metric evaluates a discriminative model by training it on a combination of synthetic
and real datasets and assessing its performance on a test set.
:param d_hist: Real dataset.
:type d_hist: tsgm.dataset.DatasetOrTensor
:param d_syn: Synthetic dataset.
:type d_syn: tsgm.dataset.DatasetOrTensor
:param model: Discriminative model to be evaluated.
:type model: T.Callable
:param test_size: Proportion of the dataset to include in the test split
or the absolute number of test samples.
:type test_size: T.Union[float, int]
:param n_epochs: Number of training epochs for the model.
:type n_epochs: int
:param metric: Optional evaluation metric to use (default: accuracy).
:type metric: T.Optional[T.Callable]
:param random_seed: Optional random seed for reproducibility.
:type random_seed: T.Optional[int]
:return: Discriminative performance metric.
:rtype: float
Example:
--------
>>> from my_module import DiscriminativeMetric, MyDiscriminativeModel
>>> import tsgm.dataset
>>> import numpy as np
>>> import sklearn
>>>
>>> # Create real and synthetic datasets
>>> real_dataset = tsgm.dataset.Dataset(...) # Replace ... with appropriate arguments
>>> synthetic_dataset = tsgm.dataset.Dataset(...) # Replace ... with appropriate arguments
>>>
>>> # Create a discriminative model
>>> model = MyDiscriminativeModel() # Replace with the actual discriminative model class
>>>
>>> # Create and use the DiscriminativeMetric
>>> metric = DiscriminativeMetric()
>>> result = metric(real_dataset, synthetic_dataset, model, test_size=0.2, n_epochs=10)
>>> print(result)
"""
def __call__(self, d_hist: tsgm.dataset.DatasetOrTensor, d_syn: tsgm.dataset.DatasetOrTensor, model: T.Callable, test_size: T.Union[float, int], n_epochs: int, metric: T.Optional[T.Callable] = None, random_seed: T.Optional[int] = None) -> float:
X_hist, X_syn = _dataset_or_tensor_to_tensor(d_hist), _dataset_or_tensor_to_tensor(d_syn)
Expand All @@ -225,3 +281,51 @@ def __call__(self, d_hist: tsgm.dataset.DatasetOrTensor, d_syn: tsgm.dataset.Dat
return sklearn.metrics.accuracy_score(y_test, y_pred)
else:
return metric(y_test, y_pred)


def _spectral_entropy_per_feature(X: TensorLike) -> TensorLike:
return antropy.spectral_entropy(X.ravel(), sf=1, method='welch', normalize=True)


def _spectral_entropy_per_sample(X: TensorLike) -> TensorLike:
if len(X.shape) == 1:
X = X[:, None]
return np.apply_along_axis(_spectral_entropy_per_feature, 0, X)


def _spectral_entropy_sum(X: TensorLike) -> TensorLike:
return np.apply_along_axis(_spectral_entropy_per_sample, 1, X)


class EntropyMetric(Metric):
"""
Calculates the spectral entropy of a dataset or tensor.
This metric measures the randomness or disorder in a dataset or tensor
using spectral entropy, which is a measure of the distribution of energy
in the frequency domain.
Args:
d (tsgm.dataset.DatasetOrTensor): The input dataset or tensor.
Returns:
float: The computed spectral entropy.
Example:
>>> metric = EntropyMetric()
>>> dataset = tsgm.dataset.Dataset(...)
>>> result = metric(dataset)
>>> print(result)
"""
def __call__(self, d: tsgm.dataset.DatasetOrTensor) -> float:
"""
Calculate the spectral entropy of the input dataset or tensor.
Args:
d (tsgm.dataset.DatasetOrTensor): The input dataset or tensor.
Returns:
float: The computed spectral entropy.
"""
X = _dataset_or_tensor_to_tensor(d)
return np.sum(_spectral_entropy_sum(X), axis=None)
2 changes: 1 addition & 1 deletion tutorials/augmentations.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@
"import seaborn as sns\n",
"import numpy as np\n",
"import random\n",
"from tensorflow import kera\n",
"from tensorflow import keras\n",
"\n",
"import tsgm\n",
"\n",
Expand Down

0 comments on commit 6faf01b

Please sign in to comment.