Airdrop

src/archive_evaluation.py

@@ -4,7 +4,7 @@
 from zipfile import ZipFile
 
 import log_config
-from config import ACCOUNT_STATMENTS_PATH, BASE_PATH, DATA_PATH, EXPORT_PATH, TAX_YEAR
+from config import ACCOUNT_STATEMENTS_PATH, BASE_PATH, DATA_PATH, EXPORT_PATH, TAX_YEAR
 
 log = log_config.getLogger(__name__)
 IGNORE_FILES = [".gitkeep"]
@@ -19,10 +19,10 @@ def append_files(basedir: Path, filenames: list[str]) -> None:
 # Account statements
 log.debug("Archive account statements")
 account_statements = [
-    f for f in os.listdir(ACCOUNT_STATMENTS_PATH) if f not in IGNORE_FILES
+    f for f in os.listdir(ACCOUNT_STATEMENTS_PATH) if f not in IGNORE_FILES
 ]
 log.debug("Found: %s", ", ".join(account_statements))
-append_files(ACCOUNT_STATMENTS_PATH, account_statements)
+append_files(ACCOUNT_STATEMENTS_PATH, account_statements)
 
 # Price database
 log.debug("Archive price database")

src/book.py

@@ -45,6 +45,8 @@ def __init__(self, price_data: PriceData) -> None:
         self.price_data = price_data
 
         self.operations: list[tr.Operation] = []
+        # list of airdrops with user-configured taxation
+        self.airdrop_configs: list[tr.Airdrop] = []
 
     def __bool__(self) -> bool:
         return bool(self.operations)
@@ -1271,20 +1273,20 @@ def read_file(self, file_path: Path) -> None:
                 "Skipping file."
             )
 
-    def get_account_statement_paths(self, statements_dir: Path) -> list[Path]:
-        """Return file paths of all account statements in `statements_dir`.
+    def get_file_paths(self, folder_dir: Path) -> list[Path]:
+        """Return file paths of all input files in `folder_dir`.
+        For example, this function can return all account statement file paths.
 
         Args:
-            statements_dir (str): Folder in which account statements
-                                  will be searched.
+            folder_dir (str): Folder in input files will be searched.
 
         Returns:
-            list[Path]: List of account statement file paths.
+            list[Path]: List of input file paths.
         """
         file_paths: list[Path] = []
 
-        if statements_dir.is_dir():
-            for file_path in statements_dir.iterdir():
+        if folder_dir.is_dir():
+            for file_path in folder_dir.iterdir():
                 # Ignore .gitkeep and temporary excel files.
                 filename = file_path.stem
                 if filename == ".gitkeep" or filename.startswith("~$"):
@@ -1299,12 +1301,12 @@ def read_files(self) -> bool:
         Returns:
             bool: Return True if everything went as expected.
         """
-        paths = self.get_account_statement_paths(config.ACCOUNT_STATMENTS_PATH)
+        paths = self.get_file_paths(config.ACCOUNT_STATEMENTS_PATH)
 
         if not paths:
             log.warning(
                 "No account statement files located in %s.",
-                config.ACCOUNT_STATMENTS_PATH,
+                config.ACCOUNT_STATEMENTS_PATH,
             )
             return False
 
@@ -1316,3 +1318,69 @@ def read_files(self) -> bool:
             return False
 
         return True
+
+    def get_airdrops(self) -> bool:
+        """Read all airdrop configurations from the folder specified in the config.
+        This allows the user to configure the taxation of each airdrop individually.
+
+        Returns:
+            bool: Return True if everything went as expected.
+        """
+        paths = self.get_file_paths(config.AIRDROP_STATEMENTS_PATH)
+
+        if not paths:
+            log.debug(
+                "No airdrop configuration files located in %s.",
+                config.AIRDROP_STATEMENTS_PATH,
+            )
+            return False
+
+        for file_path in paths:
+            with open(file_path, encoding="utf8") as f:
+                reader = csv.reader(f)
+                line = next(reader)
+
+                if line == ["platform", "utc_time", "coin", "value", "taxable"]:
+                    log.info(f"Reading airdrop configurations from {file_path}")
+                else:
+                    log.error(
+                        f"Airdrop configuration header is not correct in {file_path}"
+                    )
+                    raise RuntimeError
+
+                for (
+                    platform,
+                    csv_utc_time,
+                    coin,
+                    csv_change,
+                    csv_taxable,
+                ) in reader:
+                    row = reader.line_num
+
+                    utc_time = datetime.datetime.fromisoformat(csv_utc_time)
+                    change = misc.force_decimal(csv_change)
+
+                    airdrop = tr.Airdrop(
+                        utc_time,
+                        platform,
+                        change,
+                        coin,
+                        row,
+                        file_path,
+                    )
+
+                    if csv_taxable.lower() == "true":
+                        taxable = True
+                    elif csv_taxable.lower() == "false":
+                        taxable = False
+                    else:
+                        log.error(
+                            f"Unkown taxable parameter '{csv_taxable}' in row {row} "
+                            f"of {file_path}: Should be 'true' or 'false'"
+                        )
+                        raise RuntimeError
+                    airdrop.set_taxable(taxable)
+
+                    self.airdrop_configs.append(airdrop)
+
+        return True

src/config.py

@@ -50,13 +50,13 @@
     def IS_LONG_TERM(buy: datetime, sell: datetime) -> bool:
         return buy + relativedelta(years=1) < sell
 
-
 else:
     raise NotImplementedError(f"Your country {COUNTRY} is not supported.")
 
 # Program specific constants.
 BASE_PATH = Path(__file__).parent.parent.absolute()
-ACCOUNT_STATMENTS_PATH = Path(BASE_PATH, "account_statements")
+ACCOUNT_STATEMENTS_PATH = Path(BASE_PATH, "account_statements")
+AIRDROP_STATEMENTS_PATH = Path(BASE_PATH, "airdrop_configs")
 DATA_PATH = Path(BASE_PATH, "data")
 EXPORT_PATH = Path(BASE_PATH, "export")
 TMP_LOG_FILEPATH = Path(EXPORT_PATH, "tmp.log")

src/main.py

@@ -40,6 +40,7 @@ def main() -> None:
         return
 
     book.get_price_from_csv()
+    book.get_airdrops()
     taxman.evaluate_taxation()
     evaluation_file_path = taxman.export_evaluation_as_csv()
     taxman.print_evaluation()

src/price_data.py

@@ -431,7 +431,7 @@ def _get_price_kraken(
 
                 if not data["error"]:
                     break
-                elif data["error"] == ['EGeneral:Invalid arguments']:
+                elif data["error"] == ["EGeneral:Invalid arguments"]:
                     # add pair to invalid pairs list
                     # leads to inversion of pair next time
                     log.warning(

src/transaction.py

@@ -108,7 +108,10 @@ class StakingInterest(Transaction):
 
 
 class Airdrop(Transaction):
-    pass
+    taxable = True
+
+    def set_taxable(self, taxable):
+        self.taxable = taxable
 
 
 class Commission(Transaction):