""" Phase C2a unit tests: Reconciler 5-case matrix + watchdog. 12 tests against an in-memory FakeBroker and synthetic state. No orchestrator, no main — just the Reconciler API. case (i): 1 test case (ii): 4 tests (no-history, win, loss, no-risk-manager) case (iii): 1 test case (iv): 1 test case (v): 2 tests (naked logged, with-SL not naked) watchdog: 2 tests (no positions early-exit, raises non-fatal) report: 1 test (mixed cases counted correctly) Run: cd ~/apex_v16 python tests/test_reconciliation.py """ from __future__ import annotations import asyncio import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from broker.broker_base import ( BrokerBase, CancelResult, ClosedTrade, Order, OrderResult, Position, ) from broker.reconciliation import ( NakedPositionsReport, ReconciliationReport, Reconciler, ) from core.contracts import ( BrainName, Direction, MarketStructure, Regime, TradeEntry, TradeRuntime, ) from persistence.state_store import ActiveTrade, SessionState def _ok(label: str) -> None: print(f" ok {label}") # ============================================================ # FAKES # ============================================================ class JsonlSink: def __init__(self): self.events: list[dict] = [] def write(self, event, **fields): self.events.append({"event": event, **fields}) class FakeLogger: def __init__(self): self.brain_log = JsonlSink() class FakeRiskManager: def __init__(self): self.pnl_calls: list[tuple[float, bool, str]] = [] self.tp_hits: list[str] = [] self.sl_hits: list[str] = [] def update_daily_pnl(self, delta_usd, *, is_win, brain): self.pnl_calls.append((delta_usd, is_win, brain)) def register_tp_hit(self, symbol): self.tp_hits.append(symbol) def register_sl_hit(self, symbol, *, now_utc=None): self.sl_hits.append(symbol) class FakeBroker(BrokerBase): """In-memory BrokerBase. Tests configure `positions`, `pendings`, `trades`.""" name = "FakeBroker" def __init__(self): self.positions: list[Position] = [] self.pendings: list[Order] = [] self.trades: list[ClosedTrade] = [] self.recent_trades_raised: bool = False 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): return [p for p in self.positions if (symbol is None or p.symbol == symbol)] async def pending_orders(self, symbol=None): if symbol is None: return list(self.pendings) return [o for o in self.pendings if o.symbol == symbol] async def recent_trades(self, symbol=None, since=None, limit=50): if self.recent_trades_raised: raise RuntimeError("history fetch failed") rows = list(self.trades) if symbol is not None: rows = [t for t in rows if t.symbol == symbol or symbol in t.symbol] return rows[:limit] async def place_market_bracket(self, **kw): return OrderResult(success=True) async def place_stop_order(self, symbol, side, contracts, stop_price): return OrderResult(success=True, sl_price=stop_price, stop_id="S-FAKE", entry_id="S-FAKE") async def place_limit_order(self, symbol, side, contracts, limit_price): return OrderResult(success=True, tp_price=limit_price, target_id="T-FAKE", entry_id="T-FAKE") 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, sl_price=new_sl_price) 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"]) # ============================================================ # FIXTURES # ============================================================ def make_entry(symbol="MES", contracts=2, brain=BrainName.TF.value, opened_minutes_ago=10) -> TradeEntry: return TradeEntry( symbol=symbol, brain_name=brain, direction=Direction.BUY.value, contracts=contracts, entry_price=5800.0, sl_price=5790.0, tp_price=5820.0, opened_at=datetime.now(timezone.utc) - timedelta(minutes=opened_minutes_ago), rsi_m5_at_entry=55.0, rsi_h1_at_entry=52.0, rsi_h4_at_entry=50.0, atr_ratio_at_entry=1.0, market_structure_at_entry=MarketStructure.BULLISH_EXPANSION.value, regime_at_entry=Regime.TRENDING.value, h1_compat_at_entry=1.0, confidence_at_entry=70, entry_order_id="E1", stop_order_id="S1", target_order_id="T1", ) def make_state_with_trade(symbol="MES", contracts=2, brain=BrainName.TF.value) -> SessionState: state = SessionState() entry = make_entry(symbol=symbol, contracts=contracts, brain=brain) runtime = TradeRuntime() state.active_trades[symbol] = ActiveTrade(entry=entry, runtime=runtime) return state def run(coro): return asyncio.run(coro) # ============================================================ # TESTS — 5-case matrix # ============================================================ def test_case_i_state_open_broker_open_match(): """state OPEN + broker OPEN + matching contracts -> case (i) OK, no mutation.""" state = make_state_with_trade("MES", contracts=2) broker = FakeBroker() broker.positions = [Position(symbol="MES", direction="BUY", contracts=2, avg_price=5800.0)] rec = Reconciler(broker, state, FakeRiskManager(), FakeLogger()) report = run(rec.reconcile_startup()) assert report.case_i_ok == ["MES"] assert "MES" in state.active_trades, "state must NOT be mutated in case (i)" _ok("case (i): state OPEN + broker OPEN + size match -> log-only, no mutation") def test_case_ii_no_history_drops_trade_without_risk_update(): """state OPEN + broker FLAT + no recent_trades -> drop, no risk hooks.""" state = make_state_with_trade("MES") broker = FakeBroker() # empty positions, empty trades risk = FakeRiskManager() rec = Reconciler(broker, state, risk, FakeLogger()) report = run(rec.reconcile_startup()) assert report.case_ii_state_open_broker_flat == ["MES"] assert report.case_ii_recovered_via_history == [] assert "MES" not in state.active_trades, "case (ii) MUST drop the orphan trade" assert risk.pnl_calls == [] and risk.tp_hits == [] and risk.sl_hits == [] _ok("case (ii): no history -> drop trade, no risk update") def test_case_ii_recovered_via_history_win(): """state OPEN + broker FLAT + recent_trades has WIN -> register_tp_hit + pnl applied.""" state = make_state_with_trade("MES", brain=BrainName.TF.value) broker = FakeBroker() broker.trades = [ClosedTrade( trade_id="T9", symbol="MES", contracts=2, side="SELL", exit_price=5810.0, pnl_usd=120.0, closed_at="2026-04-28T12:00:00+00:00", )] risk = FakeRiskManager() rec = Reconciler(broker, state, risk, FakeLogger()) report = run(rec.reconcile_startup()) assert report.case_ii_recovered_via_history == ["MES"] assert report.pnl_recovered_usd == 120.0 assert risk.pnl_calls == [(120.0, True, "TF")] assert risk.tp_hits == ["MES"] assert risk.sl_hits == [] assert "MES" not in state.active_trades _ok("case (ii) recovered WIN: pnl applied, register_tp_hit, trade dropped") def test_case_ii_recovered_via_history_loss(): """state OPEN + broker FLAT + recent_trades has LOSS -> register_sl_hit.""" state = make_state_with_trade("MES", brain=BrainName.MR.value) broker = FakeBroker() broker.trades = [ClosedTrade( trade_id="T9", symbol="MES", contracts=2, side="SELL", exit_price=5790.0, pnl_usd=-90.0, closed_at="2026-04-28T12:00:00+00:00", )] risk = FakeRiskManager() rec = Reconciler(broker, state, risk, FakeLogger()) report = run(rec.reconcile_startup()) assert report.pnl_recovered_usd == -90.0 assert risk.pnl_calls == [(-90.0, False, "MR")] assert risk.sl_hits == ["MES"] assert risk.tp_hits == [] _ok("case (ii) recovered LOSS: pnl applied, register_sl_hit") def test_case_ii_recent_trades_raises_falls_back(): """recent_trades raising -> degraded mode: drop trade, no risk update.""" state = make_state_with_trade("MES") broker = FakeBroker() broker.recent_trades_raised = True risk = FakeRiskManager() rec = Reconciler(broker, state, risk, FakeLogger()) report = run(rec.reconcile_startup()) assert report.case_ii_state_open_broker_flat == ["MES"] assert report.case_ii_recovered_via_history == [] assert "MES" not in state.active_trades, "fallback still drops the trade" assert risk.pnl_calls == [] _ok("case (ii) recent_trades raises: degraded, drop trade, no risk update") def test_case_ii_no_risk_manager_drops_without_recovery(): """risk_manager=None -> case (ii) drops trade but never attempts P&L recovery.""" state = make_state_with_trade("MES") broker = FakeBroker() broker.trades = [ClosedTrade( trade_id="T9", symbol="MES", contracts=2, side="SELL", exit_price=5810.0, pnl_usd=120.0, closed_at="2026-04-28T12:00:00+00:00", )] rec = Reconciler(broker, state, risk_manager=None, logger=FakeLogger()) report = run(rec.reconcile_startup()) assert report.case_ii_recovered_via_history == [], \ "no risk_manager -> no recovery attempted" assert "MES" not in state.active_trades, "trade still dropped" _ok("case (ii) no risk_manager: drop without recovery") def test_case_iii_state_flat_broker_open_log_only(): """state FLAT + broker OPEN -> case (iii), LOG-ONLY (no auto-adopt).""" state = SessionState() # no active trades broker = FakeBroker() broker.positions = [Position( symbol="MES", direction="BUY", contracts=1, avg_price=5800.0, )] logger = FakeLogger() rec = Reconciler(broker, state, FakeRiskManager(), logger) report = run(rec.reconcile_startup()) assert report.case_iii_state_flat_broker_open == ["MES"] assert "MES" not in state.active_trades, "case (iii) MUST NOT auto-adopt" assert any(e["event"] == "recon_case_iii_state_flat_broker_open" for e in logger.brain_log.events) _ok("case (iii): state FLAT + broker OPEN -> log-only, no auto-adopt") def test_case_iv_size_mismatch_log_only(): """state OPEN(2) + broker OPEN(1) -> case (iv), LOG-ONLY (V16 conservative).""" state = make_state_with_trade("MES", contracts=2) broker = FakeBroker() broker.positions = [Position( symbol="MES", direction="BUY", contracts=1, avg_price=5800.0, )] logger = FakeLogger() rec = Reconciler(broker, state, FakeRiskManager(), logger) report = run(rec.reconcile_startup()) assert report.case_iv_size_mismatch == ["MES"] # state.active_trades still holds the trade with original 2 contracts assert state.active_trades["MES"].entry.contracts == 2 assert any(e["event"] == "recon_case_iv_size_mismatch" for e in logger.brain_log.events) _ok("case (iv): size mismatch -> log-only, state untouched") def test_case_v_naked_position_logged(): """Position with no STOP order in pending -> case (v) reported.""" state = make_state_with_trade("MES") broker = FakeBroker() broker.positions = [Position( symbol="MES", direction="BUY", contracts=2, avg_price=5800.0, )] # only a TP order, no STOP -> naked broker.pendings = [Order( order_id="O-TP", symbol="CON.F.US.ES.M26", kind="LIMIT", price=5820.0, contracts=2, )] rec = Reconciler(broker, state, FakeRiskManager(), FakeLogger()) report = run(rec.reconcile_startup()) assert "MES" in report.case_v_naked_orders _ok("case (v): position without STOP in pending -> flagged naked") def test_case_v_position_with_stop_not_naked(): """Position with STOP order -> NOT naked.""" state = make_state_with_trade("MES") broker = FakeBroker() broker.positions = [Position( symbol="MES", direction="BUY", contracts=2, avg_price=5800.0, )] broker.pendings = [Order( order_id="O-SL", symbol="CON.F.US.ES.M26", kind="STOP", price=5790.0, contracts=2, )] rec = Reconciler(broker, state, FakeRiskManager(), FakeLogger()) report = run(rec.reconcile_startup()) assert "MES" not in report.case_v_naked_orders _ok("case (v): position with STOP -> not flagged naked") def test_watchdog_no_positions_returns_empty(): """watchdog_naked_positions with no broker positions -> empty report.""" state = SessionState() broker = FakeBroker() # zero positions rec = Reconciler(broker, state, FakeRiskManager(), FakeLogger()) report = run(rec.watchdog_naked_positions()) assert isinstance(report, NakedPositionsReport) assert report.naked_symbols == [] _ok("watchdog: no positions -> empty NakedPositionsReport") def test_watchdog_flags_naked_after_pending_drop(): """watchdog_naked_positions: position present, no STOP in pending -> flagged.""" state = SessionState() broker = FakeBroker() broker.positions = [Position( symbol="MNQ", direction="SELL", contracts=1, avg_price=20000.0, )] broker.pendings = [] # no SL, no TP rec = Reconciler(broker, state, FakeRiskManager(), FakeLogger()) report = run(rec.watchdog_naked_positions()) assert report.naked_symbols == ["MNQ"] _ok("watchdog: position without any pending order -> flagged naked") def test_report_counts_mixed_cases(): """All 5 cases coexist in a single reconcile_startup() -> all reported.""" state = SessionState() # case (i) MES: state + broker match (2 contracts) e_mes = make_entry(symbol="MES", contracts=2) state.active_trades["MES"] = ActiveTrade(entry=e_mes, runtime=TradeRuntime()) # case (ii) MNQ: state but no broker pos (no recent_trades match -> no recovery) e_mnq = make_entry(symbol="MNQ", contracts=1) state.active_trades["MNQ"] = ActiveTrade(entry=e_mnq, runtime=TradeRuntime()) # case (iv) 6E: state(2) vs broker(3) e_6e = make_entry(symbol="6E", contracts=2) state.active_trades["6E"] = ActiveTrade(entry=e_6e, runtime=TradeRuntime()) broker = FakeBroker() broker.positions = [ Position(symbol="MES", direction="BUY", contracts=2, avg_price=5800.0), Position(symbol="6E", direction="BUY", contracts=3, avg_price=1.10), # case (iii) 6B in broker but not in state Position(symbol="6B", direction="SELL", contracts=1, avg_price=1.27), ] # case (v): MES has STOP so not naked; 6E has no SL so naked broker.pendings = [ Order(order_id="OS-MES", symbol="CON.F.US.ES.M26", kind="STOP", price=5790.0, contracts=2), ] rec = Reconciler(broker, state, FakeRiskManager(), FakeLogger()) report = run(rec.reconcile_startup()) assert report.case_i_ok == ["MES"] assert report.case_ii_state_open_broker_flat == ["MNQ"] assert report.case_iii_state_flat_broker_open == ["6B"] assert report.case_iv_size_mismatch == ["6E"] # 6B and 6E are both naked (no STOP for either contract) assert set(report.case_v_naked_orders) >= {"6B", "6E"} assert "MES" not in report.case_v_naked_orders, "MES has its STOP -> not naked" # state mutated only for case (ii) assert "MES" in state.active_trades and "6E" in state.active_trades assert "MNQ" not in state.active_trades _ok("mixed: all 5 cases counted; mutation only for case (ii)") # ============================================================ # RUN # ============================================================ if __name__ == "__main__": print("test_reconciliation.py") test_case_i_state_open_broker_open_match() test_case_ii_no_history_drops_trade_without_risk_update() test_case_ii_recovered_via_history_win() test_case_ii_recovered_via_history_loss() test_case_ii_recent_trades_raises_falls_back() test_case_ii_no_risk_manager_drops_without_recovery() test_case_iii_state_flat_broker_open_log_only() test_case_iv_size_mismatch_log_only() test_case_v_naked_position_logged() test_case_v_position_with_stop_not_naked() test_watchdog_no_positions_returns_empty() test_watchdog_flags_naked_after_pending_drop() test_report_counts_mixed_cases() print("ALL 13 TESTS PASSED")