파이썬 금융 시계열 처리 최적화하기

당신의 데이터 처리를 가속시킬 몇가지 Tips & Tricks

JooHyun Jo

Published in

Qraft Technologies

19 min readJul 22, 2020

본 글은 Qraft Technologies의 Optimizing Python code on Financial Data Science를 번역한 글입니다.

금융 데이터를 파이썬으로 처리하는 것은 상대적으로 쉽고, 흔한 일입니다. 다른 개발 언어들에 비하면 말이죠. 이는 다음과 같은 특성에 기인하는데,

파이썬은 인터프리터 언어이기 때문에 런타임에서 간편하게 데이터를 분석/수정할 수 있습니다.
데이터, 특히 금융데이터를 처리하기 위한 간편한 라이브러리들이 존재합니다. 예를 들면 Pandas와 Numpy 등이 있습니다.

하지만 파이썬을 통한 금융데이터 처리는 치명적인 단점 역시 가지고 있는데요,

순수 파이썬의 상대적으로 저조한 성능
최적화되지 않은 라이브러리들의 함정
불완전한 쓰레딩 지원(GIL)
느슨한 타입 시스템과 그로 인한 런타임 에러에 대한 노출

데이터 처리 과정에서 위와 같은 단점은 최소화하고 장점은 최대한 활용하기 위해, 몇몇 테크닉을 고려할 필요가 있습니다.

본 글에서는 저희 팀에서도 사용하고 있는, 아주 간단하고 직관적인, 하지만 당신의 코드를 안정적이고 빠르게 만들어줄 강력한 팁들을 소개합니다.

라이브러리의 함정

파이썬이라는 언어의 컨셉에 맞게, 파이썬 라이브러리들 역시 성능보다는 사용의 편리섬에 중점을 맞춘 케이스가 많습니다. 이는 뉴비들에게는 진입장벽을 낮추는 좋은 특성이지만, 좀 더 빠른 실험과 프로덕션을 원하는 전문가들에게는 해결해야할 문제입니다.

Pandas는 파이썬에서 정형화된 데이터를 처리하는 가장 유명하고, 손쉬운 툴입니다. 하지만 자주 사용되는 몇몇 기능들은 코드에서 병목이 되곤 합니다.
그 중 하나는 바로 datetime object 입니다.

df = pd.read_csv(‘./data/2020-Jan.csv’, index_col=’event_time’,parse_dates=[‘event_time’])

위의 코드는 시간을 나타내는 컬럼을 가지고 있는 데이터를 불러올 때 가장 자연스럽게 사용하는 코드입니다. 이는 데이터 행렬의 행 수가 늘어남에 따라 엄청난 시간을 소요하게 됩니다.

# read / parse
CPU times: user 7min 39s, sys: 1.12 s, total: 7min 40s
Wall time: 7min 40s

parse_dates를 사용할 때, Pandas는 datetime 포맷을 각 string마다 파싱합니다(예를 들어, 첫 row는 “2020–01–01” 같은 포맷이고, 다음 row는 “Jan 2, 2020” 일 수 있다고 생각합니다 — 물론 그런일은 거의 없죠).
모든 날짜들이 같은 형식으로 기록되어 있다면, 그 형식을 pd.read_csv()에 제공함으로써 훨씬 빠르게 데이터를 읽어 올 수 있습니다.

head = pd.read_csv("./data/2020-Jan.csv", index_col='event_time', nrows=1)
print(head.index[0])"""
2020-01-01 00:00:00 UTC
"""

데이터의 첫 행을 확인해보니, 이 데이터가 iso8601 날짜 포맷을 사용하고 있음을 알 수 있습니다. 마지막 4글자 “ UTC”만 제외하고 말이죠. 이를 통해 아래와 같은 효율적인 파서를 만들어 봅시다.

import datetime as dtdef my_parser(date_str):
    return ciso8601.parse_datetime(date_str[:-4])df = pd.read_csv("./data/2020-Jan.csv", index_col='event_time', parse_dates=['event_time'], date_parser=my_parser)

아래와 같은 parse를 pd.read_csv의 date_parser 에 인자로 제공하면, 데이터를 읽어오는 데 걸리는 시간이 매우 절약됩니다.

# read / parse
CPU times: user 5.58 s, sys: 880 ms, total: 6.46 s
Wall time: 6.46 s

만약 시간 형식이 iso8601이 아닌 임의의 것이라면, datetime.datetime.strptime 을 참조하여 어떤 파서도 만들 수 있습니다.

데이터를 읽는데에만 7분이라는 시간은 지겹긴 하지만, 연구나 실험 내에서 읽기 작업은 주로 단 한 번만 일어나기 때문에 감수할만 할 수도 있습니다. 하지만 날짜를 이용해서 로드된 데이터에서 일부를 indexing 하거나 부분을 slicing하는 등의 연산은 더욱 빈번하게 일어납니다.

가장 일반적인 slicing 코드는 다음과 같습니다.

%%time
# from 2020-01-03:00:00:00 to 2020-01-09:00:00:00(open)
sliced = df.loc['2020-01-03': '2020-01-08']

이 코드의 가장 큰 문제점은, pandas.DatetimeIndex를 통해 ‘시간’으로 DataFrame을 slicing하고 있다는 점입니다.

# slice
CPU times: user 383 ms, sys: 20 ms, total: 403 ms
Wall time: 402 ms

만약 당신이 데이터에서 날짜를 필요로 하는 이유가 단순하게 데이터 행들의 순서를 보존하기 위해서이고, 다른 복잡한 시간 연산(ex. ‘다음 월말 데이터 찾기’, ‘1시간 후 날짜 찾기’ 등)이 필요한 것이 아니라면 굳이 DatetimeIndex를 사용할 필요가 없습니다. 단순 정렬된 string 만으로도 indexing/slicing이 가능하기 때문입니다.

%%time
df = pd.read_csv('./data/2020-Jan.csv')

np.searchsorted를 통해 위에서와 같이 2020–01–03 과 2020–01–09 사이의 데이터를 찾아봅시다.

(주의할 점은, df.loc 과 datetime의 조합을 사용하는 경우, slice의 뒷부분은 2020–01–08의 00시00분의 열린 괄호가 아닌, 2020–01–08 23시 59분 59초 99… 의 닫힌 괄호입니다. 따라서 2020–01–09 직전까지의 데이터들이 수집됩니다.)

%%time
idxs = np.searchsorted(df['event_time'], '2020-01-03'), np.searchsorted(df['event_time'], '2020-01-09')
df.iloc[idxs[0]: idxs[1]]

성능 결과는 다음과 같습니다.

# read
CPU times: user 3.24 s, sys: 184 ms, total: 3.42 s
Wall time: 3.43 s# slice
CPU times: user 413 µs, sys: 0 ns, total: 413 µs
Wall time: 397 µs

You don’t have to use pd.DatetimeIndex every time!

즉, 필요한 기능에 맞추어 최소한의 기능만 살려가는 것 만으로 꽤나 많은 성능을 증진시킬 수 있습니다.

20200723 추가
DatetimeIndex 에서 loc을 통해 slicing하는 연산은 ‘첫 시행’에서 급격한 성능 저하를 야기합니다. 2번째 이후 시행부터는 꽤 괜찮은 성능을 보이지만, searchsorted를 통해 number index 를 직접 찾아서 slicing하는 것보다는 낮은 성능을 보입니다.

Numba와 Cython 을 사용해서 반복문 성능을 향상시키기(fold/scan/rolling_apply 등).

금융 시계열을 사용하면서 가장 빈번하게 사용하는 연산 중 하나는 반복(iterative 혹은 sequential-연속) 연산입니다. 과거에서 현재로 행을 순회하면서 진행하는 작업을 의미합니다.

EMA(exponential moving average — 지수이동평균)은 반복 연산 중 하나로서, 한 시점의 데이터의 노이즈를 줄여주고 추세를 더 명확하게 보여주는 조작 중 하나입니다. 이 연산은 과거에서부터 현재로서 값을 수정하면서 계산되기 때문에, 병렬처리를 통해 속도를 향상시킬 수 없습니다.

(본 글에서는 설명의 편의를 위해 최적화하지않은 Pandas 연산으로서 EMA를 개선하는 과정을 설명하겠습니다. EMA는 pandas.Dataframe.ewm 함수를 사용하면 최적화된 상태로 사용하실 수 있습니다. MDD 등 Pandas에서 최적화하지않은 연산들은 아래와 같은 개선이 큰 성능 향상을 이끌 수 있습니다.)

from kirin import Kirin
api = Kirin()
api.reset_cache()
api.fred.earliest_realtime_start = '1970-01-31'data = pd.concat([api.fred.aaa_corporate(frequency='D', ffill=True), api.fred.baa_corporate(frequency='D', ffill=True)], axis=1).dropna()
print(data)

(Kirin 은 크레프트테크놀로지스의 리서치 플랫폼으로서 다양한 소스에서 수집한 금융 데이터들을 API로 제공합니다.)

DAAA   DBAA
date                   
1986-01-06  9.91  11.38
1986-01-07  9.92  11.35
1986-01-08  9.92  11.35
1986-01-09  9.92  11.35
1986-01-10  9.92  11.36
...          ...    ...
2020-07-15  2.13   3.34
2020-07-16  2.15   3.35
2020-07-17  2.14   3.31
2020-07-18  2.14   3.31
2020-07-19  2.14   3.31[12614 rows x 2 columns]

이 사례에서는 12611개 행의 AAA / BAA 회사채 데이터를 가져오고, 각 열의 EMA를 계산할 것입니다.(편의를 위해 forward fill을 가정했습니다)

가장 나이브한 구현은 Pandas Dataframe 행을 순회하면서 지수이동평균을 업데이트하는 것입니다.

%%timeit
data_copy = data.copy()
for i in range(1, len(data_copy)):
    data_copy.iloc[i] = data_copy.iloc[i-1] * 0.3 + data_copy.iloc[i] * 0.7
data_copy.values-----------------------------------------------------
13.7 s ± 329 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

만약 무거운 Pandas가 아닌 Numpy array로 순회한다면 꽤나 많은 시간을 절약할 수 있습니다.

%%time
arr = data.values.copy()
for i in range(1, len(arr)):
    arr[i] = arr[i-1]  * 0.3 + arr[i] * 0.7
arr-----------------------------------------------------
50.2 ms ± 1.15 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)

np.vectorize는 같은 연산을 다른 열(위 케이스에서는 AAA 회사채와 BAA 회사채 각각)에 적용하는 케이스에 조금 더 빠른 성능을 제공합니다. (함수를 정의해두고 mapping시킨다는 점에서도 이점이 있습니다.)

%%time
arr = data.values.copy()func = np.frompyfunc(lambda x, y: x*0.3 + y*0.7, 2, 1)
ret = func.accumulate(arr, dtype=np.object).astype(np.float64)
ret-----------------------------------------------------
5.48 ms ± 167 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

다른 옵션은 Numba가 있습니다. Numba는 함수를 런타임에 컴파일함으로써(JIT — just in time) 성능을 향상시킵니다.

@numba.jit
def ema(a):
    for i in range(1, len(a)):
        a[i] = a[i-1] * 0.3 + a[i] *0.7
    return a%time ema(data.values.copy())
---------------------------------------
CPU times: user 447 ms, sys: 12 ms, total: 459 ms
Wall time: 457 ms

Numba 를 사용하실 때 주의해야할 점 중 하나는, 위에서 언급하였듯이 첫 사용 시에 컴파일을 진행하기 때문에, 첫 사용에는 컴파일 시간이 포함된다는 사실입니다.

%%timeit
ema(data.values.copy())-----------------------------------------------------
1.86 ms ± 157 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

두 번째 사용부터는 훨씬 더 빠른 속도를 보입니다.

Time needed for calculate EMA (ms, log scale)

좀 더 많은 양의 데이터에서 실험한 결과(4264752행) 는 다음과 같습니다.

# pandas
CPU times: user 28min 52s, sys: 3.27 s, total: 28min 55s
Wall time: 28min 52s# numpy
CPU times: user 12.9 s, sys: 3.99 ms, total: 12.9 s
Wall time: 12.9 s# numpy vectorize
CPU times: user 916 ms, sys: 164 ms, total: 1.08 s
Wall time: 1.08 s# numba jit
CPU times: user 211 ms, sys: 4.01 ms, total: 215 ms
Wall time: 214 ms# numba jit after compiled
CPU times: user 2.1 ms, sys: 143 µs, total: 2.24 ms
Wall time: 1.74 ms

Time needed for calculate EMA on larger dataset(ms, log scale)

이번에는, 컴파일 시간이 포함되어 있음에도 다른 방식에 비해 Numba JIT 의 성능이 돋보입니다.

중요한 점은, 함수가 병렬처리될 수 없는 단순한 반복문 작업인 경우에도, 꽤나 많은 성능을 향상시킬 수 있다는 사실입니다.

저 수준(Low-level) 컴파일 언어를 통해 속도와 안정성 동시에 잡기

만약 당신이 더욱 빠른 속도를 확보하고 동시에 안전한 코드를 작성하고 싶다면 컴파일 언어를 사용하는 것을 고려해 볼 수 있습니다.

아래 예시는 특허데이터에서 해당 발명의 ‘신선함’을 측정하는 케이스입니다.

patent_text_paths = glob.glob('./data/patents/*_text.txt')
print(open(patent_text_paths[10]).read())

특허 내용 중 하나를 살펴봅시다.

RELATED APPLICATIONS
This application claims priority to U.S. Provisional Patent Application No. 62/580,476, filed Nov. 2, 2017, and incorporates the same herein by reference.TECHNICAL FIELD
This invention relates to a firearm cartridge and barrel chamber for it. More particularly, it relates to a cartridge and cartridge load combination for efficient, high-velocity, long-range, precision shooting.
BACKGROUND
Ballistics, the science and study of projectiles and firearms, may be divided into three categories......
...

‘신선함’ 점수를 부여하는 방식은 매우 간단합니다. ‘신선함’을 나타내는 단어가 있으면 추가점을 주고, 반대의 경우 감점을 가하는 방식이죠.

signals = {
    "useful":1,
    "new":1,
    "important":2,
    'complex':2,
    "impressive":3,
    "powerful":3,
    "special":4,
    "innovation":5,
    "same": -1,
    "trivial": -2,
    "modify":-3,
}def _search_file(fname):
    ret = 0
    text = open(fname,'r').read()
    for line in text.split('\n'):
        for word in line.split(' '):
            if word in signals:
                ret += signals[word]
    return ret%%timeit 
scores_python = []
for fname in patent_text_paths:
    scores_python.append(_search_file(fname))

_search_file 함수는 9221개 텍스트 파일에 적용되어 점수를 산정합니다. 해당 파일은 2020–04–03 부터 2020–04–30 까지 USPTO에 등록된 자료입니다.

9.58 s ± 118 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

위 작업은 각 파일마다 독립적이기에, 병렬적으로 처리될 수 있습니다. 다만 파이썬은 multi-threading을 지원하지 않기 때문에, multiprocessing이 순수 파이썬에서 지원하는 유일한 옵션입니다.

from multiprocessing import Pool%%timeit -r 30
pool = Pool(32)
scores_mp = pool.map(_search_file, patent_text_paths)-----------------------------------------------------
930 ms ± 175 ms per loop (mean ± std. dev. of 30 runs, 1 loop each)

혹은 ray와 같은 외부 라이브러리를 사용할 수 있습니다.

import ray
ray.init()@ray.remote
def _search_file_remote(fname):
    return _search_file(fname)%%timeit -r 30
scores_ray = []
scores_ray = ray.get([_search_file_remote.remote(fname) for fname in patent_text_paths])-----------------------------------------------------
965ms ± 17.6 ms per loop (mean ± std. dev. of 30 runs, 1 loop each)

Rust는 높은성능과 강력한 타입 체킹을 주 무기로 빠르고 안전한 프로그램을 제공하는 신생 컴파일 언어입니다.

pyo3를 사용하면 이 Rust 함수를 Python library로 바인딩할 수 있습니다.
(refer https://github.com/PyO3/pyo3/tree/master/examples/word-count)

python setup.py develop && yes | cp word_count/word_count.cpython-37m-x86_64-linux-gnu.so ..

이제 파이썬에서 이 함수를 로드하여 사용해봅시다.

from word_count import search_from_files%%timeit -r 30
scores_rust_parallel = search_from_files(patent_text_paths, signals)-----------------------------------------------------
648 ms ± 13.3 ms per loop (mean ± std. dev. of 30 runs, 1 loop each)