Coverage for mindsdb / integrations / handlers / coinbase_handler / coinbase_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 

5import pandas as pd 

6 

7 

8class CoinBaseAggregatedTradesTable(APITable): 

9 

10 DEFAULT_INTERVAL = 60 

11 DEFAULT_SYMBOL = 'BTC-USD' 

12 

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

14 """Selects data from the CoinBase API and returns it as a pandas DataFrame. 

15 

16 Returns dataframe representing the CoinBase API results. 

17 

18 Args: 

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

20 """ 

21 conditions = extract_comparison_conditions(query.where) 

22 

23 params = { 

24 'interval': CoinBaseAggregatedTradesTable.DEFAULT_INTERVAL, 

25 'symbol': CoinBaseAggregatedTradesTable.DEFAULT_SYMBOL, 

26 } 

27 for op, arg1, arg2 in conditions: 

28 if arg1 == 'interval': 

29 if op != '=': 

30 raise NotImplementedError 

31 params['interval'] = arg2 

32 

33 elif arg1 == 'symbol': 

34 if op != '=': 

35 raise NotImplementedError 

36 params['symbol'] = arg2 

37 

38 coinbase_candle_data = self.handler.call_coinbase_api( 

39 method_name='get_candle', 

40 params=params 

41 ) 

42 

43 return coinbase_candle_data 

44 

45 def get_columns(self): 

46 """Gets all columns to be returned in pandas DataFrame responses""" 

47 return [ 

48 'symbol', 

49 'low', 

50 'high', 

51 'open', 

52 'close', 

53 'volume', 

54 'timestamp', 

55 'current_time' 

56 ]