用Python开发数字货币交易机器人（附源码）

众所周知，币圈一天，人间一年。我们进行数字货币交易时，在交易所 APP 或者网站盯盘并手动下单非常耗时，当币价波动非常大的时候往往会错失良机。这时我们可以创建一个简单的 telegram 交易机器人，来帮助我们进行做空和做多交易。

该机器人可以实现以下功能：

做空交易 - 以指定的价格卖出持有货币并在价格下跌时回购
做多交易 - 指定的价格购买货币并在价格上涨时卖出
列出交易订单
显示可用余额

设置 Telegram 机器人

首先需要一个 Telegram 账号，如果没有的话请自己注册一个。然后与BotFather进行对话，通过输入/newbot来新建一个telegram机器人，根据指示一步步创建并记住你的token。

用Python开发数字货币交易机器人（附源码）

获取交易所的 API keys

查找你的交易所API文档，看看如何获取对订单和账户余额的访问权限和步骤，记住你的密码和API keys。本例中我们以bitfinex为例，Bitmex交易所是目前市面上交易量最大的比特币期货交易所，交易量和交易深度非常大。

用Python开发数字货币交易机器人（附源码）

安装依赖包

我们这边用的是Python 3.6版本，同时我们还需要利用CCXT框架获取Bitmex交易所数据，CCXT是一个JavaScript / Python / PHP 开发库，用于数字货币的交易，支持众多的比特币/以太币/山寨币交易市场和交易所API。

CCXT库用于连接数字货币交易所并在世界范围内进行交易和支付处理。使用 ccxt可以快速访问数字货币市场数据，可以用于存储、分析、可视化、指标开发、量化交易、策略回溯测试、交易机器人程序以及相关的软件工程。

然后我们将使用python-telegram-bot与Telegram进行通讯，对聊天消息做出反应并进行交易。

只需要用下面方法安装以上两个依赖包：

pip install python-telegram-bot ccxt

我们需要交易机器人实现的基本类功能：

1、获取交易所概况，允许创建订单，列出订单详情并获取余额。这将是以 ccxt 实现的装饰器。

2、交易执行者，因为我们希望自动执行做空和做多交易。

3、即时响应的telegram 机器人。

编写机器人

项目结构如下：

main.py  


\config  


\core  


\model   

\util

我们将从一个简单的模型开始。因为多空交易两者有很多共同点，可以在\ model中创建一个基类TradeDetails：

import abc  


class TradeDetails(metaclass=abc.ABCMeta):  


    def __init__(self, start_price: float, symbol: str, amount: float, currency: str = "USD"):  


        self.start_price = start_price  


        self.symbol = symbol.upper()  


        self.amount = amount  


        self.currency = currency  


    @property  


    def exchange_symbol(self):  


        return f"{self.symbol.upper()}/{self.currency}"  


    @property  

    @abc.abstractmethod  

    def exit_price(self):  

        pass 

    def __str__(self) -> str:  

        return f"order for {self.amount} {self.exchange_symbol} with enter price: {self.start_price:.5}, " \  

               f"exit_price: {self.exit_price:.5}"

具体的为：

LongTrade

from fasttrade.model.trade import TradeDetails  


class LongTrade(TradeDetails):  


    def __init__(self, start_price: float, symbol: str, amount: float, percent_change: float = 0.5,  


                 currency: str = "USD") -> None:  


        super().__init__(start_price, symbol, amount, currency)  


        self.end_price = start_price * (1 + percent_change / 100)  


    @property  


    def exit_price(self):  


        return self.end_price  


    def __str__(self) -> str:  

        return "Long " + super().__str__()

ShortTrade

from fasttrade.model.trade import TradeDetails  


class ShortTrade(TradeDetails):  


    def __init__(self, start_price: float, symbol: str, amount: float, percent_change: float = 0.5,  


                 currency: str = "USD") -> None:  

        super().__init__(start_price, symbol, amount, currency)  

        self.end_price = start_price * (1 - percent_change / 100)  


    @property  

    def exit_price(self):  


        return self.end_price  


    def __str__(self) -> str:  

        return "Short " + super().__str__()

接下来是获取交易所数据：

from ccxt import Exchange, OrderNotFound  


class CryptoExchange:  


    def __init__(self, exchange: Exchange):  


        self.exchange = exchange  


        self.exchange.load_markets()  


    @property  


    def free_balance(self):  


        balance = self.exchange.fetch_free_balance()  


        # surprisingly there are balances with 0, so we need to filter these out  


        return {k: v for k, v in balance.items() if v > 0}  


    def fetch_open_orders(self, symbol: str = None):  


        return self.exchange.fetch_open_orders(symbolsymbol=symbol)  


    def fetch_order(self, order_id: int):  


        return self.exchange.fetch_order(order_id)  


    def cancel_order(self, order_id: int):  


        try:  


            self.exchange.cancel_order(order_id)  


        except OrderNotFound:  


            # treat as success  

            pass 

    def create_sell_order(self, symbol: str, amount: float, price: float):  


        return self.exchange.create_order(symbolsymbol=symbol, type="limit", side="sell", amountamount=amount, priceprice=price)  


    def create_buy_order(self, symbol: str, amount: float, price: float):  

        return self.exchange.create_order(symbolsymbol=symbol, type="limit", side="buy", amountamount=amount, priceprice=price)

然后，我们将执行交易程序。程序将接受交易所数据和超时情况以检查订单是否完成。当做空时，我们以设定的价格卖出，当价格下降到一定水平时回购。我们使用asyncio协程进行编码，以使等待不会阻塞：

import asyncio  


import logging  


from ccxt import ExchangeError  


from model.longtrade import LongTrade  

from model.shorttrade import ShortTrade  


class TradeExecutor:  


    def __init__(self, exchange, check_timeout: int = 15):  


        self.check_timeout = check_timeout  


        self.exchange = exchange  


    async def execute_trade(self, trade):  


        if isinstance(trade, ShortTrade):  


            await self.execute_short_trade(trade)  


        elif isinstance(trade, LongTrade):  


            await self.execute_long_trade(trade)  


    async def execute_short_trade(self, trade: ShortTrade):  


        sell_price = trade.start_price  


        buy_price = trade.exit_price  


        symbol = trade.exchange_symbol  


        amount = trade.amount  


        order = self.exchange.create_sell_order(symbol, amount, sell_price)  


        logging.info(f'Opened sell order: {amount} of {symbol}. Target sell {sell_price}, buy price {buy_price}')  


        await self._wait_order_complete(order['id'])  


        # post buy order  

        order = self.exchange.create_buy_order(symbol, amount, buy_price)  


        await self._wait_order_complete(order['id'])  


        logging.info(f'Completed short trade: {amount} of {symbol}. Sold at {sell_price} and bought at {buy_price}')  


    async def execute_long_trade(self, trade: LongTrade):  

        buy_price = trade.start_price 

         sell_price = trade.exit_price  

        symbol = trade.exchange_symbol  


        amount = trade.amount  


        order = self.exchange.create_buy_order(symbol, amount, buy_price)  


        logging.info(f'Opened long trade: {amount} of {symbol}. Target buy {buy_price}, sell price {sell_price}')  


        await self._wait_order_complete(order.id)  


        # post sell order  


        order = self.exchange.create_sell_order(symbol, amount, sell_price)  


        await self._wait_order_complete(order.id)  

        logging.info(f'Completed long trade: {amount} of {symbol}. Bought at {buy_price} and sold at {sell_price}')  

    async def _wait_order_complete(self, order_id): 

        status = 'open'  

        while status is 'open':  


            await asyncio.sleep(self.check_timeout)  


            order = self.exchange.fetch_order(order_id)  


            status = order['status']  


        logging.info(f'Finished order {order_id} with {status} status')  


        # do not proceed further if we canceled order  


        if status == 'canceled':  

            raise ExchangeError('Trade has been canceled')

ccxt使用REST API进行数据传输。它不如某些交易所支持的WebSockets快，但是对于这个简单的机器人来说，速度或许差别。

async def _wait_order_complete(self, order_id):  


        status = 'open'  

        order = None  

        while status is 'open':  

            await asyncio.sleep(self.check_timeout)  


            order = self.exchange.fetch_order(order_id)  


            status = order['status']  

        logging.info(f'Finished order {order_id} with {status} status') 

        # do not proceed further if we canceled order  

        if status == 'canceled':  


            raise ExchangeError('Trade has been canceled')  

        return order

接下来将创建Telegram机器人，这是最有难度的部分，我们将使其拥有以下指令：

1、列出/取消有效订单

2、显示可用余额

3、建立做多或做空交易

我们还需要对机器人做一些安全限制，使其仅对你的消息做出响应，而其他人则无法使用你的帐户进行交易。

主要是进行做多和做空交易的部分：

1、选择做空或者做多

2、输入数字货币品种

3、输入交易数量

4、所占百分比

5、每个价格

6、显示确认信息

7、显示最终交易信息

我们来创建telegrambot.py并添加以下常量：

SELECTION = "selection"  


SHORT_TRADE = "short_trade"  


LONG_TRADE = "long_trade"  


OPEN_ORDERS = "open_orders"  


FREE_BALANCE = "free_balance"  


CANCEL_ORD = "cancel_order"  


PROCESS_ORD_CANCEL = "process_ord_cancel"  


COIN_NAME = "coin_select"  


PERCENT_CHANGE = "percent_select"  


AMOUNT = "amount"  


PRICE = "price"  


PROCESS_TRADE = "process_trade"  


CONFIRM = "confirm"  


CANCEL = "cancel"  

END_CONVERSATION = ConversationHandler.END

我们可以通过扩展BaseFilter来实现对user_id的限制。这样机器人必须接受被允许用户的token、id才能执行操作。

class TelegramBot:  


    class PrivateUserFiler(BaseFilter):  


        def __init__(self, user_id):  


            self.user_id = int(user_id)  


        def filter(self, message):  


            return message.from_user.id == self.user_id  


    def __init__(self, token: str, allowed_user_id, trade_executor: TradeExecutor):  


        self.updater = Updater(tokentoken=token)  


        selfself.dispatcher = self.updater.dispatcher  


        self.trade_executor = trade_executor  


        selfself.exchange = self.trade_executor.exchange  


        selfself.private_filter = self.PrivateUserFiler(allowed_user_id)  

        self._prepare()

在_prepare（）函数中，我们将创建所有处理函数并将其附加到调度程序。我们开始与机器人聊天时希望显示的基本选项：

def _prepare(self):  


      # Create our handlers  


      def show_help(bot, update):  


          update.effective_message.reply_text('Type /trade to show options ')  


      def show_options(bot, update):  

          button_list = [  

              [InlineKeyboardButton("Short trade", callback_data=SHORT_TRADE),

                InlineKeyboardButton("Long trade", callback_data=LONG_TRADE), ],  


              [InlineKeyboardButton("Open orders", callback_data=OPEN_ORDERS),  


               InlineKeyboardButton("Available balance", callback_data=FREE_BALANCE)],  


          ]  


          update.message.reply_text("Trade options:", reply_markup=InlineKeyboardMarkup(button_list))  

          return TRADE_SELECT

InlineKeyboardButton允许我们将文本选项显示为键盘。这比键入所有命令更为直观。callback_data允许在按下按钮时传递其他数据。show_options返回下一个继续进行对话的处理函数的名称。其他处理函数将使用类似的方法。然后我们执行用户选择的处理程序。在这里，我们主要从一个问题转到另一个问题：

def process_trade_selection(bot, update, user_data):  


         query = update.callback_query  

         selection = query.data 

         if selection == OPEN_ORDERS:  


             orders = self.exchange.fetch_open_orders()  


             if len(orders) == 0:  


                 bot.edit_message_text(text="You don't have open orders",  


                                       chat_id=query.message.chat_id,  


                                       message_id=query.message.message_id)  


                 return END_CONVERSATION  


             # show the option to cancel active orders  

             keyboard = [ 

                  [InlineKeyboardButton("Ok", callback_data=CONFIRM),  

                  InlineKeyboardButton("Cancel order", callback_data=CANCEL)]  


             ]  


             bot.edit_message_text(text=formatter.format_open_orders(orders),  


                                   chat_id=query.message.chat_id,  


                                   message_id=query.message.message_id,  


                                   reply_markup=InlineKeyboardMarkup(keyboard))  


             # attach opened orders, so that we can cancel by index  


             user_data[OPEN_ORDERS] = orders  

             return CANCEL_ORD  


         elif selection == FREE_BALANCE:  


             balance = self.exchange.free_balance  


             msg = "You don't have any available balance" if len(balance) == 0 \  


                 else f"Your available balance:\n{formatter.format_balance(balance)}"  


             bot.edit_message_text(text=msg,  


                                   chat_id=query.message.chat_id,  


                                   message_id=query.message.message_id)  


             return END_CONVERSATION  


         user_data[TRADE_SELECT] = selection  

         bot.edit_message_text(text=f'Enter coin name for {selection}',  

                               chat_id=query.message.chat_id,  


                               message_id=query.message.message_id)  

         return COIN_NAME 

     def cancel_order(bot, update):  


         query = update.callback_query  


         if query.data == CANCEL:  


             query.message.reply_text('Enter order index to cancel: ')  


             return PROCESS_ORD_CANCEL  


         show_help(bot, update)  


         return END_CONVERSATION  


     def process_order_cancel(bot, update, user_data):  


         idx = int(update.message.text)  


         order = user_data[OPEN_ORDERS][idx]  


         self.exchange.cancel_order(order['id'])  


         update.message.reply_text(f'Canceled order: {formatter.format_order(order)}')  


         return END_CONVERSATION  


     def process_coin_name(bot, update, user_data):  

         user_data[COIN_NAME] = update.message.text.upper()  


         update.message.reply_text(f'What amount of {user_data[COIN_NAME]}')  


         return AMOUNT  

     def process_amount(bot, update, user_data): 

          user_data[AMOUNT] = float(update.message.text)  


         update.message.reply_text(f'What % change for {user_data[AMOUNT]} {user_data[COIN_NAME]}')  


         return PERCENT_CHANGE  


     def process_percent(bot, update, user_data):  


         user_data[PERCENT_CHANGE] = float(update.message.text)  


         update.message.reply_text(f'What price for 1 unit of {user_data[COIN_NAME]}')  


         return PRICE  


     def process_price(bot, update, user_data):  


         user_data[PRICE] = float(update.message.text)  


         keyboard = [  


             [InlineKeyboardButton("Confirm", callback_data=CONFIRM),  


              InlineKeyboardButton("Cancel", callback_data=CANCEL)]  


         ]  


         update.message.reply_text(f"Confirm the trade: '{TelegramBot.build_trade(user_data)}'",  


                                   reply_markup=InlineKeyboardMarkup(keyboard))  

         return PROCESS_TRADE

最后，我们构建会话处理程序，设置错误处理程序，并将所有处理程序添加到调度程序中。

def process_trade(bot, update, user_data):  

           query = update.callback_query 
           if query.data == CONFIRM: 

                trade = TelegramBot.build_trade(user_data)  

               self._execute_trade(trade)  


               update.callback_query.message.reply_text(f'Scheduled: {trade}')  


           else:  


               show_help(bot, update)  


           return END_CONVERSATION  


       def handle_error(bot, update, error):  


           logging.warning('Update "%s" caused error "%s"', update, error)  


           update.message.reply_text(f'Unexpected error:\n{error}')  


       # configure our handlers  

       def build_conversation_handler():  


           entry_handler = CommandHandler('trade', filters=self.private_filter, callback=show_options)  

           conversation_handler = ConversationHandler( 

                entry_points=[entry_handler],  


               fallbacks=[entry_handler],  


               states={  


                   TRADE_SELECT: [CallbackQueryHandler(process_trade_selection, pass_user_data=True)],  


                   CANCEL_ORD: [CallbackQueryHandler(cancel_order)],  


                   PROCESS_ORD_CANCEL: [MessageHandler(filters=Filters.text, callback=process_order_cancel, pass_user_data=True)],  


                   COIN_NAME: [MessageHandler(filters=Filters.text, callback=process_coin_name, pass_user_data=True)],  

                   AMOUNT: [MessageHandler(Filters.text, callback=process_amount, pass_user_data=True)], 

                    PERCENT_CHANGE: [MessageHandler(Filters.text, callback=process_percent, pass_user_data=True)],  


                   PRICE: [MessageHandler(Filters.text, callback=process_price, pass_user_data=True)],  


                   PROCESS_TRADE: [CallbackQueryHandler(process_trade, pass_user_data=True)],  


               },  

           ) 

            return conversation_handler  


       self.dispatcher.add_handler(CommandHandler('start', filters=self.private_filter, callback=show_help))  


       self.dispatcher.add_handler(build_conversation_handler())  

       self.dispatcher.add_error_handler(handle_error)

传递用户数据时允许我们向处理程序提供其他user_data参数。这样可以确保机器人从一个处理程序传递到另一个处理程序时，保持所有答复的对话状态。我们需要run_async装饰器在后台执行交易，而又不会阻止机器人对新消息进行响应：

def start_bot(self):  


      self.updater.start_polling()  


  @run_async  


  def _execute_trade(self, trade):  


      loop = asyncio.new_event_loop()  


      task = loop.create_task(self.trade_executor.execute_trade(trade))  


      loop.run_until_complete(task)  


  @staticmethod  


  def build_trade(user_data):  


      current_trade = user_data[TRADE_SELECT]  


      price = user_data[PRICE]  


      coin_name = user_data[COIN_NAME]  


      amount = user_data[AMOUNT]  


      percent_change = user_data[PERCENT_CHANGE]  


      if current_trade == LONG_TRADE:  


          return LongTrade(price, coin_name, amount, percent_change)  


      elif current_trade == SHORT_TRADE:  


          return ShortTrade(price, coin_name, amount, percent_change)  


      else:  
          raise NotImplementedError

这是用于订单和余额显示的格式化程序：

TITLES = ['idx', 'type', 'remaining', 'symbol', 'price']  

SPACING = [4, 6, 8, 10, 8] 

def format_open_orders(orders) -> str:  

    def join_line(ln): 

        return ' | '.join(str(item).center(SPACING[i]) for i, item in enumerate(ln))  

    title_line = join_line(TITLES) 

    lines = [title_line]  

    for idx, order in enumerate(orders): 

        line = [idx, order['side'], order['remaining'], order['symbol'], order['price']]  

        lines.append(join_line(line))  


    separator_line = '-' * len(title_line)  


    return f"\n{separator_line}\n".join(lines)  

def format_order(order): 

    return f"{order['amount']} {order['symbol']} priced at {order['price']}"  


def format_balance(balance) -> str:  


    coin_balance_as_list = list(f"{coin}: {val}" for coin, val in balance.items())  

    return "\n".join(coin_balance_as_list)

最后，我们创建main.py并将所有内容归结在一起：

import logging  


import os  


import ccxt  


from core.exchange import CryptoExchange  

from core.telegrambot import TelegramBot  

from core.tradeexcutor import TradeExecutor 

if __name__ == '__main__':  


    logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)  


    c_dir = os.path.dirname(__file__)  


    with open(os.path.join(c_dir, "config/secrets.txt")) as key_file:  


        api_key, secret, telegram_tkn, user_id = key_file.read().splitlines()  


    ccxtccxt_ex = ccxt.bitfinex()  


    ccxt_ex.apiKey = api_key  

    ccxt_ex.secret = secret  


    exchange = CryptoExchange(ccxt_ex)  


    trade_executor = TradeExecutor(exchange)  

    telegram_bot = TelegramBot(telegram_tkn, user_id, trade_executor) 
    telegram_bot.start_bot()

我们从secrets.txt文件中获取交易所密钥，telegram的token和用户ID，构造核心类并启动机器人。使用以下内容在config文件夹中创建secrets.txt：

# YOUR_API_KEY  


# YOUR_SECRET  


# YOUR_TELEGRAM_TOKEN  

# YOUR_TELEGRAM_USER_ID

用Python开发数字货币交易机器人（附源码）

用Python开发数字货币交易机器人（附源码）

相关推荐