ai-exchange/tests/unit/test_trading_runtime_service.py

from datetime import datetime, timezone
import contextlib
import io
import json
import os
import unittest
from unittest.mock import patch

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")

from scripts import run_trading_runtime
from src.api.server import route_request
from src.services.runtime import TradingRuntimeScheduler, TradingRuntimeSupervisor


class FakeInstrumentRepository:
    def fetch_instrument_by_symbol(self, symbol):
        return {"id": 1 if symbol == "BTC-USDT" else 2, "symbol": symbol}


class FakeTradingPlanService:
    def __init__(self):
        self.repository = FakeInstrumentRepository()
        self.build_calls = 0

    def build_plan(self, **kwargs):
        self.build_calls += 1
        return type("Plan", (), {"status": "active"})()

    def persist_plan(self, candidate):
        return 101


class FakeMarketState:
    def __init__(self, *, freshness="market_data_fresh", stream="market_stream_connected", gap="market_data_gap_clear"):
        self.freshness = freshness
        self.stream = stream
        self.gap = gap

    def to_dict(self):
        return {
            "freshness": {"code": self.freshness},
            "stream_state": {"code": self.stream},
            "data_gap_state": {"code": self.gap},
        }


class FakeMarketStateService:
    def __init__(self, market_state=None):
        self.market_state = market_state or FakeMarketState()

    def build_market_state(self, **kwargs):
        return self.market_state

    def persist_market_state_snapshot(self, **kwargs):
        return 301


class FakeBrokerSnapshot:
    status = "broker_state_ok"
    error = None

    def to_dict(self):
        return {
            "status": self.status,
            "error": self.error,
            "account": {"equity": 1000.0},
            "positions": [],
            "orders": [],
            "fills": [],
            "alerts": [],
        }


class FakeBrokerReadModelService:
    def fetch_snapshot(self, **kwargs):
        return FakeBrokerSnapshot()


class FakeSetupWatcherService:
    def build_candidate(self, **kwargs):
        return type("Candidate", (), {"status": "blocked"})()

    def persist_candidate(self, candidate):
        return 401 if candidate is not None else None


class FakeDecisionPacketService:
    def build_persist_and_alert(self, **kwargs):
        return {
            "id": 501,
            "status": "actionable",
            "action_recommendation": "confirm_ticket",
            "alerts": [{"alert_type": "candidate_actionable"}],
        }


class TradingRuntimeSchedulerTests(unittest.TestCase):
    def test_run_cycle_records_successful_task_state(self) -> None:
        scheduler = TradingRuntimeScheduler(
            market_state_service=FakeMarketStateService(),
            trading_plan_service=FakeTradingPlanService(),
            setup_watcher_service=FakeSetupWatcherService(),
            broker_read_model_service=FakeBrokerReadModelService(),
            decision_packet_service=FakeDecisionPacketService(),
        )
        with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):
            payload = scheduler.run_cycle()

        self.assertEqual(payload["scheduler_state"], "running")
        self.assertEqual(payload["current_blockers"], [])
        self.assertEqual(payload["tasks"]["plan"]["last_status"], "ok")
        self.assertEqual(payload["tasks"]["broker_sync"]["details"]["status"], "broker_state_ok")
        self.assertEqual(payload["tasks"]["decision_packet"]["details"]["packet_id"], 501)
        self.assertEqual(payload["latest_broker_state_by_symbol"]["BTC-USDT"]["account"]["equity"], 1000.0)
        self.assertTrue(payload["tasks"]["watch"]["next_run_ts"])

    def test_run_cycle_enters_degraded_mode_when_market_truth_is_bad(self) -> None:
        scheduler = TradingRuntimeScheduler(
            market_state_service=FakeMarketStateService(FakeMarketState(stream="market_stream_disconnected")),
            trading_plan_service=FakeTradingPlanService(),
            setup_watcher_service=FakeSetupWatcherService(),
            broker_read_model_service=FakeBrokerReadModelService(),
            decision_packet_service=FakeDecisionPacketService(),
        )
        with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):
            payload = scheduler.run_cycle()

        self.assertTrue(payload["degraded_mode_state"]["is_degraded"])
        self.assertIn("market_stream_degraded:BTC-USDT:market_stream_disconnected", payload["current_blockers"])
        self.assertIn("watch_skipped_market_degraded:BTC-USDT", payload["current_blockers"])
        self.assertGreaterEqual(len(payload["incidents"]), 1)

    def test_supervisor_run_once_exposes_supervisor_state(self) -> None:
        scheduler = TradingRuntimeScheduler(
            market_state_service=FakeMarketStateService(),
            trading_plan_service=FakeTradingPlanService(),
            setup_watcher_service=FakeSetupWatcherService(),
            broker_read_model_service=FakeBrokerReadModelService(),
            decision_packet_service=FakeDecisionPacketService(),
        )
        supervisor = TradingRuntimeSupervisor(scheduler=scheduler)
        with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):
            payload = supervisor.run_once()

        self.assertEqual(payload["supervisor_state"], "stopped")
        self.assertEqual(payload["current_blockers"], [])


class TradingRuntimeApiTests(unittest.TestCase):
    def test_routes_runtime_status_and_run_once(self) -> None:
        status, payload = route_request(
            "/trading/runtime-status",
            {},
            dependencies={"trading_runtime_status": lambda: {"scheduler_state": "running", "current_blockers": []}},
        )
        run_status, run_payload = route_request(
            "/trading/runtime/run-once",
            {},
            method="POST",
            body={},
            dependencies={"trading_runtime_run_once": lambda: {"scheduler_state": "running", "tasks": {}}},
        )

        self.assertEqual(status, 200)
        self.assertEqual(payload["scheduler_state"], "running")
        self.assertEqual(run_status, 200)
        self.assertIn("tasks", run_payload)


class TradingRuntimeScriptTests(unittest.TestCase):
    def test_script_outputs_json(self) -> None:
        service = type("Service", (), {"run_once": lambda self: {"scheduler_state": "running", "current_blockers": []}})()
        output = io.StringIO()
        with patch("scripts.run_trading_runtime.get_trading_runtime_service", return_value=service), contextlib.redirect_stdout(output):
            exit_code = run_trading_runtime.main(env={"TRADING_RUNTIME_OUTPUT_JSON": "1"})

        self.assertEqual(exit_code, 0)
        payload = json.loads(output.getvalue())
        self.assertEqual(payload["scheduler_state"], "running")


if __name__ == "__main__":
    unittest.main()