cyb50-quant/research/dragon/v2/tests/test_no_silent_path_change.py

from __future__ import annotations

import unittest
from pathlib import Path

from dragon_branch_configs import alpha_first_glued_refined_hot_cap_config
from dragon_rc1_golden_baseline import EVENTS_CORE_COLUMNS, TRADES_CORE_COLUMNS, _df_sha256, _load_indicator_snapshot
from dragon_shared import END_DATE, START_DATE
from dragon_rule_engine_v2 import LayeredDragonRuleEngine
from dragon_strategy import DragonRuleEngine


class TestNoSilentPathChange(unittest.TestCase):
    def setUp(self) -> None:
        self.base_dir = Path(__file__).resolve().parents[1]
        self.config = alpha_first_glued_refined_hot_cap_config()

    def test_layered_orchestrator_matches_legacy_core_paths(self) -> None:
        indexed, _ = _load_indicator_snapshot(self.base_dir)

        legacy_events, legacy_trades = DragonRuleEngine(config=self.config).run(indexed)
        layered_events, layered_trades = LayeredDragonRuleEngine(config=self.config).run(indexed)

        legacy_events = legacy_events[(legacy_events["date"] >= START_DATE) & (legacy_events["date"] <= END_DATE)].copy()
        layered_events = layered_events[(layered_events["date"] >= START_DATE) & (layered_events["date"] <= END_DATE)].copy()
        legacy_trades = legacy_trades[
            (legacy_trades["buy_date"] >= START_DATE)
            & (legacy_trades["buy_date"] <= END_DATE)
            & (legacy_trades["sell_date"] >= START_DATE)
            & (legacy_trades["sell_date"] <= END_DATE)
        ].copy()
        layered_trades = layered_trades[
            (layered_trades["buy_date"] >= START_DATE)
            & (layered_trades["buy_date"] <= END_DATE)
            & (layered_trades["sell_date"] >= START_DATE)
            & (layered_trades["sell_date"] <= END_DATE)
        ].copy()

        legacy_events.sort_values(["date", "side", "layer", "reason"], inplace=True)
        layered_events.sort_values(["date", "side", "layer", "reason"], inplace=True)
        legacy_trades.sort_values(["buy_date", "sell_date", "buy_reason", "sell_reason"], inplace=True)
        layered_trades.sort_values(["buy_date", "sell_date", "buy_reason", "sell_reason"], inplace=True)

        self.assertEqual(
            _df_sha256(legacy_events[EVENTS_CORE_COLUMNS]),
            _df_sha256(layered_events[EVENTS_CORE_COLUMNS]),
        )
        self.assertEqual(
            _df_sha256(legacy_trades[TRADES_CORE_COLUMNS]),
            _df_sha256(layered_trades[TRADES_CORE_COLUMNS]),
        )


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