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

from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

import pandas as pd

from dragon_decision_types import StrategyDecision
from dragon_reason_types import RuleLayer
from dragon_rules_bridge import BridgeRuleLayer
from dragon_rules_core import CoreRuleLayer
from dragon_rules_secondary import SecondaryRuleLayer
from dragon_strategy import DragonRuleEngine
from dragon_strategy_config import StrategyConfig


@dataclass(frozen=True)
class LayerRoutingResult:
    decision: StrategyDecision
    routed_layer: str


class LayeredDragonRuleEngine:
    """
    Compatibility-first layered orchestrator.

    The current behavior path is still produced by the legacy DragonRuleEngine.
    This orchestrator adds explicit core -> secondary -> bridge routing semantics
    on top of legacy decisions without changing the trading path.
    """

    def __init__(self, config: Optional[StrategyConfig] = None):
        self.config = config or StrategyConfig()
        self.legacy = DragonRuleEngine(config=self.config)
        self.layers = (
            CoreRuleLayer(),
            SecondaryRuleLayer(),
            BridgeRuleLayer(),
        )

    def _route_decision(self, side: str, event_layer: str, reason: str) -> LayerRoutingResult:
        for layer in self.layers:
            routed = layer.route(side=side, event_layer=event_layer, reason=reason)
            if routed is not None:
                return LayerRoutingResult(decision=routed, routed_layer=layer.layer.value)

        fallback = self.legacy._build_decision(side=side, layer=event_layer, reason=reason)
        return LayerRoutingResult(decision=fallback, routed_layer=RuleLayer.UNKNOWN.value)

    def _enrich_events_with_layer_routing(self, events: pd.DataFrame) -> pd.DataFrame:
        if events.empty:
            return events
        rows = [
            self._route_decision(
                side=str(row["side"]),
                event_layer=str(row["layer"]),
                reason=str(row["reason"]),
            )
            for _, row in events.iterrows()
        ]
        enriched = events.copy()
        enriched["orchestrated_layer"] = [r.routed_layer for r in rows]
        if "reason_layer" not in enriched.columns:
            enriched["reason_layer"] = [r.decision.reason.layer.value if r.decision.reason else "unknown" for r in rows]
            enriched["reason_family"] = [r.decision.reason.family.value if r.decision.reason else "unknown" for r in rows]
            enriched["reason_code"] = [r.decision.reason.code if r.decision.reason else "" for r in rows]
        return enriched

    def run(self, df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
        events, trades = self.legacy.run(df)
        events = self._enrich_events_with_layer_routing(events)
        if not trades.empty and "orchestrated_entry_layer" not in trades.columns:
            trades = trades.copy()
            trades["orchestrated_entry_layer"] = trades.get("buy_reason_layer", pd.Series(["unknown"] * len(trades))).astype(str)
            trades["orchestrated_exit_layer"] = trades.get("sell_reason_layer", pd.Series(["unknown"] * len(trades))).astype(str)
        return events, trades