Skip to content

Instantly share code, notes, and snippets.

@ewancook
Created March 4, 2015 20:53
Show Gist options
  • Save ewancook/0da9b86275f321557684 to your computer and use it in GitHub Desktop.
Save ewancook/0da9b86275f321557684 to your computer and use it in GitHub Desktop.
a python script to track the price of a single cryptocurrency using Cryptsy's API.
from twisted.words.protocols import irc
from twisted.words.protocols.irc import assembleFormattedText as aFT, attributes as A
from twisted.internet import defer, reactor, protocol, task
import requests as r
class bot(irc.IRCClient):
last_price = 0
@property
def market_label(self):
return self.factory.coin_label
@property
def aesthetic_label(self):
return self.factory.aesthetic_label
@property
def nickname(self):
return self.factory.nickname
@property
def coin_id(self):
return self.factory.coin_id
@property
def channels(self):
return self.factory.channels
@property
def url(self):
return "http://pubapi.cryptsy.com/api.php?method=singlemarketdata&marketid={}".format(self.coin_id)
def signedOn(self):
"""
Called upon signing onto the network.
"""
self.msg("NickServ", "IDENTIFY {}".format(self.factory.password))
for c in self.channels:
reactor.callLater(10.0, self.join, c)
self.t = task.LoopingCall(self._fetch_price)
reactor.callLater(15, self.t.start, 10)
def joined(self, channel):
"""
Called upon joining a channel.
"""
self.msg(channel, "starting up price tracker...")
def _fetch_price(self):
"""
Fetches the price of the coin specified and outputs it to all channels.
"""
try:
price = r.get(self.url).json()["return"]["markets"][self.market_label]["lasttradeprice"].encode("utf-8")
except Exception as e:
print e
return
if price == self.last_price:
return
message_label = aFT(A.bold["[{}]".format(self.aesthetic_label)])
message_price = (aFT(A.fg.green[price]) if price > self.last_price else aFT(A.fg.red[price]))
for c in self.channels:
self.msg(c, "{} {}".format(message_label, message_price))
self.last_price = price
class factory(protocol.ClientFactory):
protocol = bot
def __init__(self, nickname, password, channels, alternate_servers, coin_id, coin_label, aesthetic_label):
"""
Creates an instance of the Price Bot.
Arguments:
nickname - the bot's nickname.
password - the bot's NickServ password.
channels - a list of channels for the bot to join and report to.
alternate servers - a list of alternate servers to cycle through.
coin id - the integer market number for the coin on Cryptsy.
coin label - the label of the price pair as given by Cryptsy.
aesthetic label - the label that will precede the price in the tracker messages.
"""
self.nickname = nickname
self.password = password
self.channels = channels
self.alt_servers = alternate_servers
self.coin_id = coin_id
self.coin_label = coin_label
self.aesthetic_label = aesthetic_label
def bounceReconnect(self, connector, message):
"""
Modifies the connector socket and reconnects the bot to an alternate server.
"""
try:
server = self.alt_servers.pop(0)
except IndexError:
return reactor.stop()
self.alt_servers.append(server)
connector.host = server
connector.connect()
def clientConnectionLost(self, connector, reason):
"""
Attempts to reconnect to a different server upon losing connection.
"""
self.bounceReconnect(connector, "connection lost")
def clientConnectionFailed(self, connector, reason):
"""
Attempts to reconnect to a different server upon a connection failure.
"""
self.bounceReconnect(connector, "connection failed")
if __name__ == "__main__":
f = factory("nick", "pass", ["#test"], ["irc.freenode.net"], 2, "BTC", "BTC/USD")
reactor.connectTCP("holmes.freenode.net", 6667, f)
reactor.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment