python Bitcoin 자동 매매 프로그램(8) - Upsert 구현

1. 들어가기 전에

  오늘은 자동매매 프로그램 구현에서 중요하지만 주제와는 약간 빗겨간 DB 내용중 Upsert(Update or Insert)에 대하여 이야기 하고자 합니다. pandas의 DataFrame은 Nosql적 성격을 가진 아주 휼륭한 Lib입니다. 사실 pandas가 없다면 지금의 자동매매 프로그램의 구현은 힘들다고 해도 이견이 없을 것입니다. 하지만 pandas DataFrame을 사용하다보면 계산이 다 끝나서 이제는 실제 DB로 옮겨야 하는데 기존 제공하는 to_sql은 뭔가 약간은 부족합니다. to_sql함수는 중복이 있으면 insert 실패 처리 하거나 table 전체를 replace하거나 또는 그냥 append하는 것 입니다. 저는 지금은 pandas와 mysql을 사용하고 있지만 초창기는 sqlite3을 사용하였고 mysql로 DB를 변경하면서 문제가 발생했습니다. to_sql의 replace기능이 DB를 삭제하는 과정에 프로그램이 먹통이 되고 이때까지 모아온 소중한 데이터가 다 날아 가는 거죠. sqlite사용중에도 프로그램을 중간에 내릴때 타이밍이 안좋으면 table을 그대로 날려 먹는 경우가 생깁니다. 그래서 현재 인터넷에서는 pandas의 to_sql에 새로운 옵션인 upsert가 화재 입니다. 일부 개발 git에는 upsert가 구현되어 올라와 있지만 pandas의 to_sql을 ctrl + link로 따라 가보면 근본적인 sql lib에 있는 기능을 사용하고 있어 수정하는 것이 대공사라는 것을 금방 알 수 있습니다. 오늘은 제가 구현한 Upsert에 대하여 내용을 공유하고자 합니다. 

 

2. Upsert구현

  먼저 아래 소스를 보시죠.

def trade_data_update(df, dbcon, tb_name, if_exists: str='ignore'):
    try : 
        cols = [column for column in list(df.columns)]

        str_query = ""
        if if_exists == 'fail':
            str_query += "INSERT INTO %s (" % tb_name
        elif if_exists == 'replace':
            str_query += "REPLACE INTO %s (" % tb_name
        else:
            str_query += "INSERT IGNORE INTO %s (" % tb_name

        for col in cols:
            str_query += "`" + col + "`,"
        str_query = str_query[:-1]
        str_query += ")"

        rows_val = []
        for index in df.index:
            rowval = []
            for column in df.columns:
                rowval.append(df.loc[index, column])
            rows_val.append(rowval)
        
        str_query += " values "
        for rows in rows_val:
            str_query += "("
            for row in rows:
                if str(type(row)) == "<class 'str'>":
                    str_query += "'" + str(row) + "',"
                elif str(row) == "nan":
                    str_query += "NULL,"
                else:
                    str_query += str(row) + ","
            str_query = str_query[:-1]
            str_query += "),"
        str_query = str_query[:-1]

        cursor = dbcon.cursor()
        cnt = cursor.execute(str_query)
        # logger.info("apply row cnt:%d", cnt)
        cursor.close()

    except Exception as ex:
        exc_type, exc_obj, exc_tb = sys.exc_info()
        fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
        logger.error("trade_data_update -> exception! %s  : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))

mysql기준 쿼리 입니다. mysql 에 INSERT에는 총 4가지의 방법이 있습니다. INSERT ON DUPLICATE KEY와 REPLACE INTO, INSERT IGNORE INTO, 그리고 그냥 INSERT입니다. 당연히 그냥 INSERT를 사용하면 INSERT과정에 Dup key가 발견되면 그 부분에서 바로 에러 후 다음으로 넘어 가겠죠. 나머지 3가지는 이렇게 Dup key가 발견되었을때의 예외 처리 방법을 기술한 것 입니다. 저는 보통 회사에서 프로그램할때는 INSERT INTO 보다는 REPLACE INTO를 많이 사용하는 편인데 지금 자동 매매에서는 INSERT IGNORE INTO를 많이 사용하고 있습니다. 

import pymysql
from lib.Upsert import trade_data_update

sqlcon = pymysql.connect(host='127.0.0.1', user='root', password='1234', db='main')

df = con.get_dataframe()
trade_data_update(df, sqlcon, "tb_training_data")
sqlcon.commit()

처럼 활용이 가능 합니다. 

 

3. 마치며

  다음 회차에서는 이렇게 모인 데이터를 활용하여 실제 training을 공유할까 합니다. 위의 Upsert를 이용하는 이유는 1분 데이터이다 보니 가끔 데이터 유실이 발생합니다. api나 프로그램의 문제가 아니라 upgrade를 위해서 잠시 프로그램을 내려 놓을때 수신하지 못한 데이터가 발생하는 거죠. 1회 수신 데이터를 100개 정도만 받아도 위의 방법을 사용하면 유실 없이 DB에 데이터를 저장할 수 있습니다.