vault-dash/app/domain/backtesting_math.py

from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal

from app.domain.units import BaseCurrency, Money, Weight, WeightUnit, _coerce_currency, decimal_from_float, to_decimal
from app.models.backtest import BacktestPortfolioState
from app.models.portfolio import PortfolioConfig


@dataclass(frozen=True, slots=True)
class AssetQuantity:
    amount: Decimal
    symbol: str

    def __post_init__(self) -> None:
        object.__setattr__(self, "amount", to_decimal(self.amount))
        symbol = str(self.symbol).strip().upper()
        if not symbol:
            raise ValueError("Asset symbol is required")
        object.__setattr__(self, "symbol", symbol)

    def __mul__(self, other: object) -> Money:
        if isinstance(other, PricePerAsset):
            other.assert_symbol(self.symbol)
            return Money(amount=self.amount * other.amount, currency=other.currency)
        return NotImplemented

    def __truediv__(self, other: object) -> AssetQuantity:
        if isinstance(other, bool):
            return NotImplemented
        if isinstance(other, Decimal):
            return AssetQuantity(amount=self.amount / other, symbol=self.symbol)
        if isinstance(other, int):
            return AssetQuantity(amount=self.amount / Decimal(other), symbol=self.symbol)
        return NotImplemented


@dataclass(frozen=True, slots=True)
class PricePerAsset:
    amount: Decimal
    currency: BaseCurrency | str
    symbol: str

    def __post_init__(self) -> None:
        amount = to_decimal(self.amount)
        if amount < 0:
            raise ValueError("PricePerAsset amount must be non-negative")
        object.__setattr__(self, "amount", amount)
        object.__setattr__(self, "currency", _coerce_currency(self.currency))
        symbol = str(self.symbol).strip().upper()
        if not symbol:
            raise ValueError("Asset symbol is required")
        object.__setattr__(self, "symbol", symbol)

    @property
    def _currency_typed(self) -> BaseCurrency:
        """Type-narrowed currency accessor for internal use."""
        return self.currency  # type: ignore[return-value]

    def assert_symbol(self, symbol: str) -> PricePerAsset:
        normalized = str(symbol).strip().upper()
        if self.symbol != normalized:
            raise ValueError(f"Asset symbol mismatch: {self.symbol} != {normalized}")
        return self

    def __mul__(self, other: object) -> Money | PricePerAsset:
        if isinstance(other, bool):
            return NotImplemented
        if isinstance(other, AssetQuantity):
            other_symbol = str(other.symbol).strip().upper()
            self.assert_symbol(other_symbol)
            return Money(amount=self.amount * other.amount, currency=self.currency)
        if isinstance(other, Decimal):
            return PricePerAsset(amount=self.amount * other, currency=self.currency, symbol=self.symbol)
        if isinstance(other, int):
            return PricePerAsset(amount=self.amount * Decimal(other), currency=self.currency, symbol=self.symbol)
        return NotImplemented

    def __rmul__(self, other: object) -> PricePerAsset:
        if isinstance(other, bool):
            return NotImplemented
        if isinstance(other, Decimal):
            return PricePerAsset(amount=self.amount * other, currency=self.currency, symbol=self.symbol)
        if isinstance(other, int):
            return PricePerAsset(amount=self.amount * Decimal(other), currency=self.currency, symbol=self.symbol)
        return NotImplemented


def asset_quantity_from_money(value: Money, spot: PricePerAsset) -> AssetQuantity:
    value.assert_currency(spot._currency_typed)
    if spot.amount <= 0:
        raise ValueError("Spot price per asset must be positive")
    return AssetQuantity(amount=value.amount / spot.amount, symbol=spot.symbol)


def asset_quantity_from_floats(portfolio_value: float, entry_spot: float, symbol: str) -> float:
    notional = Money(amount=decimal_from_float(portfolio_value), currency=BaseCurrency.USD)
    spot = PricePerAsset(amount=decimal_from_float(entry_spot), currency=BaseCurrency.USD, symbol=symbol)
    return float(asset_quantity_from_money(notional, spot).amount)


def asset_quantity_from_workspace_config(config: PortfolioConfig, *, entry_spot: float, symbol: str) -> float:
    if config.gold_ounces is not None and config.gold_ounces > 0:
        try:
            from app.domain.instruments import asset_quantity_from_weight

            quantity = asset_quantity_from_weight(
                symbol,
                Weight(amount=decimal_from_float(float(config.gold_ounces)), unit=WeightUnit.OUNCE_TROY),
            )
            return float(quantity.amount)
        except ValueError:
            pass
    if config.gold_value is not None:
        return asset_quantity_from_floats(float(config.gold_value), entry_spot, symbol)
    raise ValueError("Workspace config must provide collateral weight or value")


def materialize_backtest_portfolio_state(
    *,
    symbol: str,
    underlying_units: float,
    entry_spot: float,
    loan_amount: float,
    margin_call_ltv: float,
    currency: str = "USD",
    cash_balance: float = 0.0,
    financing_rate: float = 0.0,
) -> BacktestPortfolioState:
    normalized_symbol = str(symbol).strip().upper()
    quantity = AssetQuantity(amount=decimal_from_float(underlying_units), symbol=normalized_symbol)
    spot = PricePerAsset(
        amount=decimal_from_float(entry_spot),
        currency=BaseCurrency.USD,
        symbol=normalized_symbol,
    )
    loan = Money(amount=decimal_from_float(loan_amount), currency=currency)
    _ = quantity * spot
    return BacktestPortfolioState(
        currency=str(loan.currency),
        underlying_units=float(quantity.amount),
        entry_spot=float(spot.amount),
        loan_amount=float(loan.amount),
        margin_call_ltv=margin_call_ltv,
        cash_balance=cash_balance,
        financing_rate=financing_rate,
    )