add api error recognition

Author: sgerodes

setup.py

@@ -9,8 +9,9 @@
               'three_commas.api',
               'three_commas.api.ver1',
               'three_commas.api.v2',
-              'three_commas.model'],
-    version='0.0.24',
+              'three_commas.model',
+              'three_commas.utils'],
+    version='0.0.26',
     description='Python api wrapper for 3commas with extended functionality in the api, models, error handling',
     url='https://github.com/badass-blockchain/python-three-commas',
     author='Sergey Gerodes',

src/three_commas/error.py

@@ -20,51 +20,71 @@ class BoToSmallError:
     EXTRACT_PY3CW_MESSAGE_PATTERN = re.compile(r"Other error occurred: record_invalid Invalid parameters (\{.*\})\.", re.IGNORECASE)
     BOT_WAS_DELETED_ERROR_PATTERN = re.compile(r"Other error occurred: Not found None None", re.IGNORECASE)
     BOT_DID_NOT_EXISTED_OR_BELONGS_TO_OTHER_ACCOUNT_ERROR_PATTERN = re.compile(r"Other error occurred: not_found Not Found None", re.IGNORECASE)
+    API_KEY_NOT_ENOUGH_PERMISSION_PATTERN = re.compile(r"access_denied Api key doesn't have enough permissions", re.IGNORECASE)
+    API_KEY_INVALID_OR_EXPIRED_PATTERN = re.compile(r'api_key_invalid_or_expired Unauthorized. Invalid or expired api key', re.IGNORECASE)
 
     def __init__(self, error):
         self.error: dict = error
-        self.error_parsed: dict = None
         try:
             if self._has_error_message():
-                match = ThreeCommasError.EXTRACT_PY3CW_MESSAGE_PATTERN.findall(error.get('msg'))
+                match = ThreeCommasError.EXTRACT_PY3CW_MESSAGE_PATTERN.findall(self.get_msg())
                 if match:
                     self.error_parsed = eval(match[0])
         except:
             logger.warning(f'Failed to parse inner error msg {error}')
 
-    def bo_to_small_error(self) -> List[BoToSmallError]:
-        ret = list()
-        # ret = ThreeCommasError.BoToSmallError(present=False)
-        if self._has_parsed_error_message() and self.error_parsed.get('base_order_volume'):
-            # ret.present = True
-            for sub_message in self.error_parsed.get('base_order_volume'):
-                bo_min_match = ThreeCommasError.BO_TO_SMALL_ERROR_PATTERN.findall(sub_message)
-                if bo_min_match:
-                    amount = float(bo_min_match[0][0])
-                    pair = bo_min_match[0][1] or None
-                    ret.append(ThreeCommasError.BoToSmallError(amount=amount, pair=pair))
-                    #ret.matches.append((amount, pair))
-                    # ret.amount.append(float(bo_min_match[0][0]))
-                    # ret.pair.append(bo_min_match[0][1] or None)
+    def is_api_key_has_no_permission_error(self):
+        return self._has_error_message() and self.API_KEY_NOT_ENOUGH_PERMISSION_PATTERN.findall(self.get_msg())
 
-        return ret
+    def is_api_key_invalid_or_expired(self):
+        return self._has_error_message() and self.API_KEY_INVALID_OR_EXPIRED_PATTERN.findall(self.get_msg())
+
+    def is_bo_to_small_error(self):
+        return self._has_error_message() and self.BO_TO_SMALL_ERROR_PATTERN.findall(self.get_msg())
+
+    def is_not_found_error(self):
+        return self._has_error_message() and 'not_found' in self.get_msg() or 'Not found' in self.get_msg()
 
     def is_no_market_pair_error(self) -> List[str]:
+        return self._has_error_message() and self.NO_MARKET_PAIR_ERROR_PATTERN.findall(self.get_msg())
+
+    def get_no_market_pair_error(self) -> List[str]:
         if self._has_error_message():
             pairs_to_remove = ThreeCommasError.NO_MARKET_PAIR_ERROR_PATTERN.findall(self.error.get('msg'))
             if pairs_to_remove:
                 return pairs_to_remove
+        return list()
 
-    def is_not_found_error(self):
-        return self._has_error_message() and 'not_found' in self.error.get('msg') or 'Not found' in self.error.get('msg')
+    def get_bo_to_small_error(self) -> List[BoToSmallError]:
+        ret = list()
+        if self._has_error_message():
+            try:
+                match = ThreeCommasError.EXTRACT_PY3CW_MESSAGE_PATTERN.findall(self.get_msg())
+                if match:
+                    error_parsed = eval(match[0])
+                else:
+                    return list()
+            except:
+                return list()
+            if error_parsed.get('base_order_volume'):
+                for sub_message in error_parsed.get('base_order_volume'):
+                    bo_min_match = ThreeCommasError.BO_TO_SMALL_ERROR_PATTERN.findall(sub_message)
+                    if bo_min_match:
+                        amount = float(bo_min_match[0][0])
+                        pair = bo_min_match[0][1] or None
+                        ret.append(ThreeCommasError.BoToSmallError(amount=amount, pair=pair))
+        return ret
 
     def _has_error_message(self):
-        return self.error and self.error.get('msg')
+        return self.error and self.get_msg()
 
     def _has_parsed_error_message(self):
         return self.error_parsed
 
+    def get_msg(self):
+        return self.error.get('msg')
+
     def __repr__(self):
-        return f'{self.__class__}({self.error})'
+        return f'{self.__class__.__name__}({self.error})'
 
 

src/three_commas/utils/__init__.py

@@ -1 +1,2 @@
 from .bot_utils import *
+from .pairs_utils import *

src/three_commas/utils/bot_utils.py

@@ -1,5 +1,5 @@
 import logging
-from src.three_commas.model import Bot
+from ..model import Bot
 from typing import List, Union
 
 

src/three_commas/utils/pairs_utils.py

@@ -0,0 +1,40 @@
+from typing import List
+
+def get_base_from_3c_pair(tc_pair: str, account_market_code: str = None) -> str:
+    if account_market_code is None or account_market_code in {'ftx', 'binance', 'paper_trading'}:
+        return tc_pair.split('_')[1].upper()
+    elif account_market_code in {'ftx_futures'}:
+        return tc_pair.split('_')[1].split('-')[0]
+    else:
+        raise RuntimeError(f'Not known market code {account_market_code} in get_base_from_3c_pair')
+
+
+def filter_market_pairs_with_quote(market_pairs: List[str], quote: str):
+    return [pair for pair in market_pairs if pair.upper().startswith(quote.upper())]
+
+
+def pair_is_quote(tc_pair: str, quote: str) -> bool:
+    return get_quote_from_3c_pair(tc_pair).upper() == quote.upper()
+
+
+def get_quote_from_3c_pair(tc_pair: str) -> str:
+    return tc_pair.split('_')[0].upper()
+
+
+def map_spot_tc_pairs_to_bases(tc_pairs: list, account_market_code: str) -> list:
+    return list(map(lambda pair: get_base_from_3c_pair(tc_pair=pair, account_market_code=account_market_code), tc_pairs))
+
+
+def construct_pair_from_quote_and_base(quote: str, base: str) -> str:
+    return f"{quote.upper()}_{base.upper()}"
+
+
+def filter_tc_pairs_by_quote(pairs: list, quote: str) -> list:
+    return list(filter(lambda pair: pair_is_quote(tc_pair=pair, quote=quote), pairs))
+
+
+def construct_futures_pair_from_base(base: str, account_market_code: str) -> str:
+    if account_market_code in {'ftx_futures'}:
+        return f'USD_{base.upper()}-PERP'
+    else:
+        raise RuntimeError(f'Not known market code {account_market_code} in construct_futures_pair_from_quote_and_base')

test/sample_data/errors/api_key_has_no_permission_error.json

@@ -0,0 +1,4 @@
+{
+  "error": true,
+  "msg": "Other error occurred: access_denied Api key doesn't have enough permissions None."
+}

test/sample_data/errors/api_key_invalid_or_expired_error.json

@@ -0,0 +1,4 @@
+{
+  "error": true,
+  "msg": "Other error occurred: api_key_invalid_or_expired Unauthorized. Invalid or expired api key. None."
+}

test/test_three_commas_errors.py

@@ -2,50 +2,58 @@
 import json
 
 
-def test_bo_to_small_tc_error_with_pair():
-    file_path = 'test/sample_data/errors/bo_too_small_with_pair.json'
+def read_error(file_path) -> ThreeCommasError:
     with open(file_path, 'r+') as f:
         error = json.loads(f.read())
         error_model = ThreeCommasError(error)
+    return error_model
+
 
-        bo_error = error_model.bo_to_small_error()
-        assert len(bo_error) == 1
-        assert bo_error[0].amount == 33.35
-        assert bo_error[0].pair == 'USDT_YFI'
+def test_bo_to_small_tc_error_with_pair():
+    error = read_error('test/sample_data/errors/bo_too_small_with_pair.json')
+
+    assert error.is_bo_to_small_error()
+    bo_error = error.get_bo_to_small_error()
+    assert len(bo_error) == 1
+    assert bo_error[0].amount == 33.35
+    assert bo_error[0].pair == 'USDT_YFI'
 
 
 def test_multiple_bo_error():
-    file_path = 'test/sample_data/errors/multiple_bo_so_errors.json'
-    with open(file_path, 'r+') as f:
-        error = json.loads(f.read())
-        error_model = ThreeCommasError(error)
+    error = read_error('test/sample_data/errors/multiple_bo_so_errors.json')
 
-        bo_error = error_model.bo_to_small_error()
-        assert len(bo_error) == 4
-        assert set(map(lambda be: be.pair, bo_error)) == {'USDT_1INCH', 'USDT_AAVE', 'USDT_ACM', 'USDT_ADA'}
-        assert list(map(lambda be: be.amount, bo_error)) == [10.0, 10.0, 10.0, 10.0]
+    bo_error = error.get_bo_to_small_error()
+    assert len(bo_error) == 4
+    assert set(map(lambda be: be.pair, bo_error)) == {'USDT_1INCH', 'USDT_AAVE', 'USDT_ACM', 'USDT_ADA'}
+    assert list(map(lambda be: be.amount, bo_error)) == [10.0, 10.0, 10.0, 10.0]
 
 
 def test_bo_to_small_tc_error_no_pair():
-    file_path = 'test/sample_data/errors/bo_too_small_no_pair.json'
-    with open(file_path, 'r+') as f:
-        error = json.loads(f.read())
-        error_model = ThreeCommasError(error)
+    error = read_error('test/sample_data/errors/bo_too_small_no_pair.json')
 
-        bo_error = error_model.bo_to_small_error()
-        assert bo_error[0].amount == 9.4674
-        assert not bo_error[0].pair
+    bo_error = error.get_bo_to_small_error()
+    assert bo_error[0].amount == 9.4674
+    assert not bo_error[0].pair
 
 
 def test_no_bo_error():
-    file_path = 'test/sample_data/errors/signature_invalid.json'
-    with open(file_path, 'r+') as f:
-        error = json.loads(f.read())
-        error_model = ThreeCommasError(error)
+    error = read_error('test/sample_data/errors/signature_invalid.json')
 
-        bo_error = error_model.bo_to_small_error()
-        assert len(bo_error) == 0
+    bo_error = error.get_bo_to_small_error()
+    assert len(bo_error) == 0
+    assert not error.is_bo_to_small_error()
 
-    error_model = ThreeCommasError({'custom_message': 'some error occured'})
-    bo_error = error_model.bo_to_small_error()
+    error = ThreeCommasError({'custom_message': 'some error occured'})
+    bo_error = error.get_bo_to_small_error()
     assert len(bo_error) == 0
+
+
+def test_api_key_invalid_or_expired():
+    error = read_error('test/sample_data/errors/api_key_invalid_or_expired_error.json')
+    assert error.is_api_key_invalid_or_expired()
+
+
+def test_api_key_has_no_permission_error():
+    error = read_error('test/sample_data/errors/api_key_has_no_permission_error.json')
+    assert error.is_api_key_has_no_permission_error()
+