money-expert

파이썬은 참 좋은 언어인데, C나 C#같이 컴파일 방식이 아니라 인터프리터 방식이다보니 실제 코드가 실행되는 순간에 문법적인 오류가 발생하면 죽는 문제가 생깁니다. 컴파일 방식이라면 실행하기 전에 인지할 수 있겠지만 파이썬은 쉽지가 않습니다. 이 문제를 해결하기 위해서는 수정한 부분을 실제 조건을 만족시켜 돌려봐야하는데요. 이것도 쉬운 것은 아닙니다. 실제 조건문이 만족하도록 설정하는 행위가 소스를 수정하는 것이기 때문에 이 또한 문제를 야기할 수 있습니다.

이런 문제를 풀기 위하여 보통 유닛테스트를 많이 사용하는데요. 유닛테스트를 전체 프로그램을 test할 수 있으면 수정시 발생하는 오류를 가장 빨리 발견할 수 있습니다. 하지만 소스 전체를 유닛테스트로 적용하기 위해서는  최초 설계부터 반영을 해야하기 때문에 쉽지 않습니다.

결론적으로 유닛테스트를 적용하면 좋은데, 전체 소스코드에 적용하기에는 좀 어렵다인데요. 대안으로 특정 함수만 별도로 유닛테스트한 후 사용하는 것입니다.

이번에 주식 자동매매 프로그램을 만들던 중 매매 가격에 따라 유효한 주문 가격을 결정하는 함수를 유닛테스트를 적용한 예를 설명하도록 하겠습니다. 특히 주식 자동매매에서 유효한 주문가격을 구하지 못하면 주문 거부가 발생하기 때문에 꼼꼼하게 test해보아야 합니다. 모든 case를 손으로 직접 test하는 것은 쉬운 일이 아니기 때문에 유닛테스트를 적용하는 것이 바람직 합니다. 특히 해당 함수 내부 소스가 변경되는 경우에는 유닛테스트의 진가가 발휘됩니다.

올해 변경된 주식 가격별 호가 단위입니다. 자동매매 코드에서는 소숫점 계산을 한 목표 매수가가 나오고 이를 버림/반올림/올림 등의 조건을 추가하여 호가 단위에 맞는 가격으로 변경해주는 함수가 필요합니다.

우선 아래 package를 import합니다.

import unittest

그리고 원하는 함수를 작성합니다. 일단 2천원 미만은 1원 단위의 가격을 돌려주는 함수를 만들어봅시다. 

def get_order_price(price) :
    tick_size = 0
    if price < 2000:
            tick_size = int(price)
    return tick_size

다음은 unittest를 사용하는 방법입니다. unittest에서는 아래와 같은 CustomTests class를 지원합니다. 원하는 함수안에 필요한 코드를 넣으면 됩니다. test case는 test_process_statement() 함수 안에서 작성하면 됩니다.

class CustomTests(unittest.TestCase): 
    def setUp(self):    # 무시
        pass
    def tearDown(self): # 무시
        pass
    def test_process_statement(self) :
        in_price = 45
        result = get_order_price(in_price)
        self.assertEqual(result, 45) # 계산 결과와 원하는 값 입력

위와 같이 test_process_statement() 함수 안에 test case를 만들면 됩니다. 그 후 unittest.main()을 호출하면 계산값과 원하는 결과가 틀리면 self.assertEqual함수에서 fail이 뜹니다.

unittest.main()

실행을 해보면 아래와 같은 결과 나옵니다.

만약 in_price와 다른 값을 입력한다면

    def test_process_statement(self) :
    	in_price = 45
        result = get_order_price(in_price)
        self.assertEqual(result, 44) # 계산 결과와 다른 값 입력

친절하게도 틀리다는 메세지가 출력됩니다.

다음은 주식 거래시 사용하는 실제 코드입니다. 이번 전략에서는 목표 가격이 오면 가능한 빨리 매수하기를 원하기 때문에 버림을 적용합니다. float를 int로 변경하기만 하면 자동으로 버림이 적용되기 때문에 간단하게 구현할 수 있습니다.

def get_order_price(price):
    tick_size = 0.0
    if price >= 500000:
        tick_size = int(price/1000)*1000
    elif price >= 100000:
        tick_size = int(price/500)*500
    elif price >= 50000:
        tick_size = int(price/100)*100
    elif price >= 10000:
        tick_size = int(price/50)*50
    elif price >= 5000.0:
        tick_size = int(price/10)*10
    elif price >= 1000.0:
        tick_size = int(price / 5)*5
    else:
        tick_size = int(price)
    return tick_size

위 코드를 바탕으로 test case를 만들어보면 아래와 같습니다. 이외에도 모든 가격대를 test하는 문장을 추가하여야합니다.

    def test_process_statement(self) :
        result = get_order_price(999.99) # 1000원 미만 test
        self.assertEqual(result, 999)

        result = get_order_price(1000.01) # 1000원 이상 test
        self.assertEqual(result, 1000)

        result = get_order_price(4999.9999) # 5000원 이하 test
        self.assertEqual(result, 4995)

        result = get_order_price(9999.9999) # 10000원 이하 test
        self.assertEqual(result, 9990)

실행결과는 아래와 같습니다. 즉 문제가 없다는 의미입니다.

만약 함수의 출력과 예상결과가  틀린 경우에는 오류 메세지가 뜹니다. 오류 메세지가 뜨면 해당 가격에 대한 코드를 자세하게 살펴보면서 문제점을 수정하면 됩니다.

이렇게 완성된 함수를 본 프로그램에 추가하여 개발을 진행하면 됩니다. 모든 함수를 유닛테스트하기에는 시간이 많이 걸리는 문제가 있으므로, case가 복잡하거나 경우의 수가 많은 함수를 만드는 경우에는 해당 함수만 뽑아서 유닛테스트 환경하에서 test하는 습관을 들이는 것이 좋을 듯 합니다. 당장은 힘들겠지만 시간이 가면 갈수록 코드의 안정성을 유지하는 것이 유닛테스트 통과한 코드입니다.

일전에 쓴 grid trading을 기반으로 주식 자동 매매를 돌리고 있습니다.

https://money-expert.tistory.com/74?category=781442 

지난 6월 중순부터 계속 돌리고 있는데요.

중간 점검 차 현재 상태를 올려봅니다.

6월은 주가가 계속 내려가는 시기여서 평가손실이 발생하였으나 7월에는 매매가 어느 정도 이루어지면서 수익이 났습니다. 결과적으로 두달에 약 1.7% 수익이 발생하였습니다.

6/23일 부터는 kodex200도 매매 종목에 추가하였습니다. 그 이유는 거래세가 없기 때문에 grid를 좁게 가져가도 수익이 발생할 수 있기 때문입니다. 또한 kospi가 급등락을 반복하고 있어서 grid grading하기 좋은 종목으로 판단이 되었습니다.

현재 보유 종목은 총 27개이며 적당히 +- 상태를 유지하고 있습니다.

grid trading에서 수익이 많이 나오려면 매매가 많아야 합니다. 오늘까지 총 103회 매도가 되었습니다.

1회 투자금이 약 33만원. grid는 3.3%이므로 대략 1백여만원 매매 수익이 발생하였습니다.

현재 잔고 상태는 - 이지만 매매 차익 덕분에 +로 전환한 상태입니다. 

간단한 소회입니다.

- 종목 선택이 역시 중요하다. 하지만 영원히 하락하는 종목은 별로 없다. 즉 적절히 빠진 시점에 grid trading을 시작하면 결국 수익이 날 수 있다.

- 계속 빠지는 경우도 많다. 손실을 제한하기 위하며 grid 수를 제한한다. 최대로 살 수 있는 grid 수를 제한함으로써 지속적으로 우하향하는 경우에 손실을 제한할 수 있다. 현재는 grid 3개로 제한함(약 -10%)

- 매수 시작한 시점 이후 바로 상승하여 매수를 못한 경우도 있고, 한번 정도 매매하고 우상향한 경우도 있다. 이런 건 어쩔 수 없을 듯.

- 매수 종목이 많아야 매매가 자주 발생하는데, 종목을 늘리기 위해서는 투자금이 많이 들어간다. 적절한 투자 규모를 정하는 것이 필요하다. 증거금 관리가 중요한데 40% 증거금이 필요한 종목 위주로 포트폴리오를 구축하는 것도 좋다.

지금까지 운용해본 경험상 적당히 빠진 시총 높은 종목으로 grid trading을 하면 연 10% 정도는 나올 듯 합니다. 

한달에 한번씩 투자 상황을 정리해서 공유하겠습니다.

개발은 파이썬을 이용하도록 하겠습니다. 관련 예제가 블로그에 많이 있습니다. dev center에 있는 예제와 블로그들을 참고하였습니다.

개발 IDE로는 spyder나 Visual Studio Code 등 본인이 익숙한 것을 사용하면 됩니다.

전 기존에 visual studio를 사용하던 습관이 있어서 Visual Studio Code를 사용하기로 했습니다.

이베스트 xingApi와 통신을 위해서는 win32com을 사용합니다. 이를 위하여 win32com을 설치해야하는데요. 아나콘다32를 설치하면 일반적으로 함께 설치가 됩니다. 혹시라도 win32com을 찾을 수 없다고 나오면 아래와 같은 명령어로 설치하시면 됩니다.

pip install pywin32

우선 login 과정을 보도록 하겠습니다.

그 절차는 다음과 같습니다.

1. 로그인에 관련된 정보를 입력한 후 xing api에게 login 명령어를 보냅니다.
2. xingApi에서 답이 오기를 기다립니다.
3. 등록한 callback 함수로 login 결과가 옵니다. login 성공 여부를 여기에서 확인합니다.

이베스트 주식 거래를 위한 class를 만들었습니다.

앞으로 선물거래 그리고 암호화폐 거래까지 포함하기 위하여 거래소 별로 class를 만들어 보도록 하겠습니다.

거래소별 base class인 Exchange를 기반으로 이베스트주식 거래 class입니다.

class XASessionEventHandler:  
    login_state = 0

def OnLogin(self, code, msg):  
    print('on login start')  
    if code == "0000":  
        print("login succ")  
        XASessionEventHandler.login\_state = 1  
    else:  
        print("login fail")  

def wait_for_event(code) :
    while XAQueryEventHandler.query_state == 0:
        pythoncom.PumpWaitingMessages()
    if XAQueryEventHandler.query_code != code :
        print('diff code : wish(',code,')', XAQueryEventHandler.query_code)
        return 0
    XAQueryEventHandler.query_state = 0
    XAQueryEventHandler.query_code = ''
    return 1

class EBestStock(Exchange):  
    def __init__(self):
        self.operation_begin = '085000'
        self.operation_end = '153000'

    def login(self, server, id, pwd, cer_pwd, acc, acc_pwd) :
        self.instXASession = win32com.client.DispatchWithEvents("XA_Session.XASession", XASessionEventHandler)
        self.id = id
        self.passwd = pwd
        self.cert_passwd = cer_pwd
        self.account_number = acc
        self.account_pwd = acc_pwd
        self.instXASession.ConnectServer(server, 20001)
        self.instXASession.Login(self.id, self.passwd, self.cert_passwd, 0, 0)
        while XASessionEventHandler.login_state == 0:
            pythoncom.PumpWaitingMessages()

        self.login = XASessionEventHandler.login_state
        return self.login        

if __name__ == "__main__":
    print('\\nebest testing')

server = "hts.ebestsec.co.kr"  # or "demo.ebestsec.co.kr" 모의투자
id = "user id"
passwd = "user password"
cert_passwd = "공인인증서암호"
account_number = "계좌번호"
account_pwd = "계좌비밀번호"

ebest_st = EBestStock()
ret = ebest_st.login(server, id, passwd, cert_passwd, account_number, account_pwd)
if ret == 0 :
    print('fail to login')
    quit(0)

print('login ok')  

실행 결과입니다. 계정 정보를 정확하게 입력하였다면 login 성공할 것입니다.