bu5hm4nn/vault-dash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(CORE-001D3A): normalize alerts and settings service boundaries

Browse Source
This commit is contained in:
Bu5hm4nn
2026-03-26 13:10:30 +01:00
parent 91f67cd414
commit bb557009c7
7 changed files with 231 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
app/services/alerts.py
View File

@@ -2,13 +2,79 @@
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from typing import Mapping
from app.domain.portfolio_math import build_alert_context
from app.domain.units import decimal_from_float
from app.models.alerts import AlertEvent, AlertHistoryRepository, AlertStatus
from app.models.portfolio import PortfolioConfig
@dataclass(frozen=True, slots=True)
class AlertEvaluationInput:
ltv_ratio: Decimal
spot_price: Decimal
updated_at: str
warning_threshold: Decimal
critical_threshold: Decimal
email_alerts_enabled: bool
def _decimal_from_boundary_value(value: object, *, field_name: str, default: float = 0.0) -> Decimal:
if value is None:
return decimal_from_float(float(default))
if isinstance(value, bool):
raise TypeError(f"{field_name} must be numeric, got bool")
if isinstance(value, int):
parsed = float(value)
elif isinstance(value, float):
parsed = value
elif isinstance(value, str):
stripped = value.strip()
if not stripped:
parsed = float(default)
else:
try:
parsed = float(stripped)
except ValueError as exc:
raise ValueError(f"{field_name} must be numeric, got {value!r}") from exc
else:
raise TypeError(f"{field_name} must be numeric, got {type(value)!r}")
return decimal_from_float(float(parsed))
def _normalize_alert_evaluation_input(
config: PortfolioConfig,
portfolio: Mapping[str, object],
) -> AlertEvaluationInput:
return AlertEvaluationInput(
ltv_ratio=_decimal_from_boundary_value(
portfolio.get("ltv_ratio", 0.0),
field_name="portfolio.ltv_ratio",
),
spot_price=_decimal_from_boundary_value(
portfolio.get("spot_price", 0.0),
field_name="portfolio.spot_price",
),
updated_at=str(portfolio.get("quote_updated_at", "")),
warning_threshold=_decimal_from_boundary_value(
config.ltv_warning,
field_name="config.ltv_warning",
),
critical_threshold=_decimal_from_boundary_value(
config.margin_threshold,
field_name="config.margin_threshold",
),
email_alerts_enabled=bool(config.email_alerts),
)
def _ratio_text(value: Decimal) -> str:
return f"{float(value):.1%}"
def build_portfolio_alert_context(
config: PortfolioConfig,
*,
@@ -32,20 +98,19 @@ class AlertService:
self, config: PortfolioConfig, portfolio: Mapping[str, object], *, persist: bool = True
) -> AlertStatus:
history = self.repository.load() if persist else []
ltv_ratio = float(portfolio.get("ltv_ratio", 0.0))
spot_price = float(portfolio.get("spot_price", 0.0))
updated_at = str(portfolio.get("quote_updated_at", ""))
evaluation = _normalize_alert_evaluation_input(config, portfolio)
if ltv_ratio >= float(config.margin_threshold):
if evaluation.ltv_ratio >= evaluation.critical_threshold:
severity = "critical"
message = (
f"Current LTV {ltv_ratio:.1%} is above the critical threshold of "
f"{float(config.margin_threshold):.1%}."
f"Current LTV {_ratio_text(evaluation.ltv_ratio)} is above the critical threshold of "
f"{_ratio_text(evaluation.critical_threshold)}."
)
elif ltv_ratio >= float(config.ltv_warning):
elif evaluation.ltv_ratio >= evaluation.warning_threshold:
severity = "warning"
message = (
f"Current LTV {ltv_ratio:.1%} is above the warning threshold of " f"{float(config.ltv_warning):.1%}."
f"Current LTV {_ratio_text(evaluation.ltv_ratio)} is above the warning threshold of "
f"{_ratio_text(evaluation.warning_threshold)}."
)
else:
severity = "ok"
@@ -56,12 +121,12 @@ class AlertService:
event = AlertEvent(
severity=severity,
message=message,
ltv_ratio=ltv_ratio,
warning_threshold=float(config.ltv_warning),
critical_threshold=float(config.margin_threshold),
spot_price=spot_price,
updated_at=updated_at,
email_alerts_enabled=bool(config.email_alerts),
ltv_ratio=float(evaluation.ltv_ratio),
warning_threshold=float(evaluation.warning_threshold),
critical_threshold=float(evaluation.critical_threshold),
spot_price=float(evaluation.spot_price),
updated_at=evaluation.updated_at,
email_alerts_enabled=evaluation.email_alerts_enabled,
)
if persist:
if self._should_record(history, event):
@@ -73,10 +138,10 @@ class AlertService:
return AlertStatus(
severity=severity,
message=message,
ltv_ratio=ltv_ratio,
warning_threshold=float(config.ltv_warning),
critical_threshold=float(config.margin_threshold),
email_alerts_enabled=bool(config.email_alerts),
ltv_ratio=float(evaluation.ltv_ratio),
warning_threshold=float(evaluation.warning_threshold),
critical_threshold=float(evaluation.critical_threshold),
email_alerts_enabled=evaluation.email_alerts_enabled,
history=(
preview_history
if not persist

61
app/services/settings_status.py
View File

@@ -1,7 +1,10 @@
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from typing import Protocol
from app.domain.units import decimal_from_float
from app.models.portfolio import PortfolioConfig
@@ -14,14 +17,62 @@ class _SaveStatusConfig(Protocol):
margin_call_price: object
def margin_call_price_value(config: PortfolioConfig | _SaveStatusConfig) -> float:
@dataclass(frozen=True, slots=True)
class SaveStatusSnapshot:
entry_basis_mode: str
gold_value: Decimal
entry_price: Decimal
gold_ounces: Decimal
current_ltv: Decimal
margin_call_price: Decimal
def _decimal_from_boundary_value(value: object, *, field_name: str) -> Decimal:
if value is None:
raise ValueError(f"{field_name} must be present")
if isinstance(value, bool):
raise TypeError(f"{field_name} must be numeric, got bool")
if isinstance(value, int):
parsed = float(value)
elif isinstance(value, float):
parsed = value
elif isinstance(value, str):
stripped = value.strip()
if not stripped:
raise ValueError(f"{field_name} must be present")
try:
parsed = float(stripped)
except ValueError as exc:
raise ValueError(f"{field_name} must be numeric, got {value!r}") from exc
else:
raise TypeError(f"{field_name} must be numeric, got {type(value)!r}")
return decimal_from_float(float(parsed))
def _normalize_save_status_snapshot(config: PortfolioConfig | _SaveStatusConfig) -> SaveStatusSnapshot:
margin_call_price = config.margin_call_price
return float(margin_call_price() if callable(margin_call_price) else margin_call_price)
resolved_margin_call_price = margin_call_price() if callable(margin_call_price) else margin_call_price
return SaveStatusSnapshot(
entry_basis_mode=config.entry_basis_mode,
gold_value=_decimal_from_boundary_value(config.gold_value, field_name="config.gold_value"),
entry_price=_decimal_from_boundary_value(config.entry_price, field_name="config.entry_price"),
gold_ounces=_decimal_from_boundary_value(config.gold_ounces, field_name="config.gold_ounces"),
current_ltv=_decimal_from_boundary_value(config.current_ltv, field_name="config.current_ltv"),
margin_call_price=_decimal_from_boundary_value(
resolved_margin_call_price,
field_name="config.margin_call_price",
),
)
def margin_call_price_value(config: PortfolioConfig | _SaveStatusConfig) -> float:
return float(_normalize_save_status_snapshot(config).margin_call_price)
def save_status_text(config: PortfolioConfig | _SaveStatusConfig) -> str:
snapshot = _normalize_save_status_snapshot(config)
return (
f"Saved: basis={config.entry_basis_mode}, start=${config.gold_value:,.0f}, "
f"entry=${config.entry_price:,.2f}/oz, weight={config.gold_ounces:,.2f} oz, "
f"LTV={config.current_ltv:.1%}, trigger=${margin_call_price_value(config):,.2f}/oz"
f"Saved: basis={snapshot.entry_basis_mode}, start=${float(snapshot.gold_value):,.0f}, "
f"entry=${float(snapshot.entry_price):,.2f}/oz, weight={float(snapshot.gold_ounces):,.2f} oz, "
f"LTV={float(snapshot.current_ltv):.1%}, trigger=${float(snapshot.margin_call_price):,.2f}/oz"
)