Can fetch price and output in a table format to either on-screen or t…

…o a file (#12)

* updated with airdrop core logic

* Updated

* Supported table and can output to file

notebooks/airdrop.ipynb

@@ -205,7 +205,6 @@
     "from requests.exceptions import ConnectionError, Timeout, TooManyRedirects\n",
     "import json\n",
     "import math\n",
-    "import random\n",
     "import pandas as pd\n",
     "\n",
     "# API doc: https://min-api.cryptocompare.com/documentation?key=Historical&cat=dataPriceHistorical\n",

src/scripts/airdrop.py

@@ -1,22 +1,175 @@
-# from datetime import datetime, timezone
-# from requests import Request, Session
-# from requests.exceptions import ConnectionError, Timeout, TooManyRedirects
 import json
-
-# import math
-# import pandas as pd
-from rich import print
+import pandas as pd
+import sys
 import typer
+from datetime import datetime, timezone
+from requests import Session
+from requests.exceptions import ConnectionError, Timeout, TooManyRedirects
+from rich import print
+from typing import Optional
+
+
+class Airdrop:
+    @classmethod
+    def check_config(cls, config):
+        # Check that all required fields exist
+        required_keys = ["dhk_distribution", "reference_date", "tokens", "apis"]
+
+        for key in required_keys:
+            if key not in config or config[key] == "":
+                print(
+                    f"Required key [bold red]{key}[/bold red] is missing in the config file."
+                )
+                raise typer.Exit(code=1)
+
+        return True
+
+    def __init__(self, config_file):
+        config = json.loads(config_file.read())
+        if Airdrop.check_config(config):
+            self.config = config
+
+        # Set to 12:00 UTC timezone of the reference_date
+        self.reference_datetime = datetime.strptime(
+            f"{config['reference_date']} 12:00", "%Y-%m-%d %H:%M"
+        ).replace(tzinfo=timezone.utc)
+
+    # API doc:
+    #   https://min-api.cryptocompare.com/documentation?key=Historical&cat=dataPriceHistorical
+    def fetch_price(self, from_symbol, to_symbol="USD") -> Optional[float]:
+        api = self.config["apis"]["cryptocompare"]
+        endpoint, apikey = api["endpoint"], api["apikey"]
+        if endpoint is None or apikey is None:
+            raise Exception(
+                "fetch_price: cryptocompare API endpoint or key is missing."
+            )
+
+        parameters = {
+            "fsym": from_symbol,
+            "tsyms": to_symbol,
+            "calculationType": "MidHighLow",
+            "ts": self.reference_datetime.timestamp(),
+        }
+        headers = {
+            "Accepts": "application/json",
+            "authorization": f"Apikey {apikey}",
+        }
+
+        session = Session()
+        session.headers.update(headers)
+        try:
+            response = session.get(endpoint, params=parameters)
+            data = json.loads(response.text)
+        except (ConnectionError, Timeout, TooManyRedirects) as e:
+            print(f"fetch_price {from_symbol} connection error: {e}", file=sys.stderr)
+
+        if from_symbol not in data:
+            print(
+                f"Unable to fetch price for {from_symbol}, returning: {data}",
+                file=sys.stderr,
+            )
+            return None
+
+        return float(data[from_symbol][to_symbol])
+
+    # API doc: https://docs.cosmostation.io/apis/reference/utilities/staking-apr
+    def fetch_staking_apr(self, staking_network) -> Optional[float]:
+        api = self.config["apis"]["mintscan"]
+        endpoint, apikey = api["endpoint"], api["apikey"]
+        if endpoint is None or apikey is None:
+            raise Exception(
+                "fetch_staking_apr: mintscan API endpoint or key is missing."
+            )
+
+        endpoint = endpoint.replace(":network", staking_network)
+        headers = {
+            "Accepts": "application/json",
+            "authorization": f"Bearer {apikey}",
+        }
+
+        session = Session()
+        session.headers.update(headers)
+
+        try:
+            response = session.get(endpoint)
+            data = json.loads(response.text)
+        except (ConnectionError, Timeout, TooManyRedirects) as e:
+            print(
+                f"fetch_staking_apr {staking_network} connection error: {e}",
+                file=sys.stderr,
+            )
+
+        if "apr" not in data:
+            print(
+                f"Unable to fetch staking_apr for {staking_network}, returning: {data}",
+                file=sys.stderr,
+            )
+            return None
+
+        return float(data["apr"])
+
+    def monthly_alloc(self):
+        columns = [
+            "token",
+            "price",
+            "staking-amt",
+            "staking-val",
+            "staking-apr",
+            "reward",
+            "dhk-distribution-pc",
+            "dhk-distribution",
+        ]
+
+        # Used later in panda
+        lst = []
+
+        for t in self.config["tokens"]:
+            token, staking_amt = t["token"], t["qty"]
+            token_price = t["price"] if "price" in t else self.fetch_price(token)
+
+            staking_val = token_price * staking_amt
+            staking_apr = 0
+            if "staking_apr" in t:
+                staking_apr = t["staking_apr"]
+            elif "network" in t:
+                staking_apr = self.fetch_staking_apr(t["network"])
+
+            # NX> handles TypeError: unsupported operand type(s) for *: 'NoneType' and 'float'
+            reward = (
+                staking_apr * staking_val
+                if isinstance(staking_apr, float) and isinstance(staking_val, float)
+                else None
+            )
+
+            lst.append(
+                [
+                    token,
+                    token_price,
+                    staking_amt,
+                    staking_val,
+                    staking_apr,
+                    reward,
+                    0,
+                    0,
+                ]
+            )
 
-# import random
+        # Panda dataframe
+        df = pd.DataFrame(lst, columns=columns)
 
+        # Append a total row at the end of the table
+        ttl_staking_val = df["staking-val"].sum()
+        ttl_reward = df["reward"].sum()
+        ttl = pd.Series(
+            {"token": "TOTAL", "staking-val": ttl_staking_val, "reward": ttl_reward}
+        )
+        df = pd.concat([df, ttl.to_frame().T], ignore_index=True)
 
-def airdrop_monthly_alloc(config_file, output, type):
-    config = json.loads(config_file.read())
+        # Calculate DHK distribution and distribution percent
+        for idx, row in df.iterrows():
+            df.at[idx, "dhk-distribution-pc"] = (row["reward"] / ttl_reward) * 100
+            df.at[idx, "dhk-distribution"] = (
+                row["reward"] / ttl_reward * self.config["dhk_distribution"]
+            )
 
-    # Check that all required fields exist
-    required_keys = ["dhk_distribution", "reference_date", "tokens", "apis"]
-    for key in required_keys:
-        if key not in config or config[key] == "":
-            print(f"Required key {key} is missing from the input config file")
-            raise typer.Exit(code=1)
+        return df

src/scripts/main.py

@@ -1,8 +1,10 @@
 import typer
 from typing_extensions import Annotated
 from enum import Enum
+from rich import print
 
-from airdrop import airdrop_monthly_alloc
+import utils
+from airdrop import Airdrop
 
 
 class OutputType(str, Enum):
@@ -32,7 +34,24 @@ def monthly_alloc(
     """
     Compute the DHK dao monthly airdrop allocation based on staked value on various blockchains.
     """
-    airdrop_monthly_alloc(config, output, type)
+    airdrop = Airdrop(config)
+    result = utils.round_output(airdrop.monthly_alloc())
+    result_output = None
+
+    # Transform the result suitable for output
+    match type:
+        case OutputType.json:
+            pass
+        case OutputType.csv:
+            pass
+        case OutputType.table:
+            result_output = result
+        case _:  # Unexpected output type
+            raise Exception("Unexpected output type.")
+
+    # Write the output either to screen or a file
+    fh = open(output, "w") if (output is not None) else None
+    print(result_output, file=fh)
 
 
 if __name__ == "__main__":

src/scripts/utils.py

@@ -0,0 +1,15 @@
+def round_number(val):
+    if isinstance(val, float):
+        if val < 1:  # keep 4 decimals
+            return round(val, 4)
+        if val < 100:  # keep 3 decimals
+            return round(val, 3)
+        else:  # keep 2 decimals
+            return round(val, 2)
+    return val
+
+
+def round_output(df):
+    for label, col_data in df.items():
+        df[label] = col_data.apply(round_number)
+    return df

tests/fixtures/config-invalid-1.json

@@ -7,11 +7,11 @@
     "apis": {
         "cryptocompare": {
             "endpoint": "https://min-api.cryptocompare.com/data/pricehistorical",
-            "key": "cryptocompare-123456"
+            "apikey": "cryptocompare-123456"
         },
         "mintscan": {
             "endpoint": "https://apis.mintscan.io/v1/:network/apr",
-            "key": "mintscan-123456"
+            "apikey": "mintscan-123456"
         }
     }
 }

tests/fixtures/config-valid.json

@@ -8,11 +8,11 @@
     "apis": {
         "cryptocompare": {
             "endpoint": "https://min-api.cryptocompare.com/data/pricehistorical",
-            "key": "cryptocompare-123456"
+            "apikey": "cryptocompare-123456"
         },
         "mintscan": {
             "endpoint": "https://apis.mintscan.io/v1/:network/apr",
-            "key": "mintscan-123456"
+            "apikey": "mintscan-123456"
         }
     }
 }