106 lines
4.5 KiB
Python
106 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, datetime, timezone
|
|
from typing import Literal
|
|
|
|
EventType = Literal["selloff", "recovery", "stress_test"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventScenarioOverrides:
|
|
lookback_days: int | None = None
|
|
recovery_days: int | None = None
|
|
default_template_slugs: tuple[str, ...] = field(default_factory=tuple)
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.lookback_days is not None and self.lookback_days < 0:
|
|
raise ValueError("lookback_days must be non-negative")
|
|
if self.recovery_days is not None and self.recovery_days < 0:
|
|
raise ValueError("recovery_days must be non-negative")
|
|
if any(not slug for slug in self.default_template_slugs):
|
|
raise ValueError("default_template_slugs must not contain empty values")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"lookback_days": self.lookback_days,
|
|
"recovery_days": self.recovery_days,
|
|
"default_template_slugs": list(self.default_template_slugs),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, payload: dict[str, object] | None) -> EventScenarioOverrides:
|
|
if payload is None:
|
|
return cls()
|
|
return cls(
|
|
lookback_days=payload.get("lookback_days"), # type: ignore[arg-type]
|
|
recovery_days=payload.get("recovery_days"), # type: ignore[arg-type]
|
|
default_template_slugs=tuple(payload.get("default_template_slugs", [])), # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventPreset:
|
|
event_preset_id: str
|
|
slug: str
|
|
display_name: str
|
|
symbol: str
|
|
window_start: date
|
|
window_end: date
|
|
anchor_date: date | None
|
|
event_type: EventType
|
|
tags: tuple[str, ...] = field(default_factory=tuple)
|
|
description: str = ""
|
|
scenario_overrides: EventScenarioOverrides = field(default_factory=EventScenarioOverrides)
|
|
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.event_preset_id:
|
|
raise ValueError("event_preset_id is required")
|
|
if not self.slug:
|
|
raise ValueError("slug is required")
|
|
if not self.display_name:
|
|
raise ValueError("display_name is required")
|
|
if not self.symbol:
|
|
raise ValueError("symbol is required")
|
|
if self.window_start > self.window_end:
|
|
raise ValueError("window_start must be on or before window_end")
|
|
if self.anchor_date is not None and not (self.window_start <= self.anchor_date <= self.window_end):
|
|
raise ValueError("anchor_date must fall inside the event window")
|
|
if self.event_type not in {"selloff", "recovery", "stress_test"}:
|
|
raise ValueError("unsupported event_type")
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"event_preset_id": self.event_preset_id,
|
|
"slug": self.slug,
|
|
"display_name": self.display_name,
|
|
"symbol": self.symbol,
|
|
"window_start": self.window_start.isoformat(),
|
|
"window_end": self.window_end.isoformat(),
|
|
"anchor_date": self.anchor_date.isoformat() if self.anchor_date is not None else None,
|
|
"event_type": self.event_type,
|
|
"tags": list(self.tags),
|
|
"description": self.description,
|
|
"scenario_overrides": self.scenario_overrides.to_dict(),
|
|
"created_at": self.created_at.isoformat(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, payload: dict[str, object]) -> EventPreset:
|
|
anchor_date = payload.get("anchor_date")
|
|
return cls(
|
|
event_preset_id=str(payload["event_preset_id"]),
|
|
slug=str(payload["slug"]),
|
|
display_name=str(payload["display_name"]),
|
|
symbol=str(payload["symbol"]),
|
|
window_start=date.fromisoformat(str(payload["window_start"])),
|
|
window_end=date.fromisoformat(str(payload["window_end"])),
|
|
anchor_date=date.fromisoformat(str(anchor_date)) if anchor_date else None,
|
|
event_type=payload["event_type"], # type: ignore[arg-type]
|
|
tags=tuple(payload.get("tags", [])), # type: ignore[arg-type]
|
|
description=str(payload.get("description", "")),
|
|
scenario_overrides=EventScenarioOverrides.from_dict(payload.get("scenario_overrides")), # type: ignore[arg-type]
|
|
created_at=datetime.fromisoformat(str(payload["created_at"])),
|
|
)
|