(data, k=0.49, alpha=0.05, num_obs_per_period=None,
                  use_decomp=True, use_esd=False, direction="pos", verbose=False)
return {
        'anoms': R_idx,
        'stl': smoothed
    }
Detects anomalies in a time series using S-H-ESD.

    Args:
         data: Time series to perform anomaly detection on.
         k: Maximum number of anomalies that S-H-ESD will detect as a percentage of the data.
         alpha: The level of statistical significance with which to accept or reject anomalies.
         num_obs_per_period: Defines the number of observations in a single period, and used during seasonal decomposition.
         use_decomp: Use seasonal decomposition during anomaly detection.
         use_esd: Uses regular ESD instead of hybrid-ESD. Note hybrid-ESD is more statistically robust.
         one_tail: If TRUE only positive or negative going anomalies are detected depending on if upper_tail is TRUE or FALSE.
         upper_tail: If TRUE and one_tail is also TRUE, detect only positive going (right-tailed) anomalies. If FALSE and one_tail is TRUE, only detect negative (left-tailed) anomalies.
         verbose: Additionally printing for debugging.
    Returns:
       A list containing the anomalies (anoms) and decomposition components (stl).
Detects anomalies in a time series using S-H-ESD.
Detects anomalies in a time series using S - H - ESD .
def _detect_anoms(data, k=0.49, alpha=0.05, num_obs_per_period=None,
                  use_decomp=True, use_esd=False, direction="pos", verbose=False):
    """
    Detects anomalies in a time series using S-H-ESD.

    Args:
         data: Time series to perform anomaly detection on.
         k: Maximum number of anomalies that S-H-ESD will detect as a percentage of the data.
         alpha: The level of statistical significance with which to accept or reject anomalies.
         num_obs_per_period: Defines the number of observations in a single period, and used during seasonal decomposition.
         use_decomp: Use seasonal decomposition during anomaly detection.
         use_esd: Uses regular ESD instead of hybrid-ESD. Note hybrid-ESD is more statistically robust.
         one_tail: If TRUE only positive or negative going anomalies are detected depending on if upper_tail is TRUE or FALSE.
         upper_tail: If TRUE and one_tail is also TRUE, detect only positive going (right-tailed) anomalies. If FALSE and one_tail is TRUE, only detect negative (left-tailed) anomalies.
         verbose: Additionally printing for debugging.
    Returns:
       A list containing the anomalies (anoms) and decomposition components (stl).
    """

    # validation
    assert num_obs_per_period, "must supply period length for time series decomposition"
    assert direction in ['pos', 'neg',
                         'both'], 'direction options: pos | neg | both'
    assert data.size >= num_obs_per_period * \
        2, 'Anomaly detection needs at least 2 periods worth of data'
    assert data[data.isnull(
    )].empty, 'Data contains NA. We suggest replacing NA with interpolated values before detecting anomaly'

    # conversion
    one_tail = True if direction in ['pos', 'neg'] else False
    upper_tail = True if direction in ['pos', 'both'] else False

    # -- Step 1: Decompose data. This returns a univariate remainder which will be used for anomaly detection. Optionally, we might NOT decompose.
    # Note: R use stl, but here we will use MA, the result may be different TODO.. Here need improvement
    #decomposed = sm.tsa.seasonal_decompose(data, freq=num_obs_per_period, two_sided=False)
    #smoothed = data - decomposed.resid.fillna(0)
    #data = data - decomposed.seasonal - data.mean()

    data, smoothed = _get_decomposed_data_tuple(data, num_obs_per_period)

    max_outliers = _get_max_outliers(data, k)

    R_idx = pd.Series()

    n = data.size
    # Compute test statistic until r=max_outliers values have been
    # removed from the sample.
    for i in range(1, max_outliers + 1):
        if verbose:
            logger.info(i, '/', max_outliers, ' completed')

        if not data.mad():
            break

        if not one_tail:
            ares = abs(data - data.median())
        elif upper_tail:
            ares = data - data.median()
        else:
            ares = data.median() - data

        ares = ares / data.mad()

        tmp_anom_index = ares[ares.values == ares.max()].index
        cand = pd.Series(data.loc[tmp_anom_index], index=tmp_anom_index)

        data.drop(tmp_anom_index, inplace=True)

        # Compute critical value.
        p = 1 - alpha / (n - i + 1) if one_tail else (1 -
                                                      alpha / (2 * (n - i + 1)))
        t = sp.stats.t.ppf(p, n - i - 1)
        lam = t * (n - i) / np.sqrt((n - i - 1 + t ** 2) * (n - i + 1))
        if ares.max() > lam:
            R_idx = R_idx.append(cand)

    return {
        'anoms': R_idx,
        'stl': smoothed
    }
def _detect_anoms ( data , k = 0.49 , alpha = 0.05 , num_obs_per_period = None , use_decomp = True , use_esd = False , direction = "pos" , verbose = False ) : # validation assert num_obs_per_period , "must supply period length for time series decomposition" assert direction in [ 'pos' , 'neg' , 'both' ] , 'direction options: pos | neg | both' assert data . size >= num_obs_per_period * 2 , 'Anomaly detection needs at least 2 periods worth of data' assert data [ data . isnull ( ) ] . empty , 'Data contains NA. We suggest replacing NA with interpolated values before detecting anomaly' # conversion one_tail = True if direction in [ 'pos' , 'neg' ] else False upper_tail = True if direction in [ 'pos' , 'both' ] else False # -- Step 1: Decompose data. This returns a univariate remainder which will be used for anomaly detection. Optionally, we might NOT decompose. # Note: R use stl, but here we will use MA, the result may be different TODO.. Here need improvement #decomposed = sm.tsa.seasonal_decompose(data, freq=num_obs_per_period, two_sided=False) #smoothed = data - decomposed.resid.fillna(0) #data = data - decomposed.seasonal - data.mean() data , smoothed = _get_decomposed_data_tuple ( data , num_obs_per_period ) max_outliers = _get_max_outliers ( data , k ) R_idx = pd . Series ( ) n = data . size # Compute test statistic until r=max_outliers values have been # removed from the sample. for i in range ( 1 , max_outliers + 1 ) : if verbose : logger . info ( i , '/' , max_outliers , ' completed' ) if not data . mad ( ) : break if not one_tail : ares = abs ( data - data . median ( ) ) elif upper_tail : ares = data - data . median ( ) else : ares = data . median ( ) - data ares = ares / data . mad ( ) tmp_anom_index = ares [ ares . values == ares . max ( ) ] . index cand = pd . Series ( data . loc [ tmp_anom_index ] , index = tmp_anom_index ) data . drop ( tmp_anom_index , inplace = True ) # Compute critical value. p = 1 - alpha / ( n - i + 1 ) if one_tail else ( 1 - alpha / ( 2 * ( n - i + 1 ) ) ) t = sp . stats . t . ppf ( p , n - i - 1 ) lam = t * ( n - i ) / np . sqrt ( ( n - i - 1 + t ** 2 ) * ( n - i + 1 ) ) if ares . max ( ) > lam : R_idx = R_idx . append ( cand ) return { 'anoms' : R_idx , 'stl' : smoothed }