133 lines
6.4 KiB
Python
133 lines
6.4 KiB
Python
import csv
|
|
import io
|
|
from datetime import date
|
|
from typing import Annotated
|
|
from fastapi import APIRouter, Form, HTTPException, Query, Request
|
|
from fastapi.responses import RedirectResponse, StreamingResponse
|
|
from services import trading_service as service
|
|
from services.asset_service import list_assets, create_asset
|
|
from trading_models import TYPES, SOURCES, STRATEGIES, TradingValidationError
|
|
from routes.export import safe_cell
|
|
from views import render
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def choices():
|
|
return dict(trade_types=TYPES, sources=SOURCES, strategies=STRATEGIES)
|
|
|
|
|
|
def trade_form(request, data, error=None, status=200, transaction_id=None):
|
|
assets = [a for a in list_assets() if a['asset_type'] in {'stock','etf','bond','crypto'}]
|
|
return render(request, 'trading_form.html', dict(data=data, assets=assets, error=error, transaction_id=transaction_id, **choices()), status)
|
|
|
|
|
|
@router.get('/trading')
|
|
def index(request: Request, currency: Annotated[str, Query(pattern='^[A-Z]{3}$')] = 'EUR'):
|
|
return render(request, 'trading.html', dict(stats=service.trading_stats(currency), entries=service.list_transactions(limit=20), **choices()))
|
|
|
|
|
|
@router.get('/trading/positions')
|
|
def positions(request: Request):
|
|
return render(request, 'trading_positions.html', {'positions': service.positions()})
|
|
|
|
|
|
@router.get('/trading/transactions')
|
|
def history(request: Request, page: Annotated[int, Query(ge=1, le=1000000)] = 1):
|
|
filters = {}
|
|
for field, maximum in [('year',9999), ('month',12), ('asset_id',9223372036854775807)]:
|
|
value = request.query_params.get(field)
|
|
if value:
|
|
try:
|
|
parsed = int(value)
|
|
if not 1 <= parsed <= maximum:
|
|
raise ValueError
|
|
except ValueError:
|
|
raise HTTPException(422, 'Ungültiger Filter.') from None
|
|
filters[field] = parsed
|
|
for field, allowed in [('transaction_type',TYPES), ('source',SOURCES), ('strategy_tag',{**STRATEGIES,'untagged':'Ohne Tag'})]:
|
|
value = request.query_params.get(field)
|
|
if value:
|
|
if value not in allowed:
|
|
raise HTTPException(422, 'Ungültiger Filter.')
|
|
filters[field] = value
|
|
rows = service.list_transactions(**filters, limit=101, offset=(page-1)*100)
|
|
return render(request, 'trading_history.html', dict(entries=rows[:100], more=len(rows)>100, page=page,
|
|
filters=filters, assets=list_assets(), years=service.available_years(), **choices()))
|
|
|
|
|
|
@router.get('/trading/assets/new')
|
|
def new_asset(request: Request):
|
|
return render(request, 'trading_asset_form.html', {'data': {}})
|
|
|
|
|
|
@router.post('/trading/assets/new')
|
|
def save_asset(request: Request, name: Annotated[str, Form()] = '', asset_type: Annotated[str, Form()] = 'stock', ticker: Annotated[str, Form()] = ''):
|
|
try:
|
|
if asset_type not in {'stock','etf','bond','crypto'}:
|
|
raise ValueError('Bitte Aktie, ETF, Anleihe oder Krypto auswählen.')
|
|
asset = create_asset(name, asset_type, ticker, reuse=True)
|
|
except ValueError as error:
|
|
return render(request, 'trading_asset_form.html', {'data':dict(name=name,asset_type=asset_type,ticker=ticker),'error':str(error)}, 422)
|
|
return RedirectResponse(f"/trading/transactions/new?asset_id={asset['id']}", status_code=303)
|
|
|
|
|
|
@router.get('/trading/assets/{asset_id}')
|
|
def asset_detail(request: Request, asset_id: int):
|
|
return render(request, 'trading_asset.html', {**service.asset_detail(asset_id), **choices()})
|
|
|
|
|
|
@router.get('/trading/transactions/new')
|
|
def new(request: Request):
|
|
return trade_form(request, dict(date=date.today().isoformat(), currency='EUR', fees='0', transaction_type='buy', source='manual', asset_id=request.query_params.get('asset_id','')))
|
|
|
|
|
|
@router.get('/trading/transactions/{transaction_id}/edit')
|
|
def edit(request: Request, transaction_id: int):
|
|
return trade_form(request, service.get_transaction(transaction_id), transaction_id=transaction_id)
|
|
|
|
|
|
@router.post('/trading/transactions/new')
|
|
@router.post('/trading/transactions/{transaction_id}/edit')
|
|
def save(request: Request, date: Annotated[str, Form()] = '', asset_id: Annotated[str, Form()] = '',
|
|
transaction_type: Annotated[str, Form()] = 'buy', quantity: Annotated[str, Form()] = '',
|
|
price_per_unit: Annotated[str, Form()] = '', currency: Annotated[str, Form()] = 'EUR',
|
|
fees: Annotated[str, Form()] = '0', source: Annotated[str, Form()] = 'manual',
|
|
strategy_tag: Annotated[str, Form()] = '', note: Annotated[str, Form()] = '', transaction_id: int | None = None):
|
|
data = dict(date=date,asset_id=asset_id,transaction_type=transaction_type,quantity=quantity,price_per_unit=price_per_unit,
|
|
currency=currency,fees=fees,source=source,strategy_tag=strategy_tag,note=note)
|
|
try:
|
|
service.save_transaction(data, transaction_id)
|
|
except TradingValidationError as error:
|
|
return trade_form(request, data, str(error), 422, transaction_id)
|
|
return RedirectResponse('/trading?message=trade_saved', status_code=303)
|
|
|
|
|
|
@router.post('/trading/transactions/{transaction_id}/delete')
|
|
def delete(request: Request, transaction_id: int):
|
|
try:
|
|
service.delete_transaction(transaction_id)
|
|
except TradingValidationError as error:
|
|
return render(request, 'trading_error.html', {'error':str(error)}, 422)
|
|
return RedirectResponse('/trading?message=trade_deleted', status_code=303)
|
|
|
|
|
|
def export_rows():
|
|
buffer = io.StringIO(newline='')
|
|
writer = csv.writer(buffer, delimiter=';', lineterminator='\r\n')
|
|
yield '\ufeff'
|
|
writer.writerow(['Datum','Position','Typ','Stückzahl','Kurs','Währung','Gebühren','Gesamtbetrag','Source','Strategie-Tag','Notiz'])
|
|
yield buffer.getvalue()
|
|
buffer.seek(0); buffer.truncate(0)
|
|
for row in service.list_transactions():
|
|
writer.writerow([row['date'], safe_cell(row['asset']), TYPES[row['transaction_type']], row['quantity'].replace('.',','),
|
|
row['price_per_unit'].replace('.',','), row['currency'], row['fees'].replace('.',','),
|
|
row['total_amount'].replace('.',','), SOURCES[row['source']], STRATEGIES.get(row['strategy_tag'],'Ohne Tag'), safe_cell(row['note'])])
|
|
yield buffer.getvalue()
|
|
buffer.seek(0); buffer.truncate(0)
|
|
|
|
|
|
@router.get('/export/trading.csv')
|
|
def export():
|
|
return StreamingResponse(export_rows(), media_type='text/csv; charset=utf-8', headers={'Content-Disposition':'attachment; filename="trading.csv"'})
|