불규칙적으로 샘플링 된 시간 데이터를 평활화합니다

StackOverflow https://stackoverflow.com/questions/1023719

문제

첫 번째 열이 특정 참조 지점을지나 초를 초과하고 두 번째 열은 임의의 측정입니다.

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 
    ...

보시다시피, 측정은 불규칙한 시점에서 샘플링됩니다. 각 측정 (Python)에서 최대 100 초 전에 읽기를 평균하여 데이터를 매끄럽게해야합니다. 데이터 테이블이 크기 때문에 반복 기반 방법이 실제로 선호됩니다. 불행히도, 2 시간의 코딩 후에는 효율적이고 우아한 솔루션을 알 수 없습니다.

누구든지 나를 도울 수 있습니까?

편집하다에스

  1. 나는 각 원시 판독에 대해 하나의 부드러운 판독 값을 원하며, 매끄러운 판독 값은 원시 판독의 산술 평균이자 이전 100 초 (델타)의 다른 것입니다. (존, 너 맞다)

  2. 거대한 ~ 1e6-10e6 라인 + 단단한 램으로 작업해야합니다.

  3. 데이터는 대략 무작위로 걷기입니다

  4. 데이터가 정렬됩니다

해결

J Machin과 Yairchu가 제안한 솔루션을 테스트했습니다. 그러나 둘 다 내 데이터 세트에서 J Machin의 버전이 기하 급수적으로 수행되는 반면 Yairchu의 버전은 선형 적입니다. 다음은 Ipython 's에서 측정 한 실행 시간입니다 %시간 (마이크로 초) :

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

도와 주셔서 감사합니다.

도움이 되었습니까?

해결책

새 회원을 추가하고 이전 회원을 빼고있는 합계 결과를 사용하고 있습니다. 그러나 이런 식으로 부동산이 부정확 한 부정확성을 겪을 수 있습니다.

따라서 목록이있는 "deque"를 구현합니다. 그리고 내 deque가 더 작은 크기로 구입할 때마다. 나는 같은 경우에 합계를 다시 계산합니다.

또한 점 X를 포함하여 평균 X까지의 평균을 계산하므로 평균에 대한 샘플 포인트가 하나 이상 있습니다.

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)

다른 팁

출력을 원할 때 정확히 말하지 않았습니다. 나는 당신이 각각의 원시 판독에 대해 하나의 매끄러운 판독 값을 원한다고 가정하고, 매끄러운 판독 값은 원시 판독의 산술 평균이며, 이전 100 (델타) 초에 다른 것입니다.

짧은 답변 : 컬렉션을 사용하십시오. 내가 설정하는 방식은 Deque를 목록처럼 취급하고 최근 판독 값에 더 많은 무게를주는 평균 또는 멋진 기즈 모이드를 쉽게 계산할 수 있습니다.

긴 답변 :

>>> 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보다 작아야하는 경우 (<= Zero는 불법이며, 0 위는 제로 위에 거의 매끄럽게하지 않으며, 산술 평균 플러스 낭비 CPU 시간을 얻을 수 있으며, 하나보다 큰 목적을 반대합니다).

귀하의 데이터는 대략 선형 인 것 같습니다.

데이터 플롯 http://rix0r.nl/~rix0r/share/shot-20090621.144851.gif

어떤 종류의 스무딩을 찾고 있습니까? 이 데이터 세트에 라인의 최소 제곱에 맞습니까? 일종의 저역 통과 필터? 또는 다른 것?

우리는 당신에게 조금 더 잘 조언 할 수 있도록 응용 프로그램을 알려주십시오.

편집 : 예를 들어, 응용 프로그램에 따라 첫 번째 지점과 마지막 지점 사이의 선을 보간하면 목적에 충분할 수 있습니다.

이것은 선형으로 만듭니다.

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

o (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

총 평균보다는 기하 급수적으로 부패하는 평균을 제공하지만, 내가 부르는 것을 원한다고 생각합니다. 다양한 알파를 갖는 지수 이동 평균, 이것은 실제로 단일 폴 저역 통과 필터입니다. 이제 해당 질문에 대한 해결책이 있으며 데이터 포인트 수에 따라 시간이 지남에 따라 실행됩니다. 그것이 당신을 위해 작동하는지 확인하십시오.

이와 같은 것은 어떻습니까, 마지막 시간과의 시차가> 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

간단한 반올림 공식이 필요한 것 같습니다. 숫자를 임의의 간격으로 반올림하려면 :

라운드 (num/interval)*간격

바닥이나 천장으로 둥글게 대체 할 수 있습니다. SQL을 포함한 모든 언어로 작동 할 수 있습니다.

라이센스 : CC-BY-SA ~와 함께 속성
제휴하지 않습니다 StackOverflow
scroll top