from dataclasses import asdict, dataclass from hashlib import sha1 from typing import Any from app.analysis.levels import Level, Side @dataclass(slots=True) class Cluster: id: str side: Side low: float high: float center: float score: float members: list[Level] distance: float def to_dict(self) -> dict[str, Any]: value = asdict(self) value["side"] = self.side.value value["members"] = [member.to_dict() for member in self.members] return value def cluster_levels( levels: list[Level], current_t: int, current_price: float, atr15: float ) -> list[Cluster]: tolerance = 0.4 * atr15 if tolerance <= 0: return [] groups: list[list[tuple[float, Level]]] = [] positioned = [(level.price_at(current_t), level) for level in levels if not level.hidden] for positional_side in (Side.SUPPORT, Side.RESISTANCE): side_levels = sorted( ( item for item in positioned if (Side.RESISTANCE if item[0] >= current_price else Side.SUPPORT) is positional_side ), key=lambda item: item[0], ) side_groups: list[list[tuple[float, Level]]] = [] for item in side_levels: if not side_groups or item[0] - side_groups[-1][-1][0] > tolerance: side_groups.append([item]) else: side_groups[-1].append(item) groups.extend(side_groups) clusters: list[Cluster] = [] for group in groups: score = sum(level.weight for _, level in group) if len(group) < 2 and score < 8: continue low, high = group[0][0], group[-1][0] center = (low + high) / 2 side = Side.RESISTANCE if center >= current_price else Side.SUPPORT identity_bucket = round(center / tolerance) identity = sha1(f"{side.value}:{identity_bucket}".encode()).hexdigest()[:12] clusters.append( Cluster( id=f"cl_{identity}", side=side, low=low, high=high, center=center, score=score, members=[level for _, level in group], distance=center - current_price, ) ) return sorted(clusters, key=lambda cluster: abs(cluster.distance))