Files
ai-exchange/tests/unit/test_ict_doctrine.py
T
2026-05-22 09:08:04 +08:00

508 lines
23 KiB
Python

import unittest
from src.services.ict_mainline import (
DOCTRINE_KERNEL_PHASE,
DOCTRINE_LIQUIDITY_PHASE,
DOCTRINE_MODEL_PLAYBOOK_PHASE,
DOCTRINE_PLAYBOOK_PHASE,
DOCTRINE_REACTION_PHASE,
DOCTRINE_SESSION_PHASE,
build_doctrine_backtest_cohort,
build_liquidity_backtest_features,
build_model_playbook_backtest_features,
build_pd_array_backtest_features,
build_time_session_backtest_features,
build_doctrine_playbook_gate,
build_doctrine_playbook_registry,
build_doctrine_rule_matrix,
build_liquidity_kernel_registry,
build_model_playbook_kernel_registry,
build_pd_array_reaction_registry,
build_time_session_kernel_registry,
evaluate_doctrine_research_status,
evaluate_liquidity_kernel_status,
evaluate_model_playbook_kernel_status,
evaluate_pd_array_reaction_status,
evaluate_time_session_kernel_status,
summarize_liquidity_kernel_statuses,
summarize_model_playbook_kernel_statuses,
summarize_pd_array_reaction_statuses,
summarize_time_session_kernel_statuses,
summarize_doctrine_rule_gaps,
)
class IctDoctrineKernelTests(unittest.TestCase):
def test_d1_rule_matrix_exposes_required_market_language_chain(self) -> None:
matrix = build_doctrine_rule_matrix()
self.assertEqual(matrix["phase"], DOCTRINE_KERNEL_PHASE)
self.assertEqual(matrix["status"], "active")
self.assertIn("Range", matrix["required_chain"])
self.assertIn("Draw", matrix["required_chain"])
self.assertIn("Invalidation", matrix["required_chain"])
self.assertIn("read_only_analysis_no_order_submission", matrix["execution_boundary"])
self.assertGreaterEqual(len(matrix["rules"]), 7)
self.assertEqual(matrix["rules"][0]["rule_id"], "range_defined")
def test_d2_playbook_registry_exposes_first_real_setup_schema(self) -> None:
registry = build_doctrine_playbook_registry()
self.assertEqual(registry["phase"], DOCTRINE_PLAYBOOK_PHASE)
self.assertEqual(registry["playbook_count"], 1)
playbook = registry["playbooks"][0]
self.assertEqual(playbook["setup_playbook_id"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL")
self.assertIn("range_definition", playbook)
self.assertIn("target_rule", playbook)
self.assertIn("liquidity_target_side", playbook["liquidity_event_rule"]["required_fields"])
self.assertIn("post_take_reaction", playbook["liquidity_event_rule"]["required_fields"])
self.assertIn("pd_array_role", playbook["pd_array_rule"]["required_fields"])
self.assertIn("reaction_quality", playbook["pd_array_rule"]["required_fields"])
self.assertIn("killzone_state", playbook["session_timing_rule"]["required_fields"])
self.assertIn("weekly_profile", playbook["session_timing_rule"]["required_fields"])
self.assertTrue(playbook["promotion_gate"]["requires_backtest_quality_gate"])
def test_d3_liquidity_kernel_registry_exposes_target_ranking_rules(self) -> None:
registry = build_liquidity_kernel_registry()
self.assertEqual(registry["phase"], DOCTRINE_LIQUIDITY_PHASE)
self.assertEqual(registry["status"], "active")
self.assertGreaterEqual(registry["rule_count"], 6)
self.assertIn("liquidity_target_side", registry["required_backtest_feature_fields"])
self.assertIn("Stop Run / Sweep / Raid is an evidence component only, never a setup by itself.", registry["hard_rules"])
def test_sweep_without_ranked_target_stays_study_only(self) -> None:
status = evaluate_liquidity_kernel_status(
{
"liquidity_event": "sweep",
"swept_level": 100.5,
"sweep_confirmed": True,
}
)
self.assertFalse(status["passed"])
self.assertEqual(status["phase"], DOCTRINE_LIQUIDITY_PHASE)
self.assertEqual(status["status"], "study_only")
self.assertIn("missing_liquidity_target", status["reasons"])
self.assertIn("missing_liquidity_pool", status["reasons"])
self.assertIn("sweep_is_not_setup", status["reasons"])
def test_complete_liquidity_kernel_payload_maps_without_execution_permission(self) -> None:
status = evaluate_liquidity_kernel_status(
{
"liquidity_target_side": "sell_side_liquidity",
"working_draw": "lower_external_pool",
"secondary_draw": "range_midpoint",
"observation_draw": "upper_bsl",
"liquidity_pool_id": "ssl-ny-low",
"liquidity_pool_rank": "primary",
"liquidity_event": "liquidity_sweep",
"swept_level": 99.5,
"sweep_confirmed": True,
"raid_stop_run_kind": "sweep",
"post_take_reaction": "reclaim_displacement",
"manipulation_window_state": "confirmed",
"reorder_trigger_type": "",
}
)
self.assertTrue(status["passed"])
self.assertEqual(status["status"], "liquidity_mapped")
self.assertEqual(status["execution_boundary"], "read_only_analysis_no_order_submission")
def test_liquidity_summary_counts_blockers(self) -> None:
summary = summarize_liquidity_kernel_statuses(
[
{"liquidity_event": "sweep", "sweep_confirmed": True},
{
"liquidity_target_side": "buy_side_liquidity",
"working_draw": "upper_external_pool",
"liquidity_pool_id": "bsl-old-high",
"liquidity_pool_rank": "primary",
"liquidity_event": "raid",
"swept_level": 105.0,
"post_take_reaction": "acceptance_displacement",
},
]
)
self.assertEqual(summary["item_count"], 2)
self.assertEqual(summary["passed_count"], 1)
self.assertGreater(summary["reason_counts"]["sweep_is_not_setup"], 0)
self.assertEqual(summary["target_side_counts"]["buy_side_liquidity"], 1)
def test_d4_pd_array_reaction_registry_exposes_location_quality_rules(self) -> None:
registry = build_pd_array_reaction_registry()
self.assertEqual(registry["phase"], DOCTRINE_REACTION_PHASE)
self.assertEqual(registry["status"], "active")
self.assertGreaterEqual(registry["rule_count"], 5)
self.assertIn("pd_array_role", registry["required_backtest_feature_fields"])
self.assertIn("PD Array is reaction-location quality, not a direction source.", registry["hard_rules"])
def test_fvg_label_without_reaction_context_stays_study_only(self) -> None:
status = evaluate_pd_array_reaction_status(
{
"pd_array_type": "fvg",
"fvg_retrace_confirmed": True,
}
)
self.assertFalse(status["passed"])
self.assertEqual(status["phase"], DOCTRINE_REACTION_PHASE)
self.assertEqual(status["status"], "study_only")
self.assertIn("missing_pd_array_role", status["reasons"])
self.assertIn("missing_array_location", status["reasons"])
self.assertIn("fvg_is_not_setup", status["reasons"])
def test_complete_pd_array_reaction_payload_maps_without_execution_permission(self) -> None:
status = evaluate_pd_array_reaction_status(_complete_pd_array_payload())
self.assertTrue(status["passed"])
self.assertEqual(status["status"], "pd_array_mapped")
self.assertEqual(status["execution_boundary"], "read_only_analysis_no_order_submission")
def test_pd_array_summary_counts_blockers(self) -> None:
summary = summarize_pd_array_reaction_statuses(
[
{"pd_array_type": "fvg", "fvg_retrace_confirmed": True},
_complete_pd_array_payload(),
]
)
self.assertEqual(summary["item_count"], 2)
self.assertEqual(summary["passed_count"], 1)
self.assertGreater(summary["reason_counts"]["fvg_is_not_setup"], 0)
self.assertEqual(summary["array_type_counts"]["fvg"], 2)
def test_d5_time_session_kernel_registry_exposes_session_quality_rules(self) -> None:
registry = build_time_session_kernel_registry()
self.assertEqual(registry["phase"], DOCTRINE_SESSION_PHASE)
self.assertEqual(registry["status"], "active")
self.assertGreaterEqual(registry["rule_count"], 6)
self.assertIn("session_window_state", registry["required_backtest_feature_fields"])
self.assertIn("Asia / London / NY AM / NY PM are context windows, not entry signals.", registry["hard_rules"])
def test_outside_session_stays_study_only(self) -> None:
status = evaluate_time_session_kernel_status(
{
"session_code": "OFF_HOURS",
"session_window_state": "outside_window",
"allowed_sessions": ["LONDON"],
}
)
self.assertFalse(status["passed"])
self.assertEqual(status["phase"], DOCTRINE_SESSION_PHASE)
self.assertEqual(status["status"], "study_only")
self.assertIn("outside_session_window", status["reasons"])
self.assertIn("session_not_allowed", status["reasons"])
def test_complete_time_session_payload_maps_without_execution_permission(self) -> None:
status = evaluate_time_session_kernel_status(_complete_time_session_payload())
self.assertTrue(status["passed"])
self.assertEqual(status["status"], "time_session_mapped")
self.assertEqual(status["execution_boundary"], "read_only_analysis_no_order_submission")
def test_time_session_summary_counts_blockers(self) -> None:
summary = summarize_time_session_kernel_statuses(
[
{"session_code": "OFF_HOURS", "session_window_state": "outside_window"},
_complete_time_session_payload(),
]
)
self.assertEqual(summary["item_count"], 2)
self.assertEqual(summary["passed_count"], 1)
self.assertGreater(summary["reason_counts"]["outside_session_window"], 0)
self.assertEqual(summary["session_code_counts"]["LONDON"], 1)
def test_d6_model_playbook_kernel_registry_exposes_full_chain_contracts(self) -> None:
registry = build_model_playbook_kernel_registry()
self.assertEqual(registry["phase"], DOCTRINE_MODEL_PLAYBOOK_PHASE)
self.assertEqual(registry["status"], "active")
self.assertGreaterEqual(registry["rule_count"], 6)
self.assertIn("model_playbook_id", registry["required_backtest_feature_fields"])
self.assertIn("Raw model_code is not a tradable setup.", registry["hard_rules"])
def test_raw_model_without_playbook_stays_study_only(self) -> None:
status = evaluate_model_playbook_kernel_status(
{"model_code": "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE"}
)
self.assertFalse(status["passed"])
self.assertEqual(status["phase"], DOCTRINE_MODEL_PLAYBOOK_PHASE)
self.assertEqual(status["status"], "study_only")
self.assertIn("missing_model_playbook", status["reasons"])
self.assertIn("missing_model_range_contract", status["reasons"])
self.assertIn("missing_model_backtest_cohort", status["reasons"])
def test_registered_model_playbook_maps_without_execution_permission(self) -> None:
status = evaluate_model_playbook_kernel_status(
{"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"}
)
self.assertTrue(status["passed"])
self.assertEqual(status["status"], "model_playbook_mapped")
self.assertEqual(status["features"]["model_family"], "london_discount_reversal")
self.assertIn("range_id", status["features"]["range_contract"])
self.assertEqual(status["execution_boundary"], "read_only_analysis_no_order_submission")
def test_model_playbook_summary_counts_blockers(self) -> None:
summary = summarize_model_playbook_kernel_statuses(
[
{"model_code": "SWEEP_MSS_FVG"},
{"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"},
]
)
self.assertEqual(summary["item_count"], 2)
self.assertEqual(summary["passed_count"], 1)
self.assertGreater(summary["reason_counts"]["missing_model_playbook"], 0)
self.assertEqual(summary["playbook_counts"]["ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"], 1)
def test_playbook_gate_blocks_missing_and_passes_registered_schema(self) -> None:
missing = build_doctrine_playbook_gate({"model_code": "SWEEP_MSS_FVG"})
self.assertFalse(missing["passed"])
self.assertEqual(missing["reason"], "missing_doctrine_setup_playbook")
passed = build_doctrine_playbook_gate(
{"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"}
)
self.assertTrue(passed["passed"])
self.assertEqual(passed["schema_completeness"], "complete")
def test_surface_shape_candidate_without_doctrine_playbook_is_research_only(self) -> None:
status = evaluate_doctrine_research_status(
{
"model_code": "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE",
"missing_confirmations": ["fvg_retrace_confirmed"],
}
)
self.assertTrue(status["research_only"])
self.assertFalse(status["can_promote_to_cockpit"])
self.assertEqual(status["research_only_reason"], "missing_doctrine_setup_playbook")
self.assertIn("doctrine_missing_draw", status["research_only_reasons"])
self.assertIn("doctrine_missing_invalidation", status["research_only_reasons"])
def test_complete_doctrine_payload_can_be_mapped_but_not_executed(self) -> None:
status = evaluate_doctrine_research_status(
{
"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
"range_id": "range-1",
"range_high": 105.0,
"range_low": 95.0,
"equilibrium": 100.0,
"range_timeframe": "1h",
"primary_draw": "sell_side_liquidity",
"draw_side": "sell_side",
"objective_price": 96.0,
"premium_discount_zone": "discount",
"location_advantage": "discount_reclaim",
"context": "daily range expansion",
"framework": "London killzone reversal",
"reference_point": "range_low_sweep",
"setup_type": "range_sweep_mss_fvg_retrace",
"session_code": "LONDON",
"allowed_sessions": ["LONDON"],
"path_context": "sweep_then_displacement_then_retrace",
"sequence_validity": "sweep_mss_fvg_retrace",
"invalidations": ["range_low_reclaim_failed"],
"stop_loss": 94.5,
"invalid_level": 94.5,
"target_plan": {"tp1": 100.0, "tp2": 104.0, "irl": "near_fvg", "erl": "range_high"},
}
)
self.assertFalse(status["research_only"])
self.assertEqual(status["status"], "doctrine_mapped")
self.assertTrue(status["can_promote_to_cockpit"])
self.assertEqual(status["execution_boundary"], "read_only_analysis_no_order_submission")
def test_doctrine_gap_summary_counts_missing_book_fields(self) -> None:
summary = summarize_doctrine_rule_gaps(
[
{"model_code": "SWEEP_MSS_FVG"},
{"model_code": "SWEEP_MSS_FVG", "range_id": "range-1", "equilibrium": 100.0},
]
)
self.assertEqual(summary["item_count"], 2)
self.assertEqual(summary["research_only_count"], 2)
self.assertEqual(summary["promotable_count"], 0)
self.assertEqual(summary["research_only_reason_counts"]["missing_doctrine_setup_playbook"], 2)
self.assertGreater(summary["missing_rule_counts"]["draw_defined"], 0)
def test_doctrine_backtest_cohort_uses_playbook_and_completeness(self) -> None:
cohort = build_doctrine_backtest_cohort(
{
"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
"liquidity_target_side": "sell_side_liquidity",
"working_draw": "lower_external_pool",
"liquidity_pool_id": "ssl-ny-low",
"liquidity_pool_rank": "primary",
"liquidity_event": "sweep",
"swept_level": 99.5,
"sweep_confirmed": True,
"post_take_reaction": "reclaim_displacement",
**_complete_pd_array_payload(),
**_complete_time_session_payload(),
"summary": {"quality_gate": {"passed": True, "reasons": []}},
}
)
self.assertEqual(cohort["cohort_key"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete")
self.assertEqual(cohort["doctrine_completeness"], "complete")
self.assertTrue(cohort["promotion_gate"]["passed"])
self.assertEqual(cohort["liquidity_kernel_status"]["status"], "liquidity_mapped")
self.assertEqual(cohort["pd_array_reaction_status"]["status"], "pd_array_mapped")
self.assertEqual(cohort["time_session_kernel_status"]["status"], "time_session_mapped")
self.assertEqual(cohort["model_playbook_kernel_status"]["status"], "model_playbook_mapped")
self.assertTrue(cohort["promotion_gate"]["requires_model_playbook_kernel_complete"])
missing = build_doctrine_backtest_cohort({"model_code": "SWEEP_MSS_FVG"})
self.assertEqual(missing["cohort_key"], "missing_playbook::incomplete")
self.assertIn("missing_doctrine_setup_playbook", missing["promotion_gate"]["reasons"])
self.assertIn("missing_liquidity_target", missing["promotion_gate"]["reasons"])
self.assertIn("missing_pd_array", missing["promotion_gate"]["reasons"])
self.assertIn("missing_session_window", missing["promotion_gate"]["reasons"])
self.assertIn("missing_model_playbook", missing["promotion_gate"]["reasons"])
def test_liquidity_backtest_features_extract_nested_candidate_fields(self) -> None:
features = build_liquidity_backtest_features(
{
"candidate": {
"entry": {
"liquidity_target_side": "buy_side_liquidity",
"liquidity_pool_id": "bsl-old-high",
},
"confirmation_state": {"post_take_reaction": "acceptance_displacement"},
},
"sweep_confirmed": False,
"liquidity_pool_rank": "primary",
}
)
self.assertEqual(features["liquidity_target_side"], "buy_side_liquidity")
self.assertEqual(features["liquidity_pool_id"], "bsl-old-high")
self.assertEqual(features["liquidity_pool_rank"], "primary")
self.assertFalse(features["sweep_confirmed"])
def test_pd_array_backtest_features_extract_nested_candidate_fields(self) -> None:
features = build_pd_array_backtest_features(
{
"candidate": {
"entry": {
"pd_array_type": "fvg",
"pd_array_id": "fvg-1",
"pd_array_role": "entry",
},
"confirmation_state": {"reaction_quality": "rejection_confirmed"},
},
"pd_array_rank": "primary",
"array_location": "discount",
}
)
self.assertEqual(features["pd_array_type"], "fvg")
self.assertEqual(features["pd_array_id"], "fvg-1")
self.assertEqual(features["pd_array_role"], "entry")
self.assertEqual(features["pd_array_rank"], "primary")
self.assertEqual(features["array_location"], "discount")
def test_time_session_backtest_features_extract_nested_candidate_fields(self) -> None:
features = build_time_session_backtest_features(
{
"candidate": {
"session": {
"session_code": "NY_AM",
"killzone_name": "NY_AM_KILLZONE",
"opening_range_state": "defined",
},
"confirmation_state": {"session_invalidation": "ny_lunch_cutoff"},
},
"session_filter": "covered",
"weekly_profile": "tuesday_expansion",
}
)
self.assertEqual(features["session_code"], "NY_AM")
self.assertEqual(features["session_window_state"], "inside")
self.assertEqual(features["killzone_name"], "NY_AM_KILLZONE")
self.assertEqual(features["opening_range_state"], "defined")
def test_model_playbook_features_extract_nested_playbook_fields(self) -> None:
features = build_model_playbook_backtest_features(
{
"candidate": {
"entry": {
"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
}
}
}
)
self.assertEqual(features["model_playbook_id"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL")
self.assertEqual(features["model_family"], "london_discount_reversal")
self.assertIn("liquidity_target_side", features["liquidity_contract"])
def _complete_pd_array_payload() -> dict:
return {
"pd_array_type": "fvg",
"pd_array_id": "fvg-1",
"pd_array_role": "entry",
"pd_array_rank": "primary",
"array_location": "discount",
"array_timeframe": "5m",
"array_freshness": "fresh",
"formation_event": "post_sweep_displacement",
"displacement_origin": "post_sweep_displacement",
"fvg_role": "entry",
"fvg_retrace_confirmed": True,
"reaction_quality": "rejection_confirmed",
"retest_confirmed": True,
"pd_array_invalidation": "fvg_accepted_other_side",
}
def _complete_time_session_payload() -> dict:
return {
"session_code": "LONDON",
"session_window_state": "inside",
"allowed_sessions": ["LONDON", "NY_AM"],
"session_start_ts": "2026-05-22T06:00:00Z",
"session_end_ts": "2026-05-22T09:00:00Z",
"killzone_name": "LONDON_KILLZONE",
"killzone_state": "active",
"killzone_start_ts": "2026-05-22T07:00:00Z",
"killzone_end_ts": "2026-05-22T09:00:00Z",
"judas_window_state": "cleared",
"inducement_side": "sell_side",
"post_manipulation_confirmation": "mss_after_sweep",
"opening_range_state": "defined",
"opening_range_id": "or-london-20260522",
"opening_range_high": 101.0,
"opening_range_low": 99.0,
"opening_range_reaction": "rejected_low",
"weekly_profile": "midweek_expansion",
"weekly_bias": "bearish_draw_to_ssl",
"day_of_week": "Friday",
"weekly_range_state": "inside_weekly_range",
"news_window_state": "clear",
"high_impact_news": False,
"low_liquidity_window": False,
"session_invalidation": "london_close_or_reclaim",
}
if __name__ == "__main__":
unittest.main()