Converts transaction history CSV from Abucoins into a format compatible with CointTacking

convert.py

# Processes Abucoins transaction history CSV into a CSV format that CoinTracking
# can import (Custom Exchange Import).
#
# Example CoinTracking settings: https://imgur.com/4vvEiCo.
#
# Usage: convert.py input.csv output.csv
#
# Author: Michael Lynch <michael@mtlynch.io>
# License: MIT License

import collections
import csv
import decimal
import json
import sys

Transaction = collections.namedtuple('Transaction', [
    'date', 'buy_amount', 'buy_currency', 'sell_amount', 'sell_currency',
    'transaction_fee', 'transaction_fee_currency', 'trade_id', 'comment', 'type'
])
# Export private methods.
Transaction.fields = Transaction._fields
Transaction.as_dict = Transaction._asdict

transactions = []

trade_sets = collections.defaultdict(list)
withdrawal_sets = collections.defaultdict(list)

DEPOSIT = 'deposit_type'
WITHDRAWAL = 'withdrawal_type'
TRADE = 'trade_type'

with open(sys.argv[1]) as input_file:
    reader = csv.DictReader(
        input_file,
        fieldnames=[
            'created_at', 'account', 'value', 'balance_after', 'category',
            'details'
        ])
    # Skip header row.
    reader.next()
    for row in reader:
        category = row['category']
        if category == 'Deposit':
            transactions.append(
                Transaction(
                    date=row['created_at'],
                    buy_amount=row['value'],
                    buy_currency=row['account'],
                    sell_amount=None,
                    sell_currency=None,
                    transaction_fee=None,
                    transaction_fee_currency=row['account'],
                    trade_id=row['details'],
                    comment='Deposit %s %s' % (row['value'], row['account']),
                    type=DEPOSIT))
        elif category in ('Withdraw', 'Withdraw fee'):
            withdrawal_sets[row['details']].append(row)
        elif category in ('Market trade', 'Market fee'):
            trade_sets[row['details']].append(row)
        else:
            raise ValueError('Unexpected category: %s' % json.dumps(row))

# Trade sets always come in three transactions of the following order:
COIN_BOUGHT_INDEX = 0
TRANSACTION_FEE_INDEX = 1
COIN_SOLD_INDEX = 2


def sanity_check_trade_set(trade_set):
    if len(trade_set) != 3:
        raise ValueError('Unexpected trade set: %s' % json.dumps(trade_set))
    if trade_set[COIN_BOUGHT_INDEX]['category'] != 'Market trade':
        raise ValueError('Unexpected trade category: %s' % json.dumps(
            trade_set[COIN_BOUGHT_INDEX]))
    if trade_set[TRANSACTION_FEE_INDEX]['category'] != 'Market fee':
        raise ValueError('Unexpected trade category: %s' % json.dumps(
            trade_set[TRANSACTION_FEE_INDEX]))
    if trade_set[COIN_SOLD_INDEX]['category'] != 'Market trade':
        raise ValueError('Unexpected trade category: %s' % json.dumps(
            trade_set[COIN_SOLD_INDEX]))
    if ((trade_set[COIN_BOUGHT_INDEX]['created_at'] !=
         trade_set[TRANSACTION_FEE_INDEX]['created_at']) or
        (trade_set[COIN_BOUGHT_INDEX]['created_at'] !=
         trade_set[COIN_SOLD_INDEX]['created_at'])):
        raise ValueError(
            'Trade set timestamps don\'t match: %s' % json.dumps(trade_set))


for trade_set in trade_sets.itervalues():
    sanity_check_trade_set(trade_set)
    date = trade_set[COIN_BOUGHT_INDEX]['created_at']
    buy_amount = trade_set[COIN_BOUGHT_INDEX]['value']
    buy_currency = trade_set[COIN_BOUGHT_INDEX]['account']
    sell_amount = trade_set[COIN_SOLD_INDEX]['value']
    sell_currency = trade_set[COIN_SOLD_INDEX]['account']
    transaction_fee = trade_set[TRANSACTION_FEE_INDEX]['value']
    transaction_fee_currency = trade_set[TRANSACTION_FEE_INDEX]['account']
    trade_id = trade_set[COIN_BOUGHT_INDEX]['details']
    comment = 'Sell %s %s for %s %s' % (abs(decimal.Decimal(sell_amount)), sell_currency,
                                        buy_amount, buy_currency)
    transactions.append(
        Transaction(
            date=date,
            buy_amount=buy_amount,
            buy_currency=buy_currency,
            sell_amount=sell_amount,
            sell_currency=sell_currency,
            transaction_fee=transaction_fee,
            transaction_fee_currency=transaction_fee_currency,
            trade_id=trade_id,
            comment=comment,
            type=TRADE))

# Withdrawal sets always come in twos in the following order:
WITHDRAWAL_INDEX = 0
WITHDRAWAL_FEE_INDEX = 1


def sanity_check_withdrawal_set(withdrawal_set):
    if len(withdrawal_set) != 2:
        raise ValueError(
            'Unexpected withdrawal set size: %s' % json.dumps(withdrawal_set))
    if withdrawal_set[0]['category'] != 'Withdraw':
        raise ValueError(
            'Unexpected category: %s' % json.dumps(withdrawal_set[0]))
    if withdrawal_set[1]['category'] != 'Withdraw fee':
        raise ValueError(
            'Unexpected category: %s' % json.dumps(withdrawal_set[1]))
    if withdrawal_set[0]['created_at'] != withdrawal_set[1]['created_at']:
        raise ValueError('Trade set timestamps don\'t match: %s' %
                         json.dumps(withdrawal_set))


for withdrawal_set in withdrawal_sets.itervalues():
    sanity_check_withdrawal_set(withdrawal_set)
    fee = abs(decimal.Decimal(withdrawal_set[WITHDRAWAL_FEE_INDEX]['value']))
    date = withdrawal_set[WITHDRAWAL_INDEX]['created_at']
    withdrawal_amount = decimal.Decimal(withdrawal_set[WITHDRAWAL_INDEX]['value']) - fee
    withdrawal_currency = withdrawal_set[WITHDRAWAL_INDEX]['account']
    withdrawal_fee_amount = fee
    withdrawal_fee_currency = withdrawal_set[WITHDRAWAL_FEE_INDEX]['account']
    trade_id = withdrawal_set[WITHDRAWAL_INDEX]['details']
    comment = 'Withdrawal of %s %s minus fee of %s %s' % (
        abs(decimal.Decimal(withdrawal_amount)), withdrawal_currency,
        withdrawal_fee_amount, withdrawal_fee_currency)
    transactions.append(
        Transaction(
            date=date,
            buy_amount=None,
            buy_currency=None,
            sell_amount=withdrawal_amount,
            sell_currency=withdrawal_currency,
            transaction_fee=withdrawal_fee_amount,
            transaction_fee_currency=withdrawal_fee_currency,
            trade_id=trade_id,
            comment=comment,
            type=WITHDRAWAL))

transactions.sort(key=lambda t: t.date)

with open(sys.argv[2], 'w') as output_file:
    writer = csv.DictWriter(
        output_file, fieldnames=Transaction.fields, lineterminator='\n')
    writer.writeheader()
    for transaction in transactions:
        writer.writerow(transaction.as_dict())
        print transaction.as_dict()

print 'Converted %d transactions' % len(transactions)