ai-exchange/tests/unit/test_report_services.py

import contextlib
import io
import json
import os
import unittest

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

from src.services.reports import IntradayReportService, PostTradeReviewService, PreMarketReportService
from scripts import run_reports


class PreMarketReportTests(unittest.TestCase):
    def test_builds_pre_market_plan(self) -> None:
        report = PreMarketReportService().build_report(
            bias_snapshot={
                "daily_bias": "bullish",
                "draw_side": "buy_side",
                "execution_intensity": "high",
                "reason_codes": ["latest_mss", "active_range_available"],
            },
            dealing_range={"id": 1, "high": 110.0, "low": 90.0, "equilibrium": 100.0},
            current_price=98.0,
            session_code="LONDON",
        )

        self.assertEqual(report["report_type"], "pre_market_plan")
        self.assertEqual(report["daily_bias"], "bullish")
        self.assertEqual(report["working_range"]["location"], "discount")
        self.assertTrue(report["execution_permission"])


class IntradayReportTests(unittest.TestCase):
    def test_builds_intraday_explanation_with_richer_structure_and_fvg_state(self) -> None:
        report = IntradayReportService().build_report(
            sweep_event={"side_swept": "buy_side", "close_back_inside": True},
            structure_event={
                "event_type": "MSS",
                "direction": "bearish",
                "accepted": False,
                "meta": {"state": "invalidated", "status_reason": "old_control_reclaimed"},
            },
            fvg_zone={
                "direction": "bearish",
                "status": "mitigated",
                "fill_pct": 1.0,
                "meta": {
                    "status_reason": "mitigation_threshold_reached",
                    "data_quality_warnings": ["candle_gap_detected"],
                },
            },
            trade_signal={
                "status": "awaiting_confirmation",
                "side": "sell",
                "session_code": "NY_AM",
                "invalidations": ["session mismatch"],
                "meta": {
                    "structure_state": "fvg_mitigated",
                    "mss_status_reason": "old_control_reclaimed",
                    "fvg_status_reason": "mitigation_threshold_reached",
                    "fvg_data_quality_warnings": ["candle_gap_detected"],
                },
            },
        )

        self.assertEqual(report["report_type"], "intraday_explanation")
        self.assertFalse(report["signal"]["is_valid"])
        self.assertEqual(report["structure"]["state"], "invalidated")
        self.assertEqual(report["structure"]["status_reason"], "old_control_reclaimed")
        self.assertEqual(report["fvg"]["status"], "mitigated")
        self.assertEqual(report["fvg"]["status_reason"], "mitigation_threshold_reached")
        self.assertEqual(report["signal"]["structure_state"], "fvg_mitigated")
        self.assertEqual(report["signal"]["mss_status_reason"], "old_control_reclaimed")
        self.assertEqual(report["signal"]["fvg_status_reason"], "mitigation_threshold_reached")
        self.assertEqual(report["signal"]["fvg_data_quality_warnings"], ["candle_gap_detected"])
        self.assertIn("fvg_mitigated", report["invalidations"])
        self.assertIn("mss_invalidated", report["invalidations"])
        self.assertIn("candle_gap_detected", report["invalidations"])
        self.assertIn("session_mismatch", report["invalidations"])


class PostTradeReviewTests(unittest.TestCase):
    def test_builds_post_trade_review(self) -> None:
        report = PostTradeReviewService().build_report(
            trade_execution={"status": "rejected", "meta": {"risk_checks": ["session_not_allowed"]}},
            trade_signal={
                "status": "awaiting_context",
                "invalidations": ["session mismatch"],
                "meta": {
                    "structure_state": "mss_invalidated",
                    "mss_state": "invalidated",
                    "mss_status_reason": "old_control_reclaimed",
                    "fvg_status": "touched",
                    "fvg_status_reason": "partial_fill",
                    "fvg_data_quality_warnings": ["candle_gap_detected"],
                },
            },
            backtest_result={"outcome": "loss", "invalidated_before_entry": False, "error_tags": ["same_candle_entry_exit_ambiguous"]},
        )

        self.assertEqual(report["report_type"], "post_trade_review")
        self.assertFalse(report["rule_aligned"])
        self.assertIn("execution_rejected", report["error_types"])
        self.assertIn("session_not_allowed", report["error_types"])
        self.assertEqual(report["backtest_lifecycle_state"], "stopped_out")
        self.assertEqual(report["execution_audit"]["audit_status"], "review")
        self.assertEqual(report["execution_audit"]["lifecycle_state"], "rejected")
        self.assertEqual(report["execution_audit"]["backtest_lifecycle_state"], "stopped_out")
        self.assertEqual(report["signal_context"]["structure_state"], "mss_invalidated")
        self.assertEqual(report["signal_context"]["mss_status_reason"], "old_control_reclaimed")
        self.assertEqual(report["signal_context"]["fvg_status_reason"], "partial_fill")
        self.assertEqual(report["signal_context"]["invalidations_normalized"], ["session_mismatch"])
        self.assertIn("inspect_stop_and_structure_context", report["improvement_notes"])
        self.assertIn("review_signal_structure_state", report["improvement_notes"])
        self.assertIn("review_signal_data_quality_warnings", report["improvement_notes"])
        self.assertIn("review_same_candle_ambiguity_cases", report["improvement_notes"])

    def test_post_trade_review_distinguishes_missed_from_invalidated_before_entry(self) -> None:
        report = PostTradeReviewService().build_report(
            trade_execution={"status": "open", "meta": {"risk_checks": []}},
            trade_signal={"status": "awaiting_confirmation", "invalidations": []},
            backtest_result={"outcome": "missed", "invalidated_before_entry": False, "tp1_hit": False, "tp2_hit": False, "error_tags": ["entry_not_triggered"]},
        )

        self.assertEqual(report["backtest_lifecycle_state"], "missed")
        self.assertIn("review_untriggered_signal_window", report["improvement_notes"])
        self.assertNotIn("avoid_expired_or_invalidated_signals", report["improvement_notes"])


class ReportScriptTests(unittest.TestCase):
    def test_run_reports_prints_backtest_quality_gate(self) -> None:
        output = io.StringIO()

        with contextlib.redirect_stdout(output):
            exit_code = run_reports.main()

        self.assertEqual(exit_code, 0)
        self.assertIn("Backtest quality gate passed: False", output.getvalue())

    def test_run_reports_outputs_json_summary(self) -> None:
        output = io.StringIO()

        with contextlib.redirect_stdout(output):
            exit_code = run_reports.main(env={"REPORTS_OUTPUT_JSON": "1"})

        payload = json.loads(output.getvalue())
        self.assertEqual(exit_code, 0)
        self.assertTrue(payload["passed"])
        self.assertEqual(payload["pre_market"]["report_type"], "pre_market_plan")
        self.assertEqual(payload["intraday"]["report_type"], "intraday_explanation")
        self.assertEqual(payload["post_trade"]["report_type"], "post_trade_review")
        self.assertIn("quality_gate", payload["backtest_summary"])
        self.assertIn("weekly_review", payload)


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