Codex
240 lines
11 KiB
Python
240 lines
11 KiB
Python
import unittest
|
|
|
|
from src.services.ict_mainline import (
|
|
DOCTRINE_KERNEL_PHASE,
|
|
DOCTRINE_LIQUIDITY_PHASE,
|
|
DOCTRINE_PLAYBOOK_PHASE,
|
|
build_doctrine_backtest_cohort,
|
|
build_liquidity_backtest_features,
|
|
build_doctrine_playbook_gate,
|
|
build_doctrine_playbook_registry,
|
|
build_doctrine_rule_matrix,
|
|
build_liquidity_kernel_registry,
|
|
evaluate_doctrine_research_status,
|
|
evaluate_liquidity_kernel_status,
|
|
summarize_liquidity_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.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_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",
|
|
"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")
|
|
|
|
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"])
|
|
|
|
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"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|