from dataclasses import asdict, dataclass from enum import Enum from typing import Any from app.bars.models import Timeframe class LevelKind(str, Enum): MANUAL = "manual" MA = "ma" VWAP = "vwap" TRENDLINE = "trendline" HORIZONTAL = "horizontal" class Side(str, Enum): SUPPORT = "support" RESISTANCE = "resistance" @dataclass(slots=True) class Level: id: str kind: LevelKind tf: Timeframe side: Side weight: float score: float label: str anchor_t: int anchor_p: float slope: float points: list[tuple[int, float]] | None touches: int first_t: int last_t: int provisional: bool hidden: bool period: int | None = None color: str | None = None line_width: int | None = None number: int | None = None cutoff_t: int | None = None def price_at(self, t: int) -> float: return self.anchor_p + self.slope * (t - self.anchor_t) def to_dict(self) -> dict[str, Any]: value = asdict(self) value["kind"] = self.kind.value value["tf"] = self.tf.value value["side"] = self.side.value return value def summary(self) -> dict[str, Any]: """Compact form for embedding inside a cluster. Clusters go out on every closed 1m bar, and a moving average carries its whole point history — hundreds of entries reaching back years. Embedding the full level duplicated the entire levels payload once a minute. The client already holds the full levels and joins on id. """ return { "id": self.id, "kind": self.kind.value, "tf": self.tf.value, "side": self.side.value, "weight": self.weight, "label": self.label, }