78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
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 and (level.cutoff_t is None or current_t <= level.cutoff_t)
|
|
]
|
|
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))
|