Коррекция выбросов в Python

Question

У меня есть список чисел, таких как:

[10
20
2
40
50
60
70
80
0
100]

Я хочу заменить цифру, выполнив среднее из двух предыдущих значений, а следующие два значения будут меньше среднего по списку.Как и здесь, 2 будет заменено на среднее значение (10,20,40,50), т. Е. 30. Похоже, если оно достигнет 0 при втором последнем появлении, теперь у него не будет следующих двух вхождений, в этом случае оно должно принимать среднее из предыдущих трехи следующий, чтобы поддерживать счет 4 значений, то есть среднее значение (60, 70, 100, 100), т.е. 77,5.Кто-нибудь может направить меня с наилучшей логикой.Окончательный результат:

[10
20
30
40
50
60
70
80
77.6
100]

Answer 1

Попробуйте этот код:

def correct_outliers(s, threshold, n_prev=3, n_next=1):
    local_s = s.copy()
    updated_index = local_s.to_frame().apply(lambda x: correct_outlier(x, local_s, threshold), axis=1)
    return local_s

def correct_outlier(x, s, threshold, n_prev=3, n_next=1):
    if x.isna()[0] or x[0] < threshold:
        lower_index, upper_index = get_fixed_index(x.name, n_prev, n_next)
        s[x.name] = s.loc[lower_index:x.name-1].append(s.loc[x.name+1:upper_index]).mean(skipna=True)
        return True
    return False

Пояснения и соображения

1. Я использовал pandas.Series для работы с данными
2. Функция correct_outliers получает каквведите ряд панд, значение threshold и границы окна (n_prev и n_nex)
3. Функция correct_outlier вызывается correct_outliers и применяется элемент за элементом кряд во вводе с использованием series.apply
4. Функция get_fixed_bounds задает индекс текущего элемента i и значения границ окна с учетом ваших запросов, сформулированных вами в приложении
5. Суть функции correct_outlier заключается в следующем:
   Если текущее значение серии меньше значения threshold, то текущее значение серии заменяется на average рассчитано в интервале, определяемом фиксированными границами (исключая нулевые значения и текущее значение)

Пример

Приведены следующие ряды данных:

s = pd.Series([10, 20, 2, 40, 50, 60, 70, 80, 0, 100], dtypes='float')
0     10.0
1     20.0
2      2.0
3     40.0
4     50.0
5     60.0
6     70.0
7     80.0
8      0.0
9    100.0

Определить пороговое значение иwindows:

threshold = 5   # s.mean(skipna=True) in your example
n_prev = 3      # 3 element before the current
n_next = 1      # 1 element after the current

Теперь вызовите correct_outliers:

fixed_series = correct_outliers(s, n_prev, n_next, threshold), axis=1)

И дает:

0   10.0
1   20.0
2   30.0
3   40.0
4   50.0
5   60.0
6   70.0
7   80.0
8   77.5
9  100.0

Пошаговое выполнение:

Учитываяте же входные данные, что и в предыдущем примере, я покажу вам пошаговое выполнение для x = 2, как вы меня просили.

После вызова correct_outliers ряд повторяется с помощью функции apply, и к каждому элементу применяется функция correct_outlier, в случае, если рассматриваемый элемент равен x = 2, шагвыполнение шага будет следующим:

--- correct_outlier(), input: x: 2.0 threshold:5.0  n_prev: 3 n_next: 1
    step:
        if_condition: x is nan or x<threshold? True
            --- get_fixed_index(), input: current_index: 2 n_prev: 3 n_next: 1 
                step: if_condition: current_index-n_prev>=0? False 
                output: lower_index: 0 upper_index: 4 
            slice of series: [10. 20. 40. 50.] mean: 30.0
            @@@@ replace the value 2.0 with 30

---

Extra

Функции correct_outlier и correct_outliers не особенно эффективны, в основном по этим причинам:

1. Вся серия повторяется на чистом Python, и это никогда не будет хорошей идеей .Если возможно, вы всегда должны использовать библиотечные функции для анализа данных (такие как Pandas, Numpy, ...), которые реализованы на C / C ++ и, следовательно, на несколько порядков более эффективны, чем реализация на чистом Python.
2. Мы можем обойтись без функции добавления между двумя рядами, используемой в correct_outliers, мы можем решить проблему, просто выполнив взвешенное среднее (что, очевидно, намного быстрее)

Первая точка - это реальнаяУзкое место.

Как решить?

Ниже я предлагаю два оптимизированных решения для функций, которые мы видели:

def correct_outliers_opt(s, threshold, n_prev=3, n_next=1):
    tmp_s = s.copy()
    tmp_s[tmp_s < threshold].to_frame().apply(lambda x: correct_outlier4(x, tmp_s, threshold), axis=1)
    return tmp_s

def correct_outlier_opt(x, s, threshold, n_prev=3, n_next=1):
    i = x.name
    lower_index, upper_index = get_fixed_index(x.name, n_prev, n_next)
    n = upper_index - lower_index
    mean = s.loc[lower_index:i-1].mean(skipna=True)*(i-lower_index)/n + ss.loc[i+1:upper_index].mean(skipna=True)*(upper_index-i)/n
    s[i] = mean
    return mean

Ключевым моментом являетсяв пределах correct_outliers_opt и имеет следующий вид:

tmp_s[tmp_s < threshold]

Таким образом, я фильтрую ряд (используя преимущества функций Pandas вместо чистого питона), прежде чем итерировать его: таким образом, только значениякоторые удовлетворяют условию, будут повторяться.В нашем примере мы выполняем итерацию в python только с двумя значениями, которые нам нужно заменить вместо итерации, чем весь ряд.

Второе, что было оптимизировано (что оказывает гораздо меньшее влияние на производительность, чем предыдущиеточка) - это вычисление среднего значения в функции correct_outlier_opt: вместо добавления между рядами теперь среднее значение вычисляется отдельно для двух рядов, а затем делается взвешенное среднее для получения одного результата.

Сравнение времени выполнения

Оба получают одинаковые входные данные и возвращают одинаковые выходные данные, но с отчетливо разными временами выполнения.

Время выполнения было вычислено для следующего экземпляра теста:

threshold = 5
n_prev, n_next = 3, 1
N = 1000 

ss = pd.Series([10, 20, 2, 40, 50, 60, 70, 80, 0, 100] * N, dtype='float') # total len N * 10

correct_outliers:

%%timeit
correct_outliers(ss, threshold)
# Execution time: 2.95 s ± 417 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

оптимизированная версия:

%%timeit
correct_outliers_opt(ss, threshold)
#Execution time: 545 ms ± 16.3 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

Как видите, оптимизированная версия примерно в 6 раз быстрее.

Answer 2

Вы можете попробовать что-то вроде этого:

import numpy as np

def moving_average(vals):
    moving_vals = []
    size = len(vals)
    for (idx, val) in enumerate(vals):
        # determine the average for a given index
        if idx == 0 or idx == (size - 1):
            moving_vals.append(val)
            continue
        elif idx == 1:
            temp_arr = vals[2:4]
            temp_arr.append(vals[0])
            average = np.mean(temp_arr)
        elif idx == (size - 2):
            temp_arr = vals[size - 5:size - 3]
            temp_arr.append(vals[size - 1])
            average = np.mean(temp_arr)
        else:
            temp_arr = vals[idx - 2:idx]
            temp_arr1 = vals[idx + 1:idx + 3]
            average = np.mean(temp_arr + temp_arr1)

        # add to the final array based on the average
        if val < average:
            moving_vals.append(average)
        else:
            moving_vals.append(val)

    return moving_vals

Это можно исправить, но я думаю, что смысл ясен.Для решения такой проблемы вам необходимо определить свои угловые случаи и учесть их, а также обработать основной случай.В вашем примере угловой случай - это когда idx = 1 или idx = len(list) - 2.

Вы также можете по-разному обрабатывать начало и конец массива.В моем фрагменте кода эти значения всегда будут возвращены.Кроме того, если вы не хотите использовать numpy, вы можете заменить np.mean средним значением из математического модуля.