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

from __future__ import annotations

import json
import unittest
from pathlib import Path

import pandas as pd

from dragon_branch_configs import (
    alpha_first_glued_followthrough_probe_config,
    alpha_first_glued_refined_hot_cap_config,
)
from dragon_rc1_golden_baseline import (
    EVENTS_CORE_COLUMNS,
    MANIFEST_OUTPUT,
    TRADES_CORE_COLUMNS,
    _df_sha256,
    _load_indicator_snapshot,
)
from dragon_shared import END_DATE, START_DATE
from dragon_strategy import DragonRuleEngine


class TestGluedFollowthroughShadowReentry(unittest.TestCase):
    def setUp(self) -> None:
        self.base_dir = Path(__file__).resolve().parents[1]
        self.manifest_path = self.base_dir / MANIFEST_OUTPUT
        self.manifest = json.loads(self.manifest_path.read_text(encoding="utf-8"))
        indexed, _ = _load_indicator_snapshot(self.base_dir)
        self.indexed = indexed

    def _release_window_trades(self, trades: pd.DataFrame) -> pd.DataFrame:
        return trades[
            (trades["buy_date"] >= START_DATE)
            & (trades["buy_date"] <= END_DATE)
            & (trades["sell_date"] >= START_DATE)
            & (trades["sell_date"] <= END_DATE)
        ].copy()

    def test_default_branch_full_snapshot_hashes_stay_stable(self) -> None:
        events, trades = DragonRuleEngine(config=alpha_first_glued_refined_hot_cap_config()).run(self.indexed)

        events.sort_values(["date", "side", "layer", "reason"], inplace=True)
        trades.sort_values(["buy_date", "sell_date", "buy_reason", "sell_reason"], inplace=True)

        self.assertEqual(len(events), 296)
        self.assertEqual(len(trades), 98)
        self.assertEqual(
            _df_sha256(events[EVENTS_CORE_COLUMNS]),
            "5636adc78212aad094ac57e45d44ecb230e59287a58569d94d2a82ddd148b600",
        )
        self.assertEqual(
            _df_sha256(trades[TRADES_CORE_COLUMNS]),
            "1d419aafd7cbb4a88091b4a1b217d9e3ea806607c73f34650fd3b686790093ff",
        )

    def test_default_branch_release_hashes_stay_stable(self) -> None:
        events, trades = DragonRuleEngine(config=alpha_first_glued_refined_hot_cap_config()).run(self.indexed)
        events = events[(events["date"] >= START_DATE) & (events["date"] <= END_DATE)].copy()
        trades = self._release_window_trades(trades)

        events.sort_values(["date", "side", "layer", "reason"], inplace=True)
        trades.sort_values(["buy_date", "sell_date", "buy_reason", "sell_reason"], inplace=True)

        self.assertEqual(
            _df_sha256(events[EVENTS_CORE_COLUMNS]),
            self.manifest["artifacts"]["events"]["core_sha256"],
        )
        self.assertEqual(
            _df_sha256(trades[TRADES_CORE_COLUMNS]),
            self.manifest["artifacts"]["trades"]["core_sha256"],
        )

    def test_probe_branch_adds_expected_single_trade_fixture(self) -> None:
        _, base_trades = DragonRuleEngine(config=alpha_first_glued_refined_hot_cap_config()).run(self.indexed)
        _, probe_trades = DragonRuleEngine(config=alpha_first_glued_followthrough_probe_config()).run(self.indexed)

        base_trades = self._release_window_trades(base_trades)
        probe_trades = self._release_window_trades(probe_trades)

        base_keys = {
            (row.buy_date, row.sell_date, row.buy_reason, row.sell_reason)
            for row in base_trades.itertuples()
        }
        extra = probe_trades[
            ~probe_trades.apply(
                lambda row: (
                    row["buy_date"],
                    row["sell_date"],
                    row["buy_reason"],
                    row["sell_reason"],
                )
                in base_keys,
                axis=1,
            )
        ].copy()

        self.assertEqual(len(base_trades), 91)
        self.assertEqual(len(probe_trades), 92)
        self.assertEqual(len(extra), 1)

        row = extra.iloc[0]
        self.assertEqual(row["buy_date"], "2020-12-01")
        self.assertEqual(
            row["buy_reason"],
            "glued_followthrough_reentry_buy:confirmed_mid_zone_very_weak_b1",
        )
        self.assertEqual(row["sell_date"], "2020-12-09")
        self.assertEqual(row["sell_reason"], "knife_take_profit_2_glued")
        self.assertEqual(int(row["holding_days"]), 8)
        self.assertAlmostEqual(float(row["return_pct"]), -0.0024350630226196435, places=12)


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