144 lines
5.7 KiB
Python
144 lines
5.7 KiB
Python
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,
|
|
)
|