229 lines
12 KiB
Python
229 lines
12 KiB
Python
"""Trading ledger and weighted-average open cost; independent of income entries."""
|
|
from datetime import date
|
|
from decimal import Decimal, localcontext, ROUND_DOWN, ROUND_HALF_UP
|
|
from fastapi import HTTPException
|
|
from database import connect
|
|
from trading_models import (CENT, ZERO, SOURCES, STRATEGIES, TradingValidationError,
|
|
validate_trade, fixed, exact, currency_code)
|
|
|
|
FIELDS = ('date', 'asset_id', 'transaction_type', 'quantity', 'price_per_unit', 'currency',
|
|
'fees', 'source', 'strategy_tag', 'note')
|
|
SELECT = 'SELECT t.*, a.name asset, a.ticker, a.asset_type FROM transactions t JOIN assets a ON a.id=t.asset_id'
|
|
|
|
|
|
def _get(db, transaction_id):
|
|
if not 1 <= transaction_id <= 9223372036854775807:
|
|
raise HTTPException(404, 'Transaktion nicht gefunden.')
|
|
row = db.execute(SELECT + ' WHERE t.id=?', (transaction_id,)).fetchone()
|
|
if row is None:
|
|
raise HTTPException(404, 'Transaktion nicht gefunden.')
|
|
return dict(row)
|
|
|
|
|
|
def amounts(row):
|
|
with localcontext() as ctx:
|
|
ctx.prec = 60
|
|
gross = (Decimal(row['quantity']) * Decimal(row['price_per_unit'])).quantize(CENT, rounding=ROUND_HALF_UP)
|
|
fees = Decimal(row['fees'])
|
|
total = gross + fees if row['transaction_type'] == 'buy' else gross - fees
|
|
return dict(gross_amount=fixed(gross), total_amount=fixed(total),
|
|
total_cost=fixed(total) if row['transaction_type'] == 'buy' else None,
|
|
net_proceeds=fixed(total) if row['transaction_type'] == 'sell' else None)
|
|
|
|
|
|
def _reduce_buckets(buckets, removal, total):
|
|
"""Distribute disposed cost by largest remainder, keeping every cent accounted for."""
|
|
if not removal or not total:
|
|
return
|
|
if removal == total:
|
|
for key in buckets:
|
|
buckets[key] = ZERO
|
|
return
|
|
allocations = {key: (value * removal / total).quantize(CENT, rounding=ROUND_DOWN) for key, value in buckets.items()}
|
|
remainders = sorted(buckets, key=lambda key: (-(buckets[key] * removal / total - allocations[key]), key))
|
|
missing = int((removal - sum(allocations.values(), ZERO)) / CENT)
|
|
for key in remainders[:missing]:
|
|
allocations[key] += CENT
|
|
for key in buckets:
|
|
buckets[key] -= allocations[key]
|
|
|
|
|
|
def replay(rows):
|
|
"""Deterministic date/id order. Also validates historical inventory after mutations."""
|
|
with localcontext() as ctx:
|
|
ctx.prec = 60
|
|
positions, ledger = {}, []
|
|
for original in rows:
|
|
row = dict(original)
|
|
aid = row['asset_id']
|
|
if aid not in positions:
|
|
positions[aid] = dict(asset_id=aid, asset=row['asset'], ticker=row['ticker'], asset_type=row['asset_type'],
|
|
currency=row['currency'], quantity=ZERO, invested_capital=ZERO, realized_profit_loss=ZERO,
|
|
total_buys=ZERO, total_sells=ZERO, buy_count=0, sell_count=0,
|
|
first_transaction=row['date'], last_transaction=row['date'],
|
|
sources={key: ZERO for key in SOURCES}, strategies={key: ZERO for key in [*STRATEGIES, 'untagged']})
|
|
position = positions[aid]
|
|
if position['currency'] != row['currency']:
|
|
raise TradingValidationError('Eine Position muss in einer einheitlichen Währung geführt werden. Keine automatische Währungsumrechnung.')
|
|
quantity = Decimal(row['quantity'])
|
|
row.update(amounts(row))
|
|
row['quantity_before'] = exact(position['quantity'])
|
|
realized = ZERO
|
|
if row['transaction_type'] == 'buy':
|
|
cost = Decimal(row['total_cost'])
|
|
position['quantity'] += quantity
|
|
position['invested_capital'] += cost
|
|
position['total_buys'] += cost
|
|
position['buy_count'] += 1
|
|
position['sources'][row['source']] += cost
|
|
position['strategies'][row['strategy_tag'] or 'untagged'] += cost
|
|
else:
|
|
if quantity > position['quantity']:
|
|
raise TradingValidationError('Verkauf übersteigt den Bestand am Buchungsdatum. Auch spätere Verkäufe müssen nach Änderungen gedeckt bleiben.')
|
|
cost = position['invested_capital']
|
|
removed = cost if quantity == position['quantity'] else (cost * quantity / position['quantity']).quantize(CENT, rounding=ROUND_HALF_UP)
|
|
_reduce_buckets(position['sources'], removed, cost)
|
|
_reduce_buckets(position['strategies'], removed, cost)
|
|
position['quantity'] -= quantity
|
|
position['invested_capital'] -= removed
|
|
proceeds = Decimal(row['net_proceeds'])
|
|
realized = proceeds - removed
|
|
position['realized_profit_loss'] += realized
|
|
position['total_sells'] += proceeds
|
|
position['sell_count'] += 1
|
|
position['last_transaction'] = row['date']
|
|
row['quantity_after'] = exact(position['quantity'])
|
|
row['realized_profit_loss'] = fixed(realized)
|
|
ledger.append(row)
|
|
for position in positions.values():
|
|
position['average_cost'] = position['invested_capital'] / position['quantity'] if position['quantity'] else ZERO
|
|
return list(positions.values()), ledger
|
|
|
|
|
|
def _asset_history(db, asset_id):
|
|
return db.execute(SELECT + ' WHERE t.asset_id=? ORDER BY t.date,t.id', (asset_id,)).fetchall()
|
|
|
|
|
|
def save_transaction(data, transaction_id=None):
|
|
with connect() as db:
|
|
db.execute('BEGIN IMMEDIATE')
|
|
old = _get(db, transaction_id) if transaction_id is not None else None
|
|
merged = {**old, **data} if old else data
|
|
values = validate_trade(merged)
|
|
asset = db.execute('SELECT * FROM assets WHERE id=?', (values['asset_id'],)).fetchone()
|
|
if asset is None:
|
|
raise TradingValidationError('Position existiert nicht.')
|
|
if asset['asset_type'] not in {'stock', 'etf', 'bond', 'crypto'}:
|
|
raise TradingValidationError('Trading ist für Aktien, ETFs, Anleihen und Krypto möglich.')
|
|
if not asset['active'] and (old is None or old['asset_id'] != asset['id']) and values['transaction_type'] == 'buy':
|
|
raise TradingValidationError('Neue Käufe für inaktive Positionen sind nicht möglich.')
|
|
if old:
|
|
# Field names are a fixed internal tuple, never supplied by the request.
|
|
db.execute('UPDATE transactions SET ' + ','.join(field+'=?' for field in FIELDS) + ", updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id=?",
|
|
(*[values[field] for field in FIELDS], transaction_id))
|
|
else:
|
|
transaction_id = db.execute('INSERT INTO transactions ('+','.join(FIELDS)+') VALUES (?,?,?,?,?,?,?,?,?,?)',
|
|
[values[field] for field in FIELDS]).lastrowid
|
|
affected = {values['asset_id']}
|
|
if old:
|
|
affected.add(old['asset_id'])
|
|
for aid in affected:
|
|
replay(_asset_history(db, aid))
|
|
return {**_get(db, transaction_id), **amounts(values)}
|
|
|
|
|
|
def delete_transaction(transaction_id):
|
|
with connect() as db:
|
|
db.execute('BEGIN IMMEDIATE')
|
|
old = _get(db, transaction_id)
|
|
db.execute('DELETE FROM transactions WHERE id=?', (transaction_id,))
|
|
replay(_asset_history(db, old['asset_id']))
|
|
|
|
|
|
def get_transaction(transaction_id):
|
|
with connect() as db:
|
|
row = _get(db, transaction_id)
|
|
return {**row, **amounts(row)}
|
|
|
|
|
|
def list_transactions(year=None, month=None, asset_id=None, transaction_type=None, source=None, strategy_tag=None, limit=None, offset=0):
|
|
clauses, args = [], []
|
|
for clause, value in [("substr(t.date,1,4)=?", f'{year:04}' if year else None),
|
|
("substr(t.date,6,2)=?", f'{month:02}' if month else None),
|
|
('t.asset_id=?', asset_id), ('t.transaction_type=?', transaction_type), ('t.source=?', source)]:
|
|
if value is not None:
|
|
clauses.append(clause)
|
|
args.append(value)
|
|
if strategy_tag == 'untagged':
|
|
clauses.append('t.strategy_tag IS NULL')
|
|
elif strategy_tag is not None:
|
|
clauses.append('t.strategy_tag=?')
|
|
args.append(strategy_tag)
|
|
query = SELECT + (' WHERE ' + ' AND '.join(clauses) if clauses else '') + ' ORDER BY t.date DESC,t.id DESC'
|
|
if limit is not None:
|
|
query += ' LIMIT ? OFFSET ?'
|
|
args.extend([limit, offset])
|
|
with connect() as db:
|
|
return [{**dict(row), **amounts(row)} for row in db.execute(query, args)]
|
|
|
|
|
|
def available_years():
|
|
with connect() as db:
|
|
return [int(row[0]) for row in db.execute('SELECT DISTINCT substr(date,1,4) FROM transactions ORDER BY 1')]
|
|
|
|
|
|
def portfolio():
|
|
with connect() as db:
|
|
rows = db.execute(SELECT + ' ORDER BY t.date,t.id').fetchall()
|
|
return replay(rows)
|
|
|
|
|
|
def position_response(position):
|
|
return {key: (exact(value) if key == 'quantity' else fixed(value, 12 if key == 'average_cost' else 2))
|
|
if isinstance(value, Decimal) else value for key, value in position.items() if key not in {'sources', 'strategies'}}
|
|
|
|
|
|
def positions(currency=None, include_closed=False):
|
|
items, _ = portfolio()
|
|
return [position_response(p) for p in items if (include_closed or p['quantity'] > 0) and (currency is None or p['currency'] == currency)]
|
|
|
|
|
|
def asset_detail(asset_id):
|
|
from services.asset_service import get_asset
|
|
asset = get_asset(asset_id)
|
|
with connect() as db:
|
|
items, ledger = replay(_asset_history(db, asset_id))
|
|
return dict(asset=asset, position=position_response(items[0]) if items else None, entries=list(reversed(ledger)))
|
|
|
|
|
|
def trading_stats(currency='EUR', today=None):
|
|
currency = currency_code(currency)
|
|
today = today or date.today()
|
|
with localcontext() as ctx:
|
|
ctx.prec = 60
|
|
all_positions, all_ledger = portfolio()
|
|
ps = [p for p in all_positions if p['currency'] == currency]
|
|
ledger = [r for r in all_ledger if r['currency'] == currency]
|
|
current = [r for r in ledger if int(r['date'][:4]) == today.year]
|
|
invested = sum((p['invested_capital'] for p in ps), ZERO)
|
|
def breakdown(field, keys):
|
|
rows = []
|
|
for key in keys:
|
|
amount = sum((p[field][key] for p in ps), ZERO)
|
|
rows.append({'key': key, 'amount': fixed(amount), 'percentage': fixed(amount * 100 / invested) if invested else None})
|
|
return rows
|
|
years = sorted({int(r['date'][:4]) for r in ledger} | {today.year})
|
|
monthly = {y: [0]*12 for y in years}
|
|
for row in ledger:
|
|
if row['transaction_type'] == 'buy':
|
|
monthly[int(row['date'][:4])][int(row['date'][5:7])-1] += 1
|
|
return dict(currency=currency, invested_capital=fixed(invested), active_positions=sum(p['quantity'] > 0 for p in ps),
|
|
buys_current_year=sum(r['transaction_type'] == 'buy' for r in current),
|
|
sells_current_year=sum(r['transaction_type'] == 'sell' for r in current),
|
|
realized_profit_loss_current_year=fixed(sum((Decimal(r['realized_profit_loss']) for r in current), ZERO)),
|
|
transactions_total=len(ledger), by_source=breakdown('sources', SOURCES),
|
|
by_strategy=breakdown('strategies', [*STRATEGIES, 'untagged']),
|
|
monthly=[{'year': y, 'counts': monthly[y]} for y in years],
|
|
currencies=sorted({p['currency'] for p in all_positions} | {'EUR'}),
|
|
positions=[position_response(p) for p in ps if p['quantity'] > 0])
|