Coingecko - trending coin monitoring and alerting

gecko.py

#!/usr/bin/env python3

# Coingecko - trending coin monitoring and alerting
# v0.3
# March, 2021
# greg.foss[at]owasp.org

'''
	Licensed under the Apache License, Version 2.0 (the "License");
	you may not use this file except in compliance with the License.
	You may obtain a copy of the License at
		http://www.apache.org/licenses/LICENSE-2.0
	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.
'''

import os, sys, argparse, json, subprocess, requests, http.client, urllib
from datetime import datetime
from pycoingecko import CoinGeckoAPI
from twilio.rest import Client
from stringcolor import *

################ CUSTOM VARIABLES ################
# CSV file to log trending coins:
filename = '/file/location/trending-coins.csv'

# Choose between Pushbullet and Twilio notification options
# 	Enter either 'pushover, 'pushbullet', or 'twilio':
notification_option = 'pushover'

# Pushover Configuration:
pushover_access_token = 'TOKEN'
pushover_user_key = 'KEY'

# Pushbullet Configuration:
pushbullet_access_token = 'TOKEN'

# Twilio SMS Configuration:
twilio_account_sid = 'KEY'
twilio_auth_token = 'TOKEN'
sender = "FROM"
recipient = "TO"
# Twilio Free Trial and $10 Credit:
#	https://www.twilio.com/try-twilio?promo=iQbzxw

# Crontab Configuration - Run once every 15-minutes:
# 	*/15 * * * * /script/location/gecko.py -m
##################################################

# Coin Gecko
cg = CoinGeckoAPI()
all_the_datas = cg.get_search_trending()

# Yay Options
def parse_all_things():
	parser = argparse.ArgumentParser(description = 'monitor and alert on coingecko\'s trending coins')
	parser.add_argument('-t', '--trend', help = 'analyze current market trends', action = 'store_true')
	parser.add_argument('-m', '--monitor', help = 'logging and alerting mode', action = 'store_true')
		# usage: gecko.py [-h] [-m]
		# 
		# monitor and alert on trending coins
		# 
		# optional arguments:
		#   -h, --help     show this help message and exit
		#   -t, --trend    analyze current market trends
		#   -m, --monitor  logging and alerting mode
		#
		# if no arguments are supplied, this script will print out help information
	return parser

# Check current trending coins and print to screen
def raw_output():
	print("")
	print(cs("TREND	SYMBOL	RANK	PRICE	$CHANGE	%CHANGE	:-)	:-(	COIN NAME","blue").bold())
	print("==========================================================================")
	for i in range(len(all_the_datas['coins'])):
		coin_id = all_the_datas['coins'][i]['item']['id']
		name = all_the_datas['coins'][i]['item']['name']
		symbol = all_the_datas['coins'][i]['item']['symbol']
		mkp = all_the_datas['coins'][i]['item']['market_cap_rank']
		score = all_the_datas['coins'][i]['item']['score']
		coin_details = cg.get_coin_by_id(coin_id)
		price = str(round(coin_details['market_data']['current_price']['usd'], 2))
		homepage = coin_details['links']['homepage'][0]
		positive_sentiment = str(coin_details['sentiment_votes_up_percentage'])
		negative_sentiment = str(coin_details['sentiment_votes_down_percentage'])
		price_change = round(coin_details['market_data']['price_change_24h_in_currency']['usd'], 2)
		if price_change > 0:
			dollar_sign = cs('$','green')
			price_change = cs(price_change,'green')
		else:
			dollar_sign = cs('$','red')
			price_change = cs(price_change,'red')
		percent_change = round(coin_details['market_data']['price_change_percentage_24h_in_currency']['usd'], 2)
		if percent_change > 0:
			percent_sign = cs('%','green')
			percent_change = cs(percent_change,'green')
		else:
			percent_sign = cs('%','red')
			percent_change = cs(percent_change,'red')
		print("{}	{}	{}	$ {}	{} {}	{} {}	{}	{}	{}".format(score,symbol,mkp,price,dollar_sign,price_change,percent_change,percent_sign,positive_sentiment,negative_sentiment,name))
	print("")

# Push/SMS Notification
def alert(name,symbol,coin_id):
	coin_details = cg.get_coin_by_id(coin_id)
	price = str(round(coin_details['market_data']['current_price']['usd'], 2))
	homepage = coin_details['links']['homepage'][0]
	positive_sentiment = str(coin_details['sentiment_votes_up_percentage'])
	negative_sentiment = str(coin_details['sentiment_votes_down_percentage'])
	price_change = str(round(coin_details['market_data']['price_change_24h_in_currency']['usd'], 2))
	percent_change = str(round(coin_details['market_data']['price_change_percentage_24h_in_currency']['usd'], 2))
	coin_url = 'https://www.coingecko.com/en/coins/{}'.format(coin_id)
	#coin_description = str(coin_details['description']['en']).replace("\xa0", "").replace("\r", "").replace("\n", "\\n")

	#Pushover
	if 'pushover' in notification_option:
		message = 'Trending Coin ('+symbol+')\n'+name+'\n'+homepage+'\nPrice: $'+price+'\n24hr Change: $'+price_change+'\n24hr Change: '+percent_change+'%\n:-) Sentiment: '+positive_sentiment+'\n:-( Sentiment: '+negative_sentiment+'\n'+coin_url
		conn = http.client.HTTPSConnection("api.pushover.net:443")
		conn.request("POST", "/1/messages.json",
			urllib.parse.urlencode({
			"token": pushover_access_token,
			"user": pushover_user_key,
			"message": message,
		}), { "Content-type": "application/x-www-form-urlencoded" })
	# Pushbullet
	elif 'pushbullet' in notification_option:
		title = 'Trending Coin: {}'.format(symbol)
		message = name+'\\n'+homepage+'\\nPrice: $'+price+'\\n24hr Change: $'+price_change+'\\n24hr Change: '+percent_change+'%\\n:-) Sentiment: '+positive_sentiment+'\\n:-( Sentiment: '+negative_sentiment+'\\n'+coin_url
		headers = {'Access-Token': pushbullet_access_token, 'Content-Type': 'application/json',}
		data = '{"body":"'+message+'","title":"'+title+'","type":"note"}'
		response = requests.post('https://api.pushbullet.com/v2/pushes', headers=headers, data=data)
	# Twilio
	elif 'twilio' in notification_option:
		client = Client(twilio_account_sid, twilio_auth_token)
		msgBody = "New Trending Coin: {} -- Symbol: {}".format(name,symbol)
		message = client.messages.create(
			from_='{}'.format(sender),
			body='{}'.format(msgBody),
			to='{}'.format(recipient)
		),
	else:
		print(" [ ! ] Notification Options Are Not Properly Configured...")
		quit()

# Check if trending coin has been observed in past 3 script iterations
def analysis():
	line = subprocess.check_output(['tail', '-21', filename])

	for i in range(len(all_the_datas['coins'])):
		name = all_the_datas['coins'][i]['item']['name']
		symbol = all_the_datas['coins'][i]['item']['symbol']
		coin_id = all_the_datas['coins'][i]['item']['id']
		if (str(line).find(symbol) != -1):
			pass
		else:
			alert(name,symbol,coin_id)

# Ongoing logging of trending coins to CSV
def monitor():
	today = datetime.now()

	orig_stdout = sys.stdout
	f = open(filename, 'a')
	sys.stdout = f

	for i in range(len(all_the_datas['coins'])):
		coin_id = all_the_datas['coins'][i]['item']['id']
		name = all_the_datas['coins'][i]['item']['name']
		symbol = all_the_datas['coins'][i]['item']['symbol']
		mkp = all_the_datas['coins'][i]['item']['market_cap_rank']
		score = all_the_datas['coins'][i]['item']['score']
		coin_details = cg.get_coin_by_id(coin_id)
		price = str(coin_details['market_data']['current_price']['usd'])
		homepage = coin_details['links']['homepage'][0]
		positive_sentiment = coin_details['sentiment_votes_up_percentage']
		negative_sentiment = coin_details['sentiment_votes_down_percentage']
		price_change = str(coin_details['market_data']['price_change_24h_in_currency']['usd'])
		percent_change = str(coin_details['market_data']['price_change_percentage_24h_in_currency']['usd'])
		print("{},{},{},${},${},{}%,{},{},{},{}".format(score,symbol,mkp,price,price_change,percent_change,positive_sentiment,negative_sentiment,name,today))
	
	sys.stdout = orig_stdout
	f.close()

# Let's gooooooooooooo
def main():
	parser = parse_all_things()
	args = parser.parse_args()
	if args.monitor:
		analysis()
		monitor()
	elif args.trend:
		raw_output()
	else:
		print((parser.format_help()))
		quit()

if __name__ == "__main__":
	main()