This is an algorithmic trading bot I’ve been using on the Binance cryptocurrency exchange. So far I have only used it with Ethereum/USDT.

Link to full repo after the code.

   

cryptobot.py

   1 
   2 
    from binance.client import Client
   3 
    import requests
   4 
    import smtplib
   5 
    from email.message import EmailMessage
   6 
    import loguru
   7 
    import time
   8 
    import random
   9 
    import os
  10 
 
  11 
 
  12 
 
  13 
 
  14 
    class CryptoBot(Client):
  15 
       
  16 
       
  17 
       
  18 
 
  19 
        def __init__(self, currencies = None, email = None, **kwargs):
  20 
            super().__init__(**kwargs)
  21 
 
  22 
            self.currencies = currencies
  23 
            self.email = email
  24 
           
  25 
            self.pair = self.currencies["trading"] + self.currencies["wallet"]
  26 
 
  27 
            self.buy_in = {}
  28 
           
  29 
       
  30 
       
  31 
       
  32 
        @property
  33 
        def funds(self):
  34 
 
  35 
            complete_funds = round(float(self.get_asset_balance(
  36 
                asset=self.currencies["wallet"])["free"]), 2)
  37 
 
  38 
            if complete_funds:
  39 
                return complete_funds
  40 
 
  41 
 
  42 
       
  43 
       
  44 
        @property
  45 
        def liquidity_check(self):
  46 
 
  47 
            if not self.funds > 1.0:
  48 
 
  49 
                self.email["subject"] = (
  50 
                    "TRADING BALANCE TOO LOW")
  51 
               
  52 
                self.email["content"] = (
  53 
                    """Trading bot reports a balance of zero. """
  54 
                    """The bot will remain inactive until it """
  55 
                    """can access adequate funds.""")
  56 
 
  57 
                self.notifier()
  58 
 
  59 
                while True:
  60 
                   
  61 
                    if not self.funds > 1.0:
  62 
                        time.sleep(600)
  63 
                    else: 
  64 
                        break
  65 
 
  66 
                self.email["subject"] = (
  67 
                    "TRADING BALANCE INCREASED")
  68 
               
  69 
                self.email["content"] = (
  70 
                    """Trading bot reports it now has access to funds. """
  71 
                    """The bot will now resume trade.""")
  72 
 
  73 
                self.notifier()
  74 
 
  75 
                return False
  76 
 
  77 
            return True
  78 
    
  79 
       
  80 
       
  81 
 
  82 
        @property  
  83 
        def order_check(self):
  84 
 
  85 
 
  86 
            if self.get_open_orders():
  87 
 
  88 
                self.email["subject"] = (
  89 
                    "WAITING ON OPEN ORDERS")
  90 
               
  91 
                self.email["content"] = (
  92 
                    """Trading bot has detected open orders. """
  93 
                    """It will wait for all open orders to settle """
  94 
                    """before it resumes operating.""")
  95 
 
  96 
                self.notifier()
  97 
 
  98 
                while True:
  99 
 
 100 
                    if self.get_open_orders():
 101 
                        time.sleep(600)
 102 
                    else:
 103 
                        break
 104 
 
 105 
                self.email["subject"] = (
 106 
                    "OPEN ORDERS SETTLED. TRADING RESUMED")
 107 
               
 108 
                self.email["content"] = (
 109 
                    """Trading bot reports that all open orders """
 110 
                    """have been settled. It will now resume trade.""")
 111 
 
 112 
                self.notifier()
 113 
 
 114 
                return False
 115 
 
 116 
            return True
 117 
 
 118 
 
 119 
        @property
 120 
        def trend(self):
 121 
 
 122 
            candles = requests.get(
 123 
            f"https://api.binance.com/api/v3/klines?symbol={self.pair}&interval=1m&limit=60").json()
 124 
 
 125 
            old_price = float(candles[0][1])
 126 
            new_price = float(candles[-1][4])
 127 
 
 128 
            difference = (new_price - old_price) / old_price * 100
 129 
 
 130 
            return difference
 131 
 
 132 
 
 133 
 
 134 
 
 135 
        @property
 136 
        def circuit_breaker(self):
 137 
 
 138 
            if self.trend < -3.0:
 139 
 
 140 
                self.email["subject"] = (
 141 
                    "CIRCUIT BREAKER IMPOSED")
 142 
               
 143 
                self.email["content"] = (
 144 
                    """Circuit breaker imposed due to """
 145 
                    """market losses.""")
 146 
 
 147 
                self.notifier()
 148 
 
 149 
                while True:
 150 
 
 151 
                    if not price_movement > 2.0:
 152 
                        time.sleep(600)
 153 
                    else:
 154 
                        break
 155 
                   
 156 
                self.email["subject"] = (
 157 
                            "CIRCUIT BREAKER LIFTED")
 158 
                       
 159 
                self.email["content"] = (
 160 
                    """Circuit breaker lifted due to """
 161 
                    """price recovery.""")
 162 
 
 163 
                self.notifier()
 164 
 
 165 
                return False
 166 
 
 167 
            return True
 168 
 
 169 
       
 170 
 
 171 
        def buy(self):
 172 
           
 173 
            rate = round(float(self.get_symbol_ticker(symbol = self.pair)["price"]), 2)
 174 
            units = round(self.funds / rate - 0.0001, 5)
 175 
 
 176 
            self.buy_in = self.order_limit_buy(
 177 
                symbol = self.pair,
 178 
                price = str(rate),
 179 
                quantity = str(units))
 180 
 
 181 
           
 182 
            while True:
 183 
 
 184 
                if not self.get_order(
 185 
                    orderId = self.buy_in["orderId"],
 186 
                    symbol = self.pair)["status"] == "FILLED":
 187 
 
 188 
                    time.sleep(30)
 189 
               
 190 
                else:
 191 
                    break
 192 
 
 193 
 
 194 
 
 195 
        def sell(self):
 196 
           
 197 
            time.sleep(5)
 198 
           
 199 
            cost = round(float(self.buy_in["price"]) * 1.015, 2)
 200 
 
 201 
            cashout = self.order_limit_sell(
 202 
                symbol = self.buy_in["symbol"],
 203 
                quantity = round(float(self.buy_in["origQty"]) - 0.0001, 5),
 204 
                price = str(cost))
 205 
 
 206 
            while True:
 207 
 
 208 
                if not self.get_order(
 209 
                    orderId = cashout["orderId"],
 210 
                    symbol = cashout["symbol"]) == "FILLED":
 211 
 
 212 
                    time.sleep(30)
 213 
                else:
 214 
                    break
 215 
 
 216 
            self.buy_in = {}
 217 
 
 218 
    
 219 
       
 220 
 
 221 
        def trade(self):
 222 
 
 223 
            while True:
 224 
                if all([
 225 
                    self.order_check,
 226 
                    self.liquidity_check,
 227 
                    self.circuit_breaker]):
 228 
                   
 229 
                    self.buy()
 230 
                    self.sell()
 231 
                
 232 
 
 233 
 
 234 
 
 235 
        def notifier(self):
 236 
 
 237 
            notification = EmailMessage()
 238 
           
 239 
            notification["To"] = self.email["destination"]
 240 
            notification["From"] = self.email["origin"]
 241 
            notification["Subject"] = self.email["subject"]
 242 
            notification.set_content(self.email["content"])
 243 
           
 244 
            with smtplib.SMTP_SSL("smtp.gmail.com") as mailboy:
 245 
                mailboy.login(self.email["username"], self.email["password"])
 246 
                mailboy.send_message(notification)
 247 
 
 248 
 
 249 
 
 250 
 
 251 
        def event_logger(self, event):
 252 
           
 253 
            logger = loguru.logger
 254 
 
 255 
            logger.add(
 256 
                sink = "events.log",
 257 
                level = "WARNING",
 258 
                format = """\n\n\n\n{level} {time: {time:DD-MM-YYYY HH:mm:ss}}\n"""
 259 
                         """Elapsed Time: {elapsed}\n"""
 260 
                         """File: {file}\n"""
 261 
                         """Message: {message}""")
 262 
           
 263 
            logger.exception(event)
 264 
           
 265 
            self.email["subject"] = "EXCEPTION ENCOUNTERED"
 266 
            self.email["content"] = str(event)
 267 
 
 268 
            self.notifier()
 269 
 
 270 
 
 271 
 
 272 
 
 273 
    if __name__ == "__main__":
 274 
       
 275 
 
 276 
        key = os.environ.get("BINANCE_KEY")
 277 
 
 278 
        secret = os.environ.get("BINANCE_SECRET")
 279 
       
 280 
        ethereum = CryptoBot(
 281 
            api_key = key,
 282 
            api_secret = secret,
 283 
            currencies = {
 284 
                "wallet": "USDT",
 285 
                "trading": "ETH"},
 286 
            email = {
 287 
                "origin": os.environ.get("EMAIL_ORIG"),
 288 
                "destination": os.environ.get("EMAIL_DEST"),
 289 
                "subject": None,
 290 
                "content": None,
 291 
                "username": os.environ.get("EMAIL_USER"),
 292 
                "password": os.environ.get("EMAIL_PASS")})
 293 
           
 294 
      
 295 
        while True:
 296 
           
 297 
            try:
 298 
               
 299 
                ethereum.trade()
 300 
           
 301 
            except Exception as event:
 302 
               
 303 
                ethereum.event_logger(event)
 304 
                time.sleep(600)
 305 
           
 306 
           
 307 
           
 308 
           
 309 
           

   

Full code repo available on Github

 

Related posts:

Speculative Fuction Bot

EXP-RTL: Exponential Retaliation In Iterated Prisoner’s Dilemma Games

Interleaved Neighborhood Algorithm: Fully Exploratory Optimization

 

About Me