Coverage for source/data_handling/api_data_collector_base.py: 59%

1# data_handling/api_data_collector_base.py 

2 

3# global imports 

4import pandas as pd 

5from abc import ABC, abstractmethod 

6from datetime import datetime 

7 

8# local imports 

9from source.utils import Granularity 

10 

11class ApiDataCollectorBase(ABC): 

12 """ 

13 Implements a base class for collecting data from various APIs. 

14 """ 

15 

16 @abstractmethod 

17 async def _validate_ticker(self, ticker: str) -> bool: 

18 """ 

19 Validates if the ticker is supported by the API. 

20 

21 Parameters: 

22 ticker (str): Stock ticker to validate. 

23 

24 Returns: 

25 bool: True if ticker is valid, False otherwise. 

26 """ 

27 

28 pass 

29 

30 @abstractmethod 

31 async def _collect_data_for_ticker(self, ticker: str, start_date: str, end_date: str, 

32 granularity: Granularity) -> tuple[pd.DataFrame, dict[str, str]]: 

33 """ 

34 Collects data for a specific ticker from the API. 

35 

36 Parameters: 

37 ticker (str): Stock ticker to collect data for. 

38 start_date (str): Start date for the data collection. 

39 end_date (str): End date for the data collection. 

40 granularity (Granularity): Data resolution. 

41 

42 Returns: 

43 tuple[pd.DataFrame, dict[str, str]]: A tuple containing the collected data and metadata. 

44 """ 

45 

46 pass 

47 

48 def _convert_date_to_datetime(self, date_str: str, date_format: str = "%Y-%m-%d %H:%M:%S") -> datetime: 

49 """ 

50 Converts date string to datetime object. 

51 

52 Parameters: 

53 date_str (str): String representing date. 

54 date_format (str): Format of the date string. 

55 

56 Returns: 

57 datetime: Converted datetime object. 

58 """ 

59 try: 

60 return datetime.strptime(date_str, date_format) 

61 except ValueError: 

62 raise ValueError(f'Invalid date format! Expected was {date_format}.') 

63 

64 async def collect_data(self, ticker: str, start_date: str, end_date: str, 

65 granularity: Granularity) -> tuple[pd.DataFrame, dict[str, str]]: 

66 """ 

67 Collects data for a specific ticker from the API. 

68 

69 Parameters: 

70 ticker (str): Stock ticker to collect data for. 

71 start_date (str): Start date for the data collection. 

72 end_date (str): End date for the data collection. 

73 granularity (Granularity): Data resolution. 

74 

75 Raises: 

76 ValueError: If the ticker is not supported by the API or if the granularity is not valid. 

77 

78 Returns: 

79 tuple[pd.DataFrame, dict[str, str]]: A tuple containing the collected data and metadata. 

80 """ 

81 

82 if granularity not in Granularity: 

83 raise ValueError(f"{granularity} is not a valid value of Granularity enum!") 

84 

85 if not await self._validate_ticker(ticker): 

86 raise ValueError(f"Ticker {ticker} is not supported by this API.") 

87 

88 return await self._collect_data_for_ticker(ticker, start_date, end_date, granularity)