| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677 |
- 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
|
