Coverage for mindsdb / integrations / handlers / financial_modeling_prep_handler / financial_modeling_tables.py: 0%

1from mindsdb.integrations.libs.api_handler import APITable 

2from mindsdb.integrations.utilities.sql_utils import extract_comparison_conditions 

3from mindsdb_sql_parser import ast 

4 

5from typing import Dict, List 

6 

7import pandas as pd 

8 

9from mindsdb.integrations.libs.response import ( 

10 HandlerResponse as Response, 

11 RESPONSE_TYPE 

12) 

13import requests 

14 

15 

16class HistoricalPriceTable(APITable): 

17 

18 def _get_historical_price_endpoint_params_from_conditions(self, conditions: List) -> Dict: 

19 params = {} 

20 for op, arg1, arg2 in conditions: 

21 if arg1 == 'symbol': 

22 if op != '=': 

23 raise NotImplementedError 

24 params['symbol'] = arg2 

25 if arg1 == "from_date": 

26 if op != '=': 

27 raise NotImplementedError 

28 params['from'] = arg2 

29 if arg1 == "to_date": 

30 if op != '=': 

31 raise NotImplementedError 

32 params['to'] = arg2 

33 

34 return params 

35 

36 def select(self, query: ast.Select) -> pd.DataFrame: 

37 """Selects data from the FinancialModeling API and returns it as a pandas DataFrame. 

38 

39 Returns dataframe representing the FinancialModeling API results. 

40 

41 Args: 

42 query (ast.Select): Given SQL SELECT query 

43 """ 

44 conditions = extract_comparison_conditions(query.where) 

45 params = self._get_historical_price_endpoint_params_from_conditions(conditions) 

46 

47 if query.limit and query.limit.value: 

48 limit_value = query.limit.value 

49 params['limit'] = limit_value 

50 

51 historical_prices = self.get_historical_price_chart(params=params) 

52 

53 return historical_prices 

54 

55 def get_historical_price_chart(self, params: Dict = None) -> pd.DataFrame: 

56 base_url = self.handler.connect() 

57 if 'symbol' not in params: 

58 raise ValueError('Missing "symbol" param') 

59 symbol = params['symbol'] 

60 params.pop('symbol') 

61 

62 limitParam = False 

63 limit = 0 

64 if 'limit' in params: 

65 limit = params['limit'] 

66 params.pop('limit') 

67 limitParam = True 

68 

69 url = f"{base_url}{symbol}" # https://financialmodelingprep.com/api/v3/historical-price-full/<symbol> 

70 param = {'apikey': self.handler.api_key, **params} 

71 

72 response = requests.get(url, param) 

73 historical_data = response.json() 

74 historical = historical_data.get("historical") 

75 

76 if limitParam: 

77 return pd.DataFrame(historical).head(limit) 

78 

79 response = Response( 

80 RESPONSE_TYPE.TABLE, 

81 data_frame=pd.DataFrame( 

82 historical 

83 ) 

84 ) 

85 

86 if historical: 

87 return pd.DataFrame(historical) 

88 else: 

89 return pd.DataFrame()