Source code for pysad.transform.postprocessing.postprocessors

from pysad.core.base_postprocessor import BasePostprocessor
from pysad.statistics.average_meter import AverageMeter
from pysad.statistics.max_meter import MaxMeter
from pysad.statistics.median_meter import MedianMeter
from pysad.statistics.variance_meter import VarianceMeter
import numpy as np


[docs]class AveragePostprocessor(BasePostprocessor): """A postprocessor that convert a score to the average of of all previous scores. """ def __init__(self): self.meter = AverageMeter()
[docs] def fit_partial(self, score): """Fits the postprocessor to the (next) timestep's score. Args: score (float): Input score. Returns: object: self. """ self.meter.update(score) return self
[docs] def transform_partial(self, score=None): """Gets the current average. This method should be used immediately after the fit_partial method with same score. Args: score (float): The input score. Returns: float: Transformed score. """ return self.meter.get()
[docs]class MaxPostprocessor(BasePostprocessor): """A postprocessor that convert a score to the maximum of of all previous scores. """ def __init__(self): self.meter = MaxMeter()
[docs] def fit_partial(self, score): """Fits the postprocessor to the (next) timestep's score. Args: score (float): Input score. Returns: object: self. """ self.meter.update(score) return self
[docs] def transform_partial(self, score=None): """Applies postprocessing to the score. This method should be used immediately after the fit_partial method with same score. Args: score (float): The input score. Returns: float: Transformed score. """ return self.meter.get()
[docs]class MedianPostprocessor(BasePostprocessor): """A postprocessor that convert a score to the median of of all previous scores. """ def __init__(self): self.meter = MedianMeter()
[docs] def fit_partial(self, score): """Fits the postprocessor to the (next) timestep's score. Args: score (float): Input score. Returns: object: self. """ self.meter.update(score) return self
[docs] def transform_partial(self, score=None): """Applies postprocessing to the score. Args: score (float): The input score. Returns: float: Transformed score. """ return self.meter.get()
[docs]class ZScorePostprocessor(BasePostprocessor): """A postprocessor that normalize the score via Z-score normalization. """ def __init__(self): self.variance_meter = VarianceMeter() self.average_meter = AverageMeter()
[docs] def fit_partial(self, score): """Fits the postprocessor to the (next) timestep's score. Args: score (float): Input score. Returns: object: self. """ self.variance_meter.update(score) self.average_meter.update(score) return self
[docs] def transform_partial(self, score): """Applies postprocessing to the score. Args: score (float): The input score. Returns: float: Transformed score. """ zscore = (score - self.average_meter.get()) / \ np.sqrt(self.variance_meter.get()) return zscore