AH-Oil-Trader/tests/test_slot_gv.py

"""Zeigen G/V und G/V BRK wirklich NUR ihren eigenen Slot? (2026-08-20)

ANLASS (User): "ueberpruefe ob wirklich die G/V und G/V BRK Anzeigen stimmen,
besonders bei gleichzeitig laufenden Trades. Die Anzeige darf jeweils nur den
eigenen Slot anzeigen."

Geprueft wird die STELLE, an der die Trennung entstehen muss - die Auswahl der
Position in `TradeManager._refresh_locked`. Das Frontend liest nur `d.position`
bzw. `d.position_brk`; wenn schon dort dieselbe Position steht, kann keine
Anzeige das noch richten.

DIE GEFAEHRLICHE STELLE ist die ADOPTION: der Haupt-Trader greift jede Position
auf dem Symbol, damit ein von Hand eroeffneter Trade den Schutz-Stack bekommt.
Genau das darf er beim BRK-Ticket NICHT - dafuer gibt es `fremde_tickets`.
Diese Sperrliste wird hier in EXAKT der Form nachgebaut, in der die Engine sie
setzt (core/engine.py, `self.trader.fremde_tickets = ...`).
"""
import types

import pytest

from core import trader as trader_mod
from core.trader import TradeManager


class _Pos:
    def __init__(self, ticket, lots=1.0, entry=80.0, pnl=0.0, magic=0):
        self.ticket, self.volume, self.price_open = ticket, lots, entry
        self.profit, self.magic = pnl, magic
        self.type, self.symbol, self.time = 0, "SpotCrude", 1_700_000_000
        self.sl, self.tp = 79.0, 82.0


@pytest.fixture
def welt(monkeypatch):
    """Stub-Broker mit einer frei setzbaren Positionsliste."""
    offen = {}

    def positions_get(**kw):
        if "ticket" in kw:
            p = offen.get(int(kw["ticket"]))
            return (p,) if p else ()
        return tuple(offen.values())

    fake = types.SimpleNamespace(
        positions_get=positions_get,
        order_calc_margin=lambda *a, **k: 700.0,
        order_send=lambda *a, **k: None,
        ORDER_TYPE_BUY=0, ORDER_TYPE_SELL=1,
        TRADE_ACTION_SLTP=6, TRADE_RETCODE_DONE=10009,
        symbol_info_tick=lambda s: None,
        symbol_info=lambda s: None,
    )
    monkeypatch.setattr(trader_mod, "mt5", fake)
    monkeypatch.setattr(trader_mod, "MAGIC", 4711, raising=False)

    class _Lock:
        def __enter__(self): return True
        def __exit__(self, *a): return False
    monkeypatch.setattr(trader_mod, "mt5_lock", lambda *a, **k: _Lock())
    return offen


def _slots(pending=None, brk_restore=None):
    """Beide Manager wie in `TradingEngine.__init__` verdrahtet."""
    haupt, brk = TradeManager(), TradeManager(nur_ticket=True)
    haupt._broker_offset_s = lambda sym=None: 0
    brk._broker_offset_s = lambda sym=None: 0
    pend = dict(pending or {})
    haupt.fremde_tickets = lambda: (
        ({brk.ticket} if brk.ticket else set()) | set(pend)
        | ({brk_restore} if brk_restore else set())
    )
    return haupt, brk, pend


def _gv(haupt, brk):
    """Genau das, was `snapshot()` ans Frontend gibt."""
    a, b = haupt.snapshot(), brk.snapshot()
    return (a.get("ticket"), a.get("pnl")), (b.get("ticket"), b.get("pnl"))


# --------------------------------------------------------------------------
def test_zwei_positionen_jede_im_eigenen_slot(welt):
    """Der Normalfall, nach dem der User fragt: BEIDE Slots gleichzeitig."""
    haupt, brk, _ = _slots()
    welt[100] = _Pos(100, pnl=-12.50, magic=4711)   # BRK, gefuellte Pending
    brk.bind(100, "SpotCrude")
    welt[200] = _Pos(200, pnl=+31.00, magic=4711)   # manueller Trade
    haupt.refresh("SpotCrude")

    (t1, p1), (t2, p2) = _gv(haupt, brk)
    assert t1 == 200 and p1 == pytest.approx(31.00), "Slot 1 hat die falsche Position"
    assert t2 == 100 and p2 == pytest.approx(-12.50), "BRK hat die falsche Position"
    assert t1 != t2, "BEIDE Anzeigen zeigen dasselbe Ticket"


def test_haupt_adoptiert_die_brk_position_NICHT(welt):
    """Nur BRK offen, Slot 1 flat -> Slot 1 muss LEER bleiben."""
    haupt, brk, _ = _slots()
    welt[100] = _Pos(100, pnl=-5.0, magic=4711)
    brk.bind(100, "SpotCrude")
    haupt.refresh("SpotCrude")

    (t1, _), (t2, _) = _gv(haupt, brk)
    assert t1 is None, "Slot 1 hat die BRK-Position adoptiert -> G/V zeigt sie doppelt"
    assert t2 == 100


def test_pending_fill_wird_nicht_vorher_adoptiert(welt):
    """Der Fill liegt am Broker, bevor `_check_pending_fill` laeuft.

    Die Positions-Nummer ist dann identisch mit der Order-Nummer - deshalb
    haelt `_pending_tickets` die Luecke zu.
    """
    haupt, brk, _ = _slots(pending={100: "squeeze"})
    welt[100] = _Pos(100, pnl=+3.0, magic=4711)     # gerade gefuellt
    haupt.refresh("SpotCrude")                      # laeuft ZUERST im _pos_loop

    assert haupt.snapshot().get("ticket") is None, (
        "Slot 1 hat den BRK-Pending-Fill adoptiert, bevor der BRK-Slot ihn "
        "uebernehmen konnte")
    assert brk.bind(100, "SpotCrude")
    assert brk.snapshot().get("ticket") == 100


def test_brk_slot_adoptiert_niemals(welt):
    """`nur_ticket=True` darf sich NIE etwas greifen - auch nicht bei magic."""
    _, brk, _ = _slots()
    welt[200] = _Pos(200, pnl=+9.0, magic=4711)
    brk.refresh("SpotCrude")
    assert brk.snapshot().get("ticket") is None


def test_neustart_ohne_merker_in_der_sperrliste_BRICHT(welt):
    """REGRESSIONSPROBE: so sah es aus, bevor `_brk_restore` mit sperrte.

    Nach einem Neustart ist `trader_brk.ticket` None und `_pending_tickets`
    leer. Steht der Merker NICHT in der Sperrliste, greift Slot 1 die
    BRK-Position - und das passiert im `_pos_loop` schon eine Zeile VOR
    `_rebind_brk()`.
    """
    haupt, brk, _ = _slots(brk_restore=None)        # <- alter Zustand
    welt[100] = _Pos(100, pnl=-40.0, magic=4711)
    haupt.refresh("SpotCrude")
    assert haupt.snapshot().get("ticket") == 100, (
        "Der Test bildet den alten Zustand nicht mehr ab")


def test_neustart_mit_merker_bleibt_getrennt(welt):
    """Mit dem Merker in der Sperrliste ueberlebt die Trennung den Neustart."""
    haupt, brk, _ = _slots(brk_restore=100)
    welt[100] = _Pos(100, pnl=-40.0, magic=4711)
    haupt.refresh("SpotCrude")
    assert haupt.snapshot().get("ticket") is None, (
        "Slot 1 hat die BRK-Position trotz Merker adoptiert")
    assert brk.bind(100, "SpotCrude")               # _rebind_brk kommt danach
    (t1, _), (t2, _) = _gv(haupt, brk)
    assert t1 is None and t2 == 100