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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
import csv
from datetime import datetime
from decimal import Decimal
from itertools import groupby
import logging
from typing import Dict, List, Tuple
from kraken import read_ledger
from ledger_action import LedgerAction
from ledger_process import LedgerProcess
from trade import Trade
from trade_queue import FIFOQueue
logging.basicConfig(
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(currency: str, trades: List[Tuple[Trade, Trade]]):
report = []
for trade in trades:
sell_date = datetime.strptime(trade[1].date, "%Y-%m-%d")
sell_date_formatted = sell_date.strftime("%d.%m.%Y")
buy_date = datetime.strptime(trade[0].date, "%Y-%m-%d")
buy_date_formatted = buy_date.strftime("%d.%m.%Y")
holding_period = (sell_date - buy_date).days
sell_proceeds = trade[1].total_cost
short_or_long = "Short" if holding_period < 365 else "Long"
buy_cost = trade[0].total_cost
gain_or_loss = sell_proceeds - buy_cost
assert trade[0].amount == trade[1].amount
report.append(
{
"Amount": f"{trade[0].amount:.8f}",
"Currency": currency,
"Date Sold": sell_date_formatted,
"Date Acquired": buy_date_formatted,
"Short/Long": short_or_long,
"Buy/Input at": "Kraken",
"Sell/Output at": "Kraken",
"Proceeds": f"{sell_proceeds:.2f}",
"Cost Basis": f"{buy_cost:.2f}",
"Gain/Loss": f"{gain_or_loss:.2f}",
}
)
return report
report = []
def write_report(output_path: str):
# Write report to CSV
with open(output_path, "w", newline="") as csvfile:
fieldnames = [
"Amount",
"Currency",
"Date Sold",
"Date Acquired",
"Short/Long",
"Buy/Input at",
"Sell/Output at",
"Proceeds",
"Cost Basis",
"Gain/Loss",
]
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(report)
# Usage
ledger_path = "ledgers.csv" # Replace with your ledger file path
output_path = "tax_report.csv" # Replace with your desired output file path
actions = read_ledger(ledger_path)
lp = LedgerProcess()
lp.process_ledger(actions)
all_trade_pairs: Dict[str, List[Tuple[Trade, Trade]]] = {}
for curr, queue in lp.fifo_queues.items():
all_trade_pairs[curr] = []
trades_for_curr = queue.match_trades()
all_trade_pairs[curr].extend(trades_for_curr)
report.extend(generate_report(curr, trades_for_curr))
write_report(output_path)
pass
|
