ai-exchange/tests/unit/test_ict_mainline_taxonomy.py

import unittest

from src.services.ict_mainline import (
    ICT_MAINLINE_CHAIN,
    build_ict_rulebook_mapping,
    build_ict_rulebook_summary,
    build_review_now_mainline_gate,
    classify_book_rule_error,
    classify_ict_concepts,
    get_ict_concept,
    summarize_book_rule_errors,
)


class IctMainlineTaxonomyTests(unittest.TestCase):
    def test_registry_contains_required_book_concepts(self) -> None:
        summary = build_ict_rulebook_summary()

        self.assertEqual(summary["status"], "active")
        for key in (
            "dealing_range",
            "draw_on_liquidity",
            "liquidity_pool",
            "sweep",
            "displacement",
            "mss",
            "fvg",
            "retest",
            "session",
            "backtest_evidence",
            "trading_plan",
            "journal_feedback",
            "no_trade_conditions",
        ):
            with self.subTest(key=key):
                self.assertIn(key, ICT_MAINLINE_CHAIN)
                concept = get_ict_concept(key)
                self.assertTrue(concept["book_anchor"])
                self.assertTrue(concept["required_fields"])
                self.assertTrue(concept["ui_rule"])

    def test_classifies_trading_plan_concepts_from_existing_payload(self) -> None:
        mapping = build_ict_rulebook_mapping(
            "trading_plan",
            {
                "plan_id": 7,
                "narrative": "active:bullish:buy_side:liquidity_pool:accepted_mss:range_available",
                "range_row": {"id": 10, "high": 101.0, "low": 99.0, "equilibrium": 100.0},
                "primary_draw": {"source": "liquidity_pool", "side": "buy_side"},
                "liquidity_pools": [{"pool_type": "equal_highs"}],
                "mss": {"id": 30, "direction": "bullish", "broken_level": 100.0, "accepted": True},
                "fvgs": [{"id": 40, "status": "open"}],
                "allowed_sessions": ["LONDON", "NY_AM"],
                "invalidations": [{"code": "range_low_lost"}],
                "no_trade_conditions": [],
            },
        )

        self.assertEqual(mapping["status"], "mapped")
        for key in (
            "dealing_range",
            "draw_on_liquidity",
            "liquidity_pool",
            "mss",
            "fvg",
            "session",
            "trading_plan",
            "no_trade_conditions",
            "invalidation",
        ):
            self.assertIn(key, mapping["present_concepts"])

    def test_internal_reason_bucket_does_not_pass_without_ict_chain(self) -> None:
        mapping = build_ict_rulebook_mapping("trader_cockpit", {"reason": "sample_size_too_small"})

        self.assertEqual(mapping["status"], "mapped_with_gaps")
        self.assertIn("dealing_range", mapping["missing_concepts"])
        self.assertIn("draw_on_liquidity", mapping["missing_concepts"])
        self.assertIn("unclassified internal reason", mapping["unclassified_internal_reason_policy"])

    def test_strategy_backtest_mapping_uses_quality_gate_evidence(self) -> None:
        mapping = build_ict_rulebook_mapping(
            "strategy_backtest",
            {
                "model_code": "SWEEP_MSS_FVG",
                "timeframe_set": {"session_code": "LONDON"},
                "summary": {
                    "quality_gate": {
                        "passed": False,
                        "reasons": ["sample_size_too_small"],
                        "expectancy_r": -0.2,
                        "max_drawdown_r": 2.4,
                    },
                    "no_trade_conditions": ["quality_gate_failed"],
                },
            },
        )

        self.assertIn("backtest_evidence", mapping["present_concepts"])
        self.assertIn("session", mapping["present_concepts"])
        self.assertIn("mss", mapping["present_concepts"])
        self.assertIn("fvg", mapping["present_concepts"])

    def test_raw_classifier_detects_core_chain_tokens(self) -> None:
        concepts = classify_ict_concepts(
            {
                "setup": "sweep_mss_fvg",
                "missing_confirmations": ["fvg_retrace_confirmed"],
                "why_no_trade": "wait for retest in London session",
            }
        )

        self.assertIn("sweep", concepts)
        self.assertIn("mss", concepts)
        self.assertIn("fvg", concepts)
        self.assertIn("retest", concepts)
        self.assertIn("session", concepts)
        self.assertIn("no_trade_conditions", concepts)

    def test_review_now_mainline_gate_requires_plan_and_backtest(self) -> None:
        blocked = build_review_now_mainline_gate(
            active_trading_plan=None,
            backtest_quality_gate={"passed": False, "reasons": ["sample_size_too_small"]},
        )

        self.assertEqual(blocked["status"], "blocked")
        self.assertFalse(blocked["review_now_allowed"])
        self.assertIn("missing_active_trading_plan", blocked["blockers"])
        self.assertIn("sample_size_too_small", blocked["blockers"])

        passed = build_review_now_mainline_gate(
            active_trading_plan={
                "id": 10,
                "status": "active",
                "setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
                "primary_draw": {"side": "buy_side"},
                "invalidations": [{"code": "range_low_lost"}],
                "no_trade_conditions": [],
            },
            backtest_quality_gate={"passed": True, "reasons": []},
        )

        self.assertEqual(passed["status"], "passed")
        self.assertTrue(passed["review_now_allowed"])

    def test_feedback_actions_map_to_book_rule_error_classes(self) -> None:
        no_trade = classify_book_rule_error(
            {
                "action": "no_trade",
                "body": "No trade because the setup is waiting for MSS/FVG confirmation.",
            },
            source="operator_feedback",
        )

        self.assertEqual(no_trade["review_type"], "observation_review")
        self.assertEqual(no_trade["error_class"], "confirmation_gap")
        self.assertFalse(no_trade["execution_evidence_allowed"])

        paper = classify_book_rule_error(
            {
                "action": "need_more_evidence",
                "paper_ticket_id": 10,
                "body": "Backtest sample size is too small.",
            },
            source="operator_feedback",
        )

        self.assertEqual(paper["review_type"], "paper_review")
        self.assertEqual(paper["error_class"], "backtest_evidence_gap")
        self.assertTrue(paper["paper_evidence_allowed"])

        summary = summarize_book_rule_errors([no_trade, paper])
        self.assertEqual(summary["classification_count"], 2)
        self.assertEqual(summary["review_type_counts"]["observation_review"], 1)
        self.assertEqual(summary["review_type_counts"]["paper_review"], 1)
        self.assertEqual(summary["study_note_draft"]["status"], "draft_ready")


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