-
Notifications
You must be signed in to change notification settings - Fork 4
/
bot.py
205 lines (175 loc) · 8.85 KB
/
bot.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
import requests
import telegram
import _thread as thread
import os
import time
from telegram.ext import *
from telegram import *
from tracker import get_prices, get_top_coins,get_graph_info
from dotenv import load_dotenv
import os
load_dotenv()
PORT = int(os.environ.get('PORT', 8443))
bot = Bot(os.getenv('BOT_TOKEN'))
updater = Updater(token=os.getenv('BOT_TOKEN'), use_context=True)
dispatcher = updater.dispatcher
text = ""
MAX_ALERT_LIMIT = 1000
alerts_count = 0
def get(update, context):
global text
chat_id = update.effective_chat.id
text = update.message.text
coin = text.split()[1] # gets type of coin
try:
message = f"Ticker: {coin}"
keyboard = [
[
InlineKeyboardButton("Price", callback_data="price"),
InlineKeyboardButton("Hour Change", callback_data="hourChange"),
],
[
InlineKeyboardButton("Day Change", callback_data="dayChange"),
InlineKeyboardButton("Market Cap", callback_data="marketCap"),
],
[
InlineKeyboardButton("Volume Traded Today", callback_data="volumeTraded"),
InlineKeyboardButton("Day Opening Price", callback_data="dayOpening"),
],
[
InlineKeyboardButton("Day High", callback_data="dayHigh"),
InlineKeyboardButton("Day Low", callback_data="dayLow"),
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
context.bot.send_message(chat_id=chat_id, text=message)
update.message.reply_text('Please choose the data you are interested in:', reply_markup=reply_markup)
except: # request had an error
context.bot.send_message(chat_id=chat_id, text="The specified coin was not found in our API. Please check that you have the correct ticker symbol")
def graph(update, context):
global text
text = update.message.text
try:
chat_id = update.effective_chat.id
text = update.message.text
coin = text.split()[1] # gets type of coin
keyboard = [
[
InlineKeyboardButton("Minute", callback_data="min"),
InlineKeyboardButton("Hour", callback_data="hour"),
InlineKeyboardButton("Day", callback_data="day"),
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
update.message.reply_text('Please choose the interval:', reply_markup=reply_markup)
except Exception as e:
print(e)
context.bot.send_message(chat_id=chat_id, text="Error while generating graph")
# returns top X coins by market cap
def top(update,context):
chat_id = update.effective_chat.id
try:
coins = get_top_coins(10)
message = ""
for coin in coins:
message += f"Name: {coin['name']}\nTicker Symbol: {coin['ticker']}\nCurrent Price: {coin['price']}\nMarket Cap: {coin['market_cap']}\nVolume Traded Today: {coin['volume_day']}\nDay Opening Price: {coin['day_open']}\nDay High: {coin['day_high']}\nDay Low: {coin['day_low']}\n\n"
context.bot.send_message(chat_id=chat_id,text=message)
except Exception as e:
context.bot.send_message(chat_id=chat_id,text="Unknown error. Please try again later")
# polling function to be ran on a seperate thread
def thread_poller(chat_id,context,alert):
# Time in seconds to delay thread
global alerts_count
DELAY = 60
while True:
coin,isAbove,threshold_price = alert
data = get_prices(coin)
currentPrice = float(data["price"][1:].replace(',','')) # First character is the dollar sign
if isAbove:
if currentPrice > threshold_price:
message = f"Price of {data['ticker']} is now at {data['price']} and above your threshold price of ${threshold_price}"
context.bot.send_message(chat_id=chat_id,text=message)
context.bot.send_message(chat_id=chat_id,text="Alert removed from the system")
alerts_count -= 1
break
elif currentPrice < threshold_price:
message = f"Price of {data['ticker']} is now at {data['price']} and below your threshold price of ${threshold_price}"
context.bot.send_message(chat_id=chat_id,text=message)
context.bot.send_message(chat_id=chat_id,text="Alert is now removed from the system")
alerts_count -= 1
break
time.sleep(DELAY)
def alert(update,context):
global alerts_count
chat_id = update.effective_chat.id
text = update.message.text.split()
try:
if text[2].lower() == "above":
isAbove = True
elif text[2].lower() == "below":
isAbove = False
else:
raise ValueError
threshold_price = float(text[3])
alert=(text[1],isAbove,threshold_price)
if alerts_count + 1 < MAX_ALERT_LIMIT:
alerts_count += 1
context.bot.send_message(chat_id= chat_id, text= "Alert successfully set")
thread.start_new_thread(thread_poller,(chat_id,context,alert))
else:
context.bot.send_message(chat_id= chat_id, text= "There are too many alerts currently running on our system. Please try again later")
except Exception as e:
print(e)
context.bot.send_message(chat_id=chat_id,text="Invalid input format")
def help(update, context):
chat_id = update.effective_chat.id
text = update.message.text
message = f'Hello. Thanks for using the CryptoAlert Bot 🤖 \n\nCommands available:\n/get <coin> -- Retrieve pricing data 💰 for a specific coin\n<coin> -- Ticker symbol of a coin\n\n/top -- Retrieve data for the largest 10 🎖 cyptocurrencies by market cap.\n\n/graph <coin> -- Plots a line graph 📈 of closing price for a particular coin over 10 counts of the selected interval \n<coin> -- Ticker symbol of a coin\n\n/alert <coin> <direction> <threshold> -- Sets an alert ⏰ that triggers when the price of the coin crosses the specified threshold \n<coin> -- Ticker symbol of a coin\n<direction> -- Either "above" or "below"\n<threshold> -- Price to cross'
update.message.reply_text(message)
def button_click(update, context):
query : CallbackQuery = update.callback_query
chat_id = update.effective_chat.id
coin = text.split()[1]
crypto_data = get_prices(coin)
if query.data == "price":
message = f"Price is: {crypto_data['price']}"
context.bot.send_message(chat_id=chat_id, text=message)
elif query.data == "hourChange":
message = f"Hourly Change is: {crypto_data['change_hour']}"
context.bot.send_message(chat_id=chat_id, text=message)
elif query.data == "dayChange":
message = f"Daily Change is: {crypto_data['change_day']}"
context.bot.send_message(chat_id=chat_id, text=message)
elif query.data == "marketCap":
message = f"Market Cap is: {crypto_data['market_cap']}"
context.bot.send_message(chat_id=chat_id, text=message)
elif query.data == "volumeTraded":
message = f"Volume Traded Today is: {crypto_data['volume_day']}"
context.bot.send_message(chat_id=chat_id, text=message)
elif query.data == "dayOpening":
message = f"Day Opening Price is: {crypto_data['day_open']}"
context.bot.send_message(chat_id=chat_id, text=message)
elif query.data == "dayHigh":
message = f"Day High: {crypto_data['day_high']}"
context.bot.send_message(chat_id=chat_id, text=message)
elif query.data == "dayLow":
message = f"Day Low: {crypto_data['day_low']}"
context.bot.send_message(chat_id=chat_id, text=message)
if query.data == "min":
path = get_graph_info("minute",coin)
context.bot.send_photo(chat_id, photo=open(path, 'rb')) # sends a photo according to path
if query.data == "hour":
path = get_graph_info("hour",coin)
context.bot.send_photo(chat_id, photo=open(path, 'rb')) # sends a photo according to path
if query.data == "day":
path = get_graph_info("day",coin)
context.bot.send_photo(chat_id, photo=open(path, 'rb')) # sends a photo according to path
dispatcher.add_handler(CommandHandler("help", help)) # links /help and /start with the help function
dispatcher.add_handler(CommandHandler("start", help)) # links /help and /start with the help function
dispatcher.add_handler(CommandHandler("get", get)) # links /get with the get function
dispatcher.add_handler(CommandHandler("top", top))
dispatcher.add_handler(CommandHandler("graph", graph))
dispatcher.add_handler(CommandHandler("alert",alert))
updater.dispatcher.add_handler(CallbackQueryHandler(button_click))
updater.start_polling()
updater.idle()