ai-exchange/tests/unit/test_launch_readiness_service.py

import os
import unittest
from unittest.mock import patch

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

from src.services.readiness import LaunchReadinessService


class LaunchReadinessServiceTests(unittest.TestCase):
    def test_build_summary_reports_assisted_not_ready_without_real_trade_sample(self) -> None:
        service = LaunchReadinessService(
            market_state_service=_MarketStateService(),
            broker_read_model_service=_BrokerReadService(),
            weekly_review_service=_WeeklyReviewService(),
            trust_service=_TrustService(),
            live_readiness_service=_LiveReadinessService(),
            shadow_paper_validation_service=_ShadowPaperValidationService(),
        )

        with (
            patch("src.services.readiness.launch_readiness_service.get_trading_runtime_service", return_value=_RuntimeService()),
            patch("src.services.readiness.launch_readiness_service.settings.launch_require_real_trade_sample", True),
        ):
            payload = service.build_summary(instrument_id=1, timeframe="1m")

        self.assertTrue(payload["analysis_only_ready"])
        self.assertTrue(payload["live_analysis_assistant_ready"])
        self.assertEqual(payload["live_analysis_assistant_summary"]["status"], "ready_for_readonly_live_analysis")
        self.assertTrue(payload["live_analysis_assistant_summary"]["must_not_trade"])
        self.assertFalse(payload["live_analysis_assistant_summary"]["execution_allowed"])
        self.assertEqual(payload["readonly_live_health_summary"]["broker_account_context"], "available")
        self.assertTrue(payload["assisted_trading_ready"])
        self.assertTrue(payload["operator_soft_rollout_ready"])
        self.assertEqual(payload["operator_soft_rollout_mode"], "execution_assist")
        self.assertTrue(payload["real_trade_sample_gate_required"])
        self.assertEqual(payload["real_trade_sample_gate_status"], "waiting_for_real_fill_sample")
        self.assertIn("wait_for_and_verify_real_fill_sample", payload["next_actions"])
        self.assertFalse(payload["real_trade_loop_ready"])
        self.assertIn("real_trade_sample_verified", payload["warning_reasons"])
        self.assertIn("real_trade_sample_verified", payload["external_waiting_items"])
        self.assertFalse(payload["real_live_ready"])
        self.assertEqual(payload["decision_support_boundary_summary"]["status"], "execution_assist_available")
        self.assertIn("review_market_context", payload["decision_support_boundary_summary"]["allowed_operator_actions"])
        self.assertIn("submit_live_orders", payload["decision_support_boundary_summary"]["blocked_operator_actions"])
        self.assertEqual(
            payload["decision_support_promotion_summary"]["status"],
            "ready_for_decision_support_non_execution",
        )
        self.assertTrue(payload["decision_support_promotion_summary"]["strict_decision_support_ready"])
        self.assertEqual(
            payload["decision_support_promotion_summary"]["execution_boundary_after_promotion"],
            "execution_still_blocked",
        )
        self.assertEqual(payload["shadow_paper_validation_summary"]["status"], "collecting_paper_samples")
        self.assertEqual(payload["shadow_paper_validation_summary"]["next_validation_gate"], "paper_ticket_sample_gate")

    def test_build_summary_defers_real_trade_sample_gate_for_analysis_decision_phase(self) -> None:
        service = LaunchReadinessService(
            market_state_service=_MarketStateService(),
            broker_read_model_service=_BrokerReadService(),
            weekly_review_service=_WeeklyReviewService(),
            trust_service=_TrustService(),
            live_readiness_service=_LiveReadinessService(),
            shadow_paper_validation_service=_ShadowPaperValidationService(),
        )

        with (
            patch("src.services.readiness.launch_readiness_service.get_trading_runtime_service", return_value=_RuntimeService()),
            patch("src.services.readiness.launch_readiness_service.settings.launch_require_real_trade_sample", False),
        ):
            payload = service.build_summary(instrument_id=1, timeframe="1m")

        self.assertTrue(payload["analysis_only_ready"])
        self.assertFalse(payload["real_trade_sample_gate_required"])
        self.assertEqual(payload["real_trade_sample_gate_status"], "deferred_analysis_decision_only")
        self.assertEqual(payload["real_trade_sample_gate_phase"], "deferred_analysis_decision_only")
        self.assertNotIn("real_trade_sample_verified", payload["warning_reasons"])
        self.assertNotIn("real_trade_sample_verified", payload["external_waiting_items"])
        self.assertNotIn("wait_for_and_verify_real_fill_sample", payload["next_actions"])
        self.assertEqual(payload["decision_support_boundary_summary"]["live_execution_boundary"], "live_submit_blocked")
        self.assertIn("decision_support_promotion_summary", payload)
        self.assertIn("shadow_paper_validation_summary", payload)

    def test_live_analysis_assistant_ready_when_account_context_unavailable(self) -> None:
        service = LaunchReadinessService(
            market_state_service=_MarketStateService(),
            broker_read_model_service=_BrokerReadService(status="broker_state_unavailable"),
            weekly_review_service=_WeeklyReviewService(),
            trust_service=_TrustService(),
            live_readiness_service=_LiveReadinessService(),
            shadow_paper_validation_service=_ShadowPaperValidationService(),
        )

        with patch("src.services.readiness.launch_readiness_service.get_trading_runtime_service", return_value=_RuntimeService()):
            payload = service.build_summary(instrument_id=1, timeframe="1m")

        self.assertFalse(payload["analysis_only_ready"])
        self.assertTrue(payload["live_analysis_assistant_ready"])
        self.assertEqual(payload["product_operating_mode"], "readonly_live_analysis")
        self.assertEqual(payload["readonly_account_context_status"], "unavailable")
        self.assertIn("broker_state_ready", payload["blocking_reasons"])
        self.assertIn("account_context_unavailable", payload["live_analysis_assistant_summary"]["warnings"])
        self.assertEqual(payload["live_analysis_assistant_summary"]["blockers"], [])
        self.assertEqual(payload["decision_support_boundary_summary"]["status"], "readonly_live_analysis_available")
        self.assertIn("review_market_context", payload["decision_support_boundary_summary"]["allowed_operator_actions"])
        self.assertIn("submit_live_orders", payload["live_analysis_assistant_summary"]["blocked_actions"])
        self.assertFalse(payload["decision_support_ready"])
        self.assertFalse(payload["execution_assist_ready"])
        self.assertFalse(payload["real_live_ready"])

    def test_live_analysis_allows_rest_fresh_market_when_stream_and_news_are_degraded(self) -> None:
        service = LaunchReadinessService(
            market_state_service=_RestFreshMarketStateService(),
            broker_read_model_service=_BrokerReadService(status="broker_state_unavailable"),
            weekly_review_service=_WeeklyReviewService(),
            trust_service=_TrustService(),
            live_readiness_service=_LiveReadinessService(),
            shadow_paper_validation_service=_ShadowPaperValidationService(),
        )

        with patch("src.services.readiness.launch_readiness_service.get_trading_runtime_service", return_value=_RuntimeServiceWithStreamBlocker()):
            payload = service.build_summary(instrument_id=1, timeframe="1m")

        self.assertFalse(payload["analysis_only_ready"])
        self.assertTrue(payload["live_analysis_assistant_ready"])
        self.assertEqual(payload["product_operating_mode"], "readonly_live_analysis")
        self.assertIn("market_truth_ready", payload["blocking_reasons"])
        self.assertIn("news_provider_usable", payload["blocking_reasons"])
        self.assertIn("runtime_ready", payload["blocking_reasons"])
        self.assertIn("market_stream_not_connected_rest_fallback", payload["live_analysis_assistant_summary"]["warnings"])
        self.assertIn("news_provider_unavailable_context_warning", payload["live_analysis_assistant_summary"]["warnings"])
        self.assertIn("runtime_stream_degraded_rest_fallback", payload["live_analysis_assistant_summary"]["warnings"])
        self.assertEqual(payload["live_analysis_assistant_summary"]["blockers"], [])
        self.assertFalse(payload["decision_support_ready"])
        self.assertFalse(payload["execution_assist_ready"])
        self.assertFalse(payload["real_live_ready"])

    def test_live_analysis_treats_broker_sync_degraded_as_readonly_warning(self) -> None:
        service = LaunchReadinessService(
            market_state_service=_RestFreshMarketStateService(),
            broker_read_model_service=_BrokerReadService(status="broker_state_unavailable"),
            weekly_review_service=_WeeklyReviewService(),
            trust_service=_TrustService(),
            live_readiness_service=_LiveReadinessService(),
            shadow_paper_validation_service=_ShadowPaperValidationService(),
        )

        with patch("src.services.readiness.launch_readiness_service.get_trading_runtime_service", return_value=_RuntimeServiceWithBrokerSyncBlocker()):
            payload = service.build_summary(instrument_id=1, timeframe="1m")

        self.assertTrue(payload["live_analysis_assistant_ready"])
        self.assertEqual(payload["product_operating_mode"], "readonly_live_analysis")
        self.assertIn("runtime_ready", payload["blocking_reasons"])
        self.assertIn("runtime_broker_sync_degraded_readonly_warning", payload["live_analysis_assistant_summary"]["warnings"])
        self.assertIn("account_context_unavailable", payload["live_analysis_assistant_summary"]["warnings"])
        self.assertEqual(payload["live_analysis_assistant_summary"]["blockers"], [])
        self.assertFalse(payload["execution_assist_ready"])
        self.assertFalse(payload["real_live_ready"])

    def test_build_summary_exposes_restricted_soft_rollout_when_only_blacklist_models_exist(self) -> None:
        service = LaunchReadinessService(
            market_state_service=_MarketStateService(),
            broker_read_model_service=_BrokerReadService(),
            weekly_review_service=_WeeklyReviewService(),
            trust_service=_BlacklistOnlyTrustService(),
            live_readiness_service=_LiveReadinessService(),
            shadow_paper_validation_service=_ShadowPaperValidationService(),
        )

        with patch("src.services.readiness.launch_readiness_service.get_trading_runtime_service", return_value=_RuntimeService()):
            payload = service.build_summary(instrument_id=1, timeframe="1m")

        self.assertTrue(payload["analysis_only_ready"])
        self.assertFalse(payload["decision_support_ready"])
        self.assertTrue(payload["restricted_decision_support_ready"])
        self.assertTrue(payload["operator_soft_rollout_ready"])
        self.assertEqual(payload["operator_soft_rollout_mode"], "restricted_decision_support")
        self.assertIn("start_restricted_decision_support_soft_rollout", payload["next_actions"])
        self.assertIn("execution_assist_remains_disabled", payload["operator_soft_rollout_boundary_notes"])
        self.assertNotIn("trust_model_available", payload["warning_reasons"])
        self.assertEqual(payload["decision_support_boundary_summary"]["status"], "restricted_assisted_analysis_available")
        self.assertEqual(payload["decision_support_boundary_summary"]["current_model_boundary"], "observe_only_current_model")
        self.assertEqual(
            payload["decision_support_boundary_summary"]["next_boundary_gate"],
            "promote_model_out_of_observe_only_or_blacklist",
        )
        self.assertEqual(payload["decision_support_promotion_summary"]["status"], "needs_trust_promotion")
        self.assertIn(
            "decision_support_trust_candidate",
            payload["decision_support_promotion_summary"]["failed_criteria"],
        )


class _MarketState:
    freshness = type("Freshness", (), {"code": "market_data_fresh"})()
    stream_state = type("Stream", (), {"code": "market_stream_connected"})()
    data_gap_state = type("Gap", (), {"code": "market_data_gap_clear"})()
    news_window = type("News", (), {"provider_name": "cached", "freshness_code": "news_provider_cached", "error": None})()

    def to_dict(self):
        return {
            "freshness": {"code": "market_data_fresh"},
            "stream_state": {"code": "market_stream_connected"},
            "data_gap_state": {"code": "market_data_gap_clear"},
        }


class _MarketStateService:
    def build_market_state(self, **kwargs):
        return _MarketState()


class _RestFreshMarketState:
    freshness = type("Freshness", (), {"code": "market_data_fresh"})()
    stream_state = type("Stream", (), {"code": "market_stream_connecting"})()
    data_gap_state = type("Gap", (), {"code": "market_data_gap_clear"})()
    news_window = type(
        "News",
        (),
        {"provider_name": "news_provider_chain", "freshness_code": "news_provider_stale", "error": "snapshot_missing"},
    )()

    def to_dict(self):
        return {
            "freshness": {"code": "market_data_fresh"},
            "stream_state": {"code": "market_stream_connecting"},
            "data_gap_state": {"code": "market_data_gap_clear"},
        }


class _RestFreshMarketStateService:
    def build_market_state(self, **kwargs):
        return _RestFreshMarketState()


class _BrokerState:
    def __init__(self, status: str = "broker_state_ok") -> None:
        self.status = status

    def to_dict(self):
        return {"status": self.status}


class _BrokerReadService:
    def __init__(self, status: str = "broker_state_ok") -> None:
        self.status = status

    def fetch_snapshot(self, **kwargs):
        return _BrokerState(self.status)


class _WeeklyReviewService:
    def build_summary(self, **kwargs):
        return {
            "real_trade_count": 0,
            "entry_count": 5,
            "framework_execution_rate": 0.8,
            "risk_violation_count": 0,
        }


class _TrustService:
    def build_summary(self, **kwargs):
        return {
            "models": [
                {
                    "model_code": "OSOK",
                    "classification": "whitelist",
                    "sample_size": 6,
                    "quality_gate_passed": True,
                    "expectancy_r": 0.4,
                    "framework_execution_rate": 0.8,
                    "operator_mode": "trade_inside_framework",
                    "reasons": ["backtest_quality_gate_passed"],
                }
            ],
            "whitelist": ["OSOK"],
            "observe_list": [],
            "blacklist": [],
            "analysis_only_list": ["OSOK"],
            "restricted_decision_support_list": ["OSOK"],
            "decision_support_list": ["OSOK"],
            "execution_candidate_list": ["OSOK"],
        }


class _BlacklistOnlyTrustService:
    def build_summary(self, **kwargs):
        return {
            "models": [
                {
                    "model_code": "SWEEP_MSS_FVG",
                    "classification": "blacklist",
                    "sample_size": 1,
                    "quality_gate_passed": False,
                    "expectancy_r": 0.0,
                    "framework_execution_rate": 0.2,
                    "operator_mode": "observe_only",
                    "reasons": ["backtest_quality_gate_failed"],
                }
            ],
            "whitelist": [],
            "observe_list": [],
            "blacklist": ["SWEEP_MSS_FVG"],
            "analysis_only_list": ["SWEEP_MSS_FVG"],
            "restricted_decision_support_list": ["SWEEP_MSS_FVG"],
            "decision_support_list": [],
            "execution_candidate_list": [],
        }


class _LiveReadinessService:
    def build_summary(self, **kwargs):
        return {"ready_for_real_live": False, "blocking_reasons": ["adapter_real_live_not_ready"]}


class _RuntimeService:
    def status(self):
        return {"scheduler_state": "running", "current_blockers": []}


class _RuntimeServiceWithStreamBlocker:
    def status(self):
        return {
            "scheduler_state": "running",
            "current_blockers": ["market_stream_degraded:BTC-USDT:market_stream_connecting"],
        }


class _RuntimeServiceWithBrokerSyncBlocker:
    def status(self):
        return {
            "scheduler_state": "running",
            "current_blockers": ["broker_sync_degraded:BTC-USDT:okx_read_credentials_missing"],
        }


class _ShadowPaperValidationService:
    def build_summary(self, **kwargs):
        return {
            "status": "collecting_paper_samples",
            "next_validation_gate": "paper_ticket_sample_gate",
            "execution_boundary": "real_live_still_blocked",
            "candidate_count": 3,
            "recent_samples": {
                "candidates": [{"created_ts": "2026-05-18T01:00:00+00:00"}],
                "shadow_observations": [{"observed_ts": "2026-05-18T01:05:00+00:00"}],
            },
        }


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