feat(PORT-002): add alert status and history

This commit is contained in:
Bu5hm4nn
2026-03-24 11:04:32 +01:00
parent 7c6b8ef2c6
commit d0b1304b71
9 changed files with 525 additions and 59 deletions

64
app/models/alerts.py Normal file
View File

@@ -0,0 +1,64 @@
"""Alert notification domain models."""
from __future__ import annotations
import json
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
@dataclass
class AlertEvent:
severity: str
message: str
ltv_ratio: float
warning_threshold: float
critical_threshold: float
spot_price: float
updated_at: str
email_alerts_enabled: bool
def to_dict(self) -> dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "AlertEvent":
return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
@dataclass
class AlertStatus:
severity: str
message: str
ltv_ratio: float
warning_threshold: float
critical_threshold: float
email_alerts_enabled: bool
history: list[AlertEvent]
class AlertHistoryRepository:
"""File-backed alert history store."""
HISTORY_PATH = Path("data/alert_history.json")
def __init__(self, history_path: Path | None = None) -> None:
self.history_path = history_path or self.HISTORY_PATH
self.history_path.parent.mkdir(parents=True, exist_ok=True)
def load(self) -> list[AlertEvent]:
if not self.history_path.exists():
return []
try:
with self.history_path.open() as f:
data = json.load(f)
except (json.JSONDecodeError, OSError):
return []
if not isinstance(data, list):
return []
return [AlertEvent.from_dict(item) for item in data if isinstance(item, dict)]
def save(self, events: list[AlertEvent]) -> None:
with self.history_path.open("w") as f:
json.dump([event.to_dict() for event in events], f, indent=2)

View File

@@ -154,6 +154,8 @@ class PortfolioConfig:
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")