4599645360

Build Type

push

github

Committed by GitHub

Commit Message

Bump cryptography from 2.5 to 39.0.1

Run Details

2807 of 6557 branches covered (42.81%)

Branch coverage included in aggregate %.

12289 of 19915 relevant lines covered (61.71%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.0

/lbry/extras/daemon/exchange_rate_manager.py

import json
import time
import asyncio
import logging
from statistics import median
from decimal import Decimal
from typing import Optional, Iterable, Type
from aiohttp.client_exceptions import ContentTypeError, ClientConnectionError
from lbry.error import InvalidExchangeRateResponseError, CurrencyConversionError
from lbry.utils import aiohttp_request
from lbry.wallet.dewies import lbc_to_dewies

log = logging.getLogger(__name__)


class ExchangeRate:
    def __init__(self, market, spot, ts):
        if not int(time.time()) - ts < 600:
            raise ValueError('The timestamp is too dated.')
        if not spot > 0:
            raise ValueError('Spot must be greater than 0.')
        self.currency_pair = (market[0:3], market[3:6])
        self.spot = spot
        self.ts = ts

    def __repr__(self):
        return f"Currency pair:{self.currency_pair}, spot:{self.spot}, ts:{self.ts}"

    def as_dict(self):
        return {'spot': self.spot, 'ts': self.ts}


class MarketFeed:
    name: str = ""
    market: str = ""
    url: str = ""
    params = {}
    fee = 0

    update_interval = 300
    request_timeout = 50

    def __init__(self):
        self.rate: Optional[float] = None
        self.last_check = 0
        self._last_response = None
        self._task: Optional[asyncio.Task] = None
        self.event = asyncio.Event()

    @property
    def has_rate(self):
        return self.rate is not None

    @property
    def is_online(self):
        return self.last_check+self.update_interval+self.request_timeout > time.time()

    def get_rate_from_response(self, json_response):
        raise NotImplementedError()

    async def get_response(self):
        async with aiohttp_request(
                'get', self.url, params=self.params,
                timeout=self.request_timeout, headers={"User-Agent": "lbrynet"}
        ) as response:
            try:
                self._last_response = await response.json(content_type=None)
            except ContentTypeError as e:
                self._last_response = {}
                log.warning("Could not parse exchange rate response from %s: %s", self.name, e.message)
                log.debug(await response.text())
            return self._last_response

    async def get_rate(self):
        try:
            data = await self.get_response()
            rate = self.get_rate_from_response(data)
            rate = rate / (1.0 - self.fee)
            log.debug("Saving rate update %f for %s from %s", rate, self.market, self.name)
            self.rate = ExchangeRate(self.market, rate, int(time.time()))
            self.last_check = time.time()
            return self.rate
        except asyncio.TimeoutError:
            log.warning("Timed out fetching exchange rate from %s.", self.name)
        except json.JSONDecodeError as e:
            msg = e.doc if '<html>' not in e.doc else 'unexpected content type.'
            log.warning("Could not parse exchange rate response from %s: %s", self.name, msg)
            log.debug(e.doc)
        except InvalidExchangeRateResponseError as e:
            log.warning(str(e))
        except ClientConnectionError as e:
            log.warning("Error trying to connect to exchange rate %s: %s", self.name, str(e))
        except Exception as e:
            log.exception("Exchange rate error (%s from %s):", self.market, self.name)
        finally:
            self.event.set()

    async def keep_updated(self):
        while True:
            await self.get_rate()
            await asyncio.sleep(self.update_interval)

    def start(self):
        if not self._task:
            self._task = asyncio.create_task(self.keep_updated())

    def stop(self):
        if self._task and not self._task.done():
            self._task.cancel()
        self._task = None
        self.event.clear()


class BaseBittrexFeed(MarketFeed):
    name = "Bittrex"
    market = None
    url = None
    fee = 0.0025

    def get_rate_from_response(self, json_response):
        if 'lastTradeRate' not in json_response:
            raise InvalidExchangeRateResponseError(self.name, 'result not found')
        return 1.0 / float(json_response['lastTradeRate'])


class BittrexBTCFeed(BaseBittrexFeed):
    market = "BTCLBC"
    url = "https://api.bittrex.com/v3/markets/LBC-BTC/ticker"


class BittrexUSDFeed(BaseBittrexFeed):
    market = "USDLBC"
    url = "https://api.bittrex.com/v3/markets/LBC-USD/ticker"


class BaseCoinExFeed(MarketFeed):
    name = "CoinEx"
    market = None
    url = None

    def get_rate_from_response(self, json_response):
        if 'data' not in json_response or \
           'ticker' not in json_response['data'] or \
           'last' not in json_response['data']['ticker']:
            raise InvalidExchangeRateResponseError(self.name, 'result not found')
        return 1.0 / float(json_response['data']['ticker']['last'])


class CoinExBTCFeed(BaseCoinExFeed):
    market = "BTCLBC"
    url = "https://api.coinex.com/v1/market/ticker?market=LBCBTC"


class CoinExUSDFeed(BaseCoinExFeed):
    market = "USDLBC"
    url = "https://api.coinex.com/v1/market/ticker?market=LBCUSDT"


class BaseHotbitFeed(MarketFeed):
    name = "hotbit"
    market = None
    url = "https://api.hotbit.io/api/v1/market.last"

    def get_rate_from_response(self, json_response):
        if 'result' not in json_response:
            raise InvalidExchangeRateResponseError(self.name, 'result not found')
        return 1.0 / float(json_response['result'])


class HotbitBTCFeed(BaseHotbitFeed):
    market = "BTCLBC"
    params = {"market": "LBC/BTC"}


class HotbitUSDFeed(BaseHotbitFeed):
    market = "USDLBC"
    params = {"market": "LBC/USDT"}


class UPbitBTCFeed(MarketFeed):
    name = "UPbit"
    market = "BTCLBC"
    url = "https://api.upbit.com/v1/ticker"
    params = {"markets": "BTC-LBC"}

    def get_rate_from_response(self, json_response):
        if "error" in json_response or len(json_response) != 1 or 'trade_price' not in json_response[0]:
            raise InvalidExchangeRateResponseError(self.name, 'result not found')
        return 1.0 / float(json_response[0]['trade_price'])


FEEDS: Iterable[Type[MarketFeed]] = (
    BittrexBTCFeed,
    BittrexUSDFeed,
    CoinExBTCFeed,
    CoinExUSDFeed,
#    HotbitBTCFeed,
#    HotbitUSDFeed,
#    UPbitBTCFeed,
)


class ExchangeRateManager:
    def __init__(self, feeds=FEEDS):
        self.market_feeds = [Feed() for Feed in feeds]

    def wait(self):
        return asyncio.wait(
            [feed.event.wait() for feed in self.market_feeds],
        )

    def start(self):
        log.info("Starting exchange rate manager")
        for feed in self.market_feeds:
            feed.start()

    def stop(self):
        log.info("Stopping exchange rate manager")
        for source in self.market_feeds:
            source.stop()

    def convert_currency(self, from_currency, to_currency, amount):
        log.debug(
            "Converting %f %s to %s, rates: %s",
            amount, from_currency, to_currency,
            [market.rate for market in self.market_feeds]
        )
        if from_currency == to_currency:
            return round(amount, 8)

        rates = []
        for market in self.market_feeds:
            if (market.has_rate and market.is_online and
                    market.rate.currency_pair == (from_currency, to_currency)):
                rates.append(market.rate.spot)

        if rates:
            return round(amount * Decimal(median(rates)), 8)

        raise CurrencyConversionError(
            f'Unable to convert {amount} from {from_currency} to {to_currency}')

    def to_dewies(self, currency, amount) -> int:
        converted = self.convert_currency(currency, "LBC", amount)
        return lbc_to_dewies(str(converted))

    def fee_dict(self):
        return {market: market.rate.as_dict() for market in self.market_feeds}

1	import json	1✔
2	import time	1✔
3	import asyncio	1✔
4	import logging	1✔
5	from statistics import median	1✔
6	from decimal import Decimal	1✔
7	from typing import Optional, Iterable, Type	1✔
8	from aiohttp.client_exceptions import ContentTypeError, ClientConnectionError	1✔
9	from lbry.error import InvalidExchangeRateResponseError, CurrencyConversionError	1✔
10	from lbry.utils import aiohttp_request	1✔
11	from lbry.wallet.dewies import lbc_to_dewies	1✔
12
13	log = logging.getLogger(__name__)	1✔
14
15
16	class ExchangeRate:	1✔
17	def __init__(self, market, spot, ts):	1✔
18	if not int(time.time()) - ts < 600:	1!
19	raise ValueError('The timestamp is too dated.')	×
20	if not spot > 0:	1✔
21	raise ValueError('Spot must be greater than 0.')	1✔
22	self.currency_pair = (market[0:3], market[3:6])	1✔
23	self.spot = spot	1✔
24	self.ts = ts	1✔
25
26	def __repr__(self):	1✔
27	return f"Currency pair:{self.currency_pair}, spot:{self.spot}, ts:{self.ts}"	×
28
29	def as_dict(self):	1✔
30	return {'spot': self.spot, 'ts': self.ts}	×
31
32
33	class MarketFeed:	1✔
34	name: str = ""	1✔
35	market: str = ""	1✔
36	url: str = ""	1✔
37	params = {}	1✔
38	fee = 0	1✔
39
40	update_interval = 300	1✔
41	request_timeout = 50	1✔
42
43	def __init__(self):	1✔
44	self.rate: Optional[float] = None	1✔
45	self.last_check = 0	1✔
46	self._last_response = None	1✔
47	self._task: Optional[asyncio.Task] = None	1✔
48	self.event = asyncio.Event()	1✔
49
50	@property	1✔
51	def has_rate(self):	1✔
52	return self.rate is not None	1✔
53
54	@property	1✔
55	def is_online(self):	1✔
56	return self.last_check+self.update_interval+self.request_timeout > time.time()	1✔
57
58	def get_rate_from_response(self, json_response):	1✔
59	raise NotImplementedError()	×
60
61	async def get_response(self):	1✔
62	async with aiohttp_request(	×
63	'get', self.url, params=self.params,
64	timeout=self.request_timeout, headers={"User-Agent": "lbrynet"}
65	) as response:
66	try:	×
67	self._last_response = await response.json(content_type=None)	×
68	except ContentTypeError as e:	×
69	self._last_response = {}	×
70	log.warning("Could not parse exchange rate response from %s: %s", self.name, e.message)	×
71	log.debug(await response.text())	×
72	return self._last_response	×
73
74	async def get_rate(self):	1✔
75	try:	1✔
76	data = await self.get_response()	1✔
77	rate = self.get_rate_from_response(data)	×
78	rate = rate / (1.0 - self.fee)	×
79	log.debug("Saving rate update %f for %s from %s", rate, self.market, self.name)	×
80	self.rate = ExchangeRate(self.market, rate, int(time.time()))	×
81	self.last_check = time.time()	×
82	return self.rate	×
83	except asyncio.TimeoutError:	1!
84	log.warning("Timed out fetching exchange rate from %s.", self.name)	×
85	except json.JSONDecodeError as e:	1!
86	msg = e.doc if '<html>' not in e.doc else 'unexpected content type.'	×
87	log.warning("Could not parse exchange rate response from %s: %s", self.name, msg)	×
88	log.debug(e.doc)	×
89	except InvalidExchangeRateResponseError as e:	1!
90	log.warning(str(e))	1✔
91	except ClientConnectionError as e:	×
92	log.warning("Error trying to connect to exchange rate %s: %s", self.name, str(e))	×
93	except Exception as e:	×
94	log.exception("Exchange rate error (%s from %s):", self.market, self.name)	×
95	finally:
96	self.event.set()	1!
97
98	async def keep_updated(self):	1✔
99	while True:
100	await self.get_rate()	1✔
101	await asyncio.sleep(self.update_interval)	1✔
102
103	def start(self):	1✔
104	if not self._task:	1!
105	self._task = asyncio.create_task(self.keep_updated())	1✔
106
107	def stop(self):	1✔
108	if self._task and not self._task.done():	1!
109	self._task.cancel()	1✔
110	self._task = None	1✔
111	self.event.clear()	1✔
112
113
114	class BaseBittrexFeed(MarketFeed):	1✔
115	name = "Bittrex"	1✔
116	market = None	1✔
117	url = None	1✔
118	fee = 0.0025	1✔
119
120	def get_rate_from_response(self, json_response):	1✔
121	if 'lastTradeRate' not in json_response:	1✔
122	raise InvalidExchangeRateResponseError(self.name, 'result not found')	1✔
123	return 1.0 / float(json_response['lastTradeRate'])	1✔
124
125
126	class BittrexBTCFeed(BaseBittrexFeed):	1✔
127	market = "BTCLBC"	1✔
128	url = "https://api.bittrex.com/v3/markets/LBC-BTC/ticker"	1✔
129
130
131	class BittrexUSDFeed(BaseBittrexFeed):	1✔
132	market = "USDLBC"	1✔
133	url = "https://api.bittrex.com/v3/markets/LBC-USD/ticker"	1✔
134
135
136	class BaseCoinExFeed(MarketFeed):	1✔
137	name = "CoinEx"	1✔
138	market = None	1✔
139	url = None	1✔
140
141	def get_rate_from_response(self, json_response):	1✔
142	if 'data' not in json_response or \	×
143	'ticker' not in json_response['data'] or \
144	'last' not in json_response['data']['ticker']:
145	raise InvalidExchangeRateResponseError(self.name, 'result not found')	×
146	return 1.0 / float(json_response['data']['ticker']['last'])	×
147
148
149	class CoinExBTCFeed(BaseCoinExFeed):	1✔
150	market = "BTCLBC"	1✔
151	url = "https://api.coinex.com/v1/market/ticker?market=LBCBTC"	1✔
152
153
154	class CoinExUSDFeed(BaseCoinExFeed):	1✔
155	market = "USDLBC"	1✔
156	url = "https://api.coinex.com/v1/market/ticker?market=LBCUSDT"	1✔
157
158
159	class BaseHotbitFeed(MarketFeed):	1✔
160	name = "hotbit"	1✔
161	market = None	1✔
162	url = "https://api.hotbit.io/api/v1/market.last"	1✔
163
164	def get_rate_from_response(self, json_response):	1✔
165	if 'result' not in json_response:	×
166	raise InvalidExchangeRateResponseError(self.name, 'result not found')	×
167	return 1.0 / float(json_response['result'])	×
168
169
170	class HotbitBTCFeed(BaseHotbitFeed):	1✔
171	market = "BTCLBC"	1✔
172	params = {"market": "LBC/BTC"}	1✔
173
174
175	class HotbitUSDFeed(BaseHotbitFeed):	1✔
176	market = "USDLBC"	1✔
177	params = {"market": "LBC/USDT"}	1✔
178
179
180	class UPbitBTCFeed(MarketFeed):	1✔
181	name = "UPbit"	1✔
182	market = "BTCLBC"	1✔
183	url = "https://api.upbit.com/v1/ticker"	1✔
184	params = {"markets": "BTC-LBC"}	1✔
185
186	def get_rate_from_response(self, json_response):	1✔
187	if "error" in json_response or len(json_response) != 1 or 'trade_price' not in json_response[0]:	×
188	raise InvalidExchangeRateResponseError(self.name, 'result not found')	×
189	return 1.0 / float(json_response[0]['trade_price'])	×
190
191
192	FEEDS: Iterable[Type[MarketFeed]] = (	1✔
193	BittrexBTCFeed,
194	BittrexUSDFeed,
195	CoinExBTCFeed,
196	CoinExUSDFeed,
197	# HotbitBTCFeed,
198	# HotbitUSDFeed,
199	# UPbitBTCFeed,
200	)
201
202
203	class ExchangeRateManager:	1✔
204	def __init__(self, feeds=FEEDS):	1✔
205	self.market_feeds = [Feed() for Feed in feeds]	1✔
206
207	def wait(self):	1✔
208	return asyncio.wait(	1✔
209	[feed.event.wait() for feed in self.market_feeds],
210	)
211
212	def start(self):	1✔
213	log.info("Starting exchange rate manager")	1✔
214	for feed in self.market_feeds:	1✔
215	feed.start()	1✔
216
217	def stop(self):	1✔
218	log.info("Stopping exchange rate manager")	1✔
219	for source in self.market_feeds:	1✔
220	source.stop()	1✔
221
222	def convert_currency(self, from_currency, to_currency, amount):	1✔
223	log.debug(	1✔
224	"Converting %f %s to %s, rates: %s",
225	amount, from_currency, to_currency,
226	[market.rate for market in self.market_feeds]
227	)
228	if from_currency == to_currency:	1✔
229	return round(amount, 8)	1✔
230
231	rates = []	1✔
232	for market in self.market_feeds:	1✔
233	if (market.has_rate and market.is_online and	1✔
234	market.rate.currency_pair == (from_currency, to_currency)):
235	rates.append(market.rate.spot)	1✔
236
237	if rates:	1✔
238	return round(amount * Decimal(median(rates)), 8)	1✔
239
240	raise CurrencyConversionError(	1✔
241	f'Unable to convert {amount} from {from_currency} to {to_currency}')
242
243	def to_dewies(self, currency, amount) -> int:	1✔
244	converted = self.convert_currency(currency, "LBC", amount)	1✔
245	return lbc_to_dewies(str(converted))	1✔
246
247	def fee_dict(self):	1✔
248	return {market: market.rate.as_dict() for market in self.market_feeds}	×

lbryio / lbry-sdk / 4599645360

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous