바이비트와 파이썬을 이용한 자동매매 프로그램 - 3. 삼일고 전략

이번 글에서는 "삼일고 전략" 을 이용해보겠습니다. 해당 전략은 다른 글인 < 파이썬을 이용한 매우 간단한 백테스팅 > 에서 다루었던 전략으로, 수학적 논리는 전혀 없으나 간단한 구현을 통해 살을 붙이고 때는 것에 유용할 것이라 생각하여 선택하였습니다.

 

https://tfrecord.tistory.com/entry/%ED%8C%8C%EC%9D%B4%EC%8D%AC%EC%9D%80-%EC%9D%B4%EC%9A%A9%ED%95%9C-%EB%A7%A4%EC%9A%B0-%EA%B0%84%EB%8B%A8%ED%95%9C-%EB%B0%B1%ED%85%8C%EC%8A%A4%ED%8C%85

파이썬을 이용한 매우 간단한 백테스팅

 

전략은 다음과 같이 진행됩니다:

1. 시장 데이터를 불러와 지난 2일간의 종가와 5일 이동평균선, 그리고 현재가를 비교합니다. 지난 2일 종가보다 현재가가 높으며, 5일 이동평균선보다도 현재가가 높다면 long에 진입합니다.
2. 위 조건과 반대의 경우 short에 진입합니다.
3. 24시간, 즉 하루가 지나면 위 조건이 여전한지 확인하고, 더 이상 위 조건을 충족시키지 못할 경우 매도합니다.

 

---

 

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
import gc

 

---

 

2. 필요 함수 정의

 

def get_session(net='test'):
    if net == 'test':
        session = usdt_perpetual.HTTP(
            endpoint="https://api-testnet.bybit.com", 
            api_key='my_api_key', 
            api_secret='my_secret_key')
    elif net == 'main':
        session = usdt_perpetual.HTTP(
            endpoint="https://api.bybit.com", 
            api_key='my_api_key',
            api_secret='my_secret_key')
    return session


def get_qty(session, symbol):
    usdt_balance = session.get_wallet_balance(coin='USDT')['result']['USDT']['available_balance']
    cur_price = session.public_trading_records(symbol=symbol,limit=1)['result'][0]['price']
    amount = math.floor((usdt_balance * 1000000)/cur_price) / 1000000
    return amount


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_leverage(symbol, buy_leverage, sell_leverage):
    try:
        session.set_leverage(
            symbol=symbol,
            buy_leverage=buy_leverage,
            sell_leverage=sell_leverage
        )
    except:
        pass


def set_margin_switch(symbol, is_isolated=True):
    try:
        session.cross_isolated_margin_switch(
            symbol=symbol,
            is_isolated=is_isolated,
        )
    except:
        pass
    
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


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(3):
        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 : 코인 가격에 대한 데이터를 불러오기 위해 정의한 함수입니다.
• data_reconstruction : 한시간 단위로 일봉 데이터를 생성하기 위한 함수입니다.
• cal_amount : BTC <-> USDT 단위 변환 함수입니다. 주문을 할 경우 BTC 단위로 주문을 하는데, 계좌 잔고의 경우 USDT 단위로 되어 있습니다. 따라서 USDT 단위를 BTC 단위로 변환해주는 과정이 필요합니다.

 

---

 

3. Trading

 

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

# 필요 변수 정의.
session = get_session('test')                               # session 호출
symbol = "BTCUSDT"                                          # 코인명
interval = 60                                               # 데이터 기준
limit = 200                                                 # 데이터양
hour_split = 8                                              # 매수 매도 분할 수
fees = 0.00048                                              # 수수료
now_order = None                                            # 현재 포지션 상태
hour_list = [n for n in range(24)]                          # 실행 시간
long_clear_available = [False for n in range(24)]           # 매도 가능 여부
short_clear_available = [False for n in range(24)]          # 매도 가능 여부
long_qty_list = [0 for n in range(24)]
short_qty_list = [0 for n in range(24)]

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()
        
        # =================================================================================================
        
        # 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['close'][len(df)-1] > df['close'][len(df)-2]) & (df['close'][len(df)-1] > df['close'][len(df)-3]) & (df['ma5'][len(df)-1] < df['close'][len(df)-1]):
            now_order = 'Buy'
        elif (df['close'][len(df)-1] < df['close'][len(df)-2]) & (df['close'][len(df)-1] < df['close'][len(df)-3]) & (df['ma5'][len(df)-1] > df['close'][len(df)-1]):
            now_order = 'Sell'
        else:
            now_order = None
        
        # get qty
        long_size = session.my_position(symbol=symbol)['result'][0]['size']
        short_size = session.my_position(symbol=symbol)['result'][1]['size']
        if long_size + short_size == 0:
            qty = get_qty(session, symbol)/(hour_split+1)
        
        =================================================================================================
        
        # long
        if now_order == 'Buy':
            if long_size < hour_split*qty*buy_leverage:
                place_order(symbol, now_order, qty*buy_leverage)   # long 매수
                set_stopLoss(symbol, now_order, buy_leverage)    # long stop loss 설정
                long_clear_available[now.hour] = True
                long_qty_list[now.hour] = qty
        
        # short
        if now_order == 'Sell':
            if short_size < hour_split*qty*sell_leverage:
                place_order(symbol, now_order, qty*sell_leverage)   # short 매수
                set_stopLoss(symbol, now_order, 1)    # short stop loss 설정
                short_clear_available[now.hour] = True
                short_qty_list[now.hour] = qty
        
        # clear
        if now_order == None:
            if long_size >= qty*buy_leverage:
                if long_clear_available[now.hour]:
                    replace_order(symbol, 'Sell', long_qty_list[now.hour])   # long 매도
                    long_clear_available[now.hour] = False
            if short_size >= qty*sell_leverage:
                if short_clear_available[now.hour]:
                    replace_order(symbol, 'Buy', short_qty_list[now.hour])   # short 매도
                    short_clear_available[now.hour] = False
        
        # =================================================================================================
    
        print('='*30)
        
        del df
        del buy_leverage
        del sell_leverage
        del long_size
        del short_size
        gc.collect()
    
    time.sleep(59)