카테고리 보관물: Python

Python unittest 모든 테스트 한 번에 하는 방법

파이썬으로 유닛테스트 처음 해 보는 분들 계시죠.
파일 하나씩은 실행시켜봤는데, 한 번에 모든 테스트 파일 실행하고 싶을 때 어떻게 해야 하는지 알려드릴게요.

테스트 실행하기

아래 명령어 한 줄이면 끝납니다. 프로젝트 루트에서 실행하도록 하세요.

python -m unittest

실행했는데, 아래와 같은 결과가 나왔나요? 그렇다면 이렇게 된 2가지 가능성이 있는데, 그 두 가지 확인해 보겠습니다.

----------------------------------------------------------------------
Ran 0 tests in 0.000s

OK

해결해야 할 이슈 2가지

파일이름

테스트 파일 이름은 test_로 시작해야 합니다. 그렇지 않으면 unittest가 파일을 찾을 수 없어요.

__init__.py

테스트 파일이 있는 경로에 __init__.py가 없다면 꼭 만들어 주세요. 아래 한 줄로 끝나는 거 아시죠.

touch __init__.py

파이썬의 unittest 한 번에 하는 방법 확인해 봤습니다.
동일한 이슈 겪고 계신 분들께 도움이 되길 바랍니다.

Anaconda 가상환경 생성 및 삭제 방법

python 3.9 버전의 아나콘다 가상환경을 생성하는 방법은 다음과 같다. 3.9 대신에 원하는 버전을 입력하면 해당 버전으로 가상환경을 만들 수 있다.

conda create -n 가상환경이름 python=3.9

지금까지 생성한 가상환경 목록을 확인하고 싶다면 다음 명령어를 사용하면 된다.

conda env list

아나콘다 가상환경을 만들어 사용하다가 불필요해서 삭제해야 하는 경우가 있다. 가상환경을 삭제하는 방법은 다음과 같다.

conda env remove -n 가상환경이름

크레온 플러스 API 연동관련 IsConnect 확인 후 강제종료

크레온 플러스 API 연동시 특정 시스템에서 발생하는 문제가 있다.

결론부터 이야기하자면, 윈도우즈 운영체제에서 인식하는 CPU 개수가 1개인 경우에 CpCybos의 IsConnect 값을 확인하는 경우에 실행중인 어플리케이션이 강제 종료되는 현상이 있다. CPU 개수가 2개 이상인 경우에는 문제없이 작동하는 것을 확인했다.

원인을 파악하는 상세한 과정은 아래에 적어두었다. 과정에 대해 궁금하지 않은 사람은 읽지 않아도 괜찮다.

테스트한 머신은
Vultr 가상 머신 1 CPU, 2G RAM
python 3.9.6
pywin32 302

필자가 작성한 파이썬 코드에서 IsConnect에서 문제가 생기는 것을 확인하고, 문제를 파악하기 위해서 다시 작성한 코드는 아래와 같다.

import win32com.client
import time

print("START")
cp_cybos = win32com.client.Dispatch('CpUtil.CpCybos')
print("LOADED")
a = time.time()
is_connected = cp_cybos.IsConnect
b = time.time()
d = b - a
print("%.8f" % d)

for i in range(10):
print(i)
time.sleep(1)

print("END")

화면 출력 결과는 오류 없이 아래와 같이 출력되고 끝난다.

START
LOADED
5.55898523
0
1

너무 당황스러웠다, For Loop을 돌다가 어플리케이션이 강제종료된다는 것은 상식적인 코드의 실행결과가 아니었다.

흥미로운 것은 IsConnect 값을 얻어오는 과정이 문제가 생기지 않는 머신에서는 1초 미만의 시간만 걸리는 한 편, 문제가 생기는 머신에서는 5초 이상의 시간이 걸린다는 것이었다.

집에 있는 PC에서도 동일한 python 버전과 동일한 버전의 pywin32 라이브러리를 사용했는데, 결과가 달랐다.

1차적으로는 OS 버전이 달랐기 때문에, 혹시나 하는 마음으로 집에 있는 운영체제와 같은 버전의 윈도우를 Vultr의 동일 사양의 인스턴스에 설치해보았다. 하지만 결과는 같았다. 동일한 문제가 생기고 있었다. OS 버전의 차이가 영향을 주는 게 아니라는 것을 확인했다.

크레온 플러스 API Q&A 게시판에서 관련 게시물들을 검색하다가, “Re : 비주얼 스튜디오 닫힘 현상”이라는 게시물을 확인했고, 디버깅 하는 경우에는 “메모리 보안 프로그램 사용”을 끄고 사용해야 한다는 것이다.

그래서 동일한 시스템에서 “메모리 보안 프로그램 사용”의 체크를 해제하고 크레온 플러스를 실행하니, 정상 작동이 되는 것이다. 그러면 보안 프로그램 끄고 쓰면 되겠네 싶겠지만, 집에 있는 피씨에서는 보안 프로그램을 끄지 않고도 잘 된다. 보안 프로그램을 쓸 수 있는 PC와 쓸 수 없는 PC에는 어떤 차이가 있어서 그런 것일까 의문이 들었다.

가상 머신이라서 안 되는 것일까 싶은 생각이 들어서, 현재 개발용 머신으로 사용하는 Ubuntu 20.04에 Gnome Boxes 위에 동일한 윈도우즈를 설치하고 테스트해봤다. 전혀 문제가 생기지 않았다.

그렇다면 Vultr 서비스에 어떤 문제가 있어서 그런 것 아닐까 싶었다. 그런데 지난 번에 사용했던 $20 짜리 인스턴스는 문제가 생기지 않았었는데, 왜일까 싶은 생각이 들었다. 현재 테스트한 Vultr의 인스턴스는 CPU 1개짜리, 잘 되던 인스턴스는 CPU 2개짜리 였다. 잽싸게 떠놓은 Snapshot으로 CPU 2개짜리 인스턴스를 띄우고 테스트 해 봤고, 잘 작동하는 것을 확인했다.

이게 정말 원인이라면 Gnome Boxes에서도 동일한 문제가 생겨야 한다는 생각으로 Gnome Boxes의 CPU 개수를 1개로 변경하고 다시 테스트 해봤는데, 동일한 문제가 생겼다.

결론을 다시 정리해 보았다.

첫째, AOS는 CPU 개수가 1개인 머신에서 크레온 플러스의 IsConnect 값을 얻어오는 과정에서 문제를 일으킨다.

둘째, 가상머신이나 클라우드 서비스와는 아무 상관이 없다.

해결책
IsConnect 값을 얻어온 이후에 얼마 안 있다가 어떠한 오류 메시지도 없이 작성한 어플리케이션이 종료된다면, CPU 개수를 운영체제가 몇 개로 인식하고 있는지 확인하고, 만약 1개라면 2개 이상인 머신에서 구동한다.

끝!

numba guvectorize 활용하기

numba의 guvectorize를 사용하면 array 형태의 값 처리의 속도를 매우 빠르게 할 수 있다.

우선 실행 결과를 먼저 보겠다. aaa는 guvectorize를 사용해서 연산한 경우이고, bbb는 그냥 python 코드로 실행한 결과이다. ccc는 numpy의 벡터 연산을 실시한 결과다. guvectorize를 적용했을 때 단순 for loop를 사용했을 때에 비해 약 300배 가까이 빠른 실행을 보였다. numpy의 벡터 연산을 활용하면 for loop을 사용했을 때 보다 1400배 가까이 빠르게 실행되었다.

aaa: elapsed=0.0275056362s
bbb: elapsed=9.2325780392s
ccc: elapsed=0.0066127777s
>>> 9.2325780392 / 0.0275056362
335.6613158142476
>>> 9.2325780392 / 0.0066127777
1396.1724494685493

결론부터 말하자면, Array의 단순 연산은 numpy의 벡터연산이 가장 빠르다. 특정 로직에 따른 연산이 필요할 땐 guvectorize를 활용하자.

guvectorize의 사용법을 간단히 살펴보자. guvectorize의 경우에는 값을 return하지 않는다. 대신에 파라미터에 대한 선언을 정확히 해 주면 된다.

@guvectorize(['void(int64[:], int64[:])'], '(n) -> (n)')
def function_name(x, y):

순서대로 int64[:], int64[:]는 뒤의 (n) -> (n)과 매칭된다.

Array가 아닌 단순 값을 파라미터로 추가하는 경우에는 아래와 같이 처리하면 된다. 데이터 타입의 정의 개수와 입력, 출력 파라미터의 형식, 그리고 함수의 정의까지 모두 일치해야 한다.

@guvectorize(['void(int64[:], boolean, int64[:])'], '(n),() -> (n)')
def function_name(x, twist, y):

아래에는 위의 결과를 만든 소스 코드를 붙인다.

import time
import numpy as np
from numba import guvectorize


def timer(func):
    def measure(*args, **kwargs):
        begin = time.time()
        val = func(*args, **kwargs)
        end = time.time()
        print(f">> {func.__name__}: elapsed={'%.10f' % (end - begin)}s")
        return val

    return measure

@timer
@guvectorize(['void(int64[:], int64[:])'], '(n) -> (n)')
def aaa(x, y):
    for i, val in enumerate(x):
        y[i] = val + 3 * 3 / 5 * 1.23 / 6.53

@timer
def bbb(x):
    y = np.empty([len(x), 1])
    for i, val in enumerate(x):
        y[i] = (val + 3 * 3 / 5 * 1.23 / 6.53)
    return y

@timer
def ccc(x):
    return x + 3 * 3 / 5 * 1.23 / 6.53

a = np.int32(np.round(np.random.rand(5000000) * 1000))
out = aaa(a)
out = bbb(a)
out = ccc(a)

매매로직 재사용

빠른 배포가 중요하다는 포스팅에서 이야기 한 방법은 사실 안정적이지 않다. 더 좋은 방법은 백테스트할 때 활용한 매매 시스템의 로직을 트레이딩 시스템에서 동일하게 활용하는 방법이다. 1호 시스템을 만들고 나서, 앞으로 2호 이후의 시스템들을 만들 생각을 하니 매번 모든 코드를 작성해야 하는 것은 비효율적이라는 생각이 들었다. 매매로직을 모듈화해서 백테스트와 트레이딩 시스템에 모두 동일하게 활용할 수 있게 한다면, 백테스트에서 실전 매매 시스템 개발까지의 기간을 최소화 할 수 있을 것이라는 생각이 들었고, 이 작업에 이제 곧 착수할 예정이다.

백테스트의 성능을 저하시키지 않으면서, 트레이딩 시스템에 어떻게 장착할 수 있을지에 대한 설계가 필요하다. 특히 키움의 QEventLoop 사용방식은 코드의 복잡도를 증가시키기 때문에 ES7의 async, await 처럼 비동기를 동기식으로 사용할 수 있는 방법을 찾아보려고 한다.

그리고 앞으로는 TDD를 사용하여 시스템을 개발하려고 한다. 과거의 경험을 비추어봐도 TDD로 개발할 때 생산성이 가장 좋았고, 오류가 적었다. 1호 시스템을 개발하면서 매매로직 테스트하는 부분만 TDD를 활용하긴 했는데, case에 대한 정리가 부족한 탓에 오류 상황들이 발생했다. 그럼에도 테스트코드가 있어서 빠르게 문제를 발견하고 해결할 수 있었다.

쨌든 Do not Repeat Yourself(DRY 원칙)를 시스템 구축에도 활용해서 시스템 제작의 생산성을 높여야겠다.

Python에서 용량이 큰 테이블의 insert, update 속도 문제

테이블 하나에 약 800만개의 레코드를 가진 테이블(약 1.2GB)을 만들고 분석할 기회가 생겼는데, 매일의 데이터 변경사항에 대해 추가하거나 업데이트를 해야 한다.

약 1,000개의 레코드를 업데이트하는데 25초 정도 걸렸다. 분명 뭔가 잘못됐다는 신호다. 초당 40개 정도의 업데이트밖에 되지 않았다.

해결책은 그리 복잡하지 않았다. 필자가 했던 적용법들은 다음과 같다.

1. select query를 사용하여 레코드 하나씩 1,000번씩 쿼리를 던져서 필요한 값들을 확인하던 것을 한 번의 쿼리로 작업 단위에 필요한 만큼 불러온 후, python에서 관련 값들을 확인하고 처리하도록 변경했다.

2. insert시 1,000개의 insert query를 실행하던 것을 다음과 같이 한 개의 insert query로 변경했다. 이 방법을 사용하는 경우 query문이 너무 길어지는 경우 실행되지 않을 수 있다. my.cnf에 관련 값을 얼마로 설정했느냐에 따라 달라지는 것으로 보이는데, key값이 지금은 기억나지 않는다. 추후에 확인해 보고 추가해 놓으려 한다.

insert into 테이블명 (필드1, 필드2, ...) values (값1a, 값2a, ...), (값1b, 값2b, ...), ..., (값na, 값 nb, ...)

3. update시 1,000개의 query를 실행하는 것은 동일하지만, where절에 키 두 개를 사용해서 update 하던 것을 primary key 하나만 사용하도록 변경하였다.

위의 3가지 방법을 모두 적용한 후 1,000개 레코드 처리에 25초 정도 걸리던 것은 평균 0.6초 이내로 처리시간이 단축되었다.

적용하기 전에는 Mysqld 프로세스의 CPU 점유율이 단일코어기준으로 100%이었는데, 변경 후에는 약6% 언저리에서 작동했다. 작업 환경은 다음과 같다. CPU는 인텔 G4400, 메모리 8G, SSD 128G, OS는 Windows 10, MariaDB 버전은 10.2를 사용하였다. 그리고 프로그래밍 언어는 Python3에서 MySQLdb를 사용하였다.

테이블 용량이 조금 크다고 해서 테이블을 여러개로 쪼개는 것이 우선이 아니라는 것을 이번에 배웠다. 특히 테이블 크기가 큰 경우에는 DB의 부하를 줄이는 방향으로 최적화할 필요가 있음을 느꼈다. 재미있는 경험이었으며, 기본적인 원칙이 얼마나 중요한지 다시금 생각해 보는 계기가 됐다.