finance-dashboard/app/api/trading.py

from typing import Annotated, Literal
from fastapi import APIRouter, Query, Response
from pydantic import BeforeValidator, Field, model_validator
from api.routes import SafeAPIRoute
from api.schemas import RequestModel, Identifier, Day
from pydantic import BaseModel
from services import trading_service as service
from trading_models import decimal_value, currency_code, exact, fixed

TradeType = Literal['buy', 'sell']
Source = Literal['manual', 'savings_plan', 'roundup', 'cashback', 'rebalancing', 'other']
Strategy = Literal['core', 'income', 'conviction', 'dip_buy', 'speculation', 'rebalancing', 'other']
StrategyFilter = Literal['core', 'income', 'conviction', 'dip_buy', 'speculation', 'rebalancing', 'other', 'untagged']
Quantity = Annotated[str, BeforeValidator(lambda v: exact(decimal_value(v, 'Stückzahl', positive=True)))]
Price = Annotated[str, BeforeValidator(lambda v: exact(decimal_value(v, 'Kurs')))]
Fees = Annotated[str, BeforeValidator(lambda v: fixed(decimal_value(v, 'Gebühren', places=2)))]
Currency = Annotated[str, BeforeValidator(currency_code)]


class TradeCreate(RequestModel):
    date: Day
    asset_id: Identifier
    transaction_type: TradeType = 'buy'
    quantity: Quantity
    price_per_unit: Price
    currency: Currency = 'EUR'
    fees: Fees = '0.00'
    source: Source = 'manual'
    strategy_tag: Strategy | None = None
    note: str | None = Field(default=None, max_length=2000)


class TradePatch(RequestModel):
    date: Day | None = None
    asset_id: Identifier | None = None
    transaction_type: TradeType | None = None
    quantity: Quantity | None = None
    price_per_unit: Price | None = None
    currency: Currency | None = None
    fees: Fees | None = None
    source: Source | None = None
    strategy_tag: Strategy | None = None
    note: str | None = Field(default=None, max_length=2000)

    @model_validator(mode='before')
    @classmethod
    def required_not_null(cls, values):
        if isinstance(values, dict) and any(v is None and k not in {'strategy_tag','note'} for k,v in values.items()):
            raise ValueError('Nur Strategie-Tag und Notiz dürfen null sein.')
        return values


class TradeResponse(BaseModel):
    id: int
    date: str
    asset_id: int
    asset: str
    transaction_type: TradeType
    quantity: str
    price_per_unit: str
    currency: str
    fees: str
    gross_amount: str
    total_amount: str
    total_cost: str | None
    net_proceeds: str | None
    source: Source
    strategy_tag: Strategy | None
    note: str | None
    created_at: str
    updated_at: str


class PositionResponse(BaseModel):
    asset_id: int
    asset: str
    ticker: str | None
    asset_type: str
    currency: str
    quantity: str
    average_cost: str
    invested_capital: str
    total_buys: str
    total_sells: str
    realized_profit_loss: str
    buy_count: int
    sell_count: int
    first_transaction: str
    last_transaction: str


class TradingStatsResponse(BaseModel):
    currency: str
    invested_capital: str
    active_positions: int
    buys_current_year: int
    sells_current_year: int
    realized_profit_loss_current_year: str
    transactions_total: int


class ShareResponse(BaseModel):
    key: str
    amount: str
    percentage: str | None


class BreakdownResponse(BaseModel):
    currency: str
    items: list[ShareResponse]


router = APIRouter(route_class=SafeAPIRoute, tags=['Trading'])


@router.get('/transactions', response_model=list[TradeResponse])
def transactions(year: Annotated[int | None, Query(ge=1, le=9999)] = None,
                 month: Annotated[int | None, Query(ge=1, le=12)] = None,
                 asset_id: Annotated[int | None, Query(ge=1, le=9223372036854775807)] = None,
                 transaction_type: TradeType | None = None, source: Source | None = None,
                 strategy_tag: StrategyFilter | None = None,
                 limit: Annotated[int, Query(ge=1, le=1000)] = 100,
                 offset: Annotated[int, Query(ge=0, le=9223372036854775807)] = 0):
    return [TradeResponse(**row) for row in service.list_transactions(year, month, asset_id, transaction_type, source, strategy_tag, limit, offset)]


@router.get('/transactions/{transaction_id}', response_model=TradeResponse)
def transaction(transaction_id: int):
    return TradeResponse(**service.get_transaction(transaction_id))


@router.post('/transactions', response_model=TradeResponse, status_code=201)
def create(data: TradeCreate):
    return TradeResponse(**service.save_transaction(data.model_dump()))


@router.patch('/transactions/{transaction_id}', response_model=TradeResponse)
def update(transaction_id: int, data: TradePatch):
    return TradeResponse(**service.save_transaction(data.model_dump(exclude_unset=True), transaction_id))


@router.delete('/transactions/{transaction_id}', status_code=204)
def delete(transaction_id: int):
    service.delete_transaction(transaction_id)
    return Response(status_code=204)


@router.get('/positions', response_model=list[PositionResponse])
def positions(currency: Annotated[str | None, Query(pattern='^[A-Z]{3}$')] = None, include_closed: bool = False):
    return [PositionResponse(**row) for row in service.positions(currency, include_closed)]


@router.get('/trading/stats', response_model=TradingStatsResponse)
def stats(currency: Annotated[str, Query(pattern='^[A-Z]{3}$')] = 'EUR'):
    return TradingStatsResponse(**service.trading_stats(currency))


@router.get('/trading/by-source', response_model=BreakdownResponse)
def by_source(currency: Annotated[str, Query(pattern='^[A-Z]{3}$')] = 'EUR'):
    return BreakdownResponse(currency=currency, items=service.trading_stats(currency)['by_source'])


@router.get('/trading/by-strategy', response_model=BreakdownResponse)
def by_strategy(currency: Annotated[str, Query(pattern='^[A-Z]{3}$')] = 'EUR'):
    return BreakdownResponse(currency=currency, items=service.trading_stats(currency)['by_strategy'])