Shulou(Shulou.com)06/02 Report--

Today, I will talk to you about how to develop a digital currency trading robot with Python. Many people may not know much about it. In order to make you understand better, the editor has summarized the following content for you. I hope you can get something according to this article.

As we all know, the coin circle in one day, the world in a year. When we trade digital currencies, it is very time-consuming to mark orders on the exchange APP or website and place orders manually, and we often miss opportunities when currency prices fluctuate very much. At this point we can create a simple telegram trading robot to help us short and long trades.

The robot can perform the following functions:

Short trade-sell the holding currency at a specified price and buy it back when the price falls

Long trade-buy currency at a specified price and sell when the price goes up

List transaction orders

Show available balance

Set up the Telegram robot

First of all, you need a Telegram account. If you don't have one, please sign up for one. Then talk to BotFather, type / newbot to create a new telegram robot, and follow the instructions to create and remember your token step by step.

Get the API keys of the exchange

Check your exchange API documents to see how to get access to orders and account balances and steps, and remember your password and API keys. In this case, we take bitfinex as an example. Bitmex Exchange is currently the most traded bitcoin futures exchange on the market, with a very large trading volume and trading depth.

Install dependency packages

We are using Python version 3.6and we also need to use the CCXT framework to obtain Bitmex exchange data. CCXT is a JavaScript / Python / PHP development library for digital currency transactions, supporting a wide range of bitcoin / ethercoin / counterfeit coin exchanges and exchange API.

The CCXT library is used to connect digital currency exchanges and conduct transactions and payment processing worldwide. Using ccxt, you can quickly access digital money market data, which can be used for storage, analysis, visualization, index development, quantitative trading, strategy backtracking testing, trading robot programs and related software engineering.

Then we will use python-telegram-bot to communicate with Telegram, react to chat messages and make transactions.

You only need to install the above two dependent packages in the following ways:

Pip install python-telegram-bot ccxt

We need to trade the basic class functions that robots implement:

1. Get an overview of the exchange, allow you to create orders, list order details and get balances. This will be a decorator implemented in ccxt.

2, the transaction executor, because we want to automatically execute short selling and long trading.

3. Telegram robot with instant response.

Write a robot

The project structure is as follows:

Main.py\ config\ core\ model\ util

We'll start with a simple model. Because long and short trades have a lot in common, you can create a base class TradeDetails in\ model:

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} "

The details are as follows:

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__ ()

The next step is to obtain exchange data:

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)

Then we will execute the transaction procedure. The program will accept exchange data and timeouts to check whether the order has been completed. When shorting, we sell at a set price and buy back when the price falls to a certain level. We use the asyncio protocol to encode so that the wait does not block:

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 uses REST API for data transfer. It is not as fast as the WebSockets supported by some exchanges, but for this simple robot, the speed may be different.

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

Next, we will create the Telegram robot, which is the most difficult part, and we will make it have the following instructions:

1. List / cancel valid orders

2. Show the available balance

3. Establish a long or short trade

We also need to impose some security restrictions on robots so that they only respond to your messages, while others cannot use your account to trade.

Mainly the part of long and short trades:

1. Choose to short or long

2. Input digital currency

3. Enter the number of transactions

4. Percentage

5. Each price

6. Display confirmation information

7. Display the final transaction information

Let's create telegrambot.py and add the following constants:

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

We can implement the restrictions on user_id by extending BaseFilter. In this way, the robot must accept the token and id of the allowed user before it can perform the operation.

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 ()

In the _ prepare () function, we will create all the handlers and attach them to the scheduler. The basic options we want to display when we start chatting with the robot:

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 allows us to display text options as keyboards. This is more intuitive than typing all commands. Callback_data allows additional data to be passed when the button is pressed. Show_options returns the name of the next handler that continues the conversation. Other handlers will use a similar approach. Then we execute the handler selected by the user. Here, we mainly move from one question to another:

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 [Con _ 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 [Con _ 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

Finally, we build the session handler, set up the error handler, and add all handlers to the scheduler.

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)

Passing user data allows us to provide additional user_data parameters to the handler. This ensures that the robot maintains a conversational state of all replies as it passes from one handler to another. We need the run_async decorator to perform the transaction in the background without preventing the robot from responding to new messages:

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 [Con _ 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

This is the formatter for order and balance display:

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)

Finally, we create the main.py and put it all together:

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) slots, 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 ()

We get the exchange key, telegram token and user ID from the secrets.txt file, construct the core class and start the robot. Create a secrets.txt in the config folder using the following:

# YOUR_API_KEY # YOUR_SECRET # YOUR_TELEGRAM_TOKEN # YOUR_TELEGRAM_USER_ID

Summary

For people who want to simplify transactions and have a better experience, the robot is more like an auxiliary tool. It is not the most advanced algorithm trading robot. The following improvements can be made later:

Get the available balance when making a short transaction and display the maximum value that the user can short based on the balance

Requires that the created order be validated before the exchange executes

Add TA metrics and signals to notify the best trading time

Stop profit / stop loss operations and other statistics

Cancel orders strategically according to timeouts and other reasons

After reading the above, do you have any further understanding of how to develop a digital currency trading robot with Python? If you want to know more knowledge or related content, please follow the industry information channel, thank you for your support.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.