86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
from dataclasses import asdict, dataclass
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from app.bars.models import Timeframe
|
|
from app.instrument import DEFAULT_SYMBOL
|
|
|
|
|
|
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
|
|
# Hand-placed levels fire once and disarm themselves; everything derived
|
|
# (averages, prior-day, VWAP) is permanently armed.
|
|
armed: bool = True
|
|
# Sloped lines are evaluated across bars, not seconds (see bar_space). The
|
|
# runtime fills this in where the bar series is available; price_at() is the
|
|
# fallback for levels that are already flat or have no series to measure.
|
|
current_p: float | None = None
|
|
# Optional fixed distance for a hand-placed level. None keeps the global
|
|
# ATR-based trigger; a value alerts that many points before the level.
|
|
alert_early_points: float | None = None
|
|
# False means the attributed timeframe does not hold enough history to
|
|
# price this line safely. Such a line remains visible but cannot cluster or
|
|
# alert using the absolute-time fallback.
|
|
geometry_resolved: bool = True
|
|
symbol: str = DEFAULT_SYMBOL
|
|
|
|
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,
|
|
}
|