blob: 0252431e8bfc6f9d1ec46b4acd46efb5f94e094c (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
from collections import deque
from decimal import Decimal
from typing import Deque, List
from trade import Trade
class FIFOQueue:
"""
Crypto trading FIFO queue.
Will track trades.
"""
def __init__(self) -> None:
self.queue: Deque[Trade] = deque()
def add(self, amount: float|Decimal, total_cost: float|Decimal, date: str) -> None:
"""
Add a trade to the queue.
"""
trade = Trade(amount, total_cost, date)
self.queue.append(trade)
def remove_coins(self, amount: float|Decimal) -> List[Trade]:
"""
Remove a specified amount of coins from the queue, returning the
trades used to buy. This can be used to calculate profit/loss.
"""
if amount <= 0:
raise ValueError("The amount to remove must be positive.")
amount = Decimal(amount)
if amount > self.get_remaining_amount():
raise ValueError(f"Insufficient assets in queue to process sale of {amount}.")
remaining: Decimal = amount
entries: List[Trade] = []
while remaining > 0:
trade = self.queue[0]
if trade.amount > remaining:
ppc = trade.price_per_coin
trade.remove_coins(remaining)
entries.append(Trade(remaining, remaining * ppc, trade.date))
remaining = Decimal(0)
else:
remaining -= trade.amount
entries.append(trade)
self.queue.popleft()
return entries
def get_remaining_amount(self) -> Decimal:
"""
Calculate the total remaining amount in the queue.
"""
return sum((trade.amount for trade in self.queue), Decimal(0))
|
