Skip to content

Instantly share code, notes, and snippets.

@scvnc

scvnc/cryptocurrency_report.py

Last active August 29, 2015 14:02
Show Gist options
  • Save scvnc/67c7353705e00c307b37 to your computer and use it in GitHub Desktop.
Save scvnc/67c7353705e00c307b37 to your computer and use it in GitHub Desktop.
Download ZIP
This pulls data from cryptocurrency sources and makes a report that I'm interested in. I currently have conky display the report.
Raw
cryptocurrency_report.py
#!/usr/bin/env python
from __future__ import print_function
"""
This pulls data from cryptocurrency sources and makes a report
that I'm interested in. I currently have conky display the report.
requirements:
pip install requests
dev-requirements:
pip install pytest pytest-xdist
test:
py.test --looponfail cryptocurrency_report.py
"""
import sys
import requests
import json
import shelve
import os
import tempfile
shelf_name = os.path.join(tempfile.gettempdir(), __name__ +'_cache')
cache = shelve.open(shelf_name)
def traverse_endpoint(name, url):
# Find the next hyperlink at the url given the resource name
# Assuming the result is json parsable and the links are in
# the root of the object.
endpoint_url = cache.get(name)
if endpoint_url == None: # Cache miss
# Get next resource from endpoint
data = requests.get(url).json()
endpoint_url = data.get(name)
if endpoint_url != None:
cache[name] = endpoint_url
return endpoint_url
def get_ticker_USD_info():
tickers_url = traverse_endpoint('tickers',
'https://api.bitcoinaverage.com/')
usd_ticker_url = traverse_endpoint('USD', tickers_url)
data = requests.get(usd_ticker_url).json()
return data
def get_ticker_DOGE_price():
markets = requests.get('http://dogecoinaverage.com/BTC.json')\
.json()['markets']
cryptsy_price = [
market['price'] \
for market in markets \
if market['exchange_name'] == "Cryptsy"][0]
return cryptsy_price
def main():
d_line = "{:>11} DOGE / 1 BTC\n"
d_line += "{:>11} USD / 1 kDOGE"
b_line = "{:>11} USD / 1 BTC"
USD_BTC_ticker = get_ticker_USD_info()
USD_BTC_rate = float(USD_BTC_ticker['24h_avg'])
DOGE_BTC_rate = float(get_ticker_DOGE_price())
DOGE_USD_rate = DOGE_BTC_rate * USD_BTC_rate * 1000
print(d_line.format(get_ticker_DOGE_price(), DOGE_USD_rate))
print(b_line.format(USD_BTC_ticker['24h_avg']))
if __name__ == "__main__":
try:
main()
except Exception as e:
print("Could not retrieve information.")
print(e, file=sys.stderr)
exit(1)
exit(0)
######## BEGIN TESTS ##########
import unittest
from mock import patch
from requests.models import Response
class MainTests(unittest.TestCase):
def setUp(self):
patcher = patch(__name__+'.get_ticker_DOGE_price',
return_value='0.00000052')
self.requests_get_patch = patcher.start()
self.addCleanup(patcher.stop)
patcher = patch(__name__+'.get_ticker_USD_info',
return_value={'24h_avg':'600.00'})
self.requests_get_patch = patcher.start()
self.addCleanup(patcher.stop)
def test_main(self):
main()
self.requests_get_patch.assert_called_once_with()
class TraverseEndpointTests(unittest.TestCase):
def setUp(self):
cache.clear()
response = Response()
response._content = """{
"cookies" : "http://cookies.com/api/cookies"
}"""
# Patch requests
patcher = patch('requests.get',
return_value=response)
self.requests_get_patch = patcher.start()
self.addCleanup(patcher.stop)
def test_returns_value_from_content(self):
retval = traverse_endpoint('cookies', 'http://cookies.com/api')
self.assertEqual(retval, 'http://cookies.com/api/cookies')
# Cache/errors not verified
class Get_ticker_USD_info_Tests(unittest.TestCase):
def setUp(self):
cache.clear()
#
## Provide and patch a fake requests.get()
#
def mock_traverse_endpoint(key, url):
actions = {
'https://api.bitcoinaverage.com/': json.loads("""
{
"all": "https://api.bitcoinaverage.com/all",
"exchanges": "https://api.bitcoinaverage.com/exchanges/",
"global_tickers": "https://api.bitcoinaverage.com/ticker/global/",
"history": "https://api.bitcoinaverage.com/history/",
"ignored": "https://api.bitcoinaverage.com/ignored",
"tickers": "https://api.bitcoinaverage.com/ticker/"
} """),
'https://api.bitcoinaverage.com/ticker/': json.loads("""
{
"AUD": "https://api.bitcoinaverage.com/ticker/AUD",
"BRL": "https://api.bitcoinaverage.com/ticker/BRL",
"CAD": "https://api.bitcoinaverage.com/ticker/CAD",
"CHF": "https://api.bitcoinaverage.com/ticker/CHF",
"CNY": "https://api.bitcoinaverage.com/ticker/CNY",
"EUR": "https://api.bitcoinaverage.com/ticker/EUR",
"GBP": "https://api.bitcoinaverage.com/ticker/GBP",
"HKD": "https://api.bitcoinaverage.com/ticker/HKD",
"IDR": "https://api.bitcoinaverage.com/ticker/IDR",
"ILS": "https://api.bitcoinaverage.com/ticker/ILS",
"MXN": "https://api.bitcoinaverage.com/ticker/MXN",
"NOK": "https://api.bitcoinaverage.com/ticker/NOK",
"NZD": "https://api.bitcoinaverage.com/ticker/NZD",
"PLN": "https://api.bitcoinaverage.com/ticker/PLN",
"RON": "https://api.bitcoinaverage.com/ticker/RON",
"RUB": "https://api.bitcoinaverage.com/ticker/RUB",
"SEK": "https://api.bitcoinaverage.com/ticker/SEK",
"SGD": "https://api.bitcoinaverage.com/ticker/SGD",
"TRY": "https://api.bitcoinaverage.com/ticker/TRY",
"USD": "https://api.bitcoinaverage.com/ticker/USD",
"ZAR": "https://api.bitcoinaverage.com/ticker/ZAR",
"all": "https://api.bitcoinaverage.com/ticker/all"
}
""")
}
return actions.get(url).get(key)
def mock_requests_get(url):
mockResponse = Response()
mockResponse.status_code = 200
mockResponse._content = """
{
"24h_avg": 594.36,
"ask": 591.87,
"bid": 590.79,
"last": 591.45,
"timestamp": "Mon, 23 Jun 2014 19:01:14 -0000",
"total_vol": 16258.87
}
"""
mockResponse._content_consumed = True
return mockResponse
# Patch traverse_endpoint
patcher = patch(__name__+'.traverse_endpoint',
side_effect=mock_traverse_endpoint)
self.traverse_endpoint_patch = patcher.start()
self.addCleanup(patcher.stop)
# Patch requests
patcher = patch('requests.get',
side_effect=mock_requests_get)
self.requests_get_patch = patcher.start()
self.addCleanup(patcher.stop)
def test_information_retrieved(self):
retval = get_ticker_USD_info()
# Assert calls to traverse_endpoint(..)
call1_args, call1_kwargs = self.get_call(
self.traverse_endpoint_patch, 0)
call2_args, call2_kwargs = self.get_call(
self.traverse_endpoint_patch, 1)
self.assertEqual(call1_args,
('tickers', 'https://api.bitcoinaverage.com/',))
self.assertEqual(call2_args,
('USD', 'https://api.bitcoinaverage.com/ticker/',))
# Assert calls by request.get(..)
call1_args, call_kwargs = self.get_call(
self.requests_get_patch, 0)
self.assertEqual(call1_args,
('https://api.bitcoinaverage.com/ticker/USD',),
"Did not query the expected endpoint.")
# Assert dictionary returned ...
self.assertEqual(retval, {
"24h_avg": 594.36,
"ask": 591.87,
"bid": 590.79,
"last": 591.45,
"timestamp": "Mon, 23 Jun 2014 19:01:14 -0000",
"total_vol": 16258.87
}, "Did not return data as expected from the api.")
def get_call(self, mock, orderNDX):
try:
return mock.call_args_list[orderNDX]
except IndexError:
self.fail('{0} was not called {1} time(s)'.format(
mock,
orderNDX+1
))
"""
Copyright (c) 2014 Vincent Schramer
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment