50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
import csv
|
|
import dataclasses
|
|
from pathlib import Path
|
|
|
|
import fin_depo
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class CsvFileLogger:
|
|
"""Outputs the given `TradeOrderDetails` to the CSV file at the specified
|
|
path.
|
|
"""
|
|
|
|
path: Path
|
|
|
|
def __call__(self, trade_order: fin_depo.data.TradeOrderDetails):
|
|
fieldnames: list[str] = [
|
|
'executed_time',
|
|
'input_asset',
|
|
'input_amount',
|
|
'output_asset',
|
|
'output_amount',
|
|
'fee_asset',
|
|
'fee_amount',
|
|
'order_id',
|
|
'raw_order_details',
|
|
]
|
|
|
|
# Select data to write
|
|
d = {
|
|
'executed_time': trade_order.executed_time,
|
|
'input_asset': trade_order.input.asset,
|
|
'input_amount': trade_order.input.amount,
|
|
'output_asset': trade_order.output.asset,
|
|
'output_amount': trade_order.output.amount,
|
|
'fee_asset': trade_order.fee.asset,
|
|
'fee_amount': trade_order.fee.amount,
|
|
'order_id': trade_order.order_id,
|
|
'raw_order_details': trade_order.raw_order_details,
|
|
}
|
|
|
|
# Ensure that directory exists
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Write row in CSV file
|
|
with open(self.path, 'a') as f:
|
|
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
|
writer.writerow(d)
|
|
del d
|