158 lines
6.2 KiB
Python
158 lines
6.2 KiB
Python
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from app.analysis.confluence import Cluster
|
|
from app.analysis.levels import LevelKind
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Alert:
|
|
cluster: Cluster
|
|
message: str
|
|
# Hand-placed levels that caused this alert; the caller disarms them so a
|
|
# one-shot alert stays one-shot.
|
|
tripped: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class _Fired:
|
|
center: float
|
|
at: int
|
|
|
|
|
|
class AlertEngine:
|
|
"""Fires once per price zone, then stays quiet until price genuinely leaves.
|
|
|
|
Suppression is by proximity rather than cluster identity. Membership churns
|
|
constantly — a moving average drifts in and out of a group, changing the
|
|
cluster's identity while a human still sees one zone sitting at the prior
|
|
day's close. Keying on identity let every reshuffle through as a fresh
|
|
alert; keying on where the zone *is* does not.
|
|
|
|
Suppression is persisted when given a ``state_path``. Without it the list
|
|
lives only in memory, so every restart re-fires every zone that currently
|
|
qualifies — with a four-hour cooldown that turned each deploy into a burst
|
|
of pushes for zones that had already had their say.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
min_score: float,
|
|
cooldown_seconds: int = 900,
|
|
state_path: Path | None = None,
|
|
):
|
|
self.min_score = min_score
|
|
self.cooldown_seconds = cooldown_seconds
|
|
self.state_path = Path(state_path) if state_path else None
|
|
self._fired: list[_Fired] = self._load()
|
|
|
|
def _load(self) -> list[_Fired]:
|
|
if not self.state_path or not self.state_path.exists():
|
|
return []
|
|
try:
|
|
payload = json.loads(self.state_path.read_text(encoding="utf-8"))
|
|
return [_Fired(float(item["center"]), int(item["at"])) for item in payload]
|
|
except Exception:
|
|
# Corrupt state costs one burst of duplicate alerts, which is a far
|
|
# better failure than refusing to start the stream.
|
|
logger.warning("Could not read alert state; starting empty", exc_info=True)
|
|
return []
|
|
|
|
def _save(self) -> None:
|
|
if not self.state_path:
|
|
return
|
|
try:
|
|
self.state_path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = self.state_path.with_suffix(self.state_path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
json.dumps(
|
|
[{"center": entry.center, "at": entry.at} for entry in self._fired],
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary.replace(self.state_path)
|
|
except Exception:
|
|
# Losing a write means duplicate alerts later, never a missed one.
|
|
logger.warning("Could not persist alert state", exc_info=True)
|
|
|
|
def evaluate(
|
|
self,
|
|
clusters: list[Cluster],
|
|
current_price: float,
|
|
atr15: float,
|
|
now: int,
|
|
symbol: str,
|
|
) -> list[Alert]:
|
|
tolerance = 0.5 * atr15
|
|
if tolerance <= 0:
|
|
return []
|
|
# Two zones within an ATR of each other are the same zone as far as
|
|
# being told about them goes.
|
|
merge_distance = 2 * tolerance
|
|
|
|
# Re-arming needs both elapsed time and real separation. Time alone lets
|
|
# price oscillating on a level alert forever.
|
|
before = len(self._fired)
|
|
self._fired = [
|
|
entry
|
|
for entry in self._fired
|
|
if not (
|
|
now - entry.at >= self.cooldown_seconds
|
|
and abs(entry.center - current_price) > merge_distance
|
|
)
|
|
]
|
|
changed = len(self._fired) != before
|
|
|
|
alerts: list[Alert] = []
|
|
# Strongest first, so when several overlapping zones qualify at once the
|
|
# one that survives suppression is the most significant.
|
|
for cluster in sorted(clusters, key=lambda item: item.score, reverse=True):
|
|
drawn = [
|
|
member
|
|
for member in cluster.members
|
|
if member.kind is LevelKind.MANUAL and member.armed
|
|
]
|
|
# A drawn line bypasses the score threshold entirely. Weights run
|
|
# from 1 (5m) to 4 (1h) against a threshold of 28, so gating on
|
|
# score would mean a line you deliberately drew could never alert.
|
|
# A disarmed one has already had its say and no longer qualifies.
|
|
if not drawn and cluster.score < self.min_score:
|
|
continue
|
|
if abs(cluster.center - current_price) > tolerance:
|
|
continue
|
|
# Deliberately not matched on side. A level sitting at price flips
|
|
# between support and resistance every time price ticks across it,
|
|
# because the side is positional. Matching on it meant a zone price
|
|
# was oscillating on re-alerted on every crossing — which is exactly
|
|
# when a level is least newsworthy, not most.
|
|
if any(
|
|
abs(entry.center - cluster.center) <= merge_distance for entry in self._fired
|
|
):
|
|
continue
|
|
self._fired.append(_Fired(cluster.center, now))
|
|
changed = True
|
|
direction = "BEARISH" if cluster.side.value == "resistance" else "BULLISH"
|
|
timeframes = ", ".join(dict.fromkeys(member.tf.value for member in cluster.members))
|
|
# Naming the line matters: "your line" is actionable in a way that
|
|
# "confluence 4" is not, and it says which drawing to go look at.
|
|
headline = "LINE" if drawn and len(cluster.members) == len(drawn) else "ZONE"
|
|
detail = (
|
|
f"{cluster.side.value.title()} confluence {cluster.score:g} "
|
|
f"@ {cluster.low:.2f}-{cluster.high:.2f}"
|
|
)
|
|
if drawn:
|
|
detail += "\n" + ", ".join(member.label for member in drawn)
|
|
message = (
|
|
f"{direction} {headline} {symbol} {current_price:.2f}\n{detail}\n{timeframes}"
|
|
)
|
|
alerts.append(Alert(cluster, message, tuple(member.id for member in drawn)))
|
|
if changed:
|
|
self._save()
|
|
return alerts
|