Skip to content

Instantly share code, notes, and snippets.

@gfoss
Last active December 29, 2021 15:23
Show Gist options
  • Save gfoss/3d693f3e3147792c73e0703309354076 to your computer and use it in GitHub Desktop.
Save gfoss/3d693f3e3147792c73e0703309354076 to your computer and use it in GitHub Desktop.
Coingecko - trending coin monitoring and alerting
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment