AH-Oil-Trader/tests/test_close_notify.py

"""Close-Benachrichtigung: feuert sie genau dann, wenn sie soll?

ANLASS (2026-08-11, User: „ich bekomme keine Telegram-Nachrichten mehr"). Der
Kanal war intakt — eine Testnachricht wurde zugestellt —, aber ALLE NEUN
sendenden Ausloeser waren aus. Jede Abschaltung war fuer sich begruendet;
zusammen ergaben sie eine Stille, die niemand beschlossen hatte.

⚠ Der frueher entfernte Trade-Abschluss-Push scheiterte an fehlender SCHWELLE
(jeder Close, ~17/Tag). Genau die wird hier geprueft — und der Renner-Fall, in
dem die Engine dem DB-Schreiber einen Tick voraus ist.
"""
from __future__ import annotations

import pytest

import core.engine as eng
from core.engine import TradingEngine


class _Hist:
    def __init__(self, trade):
        self.trade = trade
        self.aufrufe = 0

    def trade_by_ticket(self, ticket):
        self.aufrufe += 1
        return self.trade


class _Stub:
    """Minimal-Engine — nur was `_check_close_notify` wirklich liest."""

    _check_close_notify = TradingEngine._check_close_notify

    def __init__(self, trade, min_eur=50.0, an=True):
        self.history = _Hist(trade)
        self._close_notify_min = min_eur
        self._close_notify_pending = [4711, 0]
        self.cfg = {"telegram": {"enabled": "true" if an else "false",
                                 "bot_token": "T", "chat_id": "C"}}


def _trade(pnl=-87.03, fertig=True):
    return {"ticket": 4711, "symbol": "SpotCrude", "direction": "BUY",
            "lots": 1.77, "entry_price": 79.065, "exit_price": 78.497,
            "entry_time": 1000, "exit_time": 7180 if fertig else None,
            "pnl": pnl if fertig else None, "closed_by": "sl", "setup": "WAVE"}


@pytest.fixture(autouse=True)
def _abfangen(monkeypatch):
    """`send_telegram` abfangen — im Test geht NICHTS ins Netz."""
    gesendet = []
    monkeypatch.setattr(eng, "send_telegram",
                        lambda msg, tok, cid: gesendet.append(msg))
    return gesendet


def test_ueber_schwelle_wird_gesendet(_abfangen):
    """Der reale −87,03-€-SL-Trade vom 10.08. — der kam bisher wortlos."""
    s = _Stub(_trade(-87.03))
    s._check_close_notify()
    assert len(_abfangen) == 1
    t = _abfangen[0]
    assert "-87.03" in t and "Broker-SL" in t and "🔴" in t
    assert s._close_notify_pending is None


def test_unter_schwelle_bleibt_still(_abfangen):
    """Genau der Grund, warum der alte Push als Spam entfernt wurde."""
    s = _Stub(_trade(12.40))
    s._check_close_notify()
    assert _abfangen == []
    assert s._close_notify_pending is None, "Vormerkung trotzdem aufraeumen"


def test_gewinn_bekommt_gruenes_icon(_abfangen):
    s = _Stub(_trade(+229.41))
    s._check_close_notify()
    assert "🟢" in _abfangen[0] and "+229.41" in _abfangen[0]


def test_db_noch_nicht_fertig_wird_erneut_versucht(_abfangen):
    """⚠ Der gefaehrlichste Fall: der Flat-Zustand kommt aus `trader.snapshot()`
    und ist dem DB-Schreiber gelegentlich einen Tick voraus. `exit_time IS NULL`
    heisst „noch nicht fertig", NICHT „kein Trade" — wer das verwechselt, meldet
    einen Close ohne P&L."""
    s = _Stub(_trade(fertig=False))
    s._check_close_notify()
    assert _abfangen == []
    assert s._close_notify_pending == [4711, 1], "muss vorgemerkt BLEIBEN"
    # naechster Tick: DB ist durch
    s.history.trade = _trade(-87.03)
    s._check_close_notify()
    assert len(_abfangen) == 1 and s._close_notify_pending is None


def test_gibt_nach_30_versuchen_auf(_abfangen):
    """Ein Fremd-Trade ohne DB-Zeile darf nicht ewig gepollt werden."""
    s = _Stub(None)
    for _ in range(31):
        s._check_close_notify()
    assert _abfangen == []
    assert s._close_notify_pending is None


def test_telegram_aus_sendet_nicht(_abfangen):
    s = _Stub(_trade(-200.0), an=False)
    s._check_close_notify()
    assert _abfangen == []


def test_ohne_vormerkung_passiert_nichts(_abfangen):
    s = _Stub(_trade(-200.0))
    s._close_notify_pending = None
    s._check_close_notify()
    assert _abfangen == [] and s.history.aufrufe == 0, "keine DB-Last ohne Anlass"


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-v"]))