blob: b8c594b7bd705243f57c97f71c27faad39a37fcf (
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
|
from collections import deque
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, total_cost: float, date: str) -> None:
"""
Add a trade to the queue.
"""
trade = Trade(amount, total_cost, date)
self.queue.append(trade)
def remove(self, amount: float) -> List[Trade]:
"""
Remove a specified amount from the queue, returning the
trades used to buy.
"""
if amount <= 0:
raise ValueError("The amount to remove must be positive.")
remaining: float = amount
entries: List[Trade] = []
while remaining > 0:
if not self.queue:
raise ValueError(f"Insufficient assets in queue to process sale of {amount}.")
trade = self.queue[0]
if trade.amount > remaining:
trade.amount -= remaining
entries.append(Trade(remaining, trade.total_cost, trade.date))
remaining = 0
else:
remaining -= trade.amount
entries.append(trade)
self.queue.popleft()
return entries
|
