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) 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) 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, )