""" Phase A smoke tests for orchestrator.py + brain/brain_selector.py. Phase A scope: orchestrator runs the loop, builds tech, resolves bias, chooses brain, calls evaluate_entry / manage_exit, and LOGS proposals without acting on them. No opener / closer / sizing / risk_manager invoked yet (Phase B). state.active_trades is observed, never mutated. Patterns: FakeMarketDataProvider (canned bars), FakeAIClient (canned AIResponse), FakeStateStore (in-memory save count), and a fake BrainBundle exposing evaluate_entry / manage_exit only. Run: cd ~/apex_v16 python -m tests.test_orchestrator_phase_a """ from __future__ import annotations import asyncio import json import sys import tempfile from pathlib import Path from datetime import datetime, timezone import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from analysis.bias import BiasData, BiasResolver from analysis.tech_snapshot import TechSnapshot from brain.ai_client import AIResponse from brain.brain_selector import ( INDICI_FUTURES, MR_EXCLUDED, TF_ALLOWED_SYMBOLS, choose_brain, ) from core.config import RuntimeConfig, RunMode, AccountKind from core.contracts import EntryDecision, BrainDecision, EntryEvalResult from orchestrator import Orchestrator from persistence.state_store import SessionState, StateStore def _ok(label: str) -> None: print(f" ok {label}") # ============================================================ # FAKES / FIXTURES # ============================================================ class FakeProvider: """Returns whatever DataFrame is supplied per (symbol, timeframe).""" def __init__(self, bars: dict | None = None) -> None: self.bars = bars or {} self.calls: list[tuple[str, str, int]] = [] async def get_bars(self, symbol: str, timeframe: str, n: int) -> pd.DataFrame: self.calls.append((symbol, timeframe, n)) df = self.bars.get((symbol, timeframe)) if df is None or df.empty: return pd.DataFrame(columns=["open", "high", "low", "close", "volume"]) return df.tail(n).reset_index(drop=True) class FakeAI: async def ask(self, prompt: str, temperature: float = 0.2, max_tokens: int | None = None) -> AIResponse: return AIResponse(text=None, error_kind="unknown") async def ask_for_decision( self, prompt: str, max_tokens: int = 600, where=None, ) -> AIResponse: return AIResponse(text=None, error_kind="unknown") class FakeBrain: """Records calls; returns canned EntryDecision / BrainDecision or None.""" def __init__( self, entry_decision: EntryDecision | None = None, exit_decision: BrainDecision | None = None, ) -> None: self.entry_decision = entry_decision self.exit_decision = exit_decision or BrainDecision(action="HOLD", reason="default") self.evaluate_calls: list[tuple[str, TechSnapshot]] = [] self.manage_calls: list[Any] = [] async def evaluate_entry(self, symbol, tech, *, last_entry_eval_time=0.0): self.evaluate_calls.append((symbol, tech)) # Mirror the production contract: when AI was "called" (we mock # success), set evaluated_candle_time so orchestrator updates # the cache. candle_time may be 0.0 in fixtures -> None. ct = float(getattr(tech, "candle_time", 0) or 0) or None return EntryEvalResult( decision=self.entry_decision, evaluated_candle_time=ct, ) async def manage_exit(self, ctx): self.manage_calls.append(ctx) return self.exit_decision class JsonlSink: """In-memory replacement for a JsonlLogger.write.""" def __init__(self) -> None: self.events: list[dict] = [] def write(self, event: str, **fields) -> None: self.events.append({"event": event, **fields}) class FakeLogger: """Mimics LoggerBundle minimally for Orchestrator's hooks.""" def __init__(self) -> None: import logging self.brain_log = JsonlSink() self.session_log = JsonlSink() self.error_log = JsonlSink() self.system = logging.getLogger("test.orch") def log_session_event(self, event: str, **fields) -> None: self.session_log.write(event, **fields) def log_error(self, where: str, error: str, **extra) -> None: self.error_log.write("error", where=where, error=error, **extra) def make_config( *, asset_filter=None, loop_sleep_seconds: int = 0, ) -> RuntimeConfig: cfg = RuntimeConfig(mode=RunMode.DRY, account=AccountKind.INELIGIBLE) cfg.asset_filter = asset_filter cfg.loop_sleep_seconds = loop_sleep_seconds # V16 3-loop test mode: 0-interval everywhere; per-loop max_iterations # counter terminates each loop after N iters. cfg.scan_loop_phase_offset_seconds = 0.0 cfg.manage_loop_interval_seconds = 0 cfg.maintenance_loop_interval_seconds = 0 return cfg def make_store(tmp: Path) -> StateStore: return StateStore(tmp / "state.json") def _to_snap(d: dict) -> TechSnapshot: """TechSnapshot factory (mirror of test_brain_tf helper).""" return TechSnapshot( symbol=d["symbol"], price=float(d.get("price", 0.0)), open=float(d.get("open", d.get("price", 0.0))), candle_time=int(d.get("candle_time", 0) or 0), is_candle_closed=bool(d.get("is_candle_closed", True)), candle_age_seconds=float(d.get("candle_age_seconds", 10.0)), rsi=float(d.get("rsi", 50.0)), rsi_prev=float(d.get("rsi_prev", d.get("rsi", 50.0))), rsi_h1=float(d.get("rsi_h1", 50.0)), rsi_h4=float(d.get("rsi_h4", 50.0)), atr_m5_points=float(d.get("atr_m5_points", 0.0)), atr_ratio=float(d.get("atr_ratio", 1.0)), vol_regime=d.get("vol_regime", "NORMAL"), vol_spike=bool(d.get("vol_spike", False)), market_structure=d.get("market_structure", "RANGING"), h1_struct_bull=bool(d.get("h1_struct_bull", False)), h1_struct_bear=bool(d.get("h1_struct_bear", False)), trend_maturity=int(d.get("trend_maturity", 0)), regime=d.get("regime", "RANGING"), regime_reason=d.get("regime_reason", ""), regime_near_trending=list(d.get("regime_near_trending", [])), deviation_pct=float(d.get("deviation_pct", 0.0)), divergence=d.get("divergence", "NONE"), macd_decelerating=bool(d.get("macd_decelerating", False)), macd_hist_last=float(d.get("macd_hist_last", 0.0)), candle_strength=float(d.get("candle_strength", 1.0)), hammer=bool(d.get("hammer", False)), shooting_star=bool(d.get("shooting_star", False)), bull_engulfing=bool(d.get("bull_engulfing", False)), bear_engulfing=bool(d.get("bear_engulfing", False)), doji=bool(d.get("doji", False)), doji_type=d.get("doji_type"), piercing=bool(d.get("piercing", False)), dark_cloud=bool(d.get("dark_cloud", False)), morning_star=bool(d.get("morning_star", False)), evening_star=bool(d.get("evening_star", False)), volume_weak=bool(d.get("volume_weak", False)), buy_absorption=bool(d.get("buy_absorption", False)), sell_absorption=bool(d.get("sell_absorption", False)), vwap=float(d.get("vwap", d.get("price", 0.0))), vwap_deviation_pct=float(d.get("vwap_deviation_pct", 0.0)), bias=d.get("bias", "NEUTRO"), allowed_direction=d.get("allowed_direction", "NONE"), h1_compatibility=float(d.get("h1_compatibility", 1.0)), h1_reason=d.get("h1_reason", ""), swing_data=dict(d.get("swing_data", {})), consecutive_sl_count=int(d.get("consecutive_sl_count", 0)), tick_size=float(d.get("tick_size", 0.0)), tick_value=float(d.get("tick_value", 0.0)), ) # Type-import safety from typing import Any # ============================================================ # 1-5: Orchestrator loop integration # ============================================================ def test_run_completes_max_iterations(): with tempfile.TemporaryDirectory() as tmp: cfg = make_config(asset_filter=["MES"]) cfg.scan_loop_phase_offset_seconds = 0.0 cfg.manage_loop_interval_seconds = 0 cfg.maintenance_loop_interval_seconds = 0 state = SessionState() store = make_store(Path(tmp)) provider = FakeProvider() # all empty bars ai = FakeAI() logger = FakeLogger() orch = Orchestrator( config=cfg, ai_client=ai, market_data_provider=provider, state=state, store=store, logger=logger, max_iterations=3, ) rc = asyncio.run(orch.run()) assert rc == 0 assert orch._iteration == 3 _ok("run(): max_iterations=3 -> exit code 0, 3 scan iterations") def test_dry_run_does_not_mutate_active_trades(): """Even when brain returns an EntryDecision, Phase A must NOT create a trade.""" with tempfile.TemporaryDirectory() as tmp: cfg = make_config(asset_filter=["MES"]) state = SessionState() store = make_store(Path(tmp)) # An entry-bearing brain — orchestrator must still skip mutation brain = FakeBrain(entry_decision=EntryDecision( direction="BUY", entry_price=5800.0, sl_price=5797.5, tp_price=5805.0, rr_multiplier=0.50, confidence=80, rationale="test", )) orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=state, store=store, logger=FakeLogger(), brain_dispatch={"TF": brain, "MR": brain}, max_iterations=2, ) asyncio.run(orch.run()) assert state.active_trades == {}, "Phase A must NOT mutate active_trades" _ok("Phase A dry-run: brain returns Entry but state.active_trades stays empty") def test_dry_run_logs_proposal_when_brain_emits(): """When evaluate_entry returns a decision, brain_log gets a 'dry_run_proposal'.""" with tempfile.TemporaryDirectory() as tmp: cfg = make_config(asset_filter=["MES"]) state = SessionState() # Inject a tech that routes to MR (RANGING + RSI<32 + non-index) # but MES is INDEX so it returns None; force a non-index symbol cfg.asset_filter = ["6E"] # 6E is FX, not in INDICI_FUTURES # Resolver path: provider is empty -> resolver returns BiasData BOTH # tech path: empty bars -> build_tech_snapshot returns None, # we won't reach evaluate_entry. For this test assert via # FakeBrain that we DO log a 'scan_skip', and an integration # test elsewhere covers the proposal path. brain = FakeBrain(entry_decision=None) logger = FakeLogger() orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=state, store=make_store(Path(tmp)), logger=logger, brain_dispatch={"TF": brain, "MR": brain}, max_iterations=1, ) asyncio.run(orch.run()) events = [e["event"] for e in logger.brain_log.events] # With empty bars build_tech_snapshot returns None -> scan_skip # is NOT emitted (we early-return on None tech). At minimum we # must NOT have crashed the loop. assert "dry_run_proposal" not in events _ok("Phase A: empty-bars path skips evaluate_entry without crashing") def test_iteration_order_and_save_each_tick(): """save() runs in maintenance_loop + final save in finally — at least 1.""" with tempfile.TemporaryDirectory() as tmp: cfg = make_config(asset_filter=["MES"]) cfg.scan_loop_phase_offset_seconds = 0.0 cfg.manage_loop_interval_seconds = 0 cfg.maintenance_loop_interval_seconds = 0 state = SessionState() store = make_store(Path(tmp)) orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=state, store=store, logger=FakeLogger(), max_iterations=4, ) save_count = {"n": 0} original_save = store.save def counting_save(s): save_count["n"] += 1 original_save(s) store.save = counting_save # type: ignore[assignment] asyncio.run(orch.run()) # V16 3-loop: save() in maintenance ticks (≥1) + 1 final assert save_count["n"] >= 1, f"expected ≥1 saves, got {save_count['n']}" _ok("save(): runs in maintenance_loop + final save in finally") def test_orchestrator_skips_when_no_brain_dispatch(): """No brains wired -> log scan_skip with brain_dispatch_missing.""" # Build tech that would route to TF/MR (skipped because dispatch empty) # We can't easily produce a non-None tech without real bars; the test # asserts no crash + no proposals (brain dispatch never called). with tempfile.TemporaryDirectory() as tmp: cfg = make_config(asset_filter=["MES"]) state = SessionState() logger = FakeLogger() orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=state, store=make_store(Path(tmp)), logger=logger, brain_dispatch={}, # empty max_iterations=1, ) rc = asyncio.run(orch.run()) assert rc == 0 # No 'dry_run_proposal' must appear (brains can't be called) assert all(e["event"] != "dry_run_proposal" for e in logger.brain_log.events) _ok("orchestrator: empty brain_dispatch -> no proposals, clean exit") # ============================================================ # 6-10: brain_selector tabular tests (pure function) # ============================================================ def test_brain_selector_ranging_extreme_rsi_picks_mr(): bias = BiasData(bias="RIALZISTA", allowed_direction="BUY", h1_compatibility=1.0, h1_reason="") tech = _to_snap({ "symbol": "6E", "rsi": 28.0, "rsi_prev": 30.0, "regime": "RANGING", "trend_maturity": 0, "market_structure": "RANGING", }) res = choose_brain(symbol="6E", tech=tech, bias=bias) assert res.chosen == "MR" assert res.reject_reason is None _ok("brain_selector: RANGING + RSI<32 + 6E (non-index) -> MR") def test_brain_selector_trending_pullback_picks_tf(): bias = BiasData(bias="RIALZISTA", allowed_direction="BUY", h1_compatibility=1.0, h1_reason="") tech = _to_snap({ "symbol": "6E", "rsi": 50.0, "rsi_prev": 48.0, # bouncing "regime": "TRENDING", "trend_maturity": 4, "market_structure": "BULLISH_EXPANSION", }) # Force daytime to avoid night_tf_block daytime = datetime(2026, 4, 29, 13, 0, 0, tzinfo=timezone.utc) res = choose_brain(symbol="6E", tech=tech, bias=bias, now_utc=daytime) assert res.chosen == "TF" assert res.reject_reason is None _ok("brain_selector: TRENDING + RSI 42-58 bouncing + non-index -> TF") def test_brain_selector_breakout_returns_none(): bias = BiasData(bias="RIALZISTA", allowed_direction="BUY", h1_compatibility=1.0, h1_reason="") tech = _to_snap({ "symbol": "MES", "rsi": 50.0, "rsi_prev": 48.0, "regime": "BREAKOUT", "trend_maturity": 4, }) res = choose_brain(symbol="MES", tech=tech, bias=bias) assert res.chosen is None assert res.reject_reason == "REGIME_BREAKOUT" _ok("brain_selector: BREAKOUT regime -> REGIME_BREAKOUT") def test_brain_selector_bias_none_returns_none(): bias = BiasData(bias="NEUTRO", allowed_direction="NONE", h1_compatibility=0.5, h1_reason="") tech = _to_snap({ "symbol": "6E", "rsi": 25.0, "rsi_prev": 30.0, # would-be MR "regime": "RANGING", "trend_maturity": 0, }) res = choose_brain(symbol="6E", tech=tech, bias=bias) assert res.chosen is None assert res.reject_reason == "BIAS_NONE" _ok("brain_selector: bias allowed=NONE -> BIAS_NONE") def test_brain_selector_mr_excluded_blocks_mr(): """MYM in MR_EXCLUDED -> never MR even if RSI extreme.""" assert "MYM" in MR_EXCLUDED bias = BiasData(bias="RIALZISTA", allowed_direction="BUY", h1_compatibility=1.0, h1_reason="") # MYM RANGING + extreme: V15 patch #20 says indices skip RANGING anyway, # so use TRENDING_SOFT to land in the MR widened branch — still excluded. tech = _to_snap({ "symbol": "MYM", "rsi": 30.0, "rsi_prev": 31.0, "regime": "TRENDING_SOFT", "trend_maturity": 4, "market_structure": "BULLISH_EXPANSION", }) # MYM is in TF_ALLOWED_SYMBOLS and in INDICI_FUTURES; index path won't # offer MR (only pro-trend MR with RSI<32, here MR_EXCLUDED is honoured) assert "MYM" in INDICI_FUTURES res = choose_brain(symbol="MYM", tech=tech, bias=bias) assert res.chosen is None # MYM is an index in TRENDING_SOFT -> lands in the index branch and # falls through INDICES_NO_PRO_TREND_SETUP. mr_excluded flag tells the # rest of the story. assert res.reject_reason == "INDICES_NO_PRO_TREND_SETUP" assert res.details.get("mr_excluded") is True _ok("brain_selector: MR_EXCLUDED symbol (MYM) never routes to MR") def test_brain_selector_indices_ranging_skipped(): """V15 patch #20: indices SKIP RANGING regime.""" bias = BiasData(bias="RIALZISTA", allowed_direction="BUY", h1_compatibility=1.0, h1_reason="") tech = _to_snap({ "symbol": "MES", "rsi": 25.0, "rsi_prev": 30.0, "regime": "RANGING", "trend_maturity": 0, }) res = choose_brain(symbol="MES", tech=tech, bias=bias) assert res.chosen is None assert res.reject_reason == "RANGING_INDEX_BLOCKED" _ok("brain_selector: indices in RANGING -> RANGING_INDEX_BLOCKED") # ============================================================ # 11-12: BiasResolver integration in Phase A loop # ============================================================ def test_no_h4_data_path_does_not_crash_loop(): """Empty H4 -> resolver returns NEUTRO BOTH; tech build returns None; loop just skips the symbol and survives the tick. Uses a fixed UTC clock at 14:00 so the trading-hours gate passes for MES/6E (otherwise the gate skips the symbol BEFORE the bias resolver runs and bias_cache stays empty). """ from datetime import datetime, timezone as _tz fixed = datetime(2026, 4, 28, 14, 0, 0, tzinfo=_tz.utc) with tempfile.TemporaryDirectory() as tmp: cfg = make_config(asset_filter=["MES", "6E"]) state = SessionState() provider = FakeProvider() # no bars at all orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=provider, state=state, store=make_store(Path(tmp)), logger=FakeLogger(), now_utc_provider=lambda: fixed, max_iterations=2, ) rc = asyncio.run(orch.run()) assert rc == 0 # Resolver was called -> bias_cache populated for both assert "MES" in state.bias_cache.entries assert "6E" in state.bias_cache.entries for sym in ("MES", "6E"): assert state.bias_cache.entries[sym].direction == "BOTH" _ok("no-h4-data: loop survives, BiasResolver caches NEUTRO BOTH per asset") def test_stop_event_breaks_loop_early(): """Calling orch.stop() makes run() exit at the next sleep boundary.""" async def runner(): with tempfile.TemporaryDirectory() as tmp: cfg = make_config(asset_filter=["MES"]) cfg.scan_loop_phase_offset_seconds = 0.0 cfg.manage_loop_interval_seconds = 1 cfg.maintenance_loop_interval_seconds = 1 state = SessionState() orch = Orchestrator( config=cfg, ai_client=FakeAI(), market_data_provider=FakeProvider(), state=state, store=make_store(Path(tmp)), logger=FakeLogger(), max_iterations=1000, # would loop forever without stop ) async def stopper(): await asyncio.sleep(0.1) orch.stop() await asyncio.gather(orch.run(), stopper()) asyncio.run(runner()) _ok("stop(): graceful exit interrupts the sleep window") # ============================================================ # MAIN # ============================================================ def main() -> int: print("test_orchestrator_phase_a.py") test_run_completes_max_iterations() test_dry_run_does_not_mutate_active_trades() test_dry_run_logs_proposal_when_brain_emits() test_iteration_order_and_save_each_tick() test_orchestrator_skips_when_no_brain_dispatch() test_brain_selector_ranging_extreme_rsi_picks_mr() test_brain_selector_trending_pullback_picks_tf() test_brain_selector_breakout_returns_none() test_brain_selector_bias_none_returns_none() test_brain_selector_mr_excluded_blocks_mr() test_brain_selector_indices_ranging_skipped() test_no_h4_data_path_does_not_crash_loop() test_stop_event_breaks_loop_early() print("ALL 13 TESTS PASSED") return 0 if __name__ == "__main__": sys.exit(main())