-
Notifications
You must be signed in to change notification settings - Fork 42
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
3 changed files
with
232 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| """Fetch prices from the https://api.portfolio-report.net API. | ||
| The symbols must be referred to by the symbol's UUID and the desired currency in | ||
| the format "<UUID>:<CURRENCY>". The symbol's UUIDs can be looked up here: | ||
| https://www.portfolio-report.net/search | ||
| Timezone information: Input and output datetimes are limited to dates, and I | ||
| believe the dates are presumed to be UTC (It's unclear, not documented.) | ||
| """ | ||
|
|
||
| import datetime | ||
| from decimal import Decimal | ||
| from typing import List, Tuple | ||
|
|
||
| import requests | ||
|
|
||
| from beanprice import source | ||
|
|
||
|
|
||
| class PortfolioreportError(ValueError): | ||
| "An error from the portfolio-report.net API." | ||
|
|
||
|
|
||
| def _get_price_series( | ||
| ticker: str, | ||
| currency: str, | ||
| time_begin: datetime.datetime = datetime.datetime.min, | ||
| time_end: datetime.datetime = datetime.datetime.max, | ||
| ) -> List[source.SourcePrice]: | ||
| base_url = \ | ||
| f"https://api.portfolio-report.net/securities/uuid/{ticker}/prices/{currency}" | ||
|
|
||
| query = {} | ||
| if time_begin != datetime.datetime.min: | ||
| query = { | ||
| 'from': time_begin.date().isoformat(), | ||
| } | ||
| response = requests.get(base_url, params=query) | ||
| if response.status_code != requests.codes.ok: | ||
| raise PortfolioreportError( | ||
| f"Invalid response ({response.status_code}): {response.text}" | ||
| ) | ||
| response_data = response.json() | ||
|
|
||
| def to_decimal(val: float) -> Decimal: | ||
| return Decimal(str(val)).quantize(Decimal('0.00000000')) | ||
|
|
||
| ret = [] | ||
| for entry in response_data: | ||
| date = datetime.datetime.fromisoformat(entry['date']).date() | ||
| if date < time_begin.date() or date > time_end.date(): | ||
| continue | ||
| ret.append(source.SourcePrice( | ||
| price=to_decimal(entry['close']), | ||
| time=datetime.datetime( | ||
| date.year, | ||
| date.month, | ||
| date.day, | ||
| tzinfo=datetime.timezone.utc, | ||
| ), | ||
| quote_currency=currency | ||
| )) | ||
| return ret | ||
|
|
||
|
|
||
| def _parse_ticker(ticker: str) -> Tuple[str, str]: | ||
| if ticker.count(':') != 1: | ||
| raise PortfolioreportError('ticker must be in the format "UUID:CURRENCY"') | ||
| parts = ticker.split(':', maxsplit=1) | ||
| return parts[0], parts[1] | ||
|
|
||
|
|
||
| class Source(source.Source): | ||
| def get_latest_price(self, ticker): | ||
| uuid, currency = _parse_ticker(ticker) | ||
| prices = _get_price_series(uuid, currency) | ||
| if len(prices) < 1: | ||
| return None | ||
| return prices[-1] | ||
|
|
||
| def get_historical_price(self, ticker, time): | ||
| uuid, currency = _parse_ticker(ticker) | ||
| prices = _get_price_series(uuid, currency, time) | ||
| if len(prices) < 1: | ||
| return None | ||
| return prices[0] | ||
|
|
||
| def get_prices_series(self, ticker, time_begin, time_end): | ||
| uuid, currency = _parse_ticker(ticker) | ||
| prices = _get_price_series(uuid, currency, time_begin, time_end) | ||
| if len(prices) < 1: | ||
| return None | ||
| return prices |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| import datetime | ||
| import decimal | ||
| import unittest | ||
| from decimal import Decimal | ||
| from unittest import mock | ||
|
|
||
| import requests | ||
|
|
||
| from beanprice import source | ||
| from beanprice.sources import portfolioreport | ||
|
|
||
| SYMBOL_A1JX52_UUID = '9a49754d8fd941bfa1dd2cff1922fe9b:EUR' | ||
|
|
||
|
|
||
| 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 PortfolioreportPriceFetcher(unittest.TestCase): | ||
| def setUp(self): | ||
| # reset the Decimal context since other tests override this | ||
| decimal.getcontext().prec = 10 | ||
| decimal.getcontext().rounding = decimal.ROUND_HALF_UP | ||
| self.sut = portfolioreport.Source() | ||
|
|
||
| def test_error_network(self): | ||
| with response(None, 404): | ||
| with self.assertRaises(portfolioreport.PortfolioreportError): | ||
| self.sut.get_historical_price( | ||
| SYMBOL_A1JX52_UUID, | ||
| datetime.datetime(2024, 1, 1), | ||
| ) | ||
| with self.assertRaises(portfolioreport.PortfolioreportError): | ||
| self.sut.get_latest_price(SYMBOL_A1JX52_UUID) | ||
| with self.assertRaises(portfolioreport.PortfolioreportError): | ||
| self.sut.get_prices_series( | ||
| SYMBOL_A1JX52_UUID, | ||
| datetime.datetime(2024, 1, 1), | ||
| datetime.datetime(2024, 1, 14), | ||
| ) | ||
|
|
||
| def test_valid_response(self): | ||
| contents = [ | ||
| {"date":"2024-01-02","close":88.37800000}, | ||
| {"date":"2024-01-03","close":87.99600000}, | ||
| {"date":"2024-01-04","close":87.89400000}, | ||
| {"date":"2024-01-05","close":87.72000000}, | ||
| ] | ||
| with response(contents): | ||
| price_1 = self.sut.get_historical_price( | ||
| SYMBOL_A1JX52_UUID, | ||
| datetime.datetime(2024, 1, 1), | ||
| ) | ||
| self.assertEqual( | ||
| source.SourcePrice( | ||
| Decimal('88.37800000'), | ||
| datetime.datetime(2024, 1, 2, tzinfo=datetime.timezone.utc), | ||
| 'EUR', | ||
| ), | ||
| price_1, | ||
| ) | ||
| self.assertTrue(price_1.time.tzinfo) | ||
|
|
||
| price_2 = self.sut.get_latest_price(SYMBOL_A1JX52_UUID) | ||
| self.assertEqual( | ||
| source.SourcePrice( | ||
| Decimal('87.72000000'), | ||
| datetime.datetime(2024, 1, 5, tzinfo=datetime.timezone.utc), | ||
| 'EUR', | ||
| ), | ||
| price_2, | ||
| ) | ||
| self.assertTrue(price_2.time.tzinfo) | ||
|
|
||
| price_3 = self.sut.get_prices_series( | ||
| SYMBOL_A1JX52_UUID, | ||
| datetime.datetime(2024, 1, 3), | ||
| datetime.datetime(2024, 1, 4), | ||
| ) | ||
| self.assertEqual( | ||
| [ | ||
| source.SourcePrice( | ||
| Decimal('87.99600000'), | ||
| datetime.datetime(2024, 1, 3, tzinfo=datetime.timezone.utc), | ||
| 'EUR', | ||
| ), | ||
| source.SourcePrice( | ||
| Decimal('87.89400000'), | ||
| datetime.datetime(2024, 1, 4, tzinfo=datetime.timezone.utc), | ||
| 'EUR', | ||
| ), | ||
| ], | ||
| price_3, | ||
| ) | ||
| self.assertTrue(price_3[0].time.tzinfo) | ||
| self.assertTrue(price_3[1].time.tzinfo) | ||
|
|
||
| def test_empty_response(self): | ||
| contents = [] | ||
| with response(contents): | ||
| price_1 = self.sut.get_historical_price( | ||
| SYMBOL_A1JX52_UUID, | ||
| datetime.datetime(2024, 1, 1), | ||
| ) | ||
| self.assertIsNone(price_1) | ||
|
|
||
| price_2 = self.sut.get_latest_price(SYMBOL_A1JX52_UUID) | ||
| self.assertIsNone(price_2) | ||
|
|
||
| price_3 = self.sut.get_prices_series( | ||
| SYMBOL_A1JX52_UUID, | ||
| datetime.datetime(2024, 1, 3), | ||
| datetime.datetime(2024, 1, 4), | ||
| ) | ||
| self.assertIsNone(price_3) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() |