simple python trading bot using phemex exchange
import ccxt
import config
import time
import schedule
import pandas as pd
import logging
# Create a custom logger
logger = logging.getLogger('trading_bot')
logger.setLevel(logging.INFO)
# File handler
file_handler = logging.FileHandler('trading_bot.log')
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
# Stream (console) handler
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.INFO)
stream_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
# Add handlers to the custom logger
logger.addHandler(file_handler)
logger.addHandler(stream_handler)
def initialize_phemex():
try:
phemex_client = ccxt.phemex({
'enableRateLimit': True,
'apiKey': config.PHEMEX_API_KEY,
'secret': config.PHEMEX_API_SECRET
})
logger.info("Phemex client initialized successfully.")
return phemex_client
except Exception as e:
logger.error(f"Error initializing Phemex client: {e}")
return None
# Initialize phemex globally
phemex = initialize_phemex()
if phemex is None:
logger.error("Failed to initialize Phemex client. Exiting.")
exit()
symbol = 'uBTCUSD'
size = 1
bid = 29000
params = {'timeInForce': 'PostOnly',}
def open_positions(symbol=symbol):
try:
if symbol == 'uBTCUSD':
index_pos = 4
elif symbol == 'APEUSD':
index_pos = 2
elif symbol == 'ETHUSD':
index_pos = 3
elif symbol == 'DOGEUSD':
index_pos = 1
elif symbol == 'u100000SHIBUSD':
index_pos = 0
else:
index_pos = None
params = {'type': 'swap', 'code': 'USD'}
phe_bal = phemex.fetch_balance(params=params)
open_positions_data = phe_bal['info']['data']['positions']
openpos_side = open_positions_data[index_pos]['side']
openpos_size = open_positions_data[index_pos]['size']
openpos_bool = openpos_side in ('Buy', 'Sell')
long = openpos_side == 'Buy'
logger.info(f'Open positions checked for {symbol}.')
return open_positions_data, openpos_bool, openpos_size, long, index_pos
except Exception as e:
logger.error(f"Error in open_positions for {symbol}: {e}")
return None, False, 0, None, None
def ask_bid(symbol=symbol):
try:
ob = phemex.fetch_order_book(symbol)
bid = ob['bids'][0][0]
ask = ob['asks'][0][0]
logger.info(f'Ask and bid fetched for {symbol}.')
return ask, bid
except Exception as e:
logger.error(f"Error in ask_bid for {symbol}: {e}")
return None, None
def kill_switch(symbol=symbol):
try:
logger.info(f'Starting the kill switch for {symbol}')
openposi = open_positions(symbol)[1]
long = open_positions(symbol)[3]
kill_size = open_positions(symbol)[2]
while openposi:
phemex.cancel_all_orders(symbol)
openposi = open_positions(symbol)[1]
long = open_positions(symbol)[3]
kill_size = open_positions(symbol)[2]
kill_size = int(kill_size)
ask, bid = ask_bid(symbol)
if not long:
phemex.create_limit_buy_order(symbol, kill_size, bid, params)
elif long:
phemex.create_limit_sell_order(symbol, kill_size, ask, params)
time.sleep(30)
openposi = open_positions(symbol)[1]
logger.info(f'Kill switch executed for {symbol}')
except Exception as e:
logger.error(f"Error in kill_switch for {symbol}: {e}")
def pnl_close(symbol=symbol, target=9, max_loss=-8):
try:
pos_dict = phemex.fetch_positions({'type':"swap", 'code':'USD'})
index_pos = open_positions(symbol)[4]
if index_pos is None or index_pos >= len(pos_dict):
logger.error(f"Invalid position index for {symbol}")
return False, False, 0, False
pos_data = pos_dict[index_pos]
side = pos_data['side']
size = pos_data['contracts']
entry_price = float(pos_data['entryPrice'])
leverage = float(pos_data['leverage'])
current_price = ask_bid(symbol)[1]
logger.info(f"Position data for {symbol}: side={side}, size={size}, entry_price={entry_price}, leverage={leverage}, current_price={current_price}")
if entry_price == 0:
logger.error("Entry price is zero, cannot calculate PnL.")
return False, False, 0, False
pnl_diff = (current_price - entry_price) if side == 'long' else (entry_price - current_price)
perc = round((pnl_diff / entry_price) * leverage, 10) * 100
pnlclose, in_pos = False, False
if perc > 0:
in_pos = True
if perc > target:
logger.info(f"{symbol} in profit, target reached. Executing kill switch.")
pnlclose = True
kill_switch(symbol)
else:
logger.info(f"{symbol} in profit, but target not yet reached.")
elif perc < 0:
in_pos = True
if perc <= max_loss:
logger.info(f"{symbol} at max loss, executing kill switch.")
kill_switch(symbol)
else:
logger.info(f"{symbol} in loss, but within acceptable range.")
else:
logger.info(f"{symbol} not in position.")
return pnlclose, in_pos, size, side == 'long'
except Exception as e:
logger.error(f"Error in pnl_close for {symbol}: {e}")
return False, False, 0, False
def size_kill():
try:
max_risk = 1000
all_phe_balance = phemex.fetch_balance({'type':"swap", "code": "USD"})
open_positions_data = all_phe_balance['info']['data']['positions']
pos_cost = float(open_positions_data[0]['posCost']) if open_positions_data else 0
if pos_cost > max_risk:
kill_switch(symbol)
time.sleep(30)
logger.info('Size kill checked')
except Exception as e:
logger.error(f"Error in size_kill: {e}")
def main():
try:
logger.info("Starting main execution loop")
schedule.every(1).minutes.do(lambda: pnl_close(symbol, 9, -8))
schedule.every(5).minutes.do(lambda: size_kill())
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
logger.info("Execution interrupted by user")
except Exception as e:
logger.error(f"Unexpected error in main execution loop: {e}")
finally:
logger.info("Main execution loop ended")
if __name__ == "__main__":
main()