"""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