98e3208b5e
Critical fixes: - Add math.isfinite() check to reject NaN/Infinity in _safe_quote_price - Raise TypeError instead of silent 0.0 fallback in price_feed.py - Use dict instead of Mapping for external data validation Type improvements: - Add PortfolioSnapshot TypedDict for type safety - Add DisplayMode and EntryBasisMode Literal types - Add explicit dict[str, Any] annotation in to_dict() - Remove cast() in favor of type comment validation
678 lines
27 KiB
Python
678 lines
27 KiB
Python
"""Portfolio configuration and domain portfolio models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
from app.models.position import Position, create_position
|
|
|
|
# Type aliases for display mode and entry basis
|
|
DisplayMode = Literal["GLD", "XAU"]
|
|
EntryBasisMode = Literal["value_price", "weight"]
|
|
|
|
_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
|
|
|
|
|
|
def build_default_portfolio_config(*, entry_price: float | None = None) -> "PortfolioConfig":
|
|
resolved_entry_price = float(entry_price) if entry_price is not None else _DEFAULT_ENTRY_PRICE
|
|
gold_value = resolved_entry_price * _DEFAULT_GOLD_OUNCES
|
|
return PortfolioConfig(
|
|
gold_value=gold_value,
|
|
entry_price=resolved_entry_price,
|
|
gold_ounces=_DEFAULT_GOLD_OUNCES,
|
|
entry_basis_mode="value_price",
|
|
)
|
|
|
|
|
|
@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)
|
|
positions: List of position entries (multi-position support)
|
|
"""
|
|
|
|
gold_value: float | None = None
|
|
entry_price: float | None = _DEFAULT_ENTRY_PRICE
|
|
gold_ounces: float | None = None
|
|
entry_basis_mode: EntryBasisMode = "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
|
|
|
|
# Underlying instrument selection
|
|
underlying: str = "GLD"
|
|
|
|
# Display mode: how to show positions (GLD shares vs physical gold)
|
|
display_mode: DisplayMode = "XAU" # "GLD" for share view, "XAU" for physical gold view
|
|
|
|
# Alert settings
|
|
volatility_spike: float = 0.25
|
|
spot_drawdown: float = 7.5
|
|
email_alerts: bool = False
|
|
|
|
# Multi-position support
|
|
positions: list[Position] = field(default_factory=list)
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Normalize entry basis fields and validate configuration."""
|
|
self._normalize_entry_basis()
|
|
self.validate()
|
|
|
|
def migrate_to_positions_if_needed(self) -> None:
|
|
"""Migrate legacy single-entry portfolios to multi-position format.
|
|
|
|
Call this after loading from persistence to migrate legacy configs.
|
|
If positions list is empty but gold_ounces exists, create one Position
|
|
representing the legacy single entry.
|
|
"""
|
|
if self.positions:
|
|
# Already has positions, no migration needed
|
|
return
|
|
|
|
if self.gold_ounces is None or self.entry_price is None:
|
|
return
|
|
|
|
# Create a single position from legacy fields
|
|
position = create_position(
|
|
underlying=self.underlying,
|
|
quantity=Decimal(str(self.gold_ounces)),
|
|
unit="oz",
|
|
entry_price=Decimal(str(self.entry_price)),
|
|
entry_date=date.today(),
|
|
entry_basis_mode=self.entry_basis_mode,
|
|
)
|
|
# PortfolioConfig is not frozen, so we can set directly
|
|
self.positions = [position]
|
|
|
|
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:
|
|
default = build_default_portfolio_config(entry_price=self.entry_price)
|
|
self.gold_value = default.gold_value
|
|
self.gold_ounces = default.gold_ounces
|
|
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 _migrate_legacy_to_positions(self) -> None:
|
|
"""Migrate legacy single-entry portfolios to multi-position format.
|
|
|
|
If positions list is empty but gold_ounces exists, create one Position
|
|
representing the legacy single entry.
|
|
"""
|
|
if self.positions:
|
|
# Already has positions, no migration needed
|
|
return
|
|
|
|
if self.gold_ounces is None or self.entry_price is None:
|
|
return
|
|
|
|
# Create a single position from legacy fields
|
|
position = create_position(
|
|
underlying=self.underlying,
|
|
quantity=Decimal(str(self.gold_ounces)),
|
|
unit="oz",
|
|
entry_price=Decimal(str(self.entry_price)),
|
|
entry_date=date.today(),
|
|
entry_basis_mode=self.entry_basis_mode,
|
|
)
|
|
# PortfolioConfig is not frozen, so we can set directly
|
|
self.positions = [position]
|
|
|
|
def _sync_legacy_fields_from_positions(self) -> None:
|
|
"""Sync legacy gold_ounces, entry_price, gold_value from positions.
|
|
|
|
For backward compatibility, compute aggregate values from positions list.
|
|
"""
|
|
if not self.positions:
|
|
return
|
|
|
|
# For now, assume homogeneous positions (same underlying and unit)
|
|
# Sum quantities and compute weighted average entry price
|
|
total_quantity = Decimal("0")
|
|
total_value = Decimal("0")
|
|
|
|
for pos in self.positions:
|
|
if pos.unit == "oz":
|
|
total_quantity += pos.quantity
|
|
total_value += pos.entry_value
|
|
|
|
if total_quantity > 0:
|
|
avg_entry_price = total_value / total_quantity
|
|
self.gold_ounces = float(total_quantity)
|
|
self.entry_price = float(avg_entry_price)
|
|
self.gold_value = float(total_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
|
|
# Sync legacy fields from positions before serializing
|
|
self._sync_legacy_fields_from_positions()
|
|
result: dict[str, Any] = {
|
|
"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,
|
|
"underlying": self.underlying,
|
|
"display_mode": self.display_mode,
|
|
"volatility_spike": self.volatility_spike,
|
|
"spot_drawdown": self.spot_drawdown,
|
|
"email_alerts": self.email_alerts,
|
|
}
|
|
# Include positions if any exist
|
|
if self.positions:
|
|
result["positions"] = [pos.to_dict() for pos in self.positions]
|
|
return result
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> PortfolioConfig:
|
|
"""Create configuration from dictionary."""
|
|
# Extract positions if present (may already be Position objects from deserialization)
|
|
positions_data = data.pop("positions", None)
|
|
config_data = {k: v for k, v in data.items() if k in cls.__dataclass_fields__}
|
|
|
|
# Create config without positions first (will be set in __post_init__)
|
|
config = cls(**config_data)
|
|
|
|
# Set positions after initialization
|
|
if positions_data:
|
|
if positions_data and isinstance(positions_data[0], Position):
|
|
# Already deserialized by _deserialize_value
|
|
positions = positions_data
|
|
else:
|
|
positions = [Position.from_dict(p) for p in positions_data]
|
|
config.positions = positions
|
|
|
|
return config
|
|
|
|
|
|
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",
|
|
"underlying", # optional with default "GLD"
|
|
"display_mode", # optional with default "XAU"
|
|
"volatility_spike",
|
|
"spot_drawdown",
|
|
"email_alerts",
|
|
"positions", # multi-position support
|
|
}
|
|
|
|
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]:
|
|
# Serialize positions separately before calling to_dict
|
|
positions_data = [pos.to_dict() for pos in config.positions] if config.positions else []
|
|
config_dict = config.to_dict()
|
|
# Remove positions from config_dict since we handle it separately
|
|
config_dict.pop("positions", None)
|
|
return {
|
|
"schema_version": cls.SCHEMA_VERSION,
|
|
"portfolio": {
|
|
**{key: cls._serialize_value(key, value) for key, value in config_dict.items()},
|
|
**({"positions": positions_data} if positions_data else {}),
|
|
},
|
|
}
|
|
|
|
@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")
|
|
if key == "positions" and isinstance(value, list):
|
|
# Already serialized as dicts from _to_persistence_payload
|
|
return value
|
|
return value
|
|
|
|
@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)
|
|
config = PortfolioConfig.from_dict(upgraded)
|
|
# Migrate legacy configs without positions to single position
|
|
config.migrate_to_positions_if_needed()
|
|
return config
|
|
|
|
# Fields that must be present in persisted payloads
|
|
# (underlying is optional with default "GLD")
|
|
# (positions is optional - legacy configs won't have it)
|
|
_REQUIRED_FIELDS = (_PERSISTED_FIELDS - {"underlying", "display_mode"}) - {"positions"}
|
|
|
|
@classmethod
|
|
def _validate_portfolio_fields(cls, payload: dict[str, Any]) -> None:
|
|
keys = set(payload.keys())
|
|
missing = sorted(cls._REQUIRED_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 _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")
|
|
if key == "positions" and isinstance(value, list):
|
|
return [Position.from_dict(p) for p in value]
|
|
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
|