""" Phase C2b unit tests for TradeOpener.post_order_safety_check. 3 tests: 1. orphan recovered with bracket re-attached -> stop_order_id and target_order_id populated, is_orphan_recovered=False, log warning "orphan_recovered_with_bracket_reattached". 2. orphan recovered but SL re-attach fails -> stop_order_id=None, is_orphan_recovered=True, log critical "orphan_recovered_naked". 3. no position present after sleep -> success=False, error_kind="not_found" Run: cd ~/apex_v16 python tests/test_trade_opener_safety_check.py """ from __future__ import annotations import asyncio import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) # Speed: zero out the safety check sleep so tests don't wait 2s each. import trading.trade_opener as _opener_mod _opener_mod.POST_ORDER_SAFETY_DELAY_SECONDS = 0.0 from broker.broker_base import ( BrokerBase, CancelResult, ClosedTrade, Order, OrderResult, Position, ) from core.contracts import ( BrainName, Direction, EntryDecision, MarketStructure, Regime, ) from trading.trade_opener import TradeOpener def _ok(label): print(f" ok {label}") # ============================================================ # FAKES # ============================================================ class JsonlSink: def __init__(self): self.events = [] def write(self, event, **fields): self.events.append({"event": event, **fields}) class FakeLogger: def __init__(self): import logging self.brain_log = JsonlSink() self.system = logging.getLogger("test.safety") class _Broker(BrokerBase): """In-memory BrokerBase. Configure positions + fail flags.""" name = "TestBroker" def __init__(self): self.positions: list[Position] = [] self.fail_sl = False self.fail_tp = False self.calls: list[str] = [] async def connect(self): return True async def disconnect(self): return None async def is_connected(self): return True async def get_last_price(self, symbol): return 5800.0 async def positions_get(self, symbol=None): self.calls.append("positions_get") return [p for p in self.positions if (symbol is None or p.symbol == symbol)] async def pending_orders(self, symbol=None): return [] async def recent_trades(self, symbol=None, since=None, limit=50): return [] async def place_market_bracket(self, **kw): return OrderResult(success=False, error="simulated bracket failure") async def place_stop_order(self, symbol, side, contracts, stop_price): self.calls.append("place_stop_order") if self.fail_sl: return OrderResult(success=False, error="SL re-attach failed") return OrderResult(success=True, sl_price=stop_price, stop_id="SL-NEW-1", entry_id="SL-NEW-1") async def place_limit_order(self, symbol, side, contracts, limit_price): self.calls.append("place_limit_order") if self.fail_tp: return OrderResult(success=False, error="TP re-attach failed") return OrderResult(success=True, tp_price=limit_price, target_id="TP-NEW-1", entry_id="TP-NEW-1") async def cancel_order(self, symbol, order_id): return CancelResult(success=True) async def cancel_all_for_symbol(self, symbol): return 0 async def close_position(self, symbol, contracts=None): return OrderResult(success=True) async def partial_close_via_opposite_order( self, symbol, direction, contracts_to_close, residual_contracts, new_sl_price, new_tp_price, old_stop_order_id, old_target_order_id, ): return OrderResult( success=True, entry_id="C-FAKE", stop_id="S2-FAKE", target_id="T2-FAKE", sl_price=new_sl_price, tp_price=new_tp_price, ) async def modify_stop(self, symbol, order_id, new_sl_price): return OrderResult(success=True) async def get_account_balance(self): return 50000.0 async def fetch_bars(self, symbol, timeframe, n): import pandas as pd return pd.DataFrame(columns=["open", "high", "low", "close", "volume"]) def make_decision(entry=5800.0): return EntryDecision( direction="BUY", entry_price=entry, sl_price=5790.0, tp_price=5820.0, rr_multiplier=0.50, confidence=70, rationale="test", ) def tech_now(): return { "rsi": 55.0, "rsi_h1": 52.0, "rsi_h4": 50.0, "atr_ratio": 1.0, "market_structure": MarketStructure.BULLISH_EXPANSION.value, "regime": Regime.TRENDING.value, "h1_compatibility": 1.0, } def run(coro): return asyncio.run(coro) # ============================================================ # TESTS # ============================================================ def test_orphan_recovered_with_bracket_reattached(): """Position found + SL re-attach OK + TP re-attach OK -> not naked.""" broker = _Broker() # Broker filled at 5801.25 even though Brain wanted 5800.0 -> diff=1.25 broker.positions = [Position( symbol="MES", direction="BUY", contracts=2, avg_price=5801.25, )] logger = FakeLogger() opener = TradeOpener(broker, is_paper=False, logger=logger) res = run(opener.post_order_safety_check( symbol="MES", brain_name=BrainName.TF.value, decision=make_decision(entry=5800.0), contracts=2, tech_now=tech_now(), error_context="bracket exception", )) assert res.success and res.entry is not None assert res.entry.is_orphan_recovered is False, \ "with bracket re-attached, NOT flagged as orphan" assert res.entry.stop_order_id == "SL-NEW-1" assert res.entry.target_order_id == "TP-NEW-1" assert res.entry.entry_price == 5801.25, "entry_price reflects broker avg_price" assert abs(res.entry.orphan_entry_price_diff - 1.25) < 1e-9, \ f"orphan_entry_price_diff={res.entry.orphan_entry_price_diff}" assert any(e["event"] == "orphan_recovered_with_bracket_reattached" for e in logger.brain_log.events) _ok("orphan recovered + SL+TP re-attached: stop_id/target_id set, NOT naked") def test_orphan_recovered_naked_when_sl_reattach_fails(): """Position found + SL re-attach FAILS -> registered as naked orphan.""" broker = _Broker() broker.positions = [Position( symbol="MES", direction="BUY", contracts=2, avg_price=5800.0, )] broker.fail_sl = True logger = FakeLogger() opener = TradeOpener(broker, is_paper=False, logger=logger) res = run(opener.post_order_safety_check( symbol="MES", brain_name=BrainName.TF.value, decision=make_decision(), contracts=2, tech_now=tech_now(), )) assert res.success and res.entry is not None assert res.entry.is_orphan_recovered is True, "naked orphan flag MUST be set" assert res.entry.stop_order_id is None, "naked: no stop_order_id" # TP is still attempted but the bracket-attached check requires BOTH assert any(e["event"] == "orphan_recovered_naked" for e in logger.brain_log.events) _ok("orphan + SL re-attach fails: naked orphan, stop_order_id=None, log critical") def test_no_orphan_position_returns_not_found(): """No position present after sleep -> success=False, error_kind=not_found.""" broker = _Broker() # no positions logger = FakeLogger() opener = TradeOpener(broker, is_paper=False, logger=logger) res = run(opener.post_order_safety_check( symbol="MES", brain_name=BrainName.TF.value, decision=make_decision(), contracts=2, tech_now=tech_now(), )) assert res.success is False assert res.error_kind == "not_found" assert "place_stop_order" not in broker.calls, \ "no position -> never attempt re-attach" assert any(e["event"] == "safety_check_no_position" for e in logger.brain_log.events) _ok("no orphan position: success=False not_found, no re-attach attempted") # ============================================================ # RUN # ============================================================ if __name__ == "__main__": print("test_trade_opener_safety_check.py") test_orphan_recovered_with_bracket_reattached() test_orphan_recovered_naked_when_sl_reattach_fails() test_no_orphan_position_returns_not_found() print("ALL 3 TESTS PASSED")