summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--bla.py5
-rw-r--r--kraken.py2
-rw-r--r--ledger_action.py3
-rw-r--r--test_trade.py5
-rw-r--r--test_trade_queue.py36
-rw-r--r--trade.py16
-rw-r--r--trade_queue.py18
7 files changed, 57 insertions, 28 deletions
diff --git a/bla.py b/bla.py
index 6bdefa2..4595bcb 100644
--- a/bla.py
+++ b/bla.py
@@ -11,15 +11,14 @@ from trade import Trade
from trade_queue import FIFOQueue
logging.basicConfig(
- level=logging.WARNING,
- format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'
+ level=logging.WARNING, format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"
)
# Set up a dedicated logger for FIFOQueue
logger = logging.getLogger("pnlcalc")
-def generate_report(sale_entries, proceeds: float | Decimal, crypto_asset, date_sold):
+def generate_report(sale_entries, proceeds: Decimal, crypto_asset, date_sold):
report = []
sell_date = datetime.strptime(date_sold, "%Y-%m-%d").strftime("%d.%m.%Y")
proceeds = Decimal(proceeds)
diff --git a/kraken.py b/kraken.py
index c420d89..9048a7d 100644
--- a/kraken.py
+++ b/kraken.py
@@ -4,7 +4,7 @@ from typing import List
from ledger_action import LedgerAction
-def parse_row(row: dict[str,str]) -> LedgerAction:
+def parse_row(row: dict[str, str]) -> LedgerAction:
date = row["time"].split(" ")[0]
return LedgerAction(
type=row["type"],
diff --git a/ledger_action.py b/ledger_action.py
index 952692b..cc0eb83 100644
--- a/ledger_action.py
+++ b/ledger_action.py
@@ -1,6 +1,7 @@
from dataclasses import dataclass
from decimal import Decimal
+
@dataclass
class LedgerAction:
"""
@@ -8,7 +9,7 @@ class LedgerAction:
The `LedgerAction` class encapsulates details of each action recorded in the ledger,
whether it's a deposit, trade, or other types of transactions.
-
+
Attributes:
type (str): The type of action, e.g., "trade" or "deposit".
asset (str): The cryptocurrency or fiat asset associated with the action.
diff --git a/test_trade.py b/test_trade.py
index c3d7764..ab853d1 100644
--- a/test_trade.py
+++ b/test_trade.py
@@ -3,12 +3,13 @@ from decimal import Decimal
from trade import PriceAdaption, Trade
+
class TestTrade(unittest.TestCase):
def setUp(self) -> None:
"""
Set up a Trade instance for testing.
"""
- self.trade = Trade(amount=10.0, total_cost=100.0, date="2025-04-14")
+ self.trade = Trade(amount=Decimal(10.0), total_cost=Decimal(100.0), date="2025-04-14")
def test_initialization(self):
"""
@@ -45,7 +46,7 @@ class TestTrade(unittest.TestCase):
self.assertEqual(self.trade.amount, 6.0)
self.assertEqual(self.trade.total_cost, 100.0)
- self.assertEqual(self.trade.price_per_coin, Decimal("100.0")/Decimal("6.0"))
+ self.assertEqual(self.trade.price_per_coin, Decimal("100.0") / Decimal("6.0"))
def test_remove_coins_exceeds_amount(self):
"""
diff --git a/test_trade_queue.py b/test_trade_queue.py
index 89c75e8..70a795a 100644
--- a/test_trade_queue.py
+++ b/test_trade_queue.py
@@ -1,17 +1,19 @@
+from decimal import Decimal
import unittest
from datetime import datetime
from trade_queue import FIFOQueue
+
class TestFIFOQueue(unittest.TestCase):
def setUp(self):
"""
Set up a FIFOQueue instance and some test trades.
"""
self.queue = FIFOQueue()
- self.queue.add(10.0, 100.0, "2025-04-14")
- self.queue.add(20.0, 200.0, "2025-04-15")
- self.queue.add(30.0, 300.0, "2025-04-16")
+ self.queue.add(Decimal(10.0), Decimal(100.0), "2025-04-14")
+ self.queue.add(Decimal(20.0), Decimal(200.0), "2025-04-15")
+ self.queue.add(Decimal(30.0), Decimal(300.0), "2025-04-16")
def test_add(self):
"""
@@ -48,8 +50,10 @@ class TestFIFOQueue(unittest.TestCase):
"""
trades = self.queue.remove_coins(25.0)
self.assertEqual(len(trades), 2) # Two trades should be returned
- self.assertEqual(trades[0].amount, 10.0) # The first trade should be fully consumed
- self.assertEqual(trades[1].amount, 15.0) # The second trade should be partially consumed
+ # The first trade should be fully consumed
+ self.assertEqual(trades[0].amount, 10.0)
+ # The second trade should be partially consumed
+ self.assertEqual(trades[1].amount, 15.0)
tq = self.queue.get_copy()
self.assertEqual(tq[0].amount, 5.0) # Remaining trade in queue should update
@@ -71,14 +75,16 @@ class TestFIFOQueue(unittest.TestCase):
"""
Test the remaining amount in the queue after adding trades.
"""
- self.assertEqual(self.queue.get_remaining_amount(), 60.0) # Total of all amounts: 10 + 20 + 30
+ # Total of all amounts: 10 + 20 + 30
+ self.assertEqual(self.queue.get_remaining_amount(), 60.0)
def test_get_remaining_amount_after_removal(self):
"""
Test the remaining amount after removing some assets.
"""
self.queue.remove_coins(15.0) # Remove 15 assets
- self.assertEqual(self.queue.get_remaining_amount(), 45.0) # Remaining: 60 - 15
+ # Remaining: 60 - 15
+ self.assertEqual(self.queue.get_remaining_amount(), 45.0)
def test_get_remaining_amount_empty_queue(self):
"""
@@ -108,14 +114,20 @@ class TestFIFOQueue(unittest.TestCase):
trades = self.queue.remove_coins(4.0) # Remove 4 COIN from the first trade
self.assertEqual(len(trades), 1) # Only one trade should be returned
- self.assertEqual(trades[0].price_per_coin, 10) # Coin-cost needs to stay constant
+ # Coin-cost needs to stay constant
+ self.assertEqual(trades[0].price_per_coin, 10)
self.assertEqual(trades[0].amount, 4.0) # Check the removed amount
- self.assertEqual(trades[0].total_cost, 40.0) # Total cost should be proportional: (100 * 5 / 10)
+ # Total cost should be proportional: (100 * 5 / 10)
+ self.assertEqual(trades[0].total_cost, 40.0)
tq = self.queue.get_copy()
- self.assertEqual(tq[0].price_per_coin, 10) # Original total cost remains unchanged
- self.assertEqual(tq[0].amount, 6.0) # Remaining amount in the first trade should be updated
- self.assertEqual(tq[0].total_cost, 60.0) # Original total cost remains unchanged
+ # Original total cost remains unchanged
+ self.assertEqual(tq[0].price_per_coin, 10)
+ # Remaining amount in the first trade should be updated
+ self.assertEqual(tq[0].amount, 6.0)
+ # Original total cost remains unchanged
+ self.assertEqual(tq[0].total_cost, 60.0)
+
if __name__ == "__main__":
unittest.main()
diff --git a/trade.py b/trade.py
index ed646d1..a07efe7 100644
--- a/trade.py
+++ b/trade.py
@@ -1,6 +1,7 @@
from decimal import Decimal
from enum import Enum
+
class PriceAdaption(Enum):
KeepTotalCost = 1
KeepPricePerCoin = 2
@@ -11,7 +12,8 @@ class Trade:
Represents a cryptocurrency trade, including the amount traded, total cost, and the date of trade.
Provides methods to modify the trade and access various attributes.
"""
- def __init__(self, amount: float|Decimal, total_cost: float|Decimal, date: str) -> None:
+
+ def __init__(self, amount: Decimal, total_cost: Decimal, date: str) -> None:
"""
Initialize a new Trade instance.
@@ -28,7 +30,11 @@ class Trade:
self.__total_cost: Decimal = Decimal(total_cost)
self.__date: str = date
- def remove_coins(self, amount: float|Decimal, adapt: PriceAdaption = PriceAdaption.KeepPricePerCoin) -> None:
+ def remove_coins(
+ self,
+ amount: float | Decimal,
+ adapt: PriceAdaption = PriceAdaption.KeepPricePerCoin,
+ ) -> None:
"""
Reduce the amount of cryptocurrency in the trade by a specified amount.
@@ -42,7 +48,7 @@ class Trade:
"""
if amount > self.__amount:
raise ValueError(f"Can't remove more than {self.__amount}")
-
+
if adapt == PriceAdaption.KeepPricePerCoin:
amount = Decimal(amount)
self.__total_cost -= amount * self.price_per_coin
@@ -95,7 +101,9 @@ class Trade:
ZeroDivisionError: If the current amount is zero.
"""
if self.amount == 0:
- raise ZeroDivisionError("Price per coin cannot be calculated when the amount is zero")
+ raise ZeroDivisionError(
+ "Price per coin cannot be calculated when the amount is zero"
+ )
return self.total_cost / self.amount
diff --git a/trade_queue.py b/trade_queue.py
index a0dfc47..dcc9725 100644
--- a/trade_queue.py
+++ b/trade_queue.py
@@ -7,18 +7,20 @@ from trade import Trade
# Set up a dedicated logger for FIFOQueue
logger = logging.getLogger(__name__)
+
class FIFOQueue:
"""
Crypto trading FIFO queue.
Will track trades.
"""
+
def __init__(self) -> None:
self.__queue: Deque[Trade] = deque()
self._cached_total: Decimal = Decimal(0)
self._cache_valid: bool = True
logger.info("FIFOQueue initialized with empty queue.")
-
+
def __len__(self) -> int:
return len(self.__queue)
@@ -28,7 +30,7 @@ class FIFOQueue:
"""
return [t for t in self.__queue]
- def add(self, amount: float | Decimal, total_cost: float | Decimal, date: str) -> None:
+ def add(self, amount: Decimal, total_cost: Decimal, date: str) -> None:
"""
Add a trade to the queue.
"""
@@ -50,7 +52,9 @@ class FIFOQueue:
if amount > self.get_remaining_amount():
logger.error(f"Insufficient assets to process sale of {amount}.")
- raise ValueError(f"Insufficient assets in queue to process sale of {amount}.")
+ raise ValueError(
+ f"Insufficient assets in queue to process sale of {amount}."
+ )
logger.debug(f"Removing {amount:.2f} coins from the queue.")
logger.info("Cache invalidated before removing coins.")
@@ -73,7 +77,9 @@ class FIFOQueue:
remaining -= trade.amount
entries.append(trade)
self.__queue.popleft()
- logger.info(f"Removed full trade: {trade}. Remaining coins to remove: {remaining}")
+ logger.info(
+ f"Removed full trade: {trade}. Remaining coins to remove: {remaining}"
+ )
return entries
@@ -83,7 +89,9 @@ class FIFOQueue:
"""
if not self._cache_valid:
logger.debug("Cache invalid, recalculating remaining amount.")
- self._cached_total = sum((trade.amount for trade in self.__queue), Decimal(0))
+ self._cached_total = sum(
+ (trade.amount for trade in self.__queue), Decimal(0)
+ )
self._cache_valid = True
logger.info(f"Cache recalculated: {self._cached_total:.2f}")
generated by cgit v1.2.3 (git 2.39.1) at 2025-04-26 22:52:11 +0000