Обнаружение периодических повторений в потоке данных - PullRequest
2 голосов
/ 08 апреля 2010

Допустим, у меня есть массив нулей:

a = numpy.zeros(1000)

Затем я представляю некоторые повторяющиеся «события»:

a[range(0, 1000, 30)] = 1

Вопрос в том, как мне обнаружить «сигнал» там? Потому что это далеко от идеального сигнала, если я делаю «обычный» БПФ, у меня нет четкого указания, где находится мой «истинный» сигнал:

f = abs(numpy.fft.rfft(a))

Есть ли способ обнаружить эти повторения с некоторой степенью достоверности? Особенно, если у меня есть несколько таких, например, здесь:

a[range(0, 1000, 30)] = 1
a[range(0, 1000, 110)] = 1
a[range(0, 1000, 48)] = 1

Я хотел бы получить три «всплеска» на результирующих данных ...

Ответы [ 2 ]

3 голосов
/ 08 апреля 2010

Рассматривали ли вы использование автокорреляции ?

0 голосов
/ 08 апреля 2010

В качестве аналитического метода acf / pacf / ccf используются для определения периодичности в зависящих от времени сигналах, следовательно, коррелограмма, графическое отображение acf или pacf, отображает самоподобие в сигнале как функцию различных лагов.(Так, например, если вы видите значения на пике оси Y с задержкой в ​​12, и если ваша дата указана в месяцах, это свидетельствует о годовой периодичности.)

Для расчета и построения графика «сходства» по сравнению с лагом, если вы не хотите снимать свои собственные, я не знаю о нативной опции Numpy / Scipy;Я также не смог найти его в scikit «временного ряда» (одна из библиотек в Scipy «Scikits», доменные модули, не входящие в стандартный дистрибутив Scipy), но стоит проверить еще раз.Другой вариант - установить привязки Python к R (RPy2, доступный на SourceForge), который позволит вам получить доступ к соответствующим функциям R, включая « acf », который будет вычислять и отображать коррелограмму, просто передавая вашуВременные ряды и вызов функции.

С другой стороны, если вы хотите идентифицировать в сигнале непрерывные (непрерывные) потоки заданного типа, тогда « кодирование длины серии »вероятно, что вы хотите:

import numpy as NP
signal = NP.array([3,3,3,3,3,3,3,3,3,0,0,0,0,0,0,0,0,0,0,7,7,7,7,7,4,4,1,1,1,1,1,1,1])
px, = NP.where(NP.ediff1d(signal) != 0)
px = NP.r_[(0, px+1, [len(signal)])]
# collect the run-lengths for each unique item in the signal
rx = [ (m, n, signal[m]) for (m, n) in zip(px[:-1], px[1:]) ]

# returns [(0, 9, 3), (9, 19, 0), (19, 24, 7), (24, 26, 4), (26, 33, 1)]
# so w/r/t first 3-tuple: '3' occurs continuously in the half-open interval 0 and 9, and so forth
...