Coverage for source/data_handling/data_handler.py: 94%

1# data_handling/data_handler.py 

2 

3import pandas as pd 

4from ..utils import Granularity 

5from ..coinbase import CoinBaseHandler 

6 

7class DataHandler(): 

8 """ 

9 Responsible for data handling. Including data collection and preparation. 

10 """ 

11 

12 def __init__(self, list_of_indicators_to_apply: list = []) -> None: 

13 """ 

14 Class constructor. 

15 

16 Parameters: 

17 list_of_indicators_to_apply (list): List of indicators further to apply. 

18 """ 

19 

20 self.indicators = list_of_indicators_to_apply 

21 self.coinbase = CoinBaseHandler() 

22 

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

24 """ 

25 Collects data from coinbase API and extends it with assigned list of indicators. 

26 

27 Parameters: 

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

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

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

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

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

33 

34 Raises: 

35 RuntimeError: If given traiding pair symbol is not recognized. 

36 

37 Returns: 

38 (pd.DataFrame): Collected data extended with given indicators. 

39 """ 

40 

41 possible_traiding_pairs = await self.coinbase.get_possible_pairs() 

42 if trading_pair not in possible_traiding_pairs.index: 

43 raise RuntimeError('Traiding pair not recognized!') 

44 

45 data = await self.coinbase.get_candles_for(trading_pair, start_date, end_date, granularity) 

46 if self.indicators: 

47 indicators_data = [] 

48 for indicator in self.indicators: 

49 indicators_data.append(indicator.calculate(data)) 

50 data = pd.concat([data] + indicators_data, axis=1) 

51 

52 return data 

53 

54 

55