Restructure
This commit is contained in:
parent
0e9ba95545
commit
3f1d576611
SSH Key Fingerprint:
SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
|
|
@ -2,3 +2,123 @@
|
|||
|
||||
WIP
|
||||
"""
|
||||
|
||||
import sys
|
||||
import requests
|
||||
import fin_depo
|
||||
from . import secrets
|
||||
from fin_defs import CryptoCurrency, AssetAmount, MPC, Asset, USDT
|
||||
from decimal import Decimal
|
||||
from collections import deque
|
||||
from fin_depo.data import *
|
||||
import datetime
|
||||
import dataclasses
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BoughtAndSold:
|
||||
amount: AssetAmount
|
||||
time_bought: datetime.datetime
|
||||
time_sold: datetime.datetime
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BoughtAndNotYetSold:
|
||||
amount: AssetAmount
|
||||
time_bought: datetime.datetime
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class TaxReport:
|
||||
bought_and_sold_for: list[BoughtAndSold]
|
||||
bought_and_spent_for: list[BoughtAndSold]
|
||||
current_assets: dict[Asset, deque[BoughtAndNotYetSold]]
|
||||
|
||||
|
||||
def compute_tax(transfers: list) -> TaxReport:
|
||||
transfers.sort(key=lambda t:t.executed_time)
|
||||
|
||||
|
||||
bought_and_sold_for: list[BoughtAndSold] = []
|
||||
bought_and_spent_for: list[BoughtAndSold] = []
|
||||
|
||||
prices_bought_for: dict[Asset, deque[BoughtAndNotYetSold]] = {}
|
||||
if False:
|
||||
prices_bought_for[MPC] = deque()
|
||||
prices_bought_for[MPC].append(BoughtAndNotYetSold(AssetAmount(MPC, Decimal(100)), datetime.datetime(2000, 1,1,1,1,1,1)))
|
||||
prices_bought_for[USDT] = deque()
|
||||
prices_bought_for[USDT].append(BoughtAndNotYetSold(AssetAmount(USDT, Decimal(100)), datetime.datetime(2000, 1,1,1,1,1,1)))
|
||||
|
||||
def sell(amount: AssetAmount, executed_time: datetime.datetime, spent=False):
|
||||
logger.info(f'Selling: {amount}')
|
||||
while amount.amount > 0:
|
||||
if len(prices_bought_for[amount.asset]) == 0:
|
||||
raise Exception('Miscalculation: ' + str(amount))
|
||||
partial = prices_bought_for[amount.asset].popleft()
|
||||
|
||||
logger.info(f'Fuck : {amount} - {partial.amount}')
|
||||
|
||||
amount_covered_by_partial = min(partial.amount, amount)
|
||||
amount -= amount_covered_by_partial
|
||||
new_partial_amount = partial.amount - amount_covered_by_partial
|
||||
|
||||
logger.info(f' => {amount} ({new_partial_amount}')
|
||||
|
||||
# Re-add partial
|
||||
if new_partial_amount.amount != 0:
|
||||
new_partial = BoughtAndNotYetSold(new_partial_amount, partial.time_bought)
|
||||
logger.info(f' => new partial: {new_partial}')
|
||||
prices_bought_for[amount.asset].appendleft(new_partial)
|
||||
del new_partial
|
||||
|
||||
# Add FIFO like
|
||||
if spent:
|
||||
bought_and_spent_for.append(BoughtAndSold(amount_covered_by_partial, partial.time_bought, executed_time))
|
||||
else:
|
||||
bought_and_sold_for.append(BoughtAndSold(amount_covered_by_partial, partial.time_bought, executed_time))
|
||||
|
||||
del partial, amount_covered_by_partial
|
||||
|
||||
def spend(amount: AssetAmount, executed_time: datetime.datetime):
|
||||
logger.info(f'Spending: {amount}')
|
||||
sell(amount, executed_time, spent=True)
|
||||
|
||||
for transfer in transfers:
|
||||
logger.info(transfer)
|
||||
# Buy
|
||||
if output := transfer.output:
|
||||
logger.info('Buy')
|
||||
if output.asset not in prices_bought_for:
|
||||
prices_bought_for[output.asset] = deque()
|
||||
prices_bought_for[output.asset].append(BoughtAndNotYetSold(output, transfer.executed_time))
|
||||
del output
|
||||
|
||||
# Sell
|
||||
if _input := transfer.input:
|
||||
logger.info('Sell')
|
||||
sell(_input, transfer.executed_time)
|
||||
del _input
|
||||
|
||||
if fee := transfer.fee:
|
||||
logger.info('Sell')
|
||||
spend(fee, transfer.executed_time)
|
||||
del transfer, fee
|
||||
|
||||
if True:
|
||||
logger.info('Bought for:')
|
||||
for asset, prices in prices_bought_for.items():
|
||||
price_sum = sum((p.amount for p in prices), start=AssetAmount.ZERO)
|
||||
sys.stdout.write(f' - {asset.raw_short_name():10} ({price_sum}): ')
|
||||
for price in prices:
|
||||
sys.stdout.write(str(price.amount.amount))
|
||||
sys.stdout.write(', ')
|
||||
del price
|
||||
sys.stdout.write('\n')
|
||||
del asset, prices
|
||||
|
||||
return TaxReport(
|
||||
bought_and_sold_for=bought_and_sold_for,
|
||||
bought_and_spent_for=bought_and_spent_for,
|
||||
current_assets=prices_bought_for,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -8,6 +8,9 @@ from collections import deque
|
|||
from fin_depo.data import *
|
||||
import datetime
|
||||
import dataclasses
|
||||
import logging
|
||||
|
||||
from . import compute_tax
|
||||
|
||||
KUCOIN_CLIENT = fin_depo.defi_kucoin.KucoinDepoFetcher(
|
||||
secrets.KUCOIN_KEY,
|
||||
|
|
@ -20,120 +23,26 @@ KRAKEN_CLIENT = fin_depo.defi_kraken.KrakenDepoFetcher(
|
|||
secrets.KRAKEN_SECRET,
|
||||
)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BoughtAndSold:
|
||||
amount: AssetAmount
|
||||
time_bought: datetime.datetime
|
||||
time_sold: datetime.datetime
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BoughtAndNotYetSold:
|
||||
amount: AssetAmount
|
||||
time_bought: datetime.datetime
|
||||
|
||||
|
||||
def compute_tax(transfers: list):
|
||||
transfers.sort(key=lambda t:t.executed_time)
|
||||
|
||||
|
||||
bought_and_sold_for: list[BoughtAndSold] = []
|
||||
bought_and_spent_for: list[BoughtAndSold] = []
|
||||
|
||||
prices_bought_for: dict[Asset, deque[BoughtAndNotYetSold]] = {}
|
||||
if False:
|
||||
prices_bought_for[MPC] = deque()
|
||||
prices_bought_for[MPC].append(BoughtAndNotYetSold(AssetAmount(MPC, Decimal(100)), datetime.datetime(2000, 1,1,1,1,1,1)))
|
||||
prices_bought_for[USDT] = deque()
|
||||
prices_bought_for[USDT].append(BoughtAndNotYetSold(AssetAmount(USDT, Decimal(100)), datetime.datetime(2000, 1,1,1,1,1,1)))
|
||||
|
||||
def sell(amount: AssetAmount, executed_time: datetime.datetime, spent=False):
|
||||
print(f'Selling: {amount}')
|
||||
while amount.amount > 0:
|
||||
if len(prices_bought_for[amount.asset]) == 0:
|
||||
raise Exception('Miscalculation: ' + str(amount))
|
||||
partial = prices_bought_for[amount.asset].popleft()
|
||||
|
||||
print(f'Fuck : {amount} - {partial.amount}')
|
||||
|
||||
amount_covered_by_partial = min(partial.amount, amount)
|
||||
amount -= amount_covered_by_partial
|
||||
new_partial_amount = partial.amount - amount_covered_by_partial
|
||||
|
||||
print(f' => {amount} ({new_partial_amount}')
|
||||
|
||||
# Re-add partial
|
||||
if new_partial_amount.amount != 0:
|
||||
new_partial = BoughtAndNotYetSold(new_partial_amount, partial.time_bought)
|
||||
print(f' => new partial: {new_partial}')
|
||||
prices_bought_for[amount.asset].appendleft(new_partial)
|
||||
del new_partial
|
||||
|
||||
# Add FIFO like
|
||||
if spent:
|
||||
bought_and_spent_for.append(BoughtAndSold(amount_covered_by_partial, partial.time_bought, executed_time))
|
||||
else:
|
||||
bought_and_sold_for.append(BoughtAndSold(amount_covered_by_partial, partial.time_bought, executed_time))
|
||||
|
||||
del partial, amount_covered_by_partial
|
||||
|
||||
def spend(amount: AssetAmount, executed_time: datetime.datetime):
|
||||
print(f'Spending: {amount}')
|
||||
sell(amount, executed_time, spent=True)
|
||||
|
||||
for transfer in transfers:
|
||||
print(transfer)
|
||||
# Buy
|
||||
if output := transfer.output:
|
||||
print('Buy')
|
||||
if output.asset not in prices_bought_for:
|
||||
prices_bought_for[output.asset] = deque()
|
||||
prices_bought_for[output.asset].append(BoughtAndNotYetSold(output, transfer.executed_time))
|
||||
del output
|
||||
|
||||
# Sell
|
||||
if _input := transfer.input:
|
||||
print('Sell')
|
||||
sell(_input, transfer.executed_time)
|
||||
del _input
|
||||
|
||||
if fee := transfer.fee:
|
||||
print('Sell')
|
||||
spend(fee, transfer.executed_time)
|
||||
del transfer, fee
|
||||
|
||||
if True:
|
||||
print('Bought for:')
|
||||
for asset, prices in prices_bought_for.items():
|
||||
price_sum = sum((p.amount for p in prices), start=AssetAmount.ZERO)
|
||||
sys.stdout.write(f' - {asset.raw_short_name():10} ({price_sum}): ')
|
||||
for price in prices:
|
||||
sys.stdout.write(str(price.amount.amount))
|
||||
sys.stdout.write(', ')
|
||||
del price
|
||||
sys.stdout.write('\n')
|
||||
del asset, prices
|
||||
|
||||
|
||||
print('-'*80)
|
||||
|
||||
|
||||
for bas in bought_and_sold_for:
|
||||
print(f'{bas.amount} ({bas.time_bought} ----> {bas.time_sold})')
|
||||
|
||||
|
||||
for c, l in prices_bought_for.items():
|
||||
print(c)
|
||||
print(l)
|
||||
|
||||
def main():
|
||||
"""Main function."""
|
||||
|
||||
logger = logging.getLogger('crypto_tax')
|
||||
logger.setLevel('INFO')
|
||||
|
||||
TRANSFERS = list(KRAKEN_CLIENT._get_double_registers())
|
||||
for k in TRANSFERS:
|
||||
print(k)
|
||||
#TRANSFERS += list(KUCOIN_CLIENT._get_double_registers())
|
||||
compute_tax(TRANSFERS)
|
||||
tax_report = compute_tax(TRANSFERS)
|
||||
|
||||
logger.info('-'*80)
|
||||
|
||||
|
||||
for bas in tax_report.bought_and_sold_for:
|
||||
logger.info(f'{bas.amount} ({bas.time_bought} ----> {bas.time_sold})')
|
||||
|
||||
|
||||
for c, prices in tax_report.current_assets.items():
|
||||
logger.info('%s', c)
|
||||
logger.info('%s', prices)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user