python - 不規則にサンプリングされた時間データの平滑化

Question

最初の列が特定の基準点を過ぎた秒数で、2 番目の列が任意の測定値であるテーブルを考えると、次のようになります。

6   0.738158581
21  0.801697222
39  1.797224596
49  2.77920469
54  2.839757536
79  3.832232283
91  4.676794376
97  5.18244704
100 5.521878863
118 6.316630137
131 6.778507504
147 7.020395216
157 7.331607129
176 7.637492223
202 7.848079136
223 7.989456499
251 8.76853608
278 9.092367123 
    ...

ご覧のとおり、測定値は不規則な時点でサンプリングされています。各測定の 100 秒前に (Python で) 読み取り値を平均化して、データを平滑化する必要があります。データテーブルは非常に大きいため、イテレータベースの方法が実際に推奨されます。残念ながら、2 時間のコーディングを行っても、効率的で洗練されたソリューションを見つけることができません。

誰でも私を助けることができますか？

編集_

生の読み取り値ごとに 1 つの平滑化された読み取り値が必要です。平滑化された読み取り値は、前の 100 (デルタ) 秒の生の読み取り値とその他の値の算術平均になります。（ジョン、あなたは正しい）
巨大な ~ 1e6 ～ 10e6 行 + タイトな RAM で動作する必要がある
データはほぼランダムウォークです
データはソートされています

解像度

J Machin と yairchu によって提案されたソリューションをテストしました。どちらも同じ結果でしたが、私のデータセットでは、J Machin のバージョンは指数関数的に実行されましたが、yairchu のバージョンは線形でした。以下は、IPython の%timeit (マイクロ秒単位) で測定された実行時間です。

data size   J Machin    yairchu
10        90.2        55.6
50          930         258
100         3080        514
500         64700       2660
1000        253000      5390
2000        952000      11500

助けてくれてありがとう。

score 3 · Accepted Answer

新しいメンバーを追加し、古いメンバーを減算する合計結果を使用しています。ただし、このようにして、浮動小数点の不正確さが蓄積する可能性があります。

したがって、リストを使用して「Deque」を実装します。そして、私のDequeがより小さなサイズに再割り当てされるときはいつでも。同時に合計を再計算します。

また、ポイントxを含むポイントxまでの平均を計算しているので、平均するサンプルポイントが少なくとも1つあります。

def getAvgValues(data, avgSampleTime):
  lastTime = 0
  prevValsBuf = []
  prevValsStart = 0
  tot = 0
  for t, v in data:
    avgStart = t - avgSampleTime
    # remove too old values
    while prevValsStart < len(prevValsBuf):
      pt, pv = prevValsBuf[prevValsStart]
      if pt > avgStart:
        break
      tot -= pv
      prevValsStart += 1
    # add new item
    tot += v
    prevValsBuf.append((t, v))
    # yield result
    numItems = len(prevValsBuf) - prevValsStart
    yield (t, tot / numItems)
    # clean prevVals if it's time
    if prevValsStart * 2 > len(prevValsBuf):
      prevValsBuf = prevValsBuf[prevValsStart:]
      prevValsStart = 0
      # recalculate tot for not accumulating float precision error
      tot = sum(v for (t, v) in prevValsBuf)

score 2 · Accepted Answer

出力が必要な時期を正確に言っていません。生の読み取り値ごとに 1 つの平滑化された読み取り値が必要であり、平滑化された読み取り値は、前の 100 (デルタ) 秒の生の読み取り値とその他の値の算術平均であると想定しています。

短い答え: collections.deque を使用してください。「delta」秒を超える読み取り値を保持することはありません。私が設定した方法では、両端キューをリストのように扱うことができ、最近の測定値に重みを与える平均値または派手なギズモイドを簡単に計算できます。

長い答え:

>>> the_data = [tuple(map(float, x.split())) for x in """\
... 6       0.738158581
... 21      0.801697222
[snip]
... 251     8.76853608
... 278     9.092367123""".splitlines()]
>>> import collections
>>> delta = 100.0
>>> q = collections.deque()
>>> for t, v in the_data:
...     while q and q[0][0] <= t - delta:
...         # jettison outdated readings
...         _unused = q.popleft()
...     q.append((t, v))
...     count = len(q)
...     print t, sum(item[1] for item in q) / count, count
...
...
6.0 0.738158581 1
21.0 0.7699279015 2
39.0 1.112360133 3
49.0 1.52907127225 4
54.0 1.791208525 5
79.0 2.13137915133 6
91.0 2.49500989771 7
97.0 2.8309395405 8
100.0 3.12993279856 9
118.0 3.74976297144 9
131.0 4.41385300278 9
147.0 4.99420529389 9
157.0 5.8325615685 8
176.0 6.033109419 9
202.0 7.15545189083 6
223.0 7.4342562845 6
251.0 7.9150342134 5
278.0 8.4246097095 4
>>>

編集

ワンストップショップ: ファンシーなギズモイドをここで入手してください。コードは次のとおりです。

numerator = sum(item[1] * upsilon ** (t - item[0]) for item in q)
denominator = sum(upsilon ** (t - item[0]) for item in q)
gizmoid = numerator / denominator

ここで、upsilon は 1.0 より少し小さくする必要があります (<= 0 は不正です。0 のすぐ上では平滑化はほとんど行われません。1 は算術平均と無駄な CPU 時間を加えたものであり、1 より大きいと目的の逆になります)。

score 0 · Accepted Answer

これはそれを線形にします：

def process_data(datafile):
    previous_n = 0
    previous_t = 0
    for line in datafile:
        t, number = line.strip().split()
        t = int(t)
        number = float(number)
        delta_n = number - previous_n
        delta_t = t - previous_t
        n_per_t = delta_n / delta_t
        for t0 in xrange(delta_t):
            yield previous_t + t0, previous_n + (n_per_t * t0)
        previous_n = n
        previous_t = t

f = open('datafile.dat')

for sample in process_data(f):
    print sample

score 0 · Accepted Answer

あなたのデータはほぼ線形のようです：

データのプロット http://rix0r.nl/~rix0r/share/shot-20090621.144851.gif

どのようなスムージングをお探しですか? このデータセットへの線の最小二乗適合? ローパスフィルターの一種？または、他の何か？

アプリケーションを教えてください。もう少し適切にアドバイスできます。

編集:たとえば、アプリケーションによっては、最初と最後の点の間の線を補間するだけで十分な場合があります。

score 0 · Accepted Answer

合計平均ではなく、指数関数的に減衰する平均が得られますが、実際には単極ローパスフィルターである、さまざまな alpha を使用した指数移動平均と呼ばれるものが必要になると思います。その質問に対する解決策があり、データポイントの数に比例して時間内に実行されます。それがあなたのために働くかどうか見てください。

score 0 · Accepted Answer

入力を複数回反復できる場合の O(1) メモリ - 「左」に 1 つ、「右」に 1 つのイテレータを使用できます。

def getAvgValues(makeIter, avgSampleTime):
  leftIter = makeIter()
  leftT, leftV = leftIter.next()
  tot = 0
  count = 0
  for rightT, rightV in makeIter():
    tot += rightV
    count += 1
    while leftT <= rightT - avgSampleTime:
      tot -= leftV
      count -= 1
      leftT, leftV = leftIter.next()
    yield rightT, tot / count

score -1 · Accepted Answer

このようなものはどうですか、前回との時間差が> 100になるまで値を保存し続け、平均してそのような値を生成します。

def getAvgValues(data):
    lastTime = 0
    prevValues = []
    avgSampleTime=100

    for t, v in data:
        if t - lastTime < avgSampleTime:
            prevValues.append(v)
        else:
            avgV = sum(prevValues)/len(prevValues)
            lastTime = t
            prevValues = [v]
            yield (t,avgV)

for v in getAvgValues(data):
    print v

score -2 · Accepted Answer

簡単な丸め式が必要なようです。任意の数値を任意の間隔に丸めるには:

round(数値/間隔)*間隔

「～に至るまで」または「以来」の影響の代わりに、ラウンドを床または天井に置き換えることができます。SQL を含む任意の言語で動作します。

python - 不規則にサンプリングされた時間データの平滑化

解像度

8 に答える 8

Related

Reference