# Python Binance API code for calculating ATR (WMA) and set Chandelier Trailing Stop Loss (SL) / Take Profit (TP) for an opened CCY pair order (e.g. BTCUSDT) # ver 0.10d (Date: 20211016) # import os import numpy as np import pandas_datareader as pdr import datetime as dt import pandas as pd import smtplib, ssl import schedule import time from time import sleep from datetime import datetime from binance.client import Client #from binance import ThreadedWebsocketManager from binance.enums import * from binance.exceptions import BinanceAPIException, BinanceOrderException, BinanceRequestException from requests import exceptions from twilio.rest import Client as Twilio_client from dotenv import load_dotenv ##### set input parameters ##### period = 22 # period for ATR averaging mult = 4.5 # multiplier for ATR (BTC: 4.5; ETH: 4) cal_type = "close" # use bar "high" or "close" value for calculation #cal_type = "high" # use bar "high" or "close" value for calculation lotsize = 0.029 # set order quantity CCY = 'BTCBUSD' # set the trading coin pair (e.g. BTCUSDT, BNBUSDT, ETHUSDT, etc.) server = 'live' # set connection to "test" or "live" server filename = 'OCO_orderID_'+CCY+'.txt' # define filename for orderID record retry_time = 10 # time (sec) for retry after exception recWin = 5000 # set command recvWindow time ### take environment variables from .env load_dotenv() ### set binance API keys if server == 'live': api_key = os.environ.get('binance_api') api_secret = os.environ.get('binance_secret') connect_url = 'https://api.binance.com/api' elif server == 'test': api_key = os.environ.get('binance_api_test') api_secret = os.environ.get('binance_secret_test') connect_url = 'https://testnet.binance.vision/api' ### set Twilio SMS settings account_sid = os.environ.get('twi_acct_sid') auth_token = os.environ.get('twi_auth_token') twi_number = os.environ.get('twi_phone') cell_number = os.environ.get('twi_cell') twi_client = Twilio_client(account_sid, auth_token) print('Twilio phone setting: ', twi_number, cell_number) ### Setting for email alert port = 465 # For SSL smtp_server = os.environ.get('em_smtp_server') sender_email = os.environ.get('em_sender_email') receiver_email = os.environ.get('em_receiver_email') #password = input("Type your password and press enter: ") password = os.environ.get('em_password') message = """\ Subject: Binance Exception This message is sent from Python: """ context = ssl.create_default_context() print('Alert Email setting: ', receiver_email) start_time = datetime.now() ### Function: Send sms/email alert def send_alert(msg, new_alert): print(msg) if (new_alert): print('Send SMS/Email') twi_client.messages.create(from_=twi_number, to=cell_number, body=msg) with smtplib.SMTP_SSL(smtp_server, port, context=context) as server: server.login(sender_email, password) server.sendmail(sender_email, receiver_email, message+msg) ### Function: WMA calculation def wma_calc(w): def g(x): return sum(w * x) / sum(w) return g ### Set BinanceAPI connection keys connected_flag=False new_alert_flag = True while(not connected_flag): try: client = Client(api_key, api_secret, {"timeout":30}) except: send_alert('BinanceAPI Client Connect keys Error!', new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue connected_flag=True new_alert_flag=True ### Set connection to live or testnet account client.API_URL = connect_url ### Flag for initial running first_run = True cur_orderID1 = 0 cur_orderID2 = 0 ### Check price and update while True: ## try: ## info = client.get_account(recvWindow=recWin) ## except BinanceAPIException as e: ## send_alert('BinanceAPIException alert (get_account): '+e.message, new_alert_flag) ## time.sleep(retry_time) ## new_alert_flag = False ## continue ## except BinanceRequestException as e: ## send_alert('BinanceRequestException alert (get_account): '+e.message, new_alert_flag) ## time.sleep(retry_time) ## new_alert_flag = False ## continue ## except: ## send_alert('BinanceAPI get_account error', new_alert_flag) ## time.sleep(retry_time) ## new_alert_flag = False ## continue ## ## new_alert_flag = True if (not first_run): ### check status of opened order try: prev_order1 = client.get_order(symbol=CCY, orderId=cur_orderID1, recvWindow=recWin) except: send_alert('BinanceAPI get_order error (prev_order1)', new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue new_alert_flag = True try: prev_order2 = client.get_order(symbol=CCY, orderId=cur_orderID2, recvWindow=recWin) except: send_alert('BinanceAPI get_order error (prev_order2)', new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue new_alert_flag = True ### if order filled -> quit the program, else if order still exist -> cancel the order if prev_order1['status'] == 'FILLED' or prev_order2['status'] == 'FILLED': ### Send SMS & Email alert #twi_client.messages.create(from_=twi_number, to=cell_number, body='The BinanceAPI '+CCY+' SL/TP order is executed') send_alert('The BinanceAPI '+CCY+' SL/TP order is executed', True) exit() elif prev_order1['status'] == 'NEW' and prev_order2['status'] == 'NEW': cancel = client.cancel_order(symbol=CCY, orderId=cur_orderID1) ### Cancel all existing open orders if program is first run ### Check existing open orders if (first_run): if os.path.isfile('./'+filename): # check if the orderID record file exist f = open(filename,"r") try: prev_order1 = client.get_order(symbol=CCY, orderId=f.read(), recvWindow=recWin) except: send_alert('BinanceAPI get_order error (prev_order1)', new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue new_alert_flag = True print(prev_order1) if prev_order1['status'] == 'NEW': cancel = client.cancel_order(symbol=CCY, orderId=prev_order1['orderId'], recvWindow=recWin) ## exist_order = client.get_open_orders(symbol=CCY, recvWindow=recWin) ## print('First run existing ',CCY,' orders: ',exist_order) ## while (exist_order != []): ## cancel = client.cancel_order(symbol=CCY, orderId=exist_order[0]['orderId'], recvWindow=recWin) ## exist_order = client.get_open_orders(symbol=CCY, recvWindow=recWin) ### Fetch previous D1 K-lines history try: bars = client.get_historical_klines(CCY, '1d', "100 days ago UTC") except: send_alert('BinanceAPI get_historical_klines error', new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue new_alert_falg = True for line in bars: del line[5:] btc_df = pd.DataFrame(bars, columns=['date', 'open', 'high', 'low', 'close']) btc_df = btc_df.astype({"open":float,"high":float,"low":float,"close":float,}) btc_df.set_index('date', inplace=True) btc_df.index = pd.to_datetime(btc_df.index, unit='ms') ### ATR calculation high_low = btc_df['high'] - btc_df['low'] high_close = np.abs(btc_df['high'] - btc_df['close'].shift()) low_close = np.abs(btc_df['low'] - btc_df['close'].shift()) ranges = pd.concat([high_low, high_close, low_close], axis=1) true_range = np.max(ranges, axis=1) ### calcualte atr(period) in WMA for testing # atr = true_range.rolling(22).mean() weights = list(reversed([(period - n) * period for n in range(period)])) atr = true_range.rolling(window = period).apply(wma_calc(weights), raw=True) atr_tp = true_range.rolling(window = period).apply(wma_calc(weights), raw=True) ### Last day ATR value #print(atr[-1]) ### calculate TrailingSL & TrailingTP values (e.g. 3x ATR of previous 22 day high (or close)) if cal_type=="close": btc_df['PeriodMax'] = btc_df.close.rolling(period).max() btc_df['PeriodMin'] = btc_df.close.rolling(period).min() btc_df['day_sl'] = btc_df.PeriodMax - mult*atr btc_df['day_tp'] = ((btc_df.PeriodMax + mult*atr) + (btc_df.PeriodMin + mult*atr))/2 btc_df['final_sl'] = btc_df.day_sl.rolling(period).max() btc_df['final_tp'] = btc_df.day_tp.rolling(period).max() elif cal_type=="high": btc_df['PeriodMax'] = btc_df.high.rolling(period).max() btc_df['PeriodMin'] = btc_df.low.rolling(period).min() btc_df['day_sl'] = btc_df.PeriodMax - mult*atr btc_df['day_tp'] = ((btc_df.PeriodMax + mult*atr) + (btc_df.PeriodMin + mult*atr))/2 btc_df['final_sl'] = btc_df.day_sl.rolling(period).max() btc_df['final_tp'] = btc_df.day_tp.rolling(period).max() ### if new trailingSL value is lower, keep the old value #long_sl = max(long_sl, long_sl_prev) #long_tp = max(long_tp, long_sl_prev) print(btc_df) ### Get the last day TP and SL values long_sl = btc_df.final_sl[-1] long_tp = btc_df.final_tp[-1] #print(long_sl, long_tp) ### get symbol info from exchange (Binance) for checking the range of allowable order values try: s_info = client.get_symbol_info(CCY) except BinanceAPIException as e: send_alert('BinanceAPIException alert (get_symbol_info): '+e.message, new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue except BinanceRequestException as e: send_alert('BinanceRequestException alert (get_symbol_info): '+e.message, new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue # except: # send_alert('BinanceAPI get_symbol_info error', new_alert_flag) # time.sleep(retry_time) # new_alert_flag = False # continue new_alert_flag = True s_info_filter = s_info.get('filters') multiUp = float(s_info_filter[1].get('multiplierUp')) multiDn = float(s_info_filter[1].get('multiplierDown')) try: s_price = client.get_symbol_ticker(symbol=CCY) except BinanceAPIException as e: send_alert('BinanceAPIException alert (get_symbol_ticker): '+e.message, new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue except BinanceRequestException as e: send_alert('BinanceRequestException alert (get_symbol_ticker): '+e.message, new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue # except: # send_alert('BinanceAPI get_symbol_ticker error', new_alert_flag) # time.sleep(retry_time) # new_alert_flag = False # continue new_alert_flag = True exchange_max = float(s_price['price'])*multiUp exchange_min = float(s_price['price'])*multiDn ### set SL and TP value within system limits sl = int(max(long_sl, exchange_min*1.05)) tp = int(min(long_tp, exchange_max*0.95)) print('Current',CCY,' $:',btc_df.close[-1]) print('SL & TP $:', sl, tp) ### Place OCO order for trailing SL try: order = client.create_oco_order(symbol=CCY,side='SELL',quantity=lotsize,price=tp,stopPrice=sl,stopLimitPrice=sl,stopLimitTimeInForce='GTC', recvWindow=recWin) ### error handling goes here except BinanceAPIException as e: send_alert('BinanceAPIException alert (place_order): '+e.message, new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue #print(e) #twi_client.messages.create(from_=twi_number, to=cell_number, body='BinanceAPIException alert: '+e.message) # with smtplib.SMTP_SSL(smtp_server, port, context=context) as server: # server.login(sender_email, password) # server.sendmail(sender_email, receiver_email, message+e.message) ### error handling goes here except BinanceOrderException as e: send_alert('BinanceOrderException alert (place_order): '+e.message, new_alert_flag) time.sleep(retry_time) new_alert_flag = False continue #print(e) #twi_client.messages.create(from_=twi_number, to=cell_number, body='BinanceOrderException alert: '+e.message) # with smtplib.SMTP_SSL(smtp_server, port, context=context) as server: # server.login(sender_email, password) # server.sendmail(sender_email, receiver_email, message+e.message) new_alert_flag = True cur_orderID1 = order['orders'][0]['orderId'] cur_orderID2 = order['orders'][1]['orderId'] f = open(filename,"w") f.write(str(cur_orderID1)) f.close() print('Opened OCO OrderID: ',cur_orderID1, cur_orderID2) print('Opened OCO orders:\n',client.get_open_orders(symbol=CCY)) first_run = False print("Current Time:", datetime.now(), "\n(Start Time:",start_time,")") time.sleep(3600)