[시계열] Chapter 04

유니·2022년 3월 23일
시계열
0

시계열

목록 보기
1/9

4. 시계열 데이터의 시뮬레이션

  1. 시계열 데이터 시뮬레이션과 다른 종류의 데이터 시뮬레이션을 비교
  2. 실제 코드 기반으로 시뮬레이션의 예
  3. 시계열 시뮬레이션의 동향

시계열 시뮬레이션의 특별한 점

시뮬레이션 : 실제로는 테스트해보기 어려운 초대형 프로젝트나 위험한 테스트 등을 모의실험 해보는 것
→ 시계열에서는 특정 시간에 발생 가능한 일을 예측하기 위해 사용

코드로 보는 시뮬레이션

시뮬레이션 프로그래밍을 할 때에는 시스템에 적용되는 논리적인 규칙을 명심해야 함

예 1) 누군가의 이메일 열람 행동이 기부로 이어지는지, 기부로 이어지지 않는지에 대한 상관관계

##회원 상태
years = ['2014', '2015', '2016', '2017', '2018']
memberStatus = ['bronze', 'silver', 'gold', 'inactive']

##np.random.choice의 p인수를 조절하여
## 각 부류의 발생 확률을 달리하여 데이터 생성
memberYears = np.random.choice(years, 1000, p = [0.1, 0.1, 0.15, 0.30, 0.35])
memberStats = np.random.choice(memberStatus, 1000, p = [0.5, 0.3, 0.1, 0.1])

yearJoined = pd.Dataframe({'yearJoined':memberYears, 'memberStats': memberStats})

모든 회원에게 특정 가입 연도를 무작위로 부여하고 회원의 상태 정보는 부여된 가입연도에 따라 결정된다

주별로 회원의 이메일 열람 시점을 나타내는 테이블 만들기

  • 이메일에 관한 회원들의 행동 패턴
    • 이메일을 열람한 적이 없음
    • 일정한 수준의 이메일 열람 및 참여율
    • 참여 수준의 증가 또는 감소
NUM_EMAILS_SENT_WEEKLY = 3
##서로 다른 패턴을 위한 몇 가지 함수를 정의

##이메일을 한 번도 열람하지 않은 회원
def never_opens(period_rng):
	return []

##매주 같은 양의 이메일을 열람한 회원
def constant_open_rate(period_rng):
	n, p = NUM_EMAILS_SENT_WEEKLY, np.random.uniform(0, 1)
    num_opened = np.random.binomial(n, p, len(period_rng))
    return num_opened
    
##매주 열람한 이메일의 양이 늘어나는 회원    
def increasing_open_rate(period_rng):
	return open_rate_with_factor_change(period_rng, np.random.uniform(1.01, 1.30))

##매주 열람한 이메일의 양이 줄어드는 회원
def decreasing_open_rate(period_rng):
	return open_rate_with_factor_change(period_rng, np.random.uniform(0.5, 0.99))
    
def open_rate_with_factor_change(period_rng, fac):
	if len(period_rng) < 1 :
    	return []
    times = np.random.randint(0, len(period_rng), int(0.1*len(period_rng)))
    num_opened = np.zeros(len(period_rng))
    for prd in range(0, len(period_rng), 2):
    	try:
        	n, p = NUM_EMAILS_SENT_WEEKLY, np.random.uniform(0, 1)
            num_opened[prd:(prd+2)] = np.random.binomial(n, p, 2)
            p = max(min(1, p*fac), 0)
        except :
        	num_opened[prd] = np.random.binomial(n, p, 1)
    for t in range(len(times)):
    	num_opened[times[t]] = 0
    return num_opened

기부 행동을 모델링하는 시스템

##기부행동
def produce_donations(period_rng, member_behavior, num_emails, use_id, member_join_year):
	donation_amounts = np.array([0, 25, 50, 75, 100, 250, 500, 1000, 1500, 2000])
    member_has = np.random.choice(donation_amounts)
    email_fraction = num_emails/(NUM_EMAILS_SENT_WEEKLY*len(period_rng))
    member_gives = member_has * email_fraction
    member_gives_idx = np.where(member_gives >= donation_amounts)[0][-1]
    member_gives_idx = max(min(member_gives_idx, len(donation_amounts) - 2), 1)
    num_times_gave = np.random.poisson(2) * (2018-member_join_year)
    times = np.random.randint(0, len(period_rng), num_times_gave)
    dons = pd.DataFrame({'member':[], 'amount':[], 'timestamp':[]})
    
    for n in range(num_times_gave):
    	donation = donation_amounts[member_gives_idx + np.random.binomial(1, .3)\
        ts = str(period_rng[times[n]].start_time + random_weekly_time_delta())
        dons = dons.append(pd.DataFrame({'member':[use_id], 'amount':[donation], 'timestamp':[ts]})
        
        if dons.shape[0] > 0:
        	dons = dons[dons.amount != 0]
        ##기부액이 0인 경우는 보고하지 않는다
        ##실세게에서 이런 정보는 데이터베이스에 반영되지 않는다
    return dons

특정 주 내의 시간을 무작위로 고르기 위한 유틸리티 함수

def random_weekly_time_delta():
	days_of_week = [d for d in range(7)]
    hours_of_day = [h for h in range(11, 23)]
    minute_of_hour = [m for m in range(50)]
    second_of_minute = [s for s in range(60)]
    return pd.Timedelta(str(np.random.choice(days_of_week)) + " days") + pd.Timedelta(str(np.random.choice(hours_of_day) + " hours") + pd.Timedelta(str(np.random.choice(minute_of_hour)) + " minutes") + pd.Timedelta(str(np.random.choice(second_of minute) + " seconds")

특정 회원의 사건을 시뮬레이션 하는데 필요한 모든 코드 요소
이 때 가입 이후의 시점에만 모든 사건이 발생할 수 있고, 이메일 열람 사건과 기부 사건이 약간의 관계를 갖도록 해줌

behaviors = [never_opens, constant_open_rate, increasing_openrate, decreasing_open_rate]
member_behaviors = np.random.choice(behaviors, 1000, [0.2, 0.5, 0.1, 0.2])

rng = pd.period_range('2015-02-14', '2018-06-01', freq = 'W')
emails = pd.DataFrame({'member':[], 'week':[], 'emailsOpened' : []})
donations = pd.DataFrame({'member':[], 'amount':[], 'timestamp':[]})

for idx in range(yearJoined.shape[0]):
	##회원이 가입한 시기를 무작위로 생성
	join_date = pd.Timestamp(yearJoined.iloc[idx].yearJoined)+pd.Timedelta(str(np.random.randint(0, 365)) + ' days')
    join_date = min(join_date, pd.Timestamp('2018-06-01'))
    
    ##가입 전에는 어떤 행동에 대한 타임스탬프가 없어야 한다
    member_rng = rng[rng > join_date]
    
    if len(member_rng) < 1:
    	continue
    
    info = member_behaviors[idx](member_rng)
    if len(info) == len(member_rng):
    emails = emails.append(pd.DataFrame({'member':[idx]*len(info), 'week':[str(r.start_time) for r in member_rng], 'emailsOpened':info}))
    donations = donations.append(produce_donations(member_rng, member_behaviors[idx], sum(info), idx, join_date.year))

월마다 발생한 기부의 총합

→ 2015년에서 2018년으로 시간이 흐름에 따라 기부 및 이메일 열람의 횟수가 증가하는 것처럼 보임

간단한 시계열 시뮬레이션을 만들기 위해

  • 시계열의 개수를 결정해야 함
  • 시간에 따라 모델링해야 하는 추세를 결정해야 함

예 2) 택시 기사의 교대 시간과 하루 동안 탑승객 빈도에 대한 합성 데이터를 시뮬레이션

파이썬의 제너레이터
: 시간에 따른 종합적인 측정 기준으로 개별 에이전트의 기여 방식을 확인
→ 독립적이거나 의존적인 일련의 액터를 생성하고, 시간을 돌려가며 각 액터가 하는 일을 관찰할 수 있게 해줌

택시의 식별 번호를 생성하는 함수

import numpy as np
def taxi_id_number(num_taxis):
	arr = np.arrange(num_taxis)
    np.random.shuffle(arr)
    for i in range(num_taxis):
    	yield arr[i]

ids = taxi_id_number(10)
print(next(ids)) ## =>7
print(next(ids)) ## =>2
print(next(ids)) ## =>5

위 코드의 각 객체가 각자의 상태를 독립적으로 보관하는 1회용 객체를 생성하는 taxi_id_number() 함수를 제너레이터 함수

밤이나 새벽보다 낮에 더 많은 택시를 할당하고자 특정 시간에 교대근무가 시작될 수 있도록 다른 확률 부여

def shift_info():
	##하루의 서로 다른 세 개의 교대 시간대를 표현
    ##시간 : 0, 8, 16, 빈도: 8, 30, 15
    start_times_and_freqs = [(0, 8), (8, 30), (16, 15)]
    indices = np.arrange(len(start_times_and_freqs))
    
    while True:
    	idx = np.random.choice(indicies, p = [0.25, 0.5, 0.25])
        start = start_times_and_freqs[idx]
        yield (start[0], start[0] + 7.5, start[1]

start_times_and_freqs는 하루 중 서로 다른 세 개의 교대 시간대를 표현한다

개별 택시의 파라미터를 설정하고 시간표를 생성하는 제너레이터

def taxi_process(taxi_id_generator, shift_into_generator):
	taxi_id = next(taxi_id_generator)
    shift_start, shift_end, shift_mean_trips = next(shift_into_generator)
    actual_trips = round(np.random.normal(loc = shift_mean_trips, scale = 2))
    average_trip_time = 6.5 / shift_mean_trips * 60
    
    ##평균 운행 시간을 분 단위로 변환한다
    between_events_time = 1.0 / (shift_mean_trips - 1) * 60
    ##이 도시는 매우 효율적이라 모든 택시가 거의 항상 사용된다
    time = shift_start
    yield TimePoint(taxi_id, 'start_shift', time)
    deltaT = np.random.poisson(between_events_time) / 60
    time += deltaT
    for i in range(actual_trips):
    	yield TimePoint(taxi_id, 'pick up', time)
    	deltaT = np.random.poisson(average_trip_time) / 60
    	time += deltaT
        yield TimePoint(taxi_id, 'drop off', time)
    	deltaT = np.random.poisson(between_events_time) / 60
    	time += deltaT
    deltaT = np.random.poisson(between_events_time) / 60
    time += deltaT
    yield TimePoint(taxi_id, 'end shift ', time)

두 개의 제너레이터를 통해서 각 택시의 ID 번호, 교대 시작 시간, 해당 시간에 대한 평균 운행 횟수를 결정

택시 제너레이터가 생산하는 TimePoint 객체

from dataclasses import dataclass
class TimePoint:
	taxi_id : int
    name : str
    time : float
    
    def __lt__(self, other):
    	return self.time < other.time

택시 개수에 따른 택시 제너레이터들을 생성
각 택시 제너레레이터를 반복적으로 접근해 각각이 반환한 TimePoint가 유효하면 그 TimePoint들을 우선순위 큐에 넣어줌

import queue

class Simulatoe:
    def __init__(self, num_taxis):
    	self.time_points = queue.PriorityQueue()
        taxi_id_generator = taxi_id_number(num_taxis)
        shift_info_generator = shift_info()
        self._taxis = [taxi_process(taxi_id_generator, shift_info_generator) for i in range(num_taxis)]
        self._prepare_run()
        
     def _prepare_run(self):
     	for t in self._taxis:
        	while True:
            	try:
                	e = next(t)
                    self._time_points.put(e)
                except:
                	break
                    
     def run(self):
     	sim_time = 0
        while sim_time < 24:
        	if self._time_points.empty()
            	break
            p = self._time_points.get()
            sim_time = p.time
            print(p)

시뮬레이션 실행 결과 →

물리적인 시뮬레이션

: 시스템을 정의하는 물리적 법칙을 완전히 꿰고 있는 상황

  • 마르코프 연쇄 몬테카를로(MCMC) 방법
  1. 격자의 각 지점에 대한 시작 상태를 무작위로 선택
  2. 각 시간 단계마다 하나의 격자 지점을 선택하고, 그 지점의 방향을 뒤집음
  3. 물리법칙에 기반해 뒤집었을 때 발생하는 에너지의 변화를 계산

예 3) 점진적으로 개별 자기요소의 위치를 맞춰나가는 자성물질의 물리적 과정을 시뮬레이션

블록을 무작위로 초기화해주는 유틸리티 함수

def initRandState(N, N):
	block = np.random.choice([-1, 1], size = (N, M))
    return block

인접 상태에 비례해 중앙 정렬 상태의 에너지를 계산

def costForCenterState(state, i, j, n, m):
	centerS = state[i, j]
    neighbors = [((i+1) % n, j), ((i-1)%n, j), (i, (j+1)%m), (i, (j-1)%m)]
    
    interactionE = [state[x, y] * centerS for (x, y) in neighbors]
    return np.sum(interactionE)

주어진 상태에서 전체 블록의 자화를 결정

def magnetizationForState(state):
	return np.sum(state)

MCMC

def mcmcAdjust(state):
	n = state.shape[0]
    m = state.shape[1]
    x, y = np.random.randint(0, n), np.random.randint(0, m)
    centerS = state[x, y]
    cost = costForCenterState(state, x, y, n, m)
    
    if cost < 0:
    	centerS *= -1
    elif np.random.random() < np.exp(-cost*BETA):
    	centerS *= -1
        
    state[x, y] = centerS
    return state


왼쪽은 무작위로 생성된 초기 상태를 한 번만 조사했을 때의 결과이다. 이 초기 상태를 시뮬레이션 돌린 후 1000번의 시간 단계가 흘렀을 때 최종 저온 상태인 오른쪽 결과를 얻게 된다.

시뮬레이션에 대한 마지막 조언

시뮬레이션으로부터 얻은 양적 측정지표와 결합된 가상적인 예를 통해 데이터에 대한 인지력을 확장해나갈 수 있다

통계적인 시뮬레이션

: 시뮬레이션된 시계열 데이터를 얻는 가장 전통적인 방법

  • 시스템의 근간이 되는 확률적인 역동성을 이미 알고 있을 때, 몇가지 모르는 파라미터를 추정하거나 서로 다른 가정이 파라미터 추정 과정에 주는 영향을 알아볼 때
  • 시뮬레이션의 정확도에 대한 불확실성을 정의하는 분명한 양적 측정지표가 필요한 경우

딥러닝 시뮬레이션

: 시계열 데이터에서 매우 복잡할 수 있는 비선형적 역동성을 잡아낼 수 있음

profile
유니
_
다음 포스트

[시계열] Chapter 05

0개의 댓글