PR for OB task

.github/workflows/lint.yaml

@@ -0,0 +1,25 @@
+name: Python Linting
+
+on: [push, pull_request]
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.11'
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install flake8 black
+
+    - name: Check with Flake8
+      run: flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+
+    - name: Format code with Black
+      run: black --check .

.gitignore

@@ -1,3 +1,6 @@
+exchanges/data_folder
+exchanges/constants/.env
 /venv
 __pycache__/
 .env
+exchanges/*.json

all_feeds.py

@@ -4,11 +4,11 @@
 Test = test_feed.Test
 MCAP1000 = mcap1000.MCAP1000
 
-#NOTE: this is a dict of all feed classes that SIWA can run, keyed by feed name
+# NOTE: this is a dict of all feed classes that SIWA can run, keyed by feed name
 #     this is used in endpoint.py to route requests to the correct feed
 #
-#TO ENABLE OR DISABLE A FEED, ADD OR REMOVE IT FROM THIS DICT
+# TO ENABLE OR DISABLE A FEED, ADD OR REMOVE IT FROM THIS DICT
 all_feeds = {
     Test.NAME: Test,
     MCAP1000.NAME: MCAP1000,
-    }
+}

apis/coingecko.py

@@ -17,6 +17,7 @@ class CoinGeckoAPI(CryptoAPI):
         extract_market_cap(data: Dict[str, Any]) -> Dict[float, Dict[str, str]]:
             Extracts market cap data from API response.
     """
+
     VS_CURRENCY = "usd"
     ORDER = "market_cap_desc"
     PAGE = 1
@@ -31,8 +32,7 @@ def __init__(self) -> None:
         Constructs all the necessary attributes for the CoinGeckoAPI object.
         """
         super().__init__(
-            url="https://api.coingecko.com/api/v3/coins/markets",
-            source='coingecko'
+            url="https://api.coingecko.com/api/v3/coins/markets", source="coingecko"
         )
 
     @utils.handle_request_errors
@@ -92,7 +92,7 @@ def get_market_caps_of_list(self, tokens: List[str]) -> Dict[str, float]:
         """
         market_caps = {}
 
-        tokens_comma_sep = ','.join(tokens)
+        tokens_comma_sep = ",".join(tokens)
 
         parameters = {
             "vs_currency": self.VS_CURRENCY,
@@ -107,7 +107,7 @@ def get_market_caps_of_list(self, tokens: List[str]) -> Dict[str, float]:
                 name = d.get(self.NAME_KEY)
                 last_updated = d.get(self.LAST_UPDATED_KEY)
                 market_caps[market_cap] = {
-                    'name': name,
-                    'last_updated': last_updated,
+                    "name": name,
+                    "last_updated": last_updated,
                 }
         return market_caps

apis/coinmarketcap.py

@@ -33,10 +33,10 @@ def __init__(self) -> None:
         """
         Constructs all the necessary attributes for the CoinMarketCapAPI object.
         """
-        source = 'coinmarketcap'
+        source = "coinmarketcap"
         super().__init__(
             url="https://pro-api.coinmarketcap.com/v1/cryptocurrency/listings/latest",
-            source=source
+            source=source,
         )
         self.headers = {
             self.CMC_PRO_API_KEY: self.get_api_key(source),
@@ -53,12 +53,8 @@ def get_data(self, N: int) -> Dict[str, Any]:
         Returns:
             Dict[str, Any]: A dictionary with data fetched from API.
         """
-        parameters = {
-            self.LIMIT: N
-        }
-        response = requests.get(
-            self.url, headers=self.headers, params=parameters
-        )
+        parameters = {self.LIMIT: N}
+        response = requests.get(self.url, headers=self.headers, params=parameters)
         data = response.json()
         return data
 
@@ -96,12 +92,8 @@ def get_market_cap_of_token(self, id: int) -> Dict[str, float]:
             Dict[str, float]: A dictionary with market cap as keys and other metadata as values.
         """
         url = "https://pro-api.coinmarketcap.com/v1/cryptocurrency/quotes/latest"
-        parameters = {
-            self.ID: id
-        }
-        response = requests.get(
-            url, headers=self.headers, params=parameters
-        )
+        parameters = {self.ID: id}
+        response = requests.get(url, headers=self.headers, params=parameters)
         data = response.json()
 
         market_cap_data = {}
@@ -111,9 +103,9 @@ def get_market_cap_of_token(self, id: int) -> Dict[str, float]:
             last_updated = token_data[self.LAST_UPDATED]
             market_cap = token_data[self.QUOTE][self.USD][self.MARKET_CAP]
             market_cap_data = {
-                'name': name,
-                'market_cap': market_cap,
-                'last_updated': last_updated,
+                "name": name,
+                "market_cap": market_cap,
+                "last_updated": last_updated,
             }
         return market_cap_data
 
@@ -131,9 +123,9 @@ def get_market_caps_of_list(self, ids: List[int]) -> Dict[str, Dict[str, Any]]:
         for id in ids:
             mcap_data = self.get_market_cap_of_token(id)
             if mcap_data:
-                market_caps[mcap_data['market_cap']] = {
-                    'name': mcap_data['name'],
-                    'last_updated': mcap_data['last_updated']
+                market_caps[mcap_data["market_cap"]] = {
+                    "name": mcap_data["name"],
+                    "last_updated": mcap_data["last_updated"],
                 }
             time.sleep(0.2)  # To prevent hitting API rate limits
 

apis/coinpaprika.py

@@ -25,10 +25,11 @@ def __init__(self) -> None:
         """
         Constructs all the necessary attributes for the CoinPaprikaAPI object.
         """
-        self.ohlc_url = "https://api.coinpaprika.com/v1/coins/{coin_id}/ohlcv/latest"  # noqa E501
+        self.ohlc_url = (
+            "https://api.coinpaprika.com/v1/coins/{coin_id}/ohlcv/latest"  # noqa E501
+        )
         super().__init__(
-            url="https://api.coinpaprika.com/v1/coins",
-            source='coinpaprika'
+            url="https://api.coinpaprika.com/v1/coins", source="coinpaprika"
         )
 
     @utils.handle_request_errors
@@ -49,17 +50,18 @@ def get_data(self, N: int) -> List[Dict[str, Any]]:
             data = response.json()
         else:
             raise requests.exceptions.RequestException(
-                f"Received status code {response.status_code} "
-                f"for URL: {self.url}"
+                f"Received status code {response.status_code} " f"for URL: {self.url}"
             )
         # Sorting the coins by market cap
         # Also filtering out coins with rank 0 (junk values in API response)
-        filtered_data = [coin for coin in data if coin['rank'] != 0]
-        sorted_data = sorted(filtered_data, key=lambda coin: coin['rank'])[:N]
+        filtered_data = [coin for coin in data if coin["rank"] != 0]
+        sorted_data = sorted(filtered_data, key=lambda coin: coin["rank"])[:N]
         return sorted_data
 
     @utils.handle_request_errors
-    def extract_market_cap(self, data: List[Dict[str, Any]]) -> Dict[float, Dict[str, Any]]:
+    def extract_market_cap(
+        self, data: List[Dict[str, Any]]
+    ) -> Dict[float, Dict[str, Any]]:
         """
         Extracts market cap data from API response.
 
@@ -87,7 +89,7 @@ def extract_market_cap(self, data: List[Dict[str, Any]]) -> Dict[float, Dict[str
 
             name = coin["name"]
             last_updated = 0  # Updated every 5 mins as per docs: https://api.coinpaprika.com/#tag/Coins/paths/~1coins~1%7Bcoin_id%7D~1ohlcv~1today~1/get # noqa E501
-            market_cap = coin_info[0]['market_cap']
+            market_cap = coin_info[0]["market_cap"]
             market_data[market_cap] = {
                 "name": name,
                 "last_updated": last_updated,

apis/crypto_api.py

@@ -20,7 +20,7 @@ class CryptoAPI:
             Abstract method to extract market cap data.
     """
 
-    API_KEYS_FILE = 'api_keys.json'
+    API_KEYS_FILE = "api_keys.json"
 
     def __init__(self, url: str, source: str) -> None:
         """
@@ -52,9 +52,7 @@ def fetch_mcap_by_list(self, tokens: List[str]) -> Dict[str, float]:
 
         # Store market data in the database
         utils.create_market_cap_database()
-        utils.store_market_cap_data(
-            market_data=market_data, source=self.source
-        )
+        utils.store_market_cap_data(market_data=market_data, source=self.source)
         return market_data
 
     def fetch_mcap_by_rank(self, N: int) -> dict:
@@ -77,9 +75,7 @@ def fetch_mcap_by_rank(self, N: int) -> dict:
 
         # Store market data in the database
         utils.create_market_cap_database()
-        utils.store_market_cap_data(
-            market_data=market_data, source=self.source
-        )
+        utils.store_market_cap_data(market_data=market_data, source=self.source)
         return market_data
 
     def get_data(self, N: int) -> Any:

apis/cryptocompare.py

@@ -37,7 +37,7 @@ def __init__(self) -> None:
         """
         super().__init__(
             url="https://min-api.cryptocompare.com/data/top/mktcapfull",
-            source='cryptocompare'
+            source="cryptocompare",
         )
 
     @utils.handle_request_errors
@@ -66,8 +66,7 @@ def get_data(self, N: int, buffer: int = 2) -> Dict[str, Any]:
             data = response.json()
         else:
             raise requests.exceptions.RequestException(
-                f"Received status code {response.status_code} "
-                f"for URL: {self.url}"
+                f"Received status code {response.status_code} " f"for URL: {self.url}"
             )
         missing_count = 0
         for coin in data[self.DATA]:
@@ -118,7 +117,7 @@ def get_market_caps_of_list(self, tokens: List[str]) -> Dict[str, Dict[str, Any]
         """
         url = "https://min-api.cryptocompare.com/data/pricemultifull"
         tokens_upper = [token.upper() for token in tokens]
-        tokens_comma_sep = ','.join(tokens_upper)
+        tokens_comma_sep = ",".join(tokens_upper)
         parameters = {
             self.FSYMS: tokens_comma_sep,
             self.TSYMS: self.USD,
@@ -133,7 +132,7 @@ def get_market_caps_of_list(self, tokens: List[str]) -> Dict[str, Dict[str, Any]
                     market_cap = data[self.RAW][token][self.USD][self.MKTCAP]
                     last_updated = data[self.RAW][token][self.USD][self.LAST_UPDATE]
                     market_caps[market_cap] = {
-                        'name': token,
-                        'last_updated': last_updated,
+                        "name": token,
+                        "last_updated": last_updated,
                     }
         return market_caps