526 lines
21 KiB
Python
526 lines
21 KiB
Python
"""Portfolio configuration and domain portfolio models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_DEFAULT_GOLD_VALUE = 215_000.0
|
|
_DEFAULT_ENTRY_PRICE = 2_150.0
|
|
_LEGACY_DEFAULT_ENTRY_PRICE = 215.0
|
|
_DEFAULT_GOLD_OUNCES = 100.0
|
|
_LEGACY_DEFAULT_GOLD_OUNCES = 1_000.0
|
|
|
|
|
|
@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 = _DEFAULT_ENTRY_PRICE
|
|
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 = _DEFAULT_GOLD_VALUE
|
|
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__})
|
|
|
|
|
|
def _coerce_persisted_decimal(value: Any) -> Decimal:
|
|
if isinstance(value, bool):
|
|
raise TypeError("Boolean values are not valid decimal persistence inputs")
|
|
if isinstance(value, Decimal):
|
|
amount = value
|
|
elif isinstance(value, int):
|
|
amount = Decimal(value)
|
|
elif isinstance(value, float):
|
|
amount = Decimal(str(value))
|
|
elif isinstance(value, str):
|
|
amount = Decimal(value)
|
|
else:
|
|
raise TypeError(f"Unsupported persisted decimal input type: {type(value)!r}")
|
|
if not amount.is_finite():
|
|
raise ValueError("Decimal persistence value must be finite")
|
|
return amount
|
|
|
|
|
|
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")
|
|
SCHEMA_VERSION = 2
|
|
PERSISTENCE_CURRENCY = "USD"
|
|
PERSISTENCE_WEIGHT_UNIT = "ozt"
|
|
_WEIGHT_FACTORS = {
|
|
"g": Decimal("1"),
|
|
"kg": Decimal("1000"),
|
|
"ozt": Decimal("31.1034768"),
|
|
}
|
|
_MONEY_FIELDS = {"gold_value", "loan_amount", "monthly_budget"}
|
|
_WEIGHT_FIELDS = {"gold_ounces"}
|
|
_PRICE_PER_WEIGHT_FIELDS = {"entry_price"}
|
|
_RATIO_FIELDS = {"margin_threshold", "ltv_warning", "volatility_spike"}
|
|
_PERCENT_FIELDS = {"spot_drawdown"}
|
|
_INTEGER_FIELDS = {"refresh_interval"}
|
|
_PERSISTED_FIELDS = {
|
|
"gold_value",
|
|
"entry_price",
|
|
"gold_ounces",
|
|
"entry_basis_mode",
|
|
"loan_amount",
|
|
"margin_threshold",
|
|
"monthly_budget",
|
|
"ltv_warning",
|
|
"primary_source",
|
|
"fallback_source",
|
|
"refresh_interval",
|
|
"volatility_spike",
|
|
"spot_drawdown",
|
|
"email_alerts",
|
|
}
|
|
|
|
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."""
|
|
payload = self._to_persistence_payload(config)
|
|
tmp_path = self.config_path.with_name(f"{self.config_path.name}.tmp")
|
|
with open(tmp_path, "w") as f:
|
|
json.dump(payload, f, indent=2)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp_path, self.config_path)
|
|
|
|
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)
|
|
except json.JSONDecodeError as e:
|
|
raise ValueError(f"Invalid portfolio config JSON: {e}") from e
|
|
|
|
return self._config_from_payload(data)
|
|
|
|
@classmethod
|
|
def _to_persistence_payload(cls, config: PortfolioConfig) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": cls.SCHEMA_VERSION,
|
|
"portfolio": {key: cls._serialize_value(key, value) for key, value in config.to_dict().items()},
|
|
}
|
|
|
|
@classmethod
|
|
def _config_from_payload(cls, data: dict[str, Any]) -> PortfolioConfig:
|
|
if not isinstance(data, dict):
|
|
raise TypeError("portfolio config payload must be an object")
|
|
schema_version = data.get("schema_version")
|
|
if schema_version != cls.SCHEMA_VERSION:
|
|
raise ValueError(f"Unsupported portfolio schema_version: {schema_version}")
|
|
portfolio = data.get("portfolio")
|
|
if not isinstance(portfolio, dict):
|
|
raise TypeError("portfolio payload must be an object")
|
|
cls._validate_portfolio_fields(portfolio)
|
|
deserialized = cls._deserialize_portfolio_payload(portfolio)
|
|
upgraded = cls._upgrade_legacy_default_workspace(deserialized)
|
|
return PortfolioConfig.from_dict(upgraded)
|
|
|
|
@classmethod
|
|
def _validate_portfolio_fields(cls, payload: dict[str, Any]) -> None:
|
|
keys = set(payload.keys())
|
|
missing = sorted(cls._PERSISTED_FIELDS - keys)
|
|
unknown = sorted(keys - cls._PERSISTED_FIELDS)
|
|
if missing or unknown:
|
|
details: list[str] = []
|
|
if missing:
|
|
details.append(f"missing={missing}")
|
|
if unknown:
|
|
details.append(f"unknown={unknown}")
|
|
raise ValueError(f"Invalid portfolio payload fields: {'; '.join(details)}")
|
|
|
|
@classmethod
|
|
def _deserialize_portfolio_payload(cls, payload: dict[str, Any]) -> dict[str, Any]:
|
|
return {key: cls._deserialize_value(key, value) for key, value in payload.items()}
|
|
|
|
@classmethod
|
|
def _upgrade_legacy_default_workspace(cls, payload: dict[str, Any]) -> dict[str, Any]:
|
|
if not cls._looks_like_legacy_default_workspace(payload):
|
|
return payload
|
|
upgraded = dict(payload)
|
|
upgraded["entry_price"] = _DEFAULT_ENTRY_PRICE
|
|
upgraded["gold_ounces"] = _DEFAULT_GOLD_OUNCES
|
|
upgraded["gold_value"] = _DEFAULT_GOLD_VALUE
|
|
return upgraded
|
|
|
|
@staticmethod
|
|
def _looks_like_legacy_default_workspace(payload: dict[str, Any]) -> bool:
|
|
def _close(key: str, expected: float) -> bool:
|
|
value = payload.get(key)
|
|
return isinstance(value, (int, float)) and abs(float(value) - expected) <= 1e-9
|
|
|
|
return (
|
|
_close("gold_value", _DEFAULT_GOLD_VALUE)
|
|
and _close("entry_price", _LEGACY_DEFAULT_ENTRY_PRICE)
|
|
and _close("gold_ounces", _LEGACY_DEFAULT_GOLD_OUNCES)
|
|
and payload.get("entry_basis_mode") == "value_price"
|
|
and _close("loan_amount", 145_000.0)
|
|
and _close("margin_threshold", 0.75)
|
|
and _close("monthly_budget", 8_000.0)
|
|
and _close("ltv_warning", 0.70)
|
|
and payload.get("primary_source") == "yfinance"
|
|
and payload.get("fallback_source") == "yfinance"
|
|
and payload.get("refresh_interval") == 5
|
|
and _close("volatility_spike", 0.25)
|
|
and _close("spot_drawdown", 7.5)
|
|
and payload.get("email_alerts") is False
|
|
)
|
|
|
|
@classmethod
|
|
def _serialize_value(cls, key: str, value: Any) -> Any:
|
|
if key in cls._MONEY_FIELDS:
|
|
return {"value": cls._decimal_to_string(value), "currency": cls.PERSISTENCE_CURRENCY}
|
|
if key in cls._WEIGHT_FIELDS:
|
|
return {"value": cls._decimal_to_string(value), "unit": cls.PERSISTENCE_WEIGHT_UNIT}
|
|
if key in cls._PRICE_PER_WEIGHT_FIELDS:
|
|
return {
|
|
"value": cls._decimal_to_string(value),
|
|
"currency": cls.PERSISTENCE_CURRENCY,
|
|
"per_weight_unit": cls.PERSISTENCE_WEIGHT_UNIT,
|
|
}
|
|
if key in cls._RATIO_FIELDS:
|
|
return {"value": cls._decimal_to_string(value), "unit": "ratio"}
|
|
if key in cls._PERCENT_FIELDS:
|
|
return {"value": cls._decimal_to_string(value), "unit": "percent"}
|
|
if key in cls._INTEGER_FIELDS:
|
|
return cls._serialize_integer(value, unit="seconds")
|
|
return value
|
|
|
|
@classmethod
|
|
def _deserialize_value(cls, key: str, value: Any) -> Any:
|
|
if key in cls._MONEY_FIELDS:
|
|
return float(cls._deserialize_money(value))
|
|
if key in cls._WEIGHT_FIELDS:
|
|
return float(cls._deserialize_weight(value))
|
|
if key in cls._PRICE_PER_WEIGHT_FIELDS:
|
|
return float(cls._deserialize_price_per_weight(value))
|
|
if key in cls._RATIO_FIELDS:
|
|
return float(cls._deserialize_ratio(value))
|
|
if key in cls._PERCENT_FIELDS:
|
|
return float(cls._deserialize_percent(value))
|
|
if key in cls._INTEGER_FIELDS:
|
|
return cls._deserialize_integer(value, expected_unit="seconds")
|
|
return value
|
|
|
|
@classmethod
|
|
def _deserialize_money(cls, value: Any) -> Decimal:
|
|
if not isinstance(value, dict):
|
|
raise TypeError("money field must be an object")
|
|
currency = value.get("currency")
|
|
if currency != cls.PERSISTENCE_CURRENCY:
|
|
raise ValueError(f"Unsupported currency: {currency!r}")
|
|
return _coerce_persisted_decimal(value.get("value"))
|
|
|
|
@classmethod
|
|
def _deserialize_weight(cls, value: Any) -> Decimal:
|
|
if not isinstance(value, dict):
|
|
raise TypeError("weight field must be an object")
|
|
amount = _coerce_persisted_decimal(value.get("value"))
|
|
unit = value.get("unit")
|
|
return cls._convert_weight(amount, unit, cls.PERSISTENCE_WEIGHT_UNIT)
|
|
|
|
@classmethod
|
|
def _deserialize_price_per_weight(cls, value: Any) -> Decimal:
|
|
if not isinstance(value, dict):
|
|
raise TypeError("price-per-weight field must be an object")
|
|
currency = value.get("currency")
|
|
if currency != cls.PERSISTENCE_CURRENCY:
|
|
raise ValueError(f"Unsupported currency: {currency!r}")
|
|
amount = _coerce_persisted_decimal(value.get("value"))
|
|
unit = value.get("per_weight_unit")
|
|
return cls._convert_price_per_weight(amount, unit, cls.PERSISTENCE_WEIGHT_UNIT)
|
|
|
|
@classmethod
|
|
def _deserialize_ratio(cls, value: Any) -> Decimal:
|
|
if not isinstance(value, dict):
|
|
raise TypeError("ratio field must be an object")
|
|
amount = _coerce_persisted_decimal(value.get("value"))
|
|
unit = value.get("unit")
|
|
if unit == "ratio":
|
|
return amount
|
|
if unit == "percent":
|
|
return amount / Decimal("100")
|
|
raise ValueError(f"Unsupported ratio unit: {unit!r}")
|
|
|
|
@classmethod
|
|
def _deserialize_percent(cls, value: Any) -> Decimal:
|
|
if not isinstance(value, dict):
|
|
raise TypeError("percent field must be an object")
|
|
amount = _coerce_persisted_decimal(value.get("value"))
|
|
unit = value.get("unit")
|
|
if unit == "percent":
|
|
return amount
|
|
if unit == "ratio":
|
|
return amount * Decimal("100")
|
|
raise ValueError(f"Unsupported percent unit: {unit!r}")
|
|
|
|
@staticmethod
|
|
def _serialize_integer(value: Any, *, unit: str) -> dict[str, Any]:
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise TypeError("integer field value must be an int")
|
|
return {"value": value, "unit": unit}
|
|
|
|
@staticmethod
|
|
def _deserialize_integer(value: Any, *, expected_unit: str) -> int:
|
|
if not isinstance(value, dict):
|
|
raise TypeError("integer field must be an object")
|
|
unit = value.get("unit")
|
|
if unit != expected_unit:
|
|
raise ValueError(f"Unsupported integer unit: {unit!r}")
|
|
raw = value.get("value")
|
|
if isinstance(raw, bool) or not isinstance(raw, int):
|
|
raise TypeError("integer field value must be an int")
|
|
return raw
|
|
|
|
@classmethod
|
|
def _convert_weight(cls, amount: Decimal, from_unit: Any, to_unit: str) -> Decimal:
|
|
if from_unit not in cls._WEIGHT_FACTORS or to_unit not in cls._WEIGHT_FACTORS:
|
|
raise ValueError(f"Unsupported weight unit conversion: {from_unit!r} -> {to_unit!r}")
|
|
if from_unit == to_unit:
|
|
return amount
|
|
grams = amount * cls._WEIGHT_FACTORS[from_unit]
|
|
return grams / cls._WEIGHT_FACTORS[to_unit]
|
|
|
|
@classmethod
|
|
def _convert_price_per_weight(cls, amount: Decimal, from_unit: Any, to_unit: str) -> Decimal:
|
|
if from_unit not in cls._WEIGHT_FACTORS or to_unit not in cls._WEIGHT_FACTORS:
|
|
raise ValueError(f"Unsupported price-per-weight unit conversion: {from_unit!r} -> {to_unit!r}")
|
|
if from_unit == to_unit:
|
|
return amount
|
|
return amount * cls._WEIGHT_FACTORS[to_unit] / cls._WEIGHT_FACTORS[from_unit]
|
|
|
|
@staticmethod
|
|
def _decimal_to_string(value: Any) -> str:
|
|
decimal_value = _coerce_persisted_decimal(value)
|
|
normalized = format(decimal_value, "f")
|
|
if "." not in normalized:
|
|
normalized = f"{normalized}.0"
|
|
return normalized
|
|
|
|
|
|
_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
|