Spitzensignalerfassung in Echtzeit-Zeitreihendaten

Question

---

Aktualisierung: Der bisher beste Algorithmus bisher ist dieser.

---

Diese Frage erkundet robuste Algorithmen zur Erkennung plötzlicher Peaks in echtzeitfähigen Zeitreihendaten.

Betrachten Sie das folgende Beispiel für Daten:

_{Beispiel dieser Daten ist im Matlab-Format (aber es handelt sich bei dieser Frage nicht um die Sprache, sondern um den Algorithmus):}

p = [1 1 1.1 1 0.9 1 1 1.1 1 0.9 1 1.1 1 1 0.9 1 1 1.1 1 1 1 1 1.1 0.9 1 1.1 1 1 0.9, ...
     1 1.1 1 1 1.1 1 0.8 0.9 1 1.2 0.9 1 1 1.1 1.2 1 1.5 1 3 2 5 3 2 1 1 1 0.9 1 1, ... 
     3 2.6 4 3 3.2 2 1 1 0.8 4 4 2 2.5 1 1 1];

Es ist deutlich zu erkennen, dass es drei große Peaks und einige kleine Peaks gibt. Diese Datensatz ist ein spezifisches Beispiel aus der Klasse von Zeitreihendatensätzen, um die es in der Frage geht. Diese Datenklasse hat zwei allgemeine Eigenschaften:

1. Es gibt grundlegendes Rauschen mit einem allgemeinen Mittelwert
2. Es gibt große 'Peaks' oder 'höhere Datenpunkte', die signifikant vom Rauschen abweichen.

Angenommen wird auch folgendes:

• Die Breite der Peaks kann nicht im Voraus bestimmt werden
• Die Höhe der Peaks weicht signifikant von den anderen Werten ab
• Der Algorithmus aktualisiert sich in Echtzeit (also mit jedem neuen Datenpunkt)

In einem solchen Szenario muss ein Grenzwert erstellt werden, der Signale auslöst. Allerdings kann der Grenzwert nicht statisch sein und muss in Echtzeit mithilfe eines Algorithmus bestimmt werden.

---

Meine Frage: Welcher Algorithmus ist gut geeignet, um solche Schwellenwerte in Echtzeit zu berechnen? Gibt es spezifische Algorithmen für solche Situationen? Was sind die bekanntesten Algorithmen?

---

_{Robuste Algorithmen oder nützliche Einsichten werden sehr geschätzt. (kann in jeder Sprache antworten: es geht um den Algorithmus)}

Answer 1

Hier ist mein Versuch, eine Ruby-Lösung für den "Smoothed z-score Algo" aus der akzeptierten Antwort zu erstellen:

module ThresholdingAlgoMixin
  def mean(array)
    array.reduce(&:+) / array.size.to_f
  end

  def stddev(array)
    array_mean = mean(array)
    Math.sqrt(array.reduce(0.0) { |a, b| a.to_f + ((b.to_f - array_mean) ** 2) } / array.size.to_f)
  end

  def thresholding_algo(lag: 5, threshold: 3.5, influence: 0.5)
    return nil if size < lag * 2
    Array.new(size, 0).tap do |signals|
      filtered = Array.new(self)

      initial_slice = take(lag)
      avg_filter = Array.new(lag - 1, 0.0) + [mean(initial_slice)]
      std_filter = Array.new(lag - 1, 0.0) + [stddev(initial_slice)]
      (lag..size-1).each do |idx|
        prev = idx - 1
        if (fetch(idx) - avg_filter[prev]).abs > threshold * std_filter[prev]
          signals[idx] = fetch(idx) > avg_filter[prev] ? 1 : -1
          filtered[idx] = (influence * fetch(idx)) + ((1-influence) * filtered[prev])
        end

        filtered_slice = filtered[idx-lag..prev]
        avg_filter[idx] = mean(filtered_slice)
        std_filter[idx] = stddev(filtered_slice)
      end
    end
  end
end

Und Beispiel-Nutzung:

test_data = [
  1, 1, 1.1, 1, 0.9, 1, 1, 1.1, 1, 0.9, 1, 1.1, 1, 1, 0.9, 1,
  1, 1.1, 1, 1, 1, 1, 1.1, 0.9, 1, 1.1, 1, 1, 0.9, 1, 1.1, 1,
  1, 1.1, 1, 0.8, 0.9, 1, 1.2, 0.9, 1, 1, 1.1, 1.2, 1, 1.5,
  1, 3, 2, 5, 3, 2, 1, 1, 1, 0.9, 1, 1, 3, 2.6, 4, 3, 3.2, 2,
  1, 1, 0.8, 4, 4, 2, 2.5, 1, 1, 1
].extend(ThresholdingAlgoMixin)

puts test_data.thresholding_algo.inspect

# Output: [
#   0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
#   0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0,
#   0, 0, 0, 0, 0, 0, 1, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1,
#   1, 1, 0, 0, 0, -1, -1, 0, 0, 0, 0, 0, 0, 0, 0
# ]

Answer 2

Ich habe mir erlaubt, eine JavaScript-Version davon zu erstellen. Möglicherweise ist sie hilfreich. Der JavaScript-Code sollte eine direkte Transkription des oben gegebenen Pseudocodes sein. Verfügbar als npm-Paket und Github-Repository:

• https://github.com/crux/smoothed-z-score
• @joe_six/smoothed-z-score-peak-signal-detection

Javascript Übersetzung:

// JavaScript-Portierung von: https://stackoverflow.com/questions/22583391/peak-signal-detection-in-realtime-timeseries-data/48895639#48895639

function sum(a) {
    return a.reduce((acc, val) => acc + val)
}

function mean(a) {
    return sum(a) / a.length
}

function stddev(arr) {
    const arr_mean = mean(arr)
    const r = function(acc, val) {
        return acc + ((val - arr_mean) * (val - arr_mean))
    }
    return Math.sqrt(arr.reduce(r, 0.0) / arr.length)
}

function smoothed_z_score(y, params) {
    var p = params || {}
    // init cooefficients
    const lag = p.lag || 5
    const threshold = p.threshold || 3.5
    const influence = p.influece || 0.5

    if (y === undefined || y.length < lag + 2) {
        throw ` ## y data array to short(${y.length}) for given lag of ${lag}`
    }
    //console.log(`lag, threshold, influence: ${lag}, ${threshold}, ${influence}`)

    // init variables
    var signals = Array(y.length).fill(0)
    var filteredY = y.slice(0)
    const lead_in = y.slice(0, lag)
    //console.log("1: " + lead_in.toString())

    var avgFilter = []
    avgFilter[lag - 1] = mean(lead_in)
    var stdFilter = []
    stdFilter[lag - 1] = stddev(lead_in)
    //console.log("2: " + stdFilter.toString())

    for (var i = lag; i < y.length; i++) {
        //console.log(`${y[i]}, ${avgFilter[i-1]}, ${threshold}, ${stdFilter[i-1]}`)
        if (Math.abs(y[i] - avgFilter[i - 1]) > (threshold * stdFilter[i - 1])) {
            if (y[i] > avgFilter[i - 1]) {
                signals[i] = +1 // positive signal
            } else {
                signals[i] = -1 // negative signal
            }
            // make influence lower
            filteredY[i] = influence * y[i] + (1 - influence) * filteredY[i - 1]
        } else {
            signals[i] = 0 // no signal
            filteredY[i] = y[i]
        }

        // adjust the filters
        const y_lag = filteredY.slice(i - lag, i)
        avgFilter[i] = mean(y_lag)
        stdFilter[i] = stddev(y_lag)
    }

    return signals
}

module.exports = smoothed_z_score

Answer 3

Hier ist eine geänderte Fortran-Version des Z-Score-Algorithmus. Es wurde speziell für die Detektion von Peaks (Resonanzen) in Übertragungsfunktionen im Frequenzraum modifiziert (Jede Änderung hat einen kleinen Kommentar im Code).

Die erste Änderung warnt den Benutzer, wenn in der Nähe des unteren Randes des Eingangsvektors eine Resonanz vorhanden ist, die durch eine Standardabweichung höher als ein bestimmter Schwellenwert (in diesem Fall 10%) angezeigt wird. Dies bedeutet einfach, dass das Signal nicht flach genug für eine ordnungsgemäße Initialisierung der Filter ist.

Die zweite Änderung besteht darin, dass nur der höchste Wert eines Peaks zu den gefundenen Peaks hinzugefügt wird. Dies wird erreicht, indem jeder gefundene Peak-Wert mit der Größe seiner (Haupt-) Vorgänger und seiner (Haupt-) Nachfolger verglichen wird.

Die dritte Änderung besteht darin, dass Resonanzpeaks in der Regel eine gewisse Form von Symmetrie um die Resonanzfrequenz zeigen. Daher ist es sinnvoll, den Mittelwert und die Standardabweichung symmetrisch um den aktuellen Datenpunkt herum zu berechnen (anstatt nur für die Vorgänger). Dies führt zu einem besseren Peak-Erkennungsverhalten.

Die Änderungen haben zur Folge, dass dem Algorithmus das gesamte Signal im Voraus bekannt sein muss, was im Fall der Resonanzerkennung üblich ist (etwas wie das Matlab-Beispiel von Jean-Paul, bei dem die Datenpunkte dynamisch generiert werden, funktioniert nicht).

function PeakDetect(y,lag,threshold, influence)
    implicit none
    ! Deklarationsteil
    real, dimension(:), intent(in) :: y
    integer, dimension(size(y)) :: PeakDetect
    real, dimension(size(y)) :: filteredY, avgFilter, stdFilter
    integer :: lag, ii
    real :: threshold, influence

    ! Ausführungsteil
    PeakDetect = 0
    filteredY = 0.0
    filteredY(1:lag+1) = y(1:lag+1)
    avgFilter = 0.0
    avgFilter(lag+1) = mean(y(1:2*lag+1))
    stdFilter = 0.0
    stdFilter(lag+1) = std(y(1:2*lag+1))

    if (stdFilter(lag+1)/avgFilter(lag+1)>0.1) then ! Wenn der Variationskoeffizient 10% überschreitet, ist das Signal am Anfang zu ungleichmäßig, möglicherweise aufgrund eines Peaks.
        write(unit=*,fmt=1001)
1001        format(1X,'Warnung: Die Peak-Erkennung könnte fehlgeschlagen sein, da möglicherweise ein Peak am Rand des Frequenzbereichs vorhanden ist.',/)
    end if
    do ii = lag+2, size(y)
        if (abs(y(ii) - avgFilter(ii-1)) > threshold * stdFilter(ii-1)) then
            ! Finde nur den größten herausragenden Wert, der nur derjenige ist, der größer als sein Vorgänger und sein Nachfolger ist
            if (y(ii) > avgFilter(ii-1) .AND. y(ii) > y(ii-1) .AND. y(ii) > y(ii+1)) then
                PeakDetect(ii) = 1
            end if
            filteredY(ii) = influence * y(ii) + (1 - influence) * filteredY(ii-1)
        else
            filteredY(ii) = y(ii)
        end if
        ! Geändert im Vergleich zum Originalcode. Mittelwert und Standardabweichung werden symmetrisch um den aktuellen Punkt berechnet
        avgFilter(ii) = mean(filteredY(ii-lag:ii+lag))
        stdFilter(ii) = std(filteredY(ii-lag:ii+lag))
    end do
end function PeakDetect

real function mean(y)
    !> @brief Berechnet den Mittelwert des Vektors y
    implicit none
    ! Deklarationsteil
    real, dimension(:), intent(in) :: y
    integer :: N
    ! Ausführungsteil
    N = max(1,size(y))
    mean = sum(y)/N
end function mean

real function std(y)
    !> @brief Berechnet die Standardabweichung des Vektors y
    implicit none
    ! Deklarationsteil
    real, dimension(:), intent(in) :: y
    integer :: N
    ! Ausführungsteil
    N = max(1,size(y))
    std = sqrt((N*dot_product(y,y) - sum(y)**2) / (N*(N-1)))
end function std

Für meine Anwendung funktioniert der Algorithmus hervorragend!

Answer 4

Eine iterative Version in Python/Numpy für die Antwort https://stackoverflow.com/a/22640362/6029703 ist hier. Dieser Code ist schneller als das Berechnen von Durchschnitt und Standardabweichung für jede Verzögerung bei großen Daten (100.000+).

def peak_detection_smoothed_zscore_v2(x, lag, threshold, influence):
    '''
    Iterativer geglätteter Z-Score-Algorithmus
    Implementierung des Algorithmus von https://stackoverflow.com/a/22640362/6029703
    '''
    import numpy as np
    labels = np.zeros(len(x))
    filtered_y = np.array(x)
    avg_filter = np.zeros(len(x))
    std_filter = np.zeros(len(x))
    var_filter = np.zeros(len(x))

    avg_filter[lag - 1] = np.mean(x[0:lag])
    std_filter[lag - 1] = np.std(x[0:lag])
    var_filter[lag - 1] = np.var(x[0:lag])
    for i in range(lag, len(x)):
        if abs(x[i] - avg_filter[i - 1]) > threshold * std_filter[i - 1]:
            if x[i] > avg_filter[i - 1]:
                labels[i] = 1
            else:
                labels[i] = -1
            filtered_y[i] = influence * x[i] + (1 - influence) * filtered_y[i - 1]
        else:
            labels[i] = 0
            filtered_y[i] = x[i]
        # Aktualisierung von Durchschnitt, Varianz, Standardabweichung
        avg_filter[i] = avg_filter[i - 1] + 1. / lag * (filtered_y[i] - filtered_y[i - lag])
        var_filter[i] = var_filter[i - 1] + 1. / lag * ((filtered_y[i] - avg_filter[i - 1]) ** 2 - (
            filtered_y[i - lag] - avg_filter[i - 1]) ** 2 - (filtered_y[i] - filtered_y[i - lag]) ** 2 / lag)
        std_filter[i] = np.sqrt(var_filter[i])

    return dict(signals=labels,
                avgFilter=avg_filter,
                stdFilter=std_filter)

Answer 5

Und hier kommt die PHP-Implementierung des ZSCORE-Algorithmus:

 $threshold * $stdFilter[$lag - 1]) {
            if ($data[$i] > $avgFilter[$i-1]) {
                $signals[$i] = 1;
            }
            else {
                $signals[$i] = -1;
            }
            $filteredY[$i] = $influence * $data[$i] + (1 - $influence) * $filteredY[$i-1];
        } 
        else {
            $signals[$i] = 0;
            $filteredY[$i] = $data[$i];
        }

        $avgFilter[$i] = mean($filteredY, $i - $lag, $lag);
        $stdFilter[$i] = stddev($filteredY, $i - $lag, $lag);
    }
    return $signals;
}

$sig = zscore($y, count($y));

print_r($y); echo "";
print_r($sig); echo "";

for ($i = 0; $i < count($y); $i++) echo $i. " " . $y[$i]. " ". $sig[$i]."";