바이비트와 파이썬을 이용한 자동매매 프로그램 - 5. 나만의 전략을 위한 파이프라인

설명

나는 아주 단순한 전략을 구현하는 것까지 오는데 오랜 시간이 걸려 왔다. API를 이용하는 것이 쉽고 간편하면서도 어려운 듯 하다. 이 글을 읽는 사람은 내가 겪었던 시행착오들을 한방에 해결할 수 있었으면 하는 마음에 코드를 공유한다. 해당 코드는 다음 조건을 만족하는 전략을 구현할 때 유용하게 사용할 수 있을 것이다.

1. 한 시간 단위로 분할 매수 / 분할 매도
2. 데이터를 통해 특정 조건을 만족할 경우 매수 / 매도
3. long / short 모두 가능하며 레버리지 / stop loss 기능까지 구현

---

코드

 

1. 라이브러리 호출

import pandas as pd
import numpy as np
import datetime
import calendar
import schedule
import math
import time
import math
from pybit import usdt_perpetual

필요 라이브러리를 호출한다.

 

2. 데이터 불러오기

def get_data(symbol, interval, limit):
    unixtime = calendar.timegm(now.utctimetuple())
    since = unixtime-interval*60*limit;
    response = session.query_kline(symbol=symbol,interval=str(interval), from_time=since, limit=limit)['result']
    return pd.DataFrame(response)


def data_reconstruction(symbol, interval, limit):
    # 데이터 받아오기
    data = get_data(symbol, interval, limit)
    for n in range(5):
        unixtime = data['start_at'][0]
        since = unixtime-interval*60*limit;
        response = session.query_kline(symbol=symbol,interval=str(interval), from_time=since, limit=limit)['result']
        df = pd.DataFrame(response)
        data = pd.concat([df, data])
        time.sleep(0.2)
    data.sort_values(by='start_at', ascending=False, inplace=True)
    data.reset_index(drop=True, inplace=True)
    # 한시간봉 데이터를 일봉 데이터로 통합
    low = data['low'].rolling(window=24).min()
    high = data['high'].rolling(window=24).max()
    volume = data['volume'].rolling(window=24).sum()
    np_temp = np.zeros((round(len(data)/24), 6))
    for n in range(1, len(np_temp)):
        np_temp[n][0] = data['start_at'][(n-1)*24]
        np_temp[n][1] = data['open'][n*24-1]
        np_temp[n][2] = high[n*24-1]
        np_temp[n][3] = low[n*24-1]
        np_temp[n][4] = data['close'][(n-1)*24]
        np_temp[n][5] = volume[n*24-1]
    data = pd.DataFrame(np_temp, columns=['start_at', 'open', 'high', 'low', 'close', 'volume'])
    data = data.iloc[1:]
    data.sort_values(by='start_at', ascending=True, inplace=True)
    data.reset_index(drop=True, inplace=True)
    return data

일단 데이터를 받아 오기 전에, 내가 어떤 데이터를 받을지 정해야한다. 위에서 get_data 함수의 경우 코인명, 데이터 간격(1분봉, 15분봉 등의 간격을 의미), 데이터 수(row) 총 3 가지를 input으로 받고 있으며, 추가적으로 항상 최신 데이터를 받기 위해 현재 시간을 정의해야 하며 API 연결을 위한 session을 준비해야한다. 한번 데이터를 출력해보자.

 

now = datetime.datetime.utcnow()
session = usdt_perpetual.HTTP(
    endpoint="https://api-testnet.bybit.com", 
    api_key='my_api_key', 
    api_secret='my_api_secret',
)
get_data('BTCUSDT', 60, 200)

	id	symbol	period	interval	start_at	open_time	volume	open	high	low	close	turnover
0	113571620	BTCUSDT	60	60	1658800800	1658800800	630.670	21086.0	21218.5	20919.0	21065.5	1.330942e+07
1	113589219	BTCUSDT	60	60	1658804400	1658804400	752.338	21065.5	21218.5	20850.5	21130.5	1.586627e+07
2	113606772	BTCUSDT	60	60	1658808000	1658808000	221.835	21130.5	21242.0	21092.0	21113.0	4.692243e+06
3	113624572	BTCUSDT	60	60	1658811600	1658811600	137.551	21113.0	21208.0	21047.0	21118.0	2.902394e+06
4	113641955	BTCUSDT	60	60	1658815200	1658815200	192.917	21118.0	21162.5	20999.5	21059.5	4.067266e+06
...	...	...	...	...	...	...	...	...	...	...	...	...
195	117028879	BTCUSDT	60	60	1659502800	1659502800	575.606	22834.5	23078.5	22659.5	22926.5	1.314929e+07
196	117046489	BTCUSDT	60	60	1659506400	1659506400	307.214	22926.5	23150.0	22809.0	23042.0	7.079537e+06
197	117064654	BTCUSDT	60	60	1659510000	1659510000	379.773	23042.0	23150.0	22969.5	22980.0	8.740947e+06
198	117082255	BTCUSDT	60	60	1659513600	1659513600	2255.385	22980.0	23372.5	22900.0	23300.5	5.225689e+07
199	117100271	BTCUSDT	60	60	1659517200	1659517200	209.925	23300.5	23460.0	23199.0	23365.5	4.897239e+06
200 rows × 12 columns

get_data 함수만으로도 데이터를 받아올 수 있다. 물론 일봉 데이터도 받아올 수 있다. 하지만 나는 data_reconstruction 함수를 이용하여 나만의 데이터를 만들었다. 그 이유는 우리가 API를 이용하여 일봉 데이터를 받아 올 경우, 데이터의 open & close 시간은 정각으로 정해져 있기 때문이다. 하지만 우리는 한시간 단위로 데이터를 읽어 오며 분할 매수 매도를 진행할 것이기 때문에, 0시 기준의 데이터, 1시 기준의 데이터 ... 23시 기준의 데이터 각각을 생성할 수 있어야 한다. data_reconstruction 함수를 출력해보자.

 

data_reconstruction('BTCUSDT', 60, 200)

	start_at	open	high	low	close	volume
0	1.655370e+09	20223.0	23069.0	20113.5	21275.0	25978.070
1	1.655456e+09	21275.0	21559.5	20200.0	20974.5	19780.334
2	1.655543e+09	20974.5	21248.0	18560.0	19169.0	14161.565
3	1.655629e+09	19169.0	19553.0	17605.0	19002.5	21832.638
4	1.655716e+09	19002.5	21154.5	18885.0	20654.0	24218.560
5	1.655802e+09	20654.0	21550.0	19764.5	21206.5	20096.661

다음과 같이 필요한 정보들을 일봉 데이터셋으로 변환하였다.

 

3. 거래 시작

# 필요 변수 정의.
symbol = "BTCUSDT"                               # 코인명
interval = 60                                    # 데이터 기준
limit = 200                                      # 데이터양
qty = 0.01                                       # 주문수량
fees = 0.00048
now_order = None
hour_list = [n for n in range(24)]               # 실행 시간

# 테스트넷 api 정보
session = usdt_perpetual.HTTP(
    endpoint="https://api-testnet.bybit.com", 
    api_key='my_api_key', 
    api_secret='my_api_secret',
)

set_margin_switch(symbol, True)          # 격리 마진 설정

필요한 변수들을 정의해준다. 마진 설정은 격리로 하였다. 마진을 설정하기 위한 함수는 다음과 같다.

 

def set_margin_switch(symbol, is_isolated=True):
    try:
        session.cross_isolated_margin_switch(
            symbol=symbol,
            is_isolated=is_isolated,
        )
    except:
        pass

설정이 완료되었다면 이제 데이터를 받아올 차례이다. 우리는 특정 조건이 만족하는지를 보고 매매를 진행하기 때문이다.

 

df = data_reconstruction(symbol, interval, limit)
df['ma5'] = df['close'].rolling(window=5).mean()
df['ma15'] = df['close'].rolling(window=15).mean()
df['ma30'] = df['close'].rolling(window=30).mean()

데이터는 data_reconstruction 함수만으로 불러와졌다. 그 이후 data engineering 부분은 각자가 원하는 전략에 따라 구현하면 되겠다.

나는 여기서 특정 조건을 5일 이동평균선과 30일 이동평균선보다 현재가가 높아졌을 경우 long 포지션 진입, 현재가가 두 이평선보다 낮을 경우 short 포지션 진입, 그렇지 않을 경우 포지션을 안들고 있는 것으로 하겠다. 따라서 이평선을 구현하는 코드를 넣었다.

 

# get leverage
if df['close'][len(df)-1] > df['ma15'][len(df)-1]:
    buy_leverage = 3
    sell_leverage = 1
else:
    buy_leverage = 1
    sell_leverage = 3
    
set_leverage(symbol, buy_leverage, sell_leverage)

레버리지의 경우 15일 이동평균선보다 현재가가 높을 경우 long 포지션에 대해 3배, 현재가가 낮을 경우 short 포지션에 3배를 주는 식으로 진행하였다. 이 또한 틀을 그대로 활용하대 원하는 조건, 원하는 레버리지 배수로 정하면 된다. set_leverage 함수는 아래와 같다.

 

def set_leverage(symbol, buy_leverage, sell_leverage):
    try:
        session.set_leverage(
            symbol=symbol,
            buy_leverage=buy_leverage,
            sell_leverage=sell_leverage
        )
    except:
        pass

이제 데이터도 받아왔고, 레버리지 설정도 해주었다. 매매만 진행하면 된다.

 

# get trade condition
if (df['ma30'][len(df)-1] < df['close'][len(df)-1]) & (df['ma5'][len(df)-1] < df['close'][len(df)-1]):
    now_order = 'Buy'
elif (df['ma30'][len(df)-1] > df['close'][len(df)-1]) & (df['ma5'][len(df)-1] > df['close'][len(df)-1]):
    now_order = 'Sell'
else:
    now_order = None

# long
if now_order == 'Buy':
    if session.my_position(symbol=symbol)['result'][0]['size'] < 24*qty:
        place_order(symbol, now_order, qty)   # long 매수
        set_stopLoss(symbol, now_order, buy_leverage)    # long stop loss 설정

# short
if now_order == 'Sell':
    if session.my_position(symbol=symbol)['result'][1]['size'] < 24*qty:
        place_order(symbol, now_order, qty)   # short 매수
        set_stopLoss(symbol, now_order, sell_leverage)    # short stop loss 설정

# clear
if now_order == None:
    if session.my_position(symbol=symbol)['result'][0]['size'] >= 1*qty:
        replace_order(symbol, 'Sell', qty)   # long 매도
    if session.my_position(symbol=symbol)['result'][1]['size'] >= 1*qty:
        replace_order(symbol, 'Buy', qty)   # short 매도

now_order 변수에 집중하여 매매를 진행한다. 코드는 다음과 같이 진행한다.

1. 주석 get trade condition 부분 : 내 매매 전략에 맞는 조건문을 생성한다. long 조건일 경우 now_order을 'Buy'로, short 조건일 경우 now_order을 'Sell'로, 두 조건 모두 아닐 경우 None으로 정의한다.
2. 주석 long 부분 : now_order 상태를 확인하여 long 진입할 시점이라고 판단될 경우, 하나의 조건을 더 확인한다.바로 내가 현재 분할매수를 얼만큼 들어갔는가에 대한 수량 확인이다. 여기서는 24번의 분할 매수를 하기 때문에, session.my_position(symbol=symbol)['result'][0]['size']를 통해 현재 매수한 수량을 확인하고 해당 값이 전체 매수 수량인 qty * 24 보다 작아야한다. my_position(symbol=symbol)['result']의 경우 0이 long 정보를, 1이 short 정보를 담고 있다. 이 조건도 만족할 경우 매수를 진행하고, stop loss를 설정한다. 결과적으로 분할 매수가 시작되며, stop loss는 분할 매수를 들어갈 때마다 갱신된다. 여기서는 3% 이상 하락할 경우 stop loss가 걸리도록 설정되어있다.
3. 주석 short 부분 : short의 경우 역시 동일하게 흘러간다. my_position을 호출할 때 index 번호가 1이라는 것에만 유의하자.
4. 주석 clear 부분 : now_order 상태가 None일 경우, 하나의 조건을 더 확인한다. 그것은 현재 진입되어져 있는 포지션 수량이며, 이 역시 충족할 경우 분할매도로 포지션을 정리하기 시작한다.

아래는 위 코드를 진행하기 위해 필요한 함수들이다.

def place_order(symbol, side, qty):
    try:
        session.place_active_order(
            symbol=symbol,
            side=side,
            order_type="Market",
            qty=round(qty, 3),
            time_in_force="GoodTillCancel",
            reduce_only=False,
            close_on_trigger=False,
        )
        print(now.strftime("%H:%M:%S"), f'에 {side} 진입')
    except:
        print(now.strftime("%H:%M:%S"), f'에 {side} 진입 과정에서 error 발생')
        

def replace_order(symbol, side, qty):
    try:
        session.place_active_order(
            symbol=symbol,
            side=side,
            order_type="Market",
            qty=round(qty, 3),
            time_in_force="GoodTillCancel",
            reduce_only=True,
            close_on_trigger=True,
        )
        if side == 'Buy':
            print(now.strftime("%H:%M:%S"), f'에 short 청산')
        else:
            print(now.strftime("%H:%M:%S"), f'에  long 청산')
    except:
        if side == 'Buy':
            print(now.strftime("%H:%M:%S"), f'에 short 청산 과정에서 error 발생')
        else:
            print(now.strftime("%H:%M:%S"), f'에 long 청산 과정에서 error 발생')


def set_stopLoss(symbol, side, leverage):
    if side == 'Buy':
        stop_loss = session.my_position(symbol=symbol)['result'][0]['entry_price']*(1-0.03/leverage)
    elif side == 'Sell':
        stop_loss = session.my_position(symbol=symbol)['result'][1]['entry_price']*(1+0.03/leverage)
    try:
        session.set_trading_stop(
            symbol=symbol,
            side=side,
            stop_loss=stop_loss
        )
        print(side, 'stop loss price :', stop_loss)
    except:
        pass

지금까지의 설명된 코드들을 정리하여 자동매매 프로그램을 만든다면 다음과 같다.

 

# =================================================================================================

# 필요 변수 정의.
symbol = "BTCUSDT"                               # 코인명
interval = 60                                    # 데이터 기준
limit = 200                                      # 데이터양
qty = 0.01                                       # 주문수량
fees = 0.00048                                   # 거래 수수료
now_order = None                                 # 포지션 변수
hour_list = [n for n in range(24)]               # 실행 시간
long_qty_list = [0 for n in range(24)]           # 실행 시간
short_qty_list = [0 for n in range(24)]          # 실행 시간

# 테스트넷 api 정보
session = usdt_perpetual.HTTP(
    endpoint="https://api-testnet.bybit.com", 
    api_key='my_api_key', 
    api_secret='my_api_secret',
)

set_margin_switch(symbol, True)          # 격리 마진 설정

# =================================================================================================

# 거래 시작
while True:
    # 현재 시간을 확인
    now = datetime.datetime.utcnow()
    if (now.hour in hour_list) & (59 <= now.minute < 60):
        
        df = data_reconstruction(symbol, interval, limit)
        df['ma5'] = df['close'].rolling(window=5).mean()
        df['ma15'] = df['close'].rolling(window=15).mean()
        df['ma30'] = df['close'].rolling(window=30).mean()
        # print(datetime.datetime.fromtimestamp(df['start_at'][len(df)-1]).strftime("%Y-%m-%dT%H:%M:%S"))
        # print('ma5 :', df['ma5'][len(df)-1], '| ma15 :', df['ma15'][len(df)-1], '| ma30 :', df['ma30'][len(df)-1])
        
        # =================================================================================================
        
        # get leverage
        if df['close'][len(df)-1] > df['ma15'][len(df)-1]:
            buy_leverage = 3
            sell_leverage = 1
        else:
            buy_leverage = 1
            sell_leverage = 3
        set_leverage(symbol, buy_leverage, sell_leverage)
        
        # get trade condition
        if (df['ma30'][len(df)-1] < df['close'][len(df)-1]) & (df['ma5'][len(df)-1] < df['close'][len(df)-1]):
            now_order = 'Buy'
        elif (df['ma30'][len(df)-1] > df['close'][len(df)-1]) & (df['ma5'][len(df)-1] > df['close'][len(df)-1]):
            now_order = 'Sell'
        else:
            now_order = None
        
        # long
        if now_order == 'Buy':
            if session.my_position(symbol=symbol)['result'][0]['size'] < 24*qty*buy_leverage:
                place_order(symbol, now_order, qty*buy_leverage)   # long 매수
                set_stopLoss(symbol, now_order, buy_leverage)    # long stop loss 설정
                long_qty_list[now.hour] = qty*buy_leverage
        
        # short
        if now_order == 'Sell':
            if session.my_position(symbol=symbol)['result'][1]['size'] < 24*qty*sell_leverage:
                place_order(symbol, now_order, qty*sell_leverage)   # short 매수
                set_stopLoss(symbol, now_order, sell_leverage)    # short stop loss 설정
                short_qty_list[now.hour] = qty*sell_leverage
        
        # clear
        if now_order == None:
            if session.my_position(symbol=symbol)['result'][0]['size'] >= 1*qty:
                replace_order(symbol, 'Sell', long_qty_list[now.hour])   # long 매도
            if session.my_position(symbol=symbol)['result'][1]['size'] >= 1*qty:
                replace_order(symbol, 'Buy', short_qty_list[now.hour])   # short 매도
        
        # =================================================================================================
    
        print('='*30)
    
    time.sleep(59)