cyb50-quant/research/dragon/v2/dragon_rule_catalog.py

from __future__ import annotations

from dataclasses import dataclass

from dragon_reason_types import DecisionReason, RuleFamily, RuleLayer


@dataclass(frozen=True)
class _RulePattern:
    match_type: str  # exact or prefix
    pattern: str
    layer: RuleLayer
    family: RuleFamily
    code: str
    tags: tuple[str, ...] = ()

    def matches(self, legacy_reason: str) -> bool:
        if self.match_type == "exact":
            return legacy_reason == self.pattern
        if self.match_type == "prefix":
            return legacy_reason.startswith(self.pattern)
        raise ValueError(f"Unknown match_type: {self.match_type}")


ENTRY_PATTERNS: tuple[_RulePattern, ...] = (
    _RulePattern("exact", "glued_buy", RuleLayer.CORE, RuleFamily.GLUED, "entry_glued_buy"),
    _RulePattern("exact", "early_crash_probe_buy", RuleLayer.CORE, RuleFamily.EARLY_CRASH, "entry_early_crash_probe"),
    _RulePattern("exact", "oversold_recovery_buy", RuleLayer.CORE, RuleFamily.OVERSOLD_RECOVERY, "entry_oversold_recovery"),
    _RulePattern("exact", "dual_gold_resonance_buy", RuleLayer.SECONDARY, RuleFamily.DUAL_GOLD, "entry_dual_gold"),
    _RulePattern("prefix", "deep_oversold_rebound_buy", RuleLayer.SECONDARY, RuleFamily.DEEP_OVERSOLD, "entry_deep_oversold"),
    _RulePattern("exact", "post_sell_rebound_buy", RuleLayer.SECONDARY, RuleFamily.POST_SELL_REBOUND, "entry_post_sell_rebound"),
    _RulePattern("exact", "oversold_reversal_after_ql_buy", RuleLayer.SECONDARY, RuleFamily.OVERSOLD_RECOVERY, "entry_oversold_reversal_after_ql"),
    _RulePattern("prefix", "glued_followthrough_reentry_buy", RuleLayer.BRIDGE, RuleFamily.GLUED, "entry_glued_followthrough_reentry"),
    _RulePattern("exact", "hot_exit_reentry_buy", RuleLayer.BRIDGE, RuleFamily.BRIDGE_REENTRY, "entry_hot_exit_reentry"),
    _RulePattern("exact", "predictive_error_reentry_buy", RuleLayer.BRIDGE, RuleFamily.PREDICTIVE_BREAK, "entry_predictive_error_reentry"),
    _RulePattern("exact", "post_washout_kdj_reentry_buy", RuleLayer.BRIDGE, RuleFamily.BRIDGE_REENTRY, "entry_post_washout_reentry"),
    _RulePattern("exact", "knife_catch_1", RuleLayer.SECONDARY, RuleFamily.DEEP_OVERSOLD, "entry_knife_catch_1"),
    _RulePattern("exact", "knife_catch_2", RuleLayer.SECONDARY, RuleFamily.DEEP_OVERSOLD, "entry_knife_catch_2"),
)


EXIT_PATTERNS: tuple[_RulePattern, ...] = (
    _RulePattern("exact", "predictive_b1_break_exit", RuleLayer.BRIDGE, RuleFamily.PREDICTIVE_BREAK, "exit_predictive_b1_break"),
    _RulePattern("exact", "post_ql_decay_exit", RuleLayer.BRIDGE, RuleFamily.PREDICTIVE_BREAK, "exit_post_ql_decay"),
    _RulePattern("exact", "crash_protection_exit", RuleLayer.CORE, RuleFamily.RISK_MANAGEMENT, "exit_crash_protection"),
    _RulePattern("exact", "prewarning_reduction_exit", RuleLayer.CORE, RuleFamily.RISK_MANAGEMENT, "exit_prewarning_reduction"),
    _RulePattern("exact", "medium_hot_take_profit", RuleLayer.CORE, RuleFamily.HIGH_REGIME, "exit_medium_hot_take_profit"),
    _RulePattern("exact", "ql_high_zone_take_profit", RuleLayer.CORE, RuleFamily.HIGH_REGIME, "exit_ql_high_zone_take_profit"),
    _RulePattern("exact", "ql_mid_zone_take_profit", RuleLayer.CORE, RuleFamily.HIGH_REGIME, "exit_ql_mid_zone_take_profit"),
    _RulePattern("prefix", "high_regime_", RuleLayer.CORE, RuleFamily.HIGH_REGIME, "exit_high_regime"),
    _RulePattern("prefix", "glued_exit:", RuleLayer.CORE, RuleFamily.GLUED, "exit_glued_signal"),
    _RulePattern("prefix", "knife_take_profit_", RuleLayer.CORE, RuleFamily.GLUED, "exit_glued_take_profit"),
    _RulePattern("prefix", "hard_exit:", RuleLayer.CORE, RuleFamily.RISK_MANAGEMENT, "exit_hard_risk"),
    _RulePattern("prefix", "negative_a1_no_b1_recovery:", RuleLayer.SECONDARY, RuleFamily.POST_SELL_REBOUND, "exit_negative_a1_recovery"),
    _RulePattern("prefix", "small_positive_a1_declining:", RuleLayer.SECONDARY, RuleFamily.HIGH_REGIME, "exit_small_positive_decline"),
    _RulePattern("prefix", "low_zone_dual_gold_exit:", RuleLayer.SECONDARY, RuleFamily.DUAL_GOLD, "exit_low_zone_dual_gold"),
    _RulePattern("exact", "early_failed_rebound_exit", RuleLayer.SECONDARY, RuleFamily.POST_SELL_REBOUND, "exit_early_failed_rebound"),
    _RulePattern("exact", "good_to_take_profit_2:kdj_sell", RuleLayer.CORE, RuleFamily.HIGH_REGIME, "exit_good_to_take_profit_2"),
    _RulePattern("prefix", "negative_a1_b1_not_strong:", RuleLayer.SECONDARY, RuleFamily.POST_SELL_REBOUND, "exit_negative_a1_b1_not_strong"),
    _RulePattern("exact", "early_positive_take_profit", RuleLayer.CORE, RuleFamily.HIGH_REGIME, "exit_early_positive_take_profit"),
    _RulePattern("exact", "high_zone_post_ql_fade_exit", RuleLayer.CORE, RuleFamily.HIGH_REGIME, "exit_high_zone_post_ql_fade"),
    _RulePattern("exact", "oversold_rebound_take_profit", RuleLayer.SECONDARY, RuleFamily.DEEP_OVERSOLD, "exit_oversold_rebound_take_profit"),
)


AUX_PATTERNS: tuple[_RulePattern, ...] = (
    _RulePattern("prefix", "bearish_signal_after_exit:", RuleLayer.BRIDGE, RuleFamily.AUXILIARY, "aux_bearish_after_exit"),
    _RulePattern("exact", "bullish_signal_while_holding", RuleLayer.BRIDGE, RuleFamily.AUXILIARY, "aux_bullish_while_holding"),
)


def _classify_with_patterns(
    legacy_reason: str,
    patterns: tuple[_RulePattern, ...],
    default_layer: RuleLayer,
    default_family: RuleFamily,
    default_code: str,
) -> DecisionReason:
    normalized = str(legacy_reason or "").strip()
    for rule in patterns:
        if rule.matches(normalized):
            return DecisionReason(
                code=rule.code,
                layer=rule.layer,
                family=rule.family,
                legacy_reason=normalized,
                tags=rule.tags,
            )
    return DecisionReason(
        code=default_code,
        layer=default_layer,
        family=default_family,
        legacy_reason=normalized,
    )


def classify_entry_reason(legacy_reason: str) -> DecisionReason:
    return _classify_with_patterns(
        legacy_reason=legacy_reason,
        patterns=ENTRY_PATTERNS,
        default_layer=RuleLayer.UNKNOWN,
        default_family=RuleFamily.UNKNOWN,
        default_code="entry_unknown",
    )


def classify_exit_reason(legacy_reason: str) -> DecisionReason:
    return _classify_with_patterns(
        legacy_reason=legacy_reason,
        patterns=EXIT_PATTERNS,
        default_layer=RuleLayer.UNKNOWN,
        default_family=RuleFamily.UNKNOWN,
        default_code="exit_unknown",
    )


def classify_aux_reason(legacy_reason: str) -> DecisionReason:
    return _classify_with_patterns(
        legacy_reason=legacy_reason,
        patterns=AUX_PATTERNS,
        default_layer=RuleLayer.UNKNOWN,
        default_family=RuleFamily.UNKNOWN,
        default_code="aux_unknown",
    )