diff --git a/beanprice/sources/BUILD b/beanprice/sources/BUILD index 2ff08f2..9e8d704 100644 --- a/beanprice/sources/BUILD +++ b/beanprice/sources/BUILD @@ -1,25 +1,5 @@ package(default_visibility = ["//visibility:public"]) -py_library( - name = "iex", - srcs = ["iex.py"], - deps = [ - "//beancount/core:number", - "//beancount/prices:source", - ], -) - -py_test( - name = "iex_test", - srcs = ["iex_test.py"], - deps = [ - "//beancount/core:number", - "//beancount/prices:source", - "//beancount/prices/sources:iex", - "//beanprice:date_utils", - ], -) - py_library( name = "oanda", srcs = ["oanda.py"], diff --git a/beanprice/sources/iex.py b/beanprice/sources/iex.py deleted file mode 100644 index cbac92d..0000000 --- a/beanprice/sources/iex.py +++ /dev/null @@ -1,63 +0,0 @@ -"""Fetch prices from the IEX 1.0 public API. - -This is a really fantastic exchange API with a lot of relevant information. - -Timezone information: There is currency no support for historical prices. The -output datetime is provided as a UNIX timestamp. -""" - -__copyright__ = "Copyright (C) 2018-2020 Martin Blais" -__license__ = "GNU GPLv2" - -import datetime -from decimal import Decimal - -from dateutil import tz -import requests - -from beanprice import source - - -class IEXError(ValueError): - "An error from the IEX API." - - -def fetch_quote(ticker): - """Fetch the latest price for the given ticker.""" - - url = "https://api.iextrading.com/1.0/tops/last?symbols={}".format(ticker.upper()) - response = requests.get(url) - if response.status_code != requests.codes.ok: - raise IEXError( - "Invalid response ({}): {}".format(response.status_code, response.text) - ) - - results = response.json() - if len(results) != 1: - raise IEXError("Invalid number of responses from IEX: {}".format(response.text)) - result = results[0] - - price = Decimal(result["price"]).quantize(Decimal("0.01")) - - # IEX is American markets. - us_timezone = tz.gettz("America/New_York") - time = datetime.datetime.fromtimestamp(result["time"] / 1000) - time = time.astimezone(us_timezone) - - # As far as can tell, all the instruments on IEX are priced in USD. - return source.SourcePrice(price, time, "USD") - - -class Source(source.Source): - "IEX API price extractor." - - def get_latest_price(self, ticker): - """See contract in beanprice.source.Source.""" - return fetch_quote(ticker) - - def get_historical_price(self, ticker, time): - """See contract in beanprice.source.Source.""" - raise NotImplementedError( - "This is now implemented at https://iextrading.com/developers/docs/#hist and " - "needs to be added here." - ) diff --git a/beanprice/sources/iex_test.py b/beanprice/sources/iex_test.py deleted file mode 100644 index 73b5e1c..0000000 --- a/beanprice/sources/iex_test.py +++ /dev/null @@ -1,52 +0,0 @@ -__copyright__ = "Copyright (C) 2018-2020 Martin Blais" -__license__ = "GNU GPLv2" - -import datetime -import unittest -from unittest import mock -from decimal import Decimal - -from dateutil import tz -import requests - -from beanprice import date_utils -from beanprice import source -from beanprice.sources import iex - - -def response(contents, status_code=requests.codes.ok): - """Produce a context manager to patch a JSON response.""" - response = mock.Mock() - response.status_code = status_code - response.text = "" - response.json.return_value = contents - return mock.patch("requests.get", return_value=response) - - -class IEXPriceFetcher(unittest.TestCase): - def test_error_network(self): - with response(None, 404): - with self.assertRaises(ValueError) as exc: - iex.fetch_quote("AAPL") - self.assertRegex(exc.message, "premium") - - def _test_valid_response(self): - contents = [{"symbol": "HOOL", "price": 183.61, "size": 100, "time": 1590177596030}] - with response(contents): - srcprice = iex.fetch_quote("HOOL") - self.assertIsInstance(srcprice, source.SourcePrice) - self.assertEqual(Decimal("183.61"), srcprice.price) - self.assertEqual( - datetime.datetime(2020, 5, 22, 19, 59, 56, 30000, tzinfo=tz.tzutc()), - srcprice.time.astimezone(tz.tzutc()), - ) - self.assertEqual("USD", srcprice.quote_currency) - - def test_valid_response(self): - for tzname in "America/New_York", "Europe/Berlin", "Asia/Tokyo": - with date_utils.intimezone(tzname): - self._test_valid_response() - - -if __name__ == "__main__": - unittest.main()