如何用Python開發(fā)數(shù)字貨幣交易機(jī)器人

今天就跟大家聊聊有關(guān)如何用Python開發(fā)數(shù)字貨幣交易機(jī)器人，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

眾所周知，幣圈一天，人間一年。我們進(jìn)行數(shù)字貨幣交易時(shí)，在交易所 APP 或者網(wǎng)站 盯盤并手動下單非常耗時(shí)，當(dāng)幣價(jià)波動非常大的時(shí)候往往會錯(cuò)失良機(jī)。這時(shí)我們可以創(chuàng)建一個(gè)簡單的 telegram 交易機(jī)器人，來幫助我們進(jìn)行做空和做多交易。

該機(jī)器人可以實(shí)現(xiàn)以下功能：

•  做空交易 - 以指定的價(jià)格賣出持有貨幣并在價(jià)格下跌時(shí)回購

•  做多交易 - 指定的價(jià)格購買貨幣并在價(jià)格上漲時(shí)賣出

•  列出交易訂單

•  顯示可用余額

設(shè)置 Telegram 機(jī)器人

首先需要一個(gè) Telegram 賬號，如果沒有的話請自己注冊一個(gè)。然后與BotFather進(jìn)行對話，通過輸入/newbot來新建一個(gè)telegram機(jī)器人，根據(jù)指示一步步創(chuàng)建并記住你的token。

獲取交易所的 API keys

查找你的交易所API文檔，看看如何獲取對訂單和賬戶余額的訪問權(quán)限和步驟，記住你的密碼和API keys。本例中我們以bitfinex為例，Bitmex交易所是目前市面上交易量最大的比特幣期貨交易所，交易量和交易深度非常大。

安裝依賴包

我們這邊用的是Python 3.6版本，同時(shí)我們還需要利用CCXT框架獲取Bitmex交易所數(shù)據(jù)，CCXT是一個(gè)JavaScript / Python / PHP 開發(fā)庫，用于數(shù)字貨幣的交易，支持眾多的比特幣/以太幣/山寨幣交易市場和交易所API。

CCXT庫用于連接數(shù)字貨幣交易所并在世界范圍內(nèi)進(jìn)行交易和支付處理。使用 ccxt可以快速訪問數(shù)字貨幣市場數(shù)據(jù)，可以用于存儲、分析、可視化、指標(biāo)開發(fā)、 量化交易、策略回溯測試、交易機(jī)器人程序以及相關(guān)的軟件工程。

然后我們將使用python-telegram-bot與Telegram進(jìn)行通訊，對聊天消息做出反應(yīng)并進(jìn)行交易。

只需要用下面方法安裝以上兩個(gè)依賴包：

pip install python-telegram-bot ccxt

我們需要交易機(jī)器人實(shí)現(xiàn)的基本類功能：

1、獲取交易所概況，允許創(chuàng)建訂單，列出訂單詳情并獲取余額。這將是以 ccxt 實(shí)現(xiàn)的裝飾器。

2、交易執(zhí)行者，因?yàn)槲覀兿Ｍ詣訄?zhí)行做空和做多交易。

3、即時(shí)響應(yīng)的telegram 機(jī)器人。

編寫機(jī)器人

項(xiàng)目結(jié)構(gòu)如下：

main.py  \config  \core  \model   \util

我們將從一個(gè)簡單的模型開始。因?yàn)槎嗫战灰變烧哂泻芏喙餐c(diǎn)，可以在\ model中創(chuàng)建一個(gè)基類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__()

接下來是獲取交易所數(shù)據(jù)：

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)

然后，我們將執(zhí)行交易程序。程序?qū)⒔邮芙灰姿鶖?shù)據(jù)和超時(shí)情況以檢查訂單是否完成。當(dāng)做空時(shí)，我們以設(shè)定的價(jià)格賣出，當(dāng)價(jià)格下降到一定水平時(shí)回購。我們使用asyncio協(xié)程進(jìn)行編碼，以使等待不會阻塞：

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進(jìn)行數(shù)據(jù)傳輸。它不如某些交易所支持的WebSockets快，但是對于這個(gè)簡單的機(jī)器人來說，速度或許差別。

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

接下來將創(chuàng)建Telegram機(jī)器人，這是最有難度的部分，我們將使其擁有以下指令：

1、列出/取消有效訂單

2、顯示可用余額

3、建立做多或做空交易

我們還需要對機(jī)器人做一些安全限制，使其僅對你的消息做出響應(yīng)，而其他人則無法使用你的帳戶進(jìn)行交易。

主要是進(jìn)行做多和做空交易的部分：

1、選擇做空或者做多

2、輸入數(shù)字貨幣品種

3、輸入交易數(shù)量

4、所占百分比

5、每個(gè)價(jià)格

6、顯示確認(rèn)信息

7、顯示最終交易信息

我們來創(chuàng)建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

我們可以通過擴(kuò)展BaseFilter來實(shí)現(xiàn)對user_id的限制。這樣機(jī)器人必須接受被允許用戶的token、id才能執(zhí)行操作。

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（）函數(shù)中，我們將創(chuàng)建所有處理函數(shù)并將其附加到調(diào)度程序。我們開始與機(jī)器人聊天時(shí)希望顯示的基本選項(xiàng)：

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允許我們將文本選項(xiàng)顯示為鍵盤。這比鍵入所有命令更為直觀。callback_data允許在按下按鈕時(shí)傳遞其他數(shù)據(jù)。show_options返回下一個(gè)繼續(xù)進(jìn)行對話的處理函數(shù)的名稱。其他處理函數(shù)將使用類似的方法。然后我們執(zhí)行用戶選擇的處理程序。在這里，我們主要從一個(gè)問題轉(zhuǎn)到另一個(gè)問題：   

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

最后，我們構(gòu)建會話處理程序，設(shè)置錯(cuò)誤處理程序，并將所有處理程序添加到調(diào)度程序中。     

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)

傳遞用戶數(shù)據(jù)時(shí)允許我們向處理程序提供其他user_data參數(shù)。這樣可以確保機(jī)器人從一個(gè)處理程序傳遞到另一個(gè)處理程序時(shí)，保持所有答復(fù)的對話狀態(tài)。我們需要run_async裝飾器在后臺執(zhí)行交易，而又不會阻止機(jī)器人對新消息進(jìn)行響應(yīng)：

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)

最后，我們創(chuàng)建main.py并將所有內(nèi)容歸結(jié)在一起：

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，構(gòu)造核心類并啟動機(jī)器人。使用以下內(nèi)容在config文件夾中創(chuàng)建secrets.txt：

# YOUR_API_KEY  # YOUR_SECRET  # YOUR_TELEGRAM_TOKEN  # YOUR_TELEGRAM_USER_ID

總結(jié)

對于想要簡化交易并擁有更好使用體驗(yàn)的人來說，該機(jī)器人更像是一個(gè)輔助工具。它不是最先進(jìn)的算法交易機(jī)器人。后面可以進(jìn)行以下改進(jìn)：

•  當(dāng)進(jìn)行做空交易時(shí)獲取可用余額并顯示用戶可以根據(jù)余額做空的最大值

•  要求在交易所執(zhí)行之前驗(yàn)證創(chuàng)建的訂單

•  添加TA指標(biāo)、信號以通知最佳交易時(shí)間

•  止盈/止損操作和其他統(tǒng)計(jì)數(shù)據(jù)

•  有策略地根據(jù)超時(shí)等原因取消訂單 

看完上述內(nèi)容，你們對如何用Python開發(fā)數(shù)字貨幣交易機(jī)器人有進(jìn)一步的了解嗎？如果還想了解更多知識或者相關(guān)內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。