Coverage for source/data_handling/coinbase_handler.py: 13%

1# data_handling/coinbase_handler.py 

2 

3# global imports 

4import aiohttp 

5import asyncio 

6import math 

7import pandas as pd 

8import pytz 

9from datetime import datetime 

10 

11# local imports 

12from source.utils import Granularity 

13 

14class CoinBaseHandler: 

15 """ 

16 Responsible for handling request towards Coinbase API. 

17 """ 

18 

19 # local constants 

20 __MAX_NUMBER_OF_CANDLES_PER_REQUEST = 300 

21 __PRODUCTS_URL = 'https://api.exchange.coinbase.com/products' 

22 

23 def __convert_date_to_timestamp(self, date_str: str, date_format: str = "%Y-%m-%d %H:%M:%S", target_timezone = pytz.UTC) -> int: 

24 """ 

25 Converts date given by string into integer timestamp. 

26 

27 Parameters: 

28 date_str (str): String representing certain date. 

29 date_format (str): String representing format that certain date is written in. 

30 target_timezone (Any): Timezone given by any type of value compatible with datetime library. 

31 

32 Raises: 

33 ValueError: In case of invalid date conversion. 

34 

35 Returns: 

36 (int): Timestamp converted from input date. 

37 """ 

38 

39 try: 

40 result = int(datetime.strptime(date_str, date_format).replace(tzinfo=target_timezone).timestamp()) 

41 except: 

42 raise ValueError(f'Invalid data format! Expected was {date_format}.') 

43 

44 return result 

45 

46 async def __send_request_to_coinbase(self, session: aiohttp.ClientSession, url: str, pid: int) -> list: 

47 """ 

48 Sends request towards Coinbase API and handles exceedance of public rates by repeating 

49 request after certain time. 

50 

51 Parameters: 

52 session (aiohttp.ClientSession): Session used to send request with. 

53 url (str): URL address that certain request is sent towards. 

54 pid (int): Request indentification number. 

55 

56 Raises: 

57 RuntimeError: If public rates are exceeded. Will try to handle that and reattempt 

58 to sent request. 

59 

60 Returns: 

61 (list): List of values returned by Coinbase API for certain request. 

62 """ 

63 

64 try: 

65 async with session.get(url) as response: 

66 data = await response.json() 

67 if 'message' in data and data['message'] == 'Public rate limit exceeded': 

68 raise RuntimeError("Exceeded public rate! Retrying in 5s...") 

69 return data 

70 except: 

71 await asyncio.sleep(5) 

72 return await self.__send_request_to_coinbase(session, url, pid) 

73 

74 async def get_candles_for(self, trading_pair: str, start_date: str, end_date: str, granularity: Granularity) -> pd.DataFrame: 

75 """ 

76 Collects data from Coinbase API given starting date, ending date, granularity and trainding pair. Dependent on amount of 

77 data segments to fetch, might take some time. Especially, if request exceeds public rates. 

78 

79 Parameters: 

80 trading_pair (str): String representing unique trainding pair symbol. 

81 start_date (str): String representing date that collected data should start from. 

82 end_date (str): String representing date that collected data should finish at. 

83 granularity (Granularity): Enum specifying resolution of collected data - e.g. each 

84 15 minutes or 1 hour or 6 hours is treated separately 

85 

86 Raises: 

87 ValueError: If given granularity is not member if Granularity enum. 

88 

89 Returns: 

90 (pd.DataFrame): Collected data frame. 

91 """ 

92 

93 if granularity not in Granularity: 

94 raise ValueError(f"{granularity} is not an value of Granularity enum!") 

95 

96 start_timestamp = self.__convert_date_to_timestamp(start_date) 

97 end_timestamp = self.__convert_date_to_timestamp(end_date) 

98 granularity_seconds = granularity.value 

99 total_periods = (end_timestamp - start_timestamp) // granularity_seconds 

100 requests_needed = math.ceil(total_periods / self.__MAX_NUMBER_OF_CANDLES_PER_REQUEST) 

101 

102 async with aiohttp.ClientSession() as session: 

103 tasks = [] 

104 for i in range(requests_needed): 

105 start_period = start_timestamp + i * self.__MAX_NUMBER_OF_CANDLES_PER_REQUEST * granularity_seconds 

106 end_period = min(start_period + self.__MAX_NUMBER_OF_CANDLES_PER_REQUEST * granularity_seconds, end_timestamp) 

107 url = f'{self.__PRODUCTS_URL}/{trading_pair}/candles?start={start_period}&end={end_period}&granularity={granularity_seconds}' 

108 tasks.append(self.__send_request_to_coinbase(session, url, i)) 

109 

110 responses = await asyncio.gather(*tasks) 

111 candles = [item for sublist in responses if sublist for item in sublist] 

112 df = pd.DataFrame(candles, columns=['time', 'low', 'high', 'open', 'close', 'volume']) 

113 df['time'] = pd.to_datetime(df['time'], unit='s') 

114 df.set_index('time', inplace=True) 

115 df.sort_values(by='time', inplace=True) 

116 return df 

117 

118 async def get_possible_pairs(self) -> pd.DataFrame: 

119 """ 

120 Collects data from Coinbase API reagrding all possible traiding pairs. 

121 

122 Returns: 

123 (pd.DataFrame): Fetched possible traiding pairs inside data frame. 

124 """ 

125 

126 async with aiohttp.ClientSession() as session: 

127 response = await asyncio.gather(self.__send_request_to_coinbase(session, self.__PRODUCTS_URL, 0)) 

128 data = [[product['id'], product['base_currency'], product['quote_currency']] for product in response[0]] 

129 df = pd.DataFrame(sorted(data), columns=['id', 'base_currency', 'quote_currency']) 

130 df.set_index('id', inplace=True) 

131 return df