262 lines
9.8 KiB
Python
262 lines
9.8 KiB
Python
"""Portfolio configuration and domain portfolio models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LombardPortfolio:
|
|
"""Lombard loan portfolio backed by physical gold."""
|
|
|
|
gold_ounces: float
|
|
gold_price_per_ounce: float
|
|
loan_amount: float
|
|
initial_ltv: float
|
|
margin_call_ltv: float
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.gold_ounces <= 0:
|
|
raise ValueError("gold_ounces must be positive")
|
|
if self.gold_price_per_ounce <= 0:
|
|
raise ValueError("gold_price_per_ounce must be positive")
|
|
if self.loan_amount < 0:
|
|
raise ValueError("loan_amount must be non-negative")
|
|
if not 0 < self.initial_ltv < 1:
|
|
raise ValueError("initial_ltv must be between 0 and 1")
|
|
if not 0 < self.margin_call_ltv < 1:
|
|
raise ValueError("margin_call_ltv must be between 0 and 1")
|
|
if self.initial_ltv > self.margin_call_ltv:
|
|
raise ValueError("initial_ltv cannot exceed margin_call_ltv")
|
|
if self.loan_amount > self.gold_value:
|
|
raise ValueError("loan_amount cannot exceed current gold value")
|
|
|
|
@property
|
|
def gold_value(self) -> float:
|
|
return self.gold_ounces * self.gold_price_per_ounce
|
|
|
|
@property
|
|
def current_ltv(self) -> float:
|
|
return self.loan_amount / self.gold_value
|
|
|
|
@property
|
|
def net_equity(self) -> float:
|
|
return self.gold_value - self.loan_amount
|
|
|
|
def gold_value_at_price(self, gold_price_per_ounce: float) -> float:
|
|
if gold_price_per_ounce <= 0:
|
|
raise ValueError("gold_price_per_ounce must be positive")
|
|
return self.gold_ounces * gold_price_per_ounce
|
|
|
|
def ltv_at_price(self, gold_price_per_ounce: float) -> float:
|
|
return self.loan_amount / self.gold_value_at_price(gold_price_per_ounce)
|
|
|
|
def net_equity_at_price(self, gold_price_per_ounce: float) -> float:
|
|
return self.gold_value_at_price(gold_price_per_ounce) - self.loan_amount
|
|
|
|
def margin_call_price(self) -> float:
|
|
return self.loan_amount / (self.margin_call_ltv * self.gold_ounces)
|
|
|
|
|
|
@dataclass
|
|
class PortfolioConfig:
|
|
"""User portfolio configuration with validation.
|
|
|
|
Attributes:
|
|
gold_value: Collateral baseline value in USD at entry
|
|
entry_price: Gold entry price per ounce in USD
|
|
gold_ounces: Canonical gold collateral weight in ounces
|
|
entry_basis_mode: Preferred settings UI input mode
|
|
loan_amount: Outstanding loan amount in USD
|
|
margin_threshold: LTV threshold for margin call (default 0.75)
|
|
monthly_budget: Approved monthly hedge budget
|
|
ltv_warning: LTV warning level for alerts (default 0.70)
|
|
"""
|
|
|
|
gold_value: float | None = None
|
|
entry_price: float | None = 215.0
|
|
gold_ounces: float | None = None
|
|
entry_basis_mode: str = "value_price"
|
|
loan_amount: float = 145000.0
|
|
margin_threshold: float = 0.75
|
|
monthly_budget: float = 8000.0
|
|
ltv_warning: float = 0.70
|
|
|
|
# Data source settings
|
|
primary_source: str = "yfinance"
|
|
fallback_source: str = "yfinance"
|
|
refresh_interval: int = 5
|
|
|
|
# Alert settings
|
|
volatility_spike: float = 0.25
|
|
spot_drawdown: float = 7.5
|
|
email_alerts: bool = False
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Normalize entry basis fields and validate configuration."""
|
|
self._normalize_entry_basis()
|
|
self.validate()
|
|
|
|
def _normalize_entry_basis(self) -> None:
|
|
"""Resolve user input into canonical weight + entry price representation."""
|
|
if self.entry_basis_mode not in {"value_price", "weight"}:
|
|
raise ValueError("Entry basis mode must be 'value_price' or 'weight'")
|
|
|
|
if self.entry_price is None or self.entry_price <= 0:
|
|
raise ValueError("Entry price must be positive")
|
|
|
|
if self.gold_value is not None and self.gold_value <= 0:
|
|
raise ValueError("Gold value must be positive")
|
|
if self.gold_ounces is not None and self.gold_ounces <= 0:
|
|
raise ValueError("Gold weight must be positive")
|
|
|
|
if self.gold_value is None and self.gold_ounces is None:
|
|
self.gold_value = 215000.0
|
|
self.gold_ounces = self.gold_value / self.entry_price
|
|
return
|
|
|
|
if self.gold_value is None and self.gold_ounces is not None:
|
|
self.gold_value = self.gold_ounces * self.entry_price
|
|
return
|
|
|
|
if self.gold_ounces is None and self.gold_value is not None:
|
|
self.gold_ounces = self.gold_value / self.entry_price
|
|
return
|
|
|
|
assert self.gold_value is not None
|
|
assert self.gold_ounces is not None
|
|
derived_gold_value = self.gold_ounces * self.entry_price
|
|
tolerance = max(0.01, abs(derived_gold_value) * 1e-9)
|
|
if abs(self.gold_value - derived_gold_value) > tolerance:
|
|
raise ValueError("Gold value and weight contradict each other")
|
|
self.gold_value = derived_gold_value
|
|
|
|
def validate(self) -> None:
|
|
"""Validate configuration values."""
|
|
assert self.gold_value is not None
|
|
assert self.entry_price is not None
|
|
assert self.gold_ounces is not None
|
|
|
|
if self.gold_value <= 0:
|
|
raise ValueError("Gold value must be positive")
|
|
if self.entry_price <= 0:
|
|
raise ValueError("Entry price must be positive")
|
|
if self.gold_ounces <= 0:
|
|
raise ValueError("Gold weight must be positive")
|
|
if self.loan_amount < 0:
|
|
raise ValueError("Loan amount cannot be negative")
|
|
if self.loan_amount >= self.gold_value:
|
|
raise ValueError("Loan amount must be less than gold value (LTV < 100%)")
|
|
if not 0.1 <= self.margin_threshold <= 0.95:
|
|
raise ValueError("Margin threshold must be between 10% and 95%")
|
|
if not 0.1 <= self.ltv_warning <= 0.95:
|
|
raise ValueError("LTV warning level must be between 10% and 95%")
|
|
if self.ltv_warning >= self.margin_threshold:
|
|
raise ValueError("LTV warning level must be less than the margin threshold")
|
|
if self.refresh_interval < 1:
|
|
raise ValueError("Refresh interval must be at least 1 second")
|
|
|
|
@property
|
|
def current_ltv(self) -> float:
|
|
"""Calculate current loan-to-value ratio."""
|
|
assert self.gold_value is not None
|
|
if self.gold_value == 0:
|
|
return 0.0
|
|
return self.loan_amount / self.gold_value
|
|
|
|
@property
|
|
def margin_buffer(self) -> float:
|
|
"""Calculate margin buffer (distance to margin call)."""
|
|
return self.margin_threshold - self.current_ltv
|
|
|
|
@property
|
|
def net_equity(self) -> float:
|
|
"""Calculate net equity (gold value - loan)."""
|
|
assert self.gold_value is not None
|
|
return self.gold_value - self.loan_amount
|
|
|
|
@property
|
|
def margin_call_price(self) -> float:
|
|
"""Calculate gold price per ounce at which margin call occurs."""
|
|
assert self.gold_ounces is not None
|
|
if self.margin_threshold == 0 or self.gold_ounces == 0:
|
|
return float("inf")
|
|
return self.loan_amount / (self.margin_threshold * self.gold_ounces)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Convert configuration to dictionary."""
|
|
assert self.gold_value is not None
|
|
assert self.entry_price is not None
|
|
assert self.gold_ounces is not None
|
|
return {
|
|
"gold_value": self.gold_value,
|
|
"entry_price": self.entry_price,
|
|
"gold_ounces": self.gold_ounces,
|
|
"entry_basis_mode": self.entry_basis_mode,
|
|
"loan_amount": self.loan_amount,
|
|
"margin_threshold": self.margin_threshold,
|
|
"monthly_budget": self.monthly_budget,
|
|
"ltv_warning": self.ltv_warning,
|
|
"primary_source": self.primary_source,
|
|
"fallback_source": self.fallback_source,
|
|
"refresh_interval": self.refresh_interval,
|
|
"volatility_spike": self.volatility_spike,
|
|
"spot_drawdown": self.spot_drawdown,
|
|
"email_alerts": self.email_alerts,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> PortfolioConfig:
|
|
"""Create configuration from dictionary."""
|
|
return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
|
|
|
|
|
|
class PortfolioRepository:
|
|
"""Repository for persisting portfolio configuration.
|
|
|
|
Uses file-based storage by default. Can be extended to use Redis.
|
|
"""
|
|
|
|
CONFIG_PATH = Path("data/portfolio_config.json")
|
|
|
|
def __init__(self, config_path: Path | None = None) -> None:
|
|
self.config_path = config_path or self.CONFIG_PATH
|
|
self.config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
def save(self, config: PortfolioConfig) -> None:
|
|
"""Save configuration to disk."""
|
|
with open(self.config_path, "w") as f:
|
|
json.dump(config.to_dict(), f, indent=2)
|
|
|
|
def load(self) -> PortfolioConfig:
|
|
"""Load configuration from disk.
|
|
|
|
Returns default configuration if file doesn't exist.
|
|
"""
|
|
if not self.config_path.exists():
|
|
default = PortfolioConfig()
|
|
self.save(default)
|
|
return default
|
|
|
|
try:
|
|
with open(self.config_path) as f:
|
|
data = json.load(f)
|
|
return PortfolioConfig.from_dict(data)
|
|
except (json.JSONDecodeError, ValueError) as e:
|
|
print(f"Warning: Failed to load portfolio config: {e}. Using defaults.")
|
|
return PortfolioConfig()
|
|
|
|
|
|
_portfolio_repo: PortfolioRepository | None = None
|
|
|
|
|
|
def get_portfolio_repository() -> PortfolioRepository:
|
|
"""Get or create global portfolio repository instance."""
|
|
global _portfolio_repo
|
|
if _portfolio_repo is None:
|
|
_portfolio_repo = PortfolioRepository()
|
|
return _portfolio_repo
|