ai-exchange/tests/unit/test_shadow_paper_validation_service.py

import unittest
from datetime import datetime, timedelta, timezone

from src.services.readiness.shadow_paper_validation_service import ShadowPaperValidationService


class ShadowPaperValidationServiceTests(unittest.TestCase):
    def test_build_summary_collects_shadow_and_paper_validation_gates(self) -> None:
        now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
        service = ShadowPaperValidationService(repository=_SparseRepository(now))

        summary = service.build_summary(
            instrument_id=1,
            timeframe="1m",
            runtime_status={"current_blockers": []},
            weekly_review={"entry_count": 1, "risk_violation_count": 0, "framework_execution_rate": 0.8},
            now=now,
        )

        self.assertEqual(summary["status"], "collecting_shadow_candidates")
        self.assertEqual(summary["candidate_count"], 3)
        self.assertEqual(summary["paper_ticket_count"], 1)
        self.assertEqual(summary["paper_execution_count"], 1)
        self.assertEqual(summary["operator_feedback_count"], 1)
        self.assertIn("shadow_candidate_day_coverage", summary["failed_criteria"])
        self.assertEqual(summary["next_validation_gate"], "shadow_candidate_day_coverage")
        self.assertEqual(summary["execution_boundary"], "real_live_still_blocked")
        self.assertEqual(len(summary["daily_coverage"]), 14)

    def test_build_summary_marks_ready_when_two_week_gates_are_met(self) -> None:
        now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
        service = ShadowPaperValidationService(repository=_ReadyRepository(now))

        summary = service.build_summary(
            instrument_id=1,
            runtime_status={"current_blockers": []},
            weekly_review={"entry_count": 5, "risk_violation_count": 0, "framework_execution_rate": 0.9},
            now=now,
        )

        self.assertEqual(summary["status"], "ready_for_shadow_paper_review")
        self.assertEqual(summary["failed_criteria"], [])
        self.assertEqual(summary["next_validation_gate"], "shadow_paper_review_ready")
        self.assertGreaterEqual(summary["observed_shadow_day_count"], 10)
        self.assertGreaterEqual(summary["paper_execution_count"], 5)

    def test_runtime_blocker_prevents_shadow_validation(self) -> None:
        now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
        service = ShadowPaperValidationService(repository=_ReadyRepository(now))

        summary = service.build_summary(
            instrument_id=1,
            runtime_status={"current_blockers": ["market_stream_degraded"]},
            weekly_review={"entry_count": 5, "risk_violation_count": 0},
            now=now,
        )

        self.assertEqual(summary["status"], "blocked_by_runtime")
        self.assertEqual(summary["next_validation_gate"], "runtime_clear_for_shadow")

    def test_readonly_rest_fallback_suppresses_market_stream_blocker(self) -> None:
        now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
        service = ShadowPaperValidationService(repository=_ReadyRepository(now))

        summary = service.build_summary(
            instrument_id=1,
            runtime_status={
                "current_blockers": ["market_stream_degraded:BTC-USDT:market_stream_connecting"],
                "latest_market_state_by_symbol": {
                    "BTC-USDT": {
                        "freshness": {"code": "market_data_fresh"},
                        "data_gap_state": {"code": "market_data_gap_clear"},
                    }
                },
            },
            weekly_review={"entry_count": 5, "risk_violation_count": 0},
            now=now,
        )

        self.assertEqual(summary["status"], "ready_for_shadow_paper_review")
        self.assertEqual(summary["runtime_blockers"], [])
        self.assertEqual(
            summary["suppressed_runtime_blockers"],
            ["market_stream_degraded:BTC-USDT:market_stream_connecting"],
        )
        self.assertNotIn("runtime_clear_for_shadow", summary["failed_criteria"])

    def test_daily_observations_count_toward_shadow_coverage_and_operator_feedback(self) -> None:
        now = datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc)
        service = ShadowPaperValidationService(repository=_ObservationOnlyRepository(now))

        summary = service.build_summary(
            instrument_id=1,
            runtime_status={"current_blockers": []},
            weekly_review={"entry_count": 1, "risk_violation_count": 0},
            now=now,
        )

        self.assertEqual(summary["shadow_observation_count"], 4)
        self.assertEqual(summary["shadow_observation_day_count"], 4)
        self.assertEqual(summary["observed_shadow_day_count"], 4)
        self.assertEqual(summary["operator_observation_feedback_count"], 1)
        self.assertEqual(summary["operator_feedback_count"], 1)
        self.assertEqual(summary["daily_coverage"][-1]["shadow_observation_count"], 1)
        self.assertEqual(summary["daily_coverage"][-1]["operator_feedback_count"], 1)


class _SparseRepository:
    def __init__(self, now: datetime) -> None:
        self.now = now

    def fetch_recent_candidates(self, **kwargs):
        return [_row(index=i, created_ts=self.now - timedelta(days=i)) for i in range(3)]

    def fetch_recent_tickets(self, **kwargs):
        return [_row(id=10, execution_mode="paper", created_ts=self.now - timedelta(days=1))]

    def fetch_recent_executions(self, **kwargs):
        return [_row(id=20, execution_mode="paper", opened_ts=self.now - timedelta(days=1))]

    def fetch_recent_journal_entries(self, **kwargs):
        return [_row(id=30, journal_ts=self.now - timedelta(days=1), journal_date=(self.now - timedelta(days=1)).date())]

    def fetch_recent_operator_feedback(self, **kwargs):
        return [_row(id=40, subject_type="alert", label="useful", created_ts=self.now - timedelta(days=1))]


class _ReadyRepository(_SparseRepository):
    def fetch_recent_candidates(self, **kwargs):
        return [_row(id=i, created_ts=self.now - timedelta(days=i % 10)) for i in range(12)]

    def fetch_recent_tickets(self, **kwargs):
        return [_row(id=100 + i, execution_mode="paper", created_ts=self.now - timedelta(days=i)) for i in range(5)]

    def fetch_recent_executions(self, **kwargs):
        return [_row(id=200 + i, execution_mode="paper", opened_ts=self.now - timedelta(days=i)) for i in range(5)]

    def fetch_recent_operator_feedback(self, **kwargs):
        return [_row(id=400 + i, subject_type="alert", label="useful", created_ts=self.now - timedelta(days=i)) for i in range(5)]


class _ObservationOnlyRepository(_SparseRepository):
    def fetch_recent_candidates(self, **kwargs):
        return []

    def fetch_recent_tickets(self, **kwargs):
        return []

    def fetch_recent_executions(self, **kwargs):
        return []

    def fetch_recent_operator_feedback(self, **kwargs):
        return []

    def fetch_recent_observations(self, **kwargs):
        return [
            _row(
                id=i,
                observation_date=(self.now - timedelta(days=i)).date(),
                observed_ts=self.now - timedelta(days=i),
                observation_status="candidate_observed",
                operator_agreement="agree" if i == 0 else "pending",
            )
            for i in range(4)
        ]


def _row(**kwargs):
    return dict(kwargs)


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