stock/tests/unit/test_decision_fusion.py

"""Unit tests for decision fusion module.

Tests the DecisionFusion, AgentOpinion, and FusionResult classes.
"""

import pytest

from openclaw.fusion.decision_fusion import (
    AgentOpinion,
    AgentRole,
    DecisionFusion,
    FusionConfig,
    FusionResult,
    SignalType,
    WeightedVote,
)


class TestSignalType:
    """Test SignalType enum."""

    def test_signal_values(self):
        """Test signal type values."""
        assert SignalType.STRONG_BUY.value == 2
        assert SignalType.BUY.value == 1
        assert SignalType.HOLD.value == 0
        assert SignalType.SELL.value == -1
        assert SignalType.STRONG_SELL.value == -2


class TestAgentOpinion:
    """Test AgentOpinion dataclass."""

    def test_opinion_creation(self):
        """Test creating an agent opinion."""
        opinion = AgentOpinion(
            agent_id="market-1",
            role=AgentRole.MARKET_ANALYST,
            signal=SignalType.BUY,
            confidence=0.8,
            reasoning="Technical breakout",
            factors=["trend", "volume"],
        )

        assert opinion.agent_id == "market-1"
        assert opinion.role == AgentRole.MARKET_ANALYST
        assert opinion.signal == SignalType.BUY
        assert opinion.confidence == 0.8

    def test_confidence_clamping(self):
        """Test that confidence is clamped to 0-1 range."""
        opinion_high = AgentOpinion(
            agent_id="test",
            role=AgentRole.MARKET_ANALYST,
            signal=SignalType.BUY,
            confidence=1.5,
            reasoning="Test",
        )
        assert opinion_high.confidence == 1.0

        opinion_low = AgentOpinion(
            agent_id="test",
            role=AgentRole.MARKET_ANALYST,
            signal=SignalType.BUY,
            confidence=-0.5,
            reasoning="Test",
        )
        assert opinion_low.confidence == 0.0

    def test_opinion_to_dict(self):
        """Test converting opinion to dictionary."""
        opinion = AgentOpinion(
            agent_id="fund-1",
            role=AgentRole.FUNDAMENTAL_ANALYST,
            signal=SignalType.STRONG_BUY,
            confidence=0.9,
            reasoning="Strong earnings",
        )

        d = opinion.to_dict()
        assert d["agent_id"] == "fund-1"
        assert d["role"] == "fundamental_analyst"
        assert d["signal"] == "STRONG_BUY"
        assert "timestamp" in d


class TestFusionConfig:
    """Test FusionConfig validation."""

    def test_default_weights(self):
        """Test default role weights."""
        config = FusionConfig()

        assert config.role_weights[AgentRole.RISK_MANAGER] == 1.5
        assert config.role_weights[AgentRole.FUNDAMENTAL_ANALYST] == 1.2
        assert config.role_weights[AgentRole.MARKET_ANALYST] == 1.0

    def test_invalid_confidence_threshold(self):
        """Test invalid confidence threshold raises error."""
        with pytest.raises(ValueError):
            FusionConfig(confidence_threshold=1.5)

    def test_invalid_consensus_threshold(self):
        """Test invalid consensus threshold raises error."""
        with pytest.raises(ValueError):
            FusionConfig(consensus_threshold=-0.5)


class TestDecisionFusion:
    """Test DecisionFusion functionality."""

    @pytest.fixture
    def fusion(self):
        """Create a decision fusion instance."""
        config = FusionConfig()
        return DecisionFusion(config)

    def test_start_fusion(self, fusion):
        """Test starting fusion process."""
        fusion.start_fusion("AAPL")
        assert fusion.symbol == "AAPL"

    def test_add_opinion(self, fusion):
        """Test adding opinions."""
        fusion.start_fusion("AAPL")

        opinion = AgentOpinion(
            agent_id="market-1",
            role=AgentRole.MARKET_ANALYST,
            signal=SignalType.BUY,
            confidence=0.8,
            reasoning="Breakout",
        )
        fusion.add_opinion(opinion)

        assert len(fusion._current_opinions) == 1

    def test_fuse_single_opinion(self, fusion):
        """Test fusion with single opinion."""
        fusion.start_fusion("AAPL")
        fusion.add_opinion(
            AgentOpinion(
                agent_id="market-1",
                role=AgentRole.MARKET_ANALYST,
                signal=SignalType.BUY,
                confidence=0.8,
                reasoning="Technical breakout",
            )
        )

        result = fusion.fuse()

        assert result.symbol == "AAPL"
        assert result.final_signal == SignalType.BUY
        assert result.final_confidence > 0

    def test_fuse_multiple_opinions(self, fusion):
        """Test fusion with multiple opinions."""
        fusion.start_fusion("AAPL")

        # Bullish technical
        fusion.add_opinion(
            AgentOpinion(
                agent_id="market-1",
                role=AgentRole.MARKET_ANALYST,
                signal=SignalType.BUY,
                confidence=0.8,
                reasoning="Breakout",
            )
        )

        # Bullish fundamental
        fusion.add_opinion(
            AgentOpinion(
                agent_id="fund-1",
                role=AgentRole.FUNDAMENTAL_ANALYST,
                signal=SignalType.STRONG_BUY,
                confidence=0.9,
                reasoning="Strong earnings",
            )
        )

        # Bearish sentiment
        fusion.add_opinion(
            AgentOpinion(
                agent_id="sent-1",
                role=AgentRole.SENTIMENT_ANALYST,
                signal=SignalType.SELL,
                confidence=0.5,
                reasoning="Negative news",
            )
        )

        result = fusion.fuse()

        # Should be bullish overall due to strong fundamental + technical
        assert result.final_signal.value > 0
        assert len(result.weighted_votes) == 3

    def test_risk_manager_override(self, fusion):
        """Test risk manager override functionality."""
        fusion.config.enable_risk_override = True
        fusion.start_fusion("AAPL")

        # Bullish opinions
        fusion.add_opinion(
            AgentOpinion(
                agent_id="market-1",
                role=AgentRole.MARKET_ANALYST,
                signal=SignalType.STRONG_BUY,
                confidence=0.9,
                reasoning="Perfect setup",
            )
        )

        # Risk manager strongly warns
        fusion.add_opinion(
            AgentOpinion(
                agent_id="risk-1",
                role=AgentRole.RISK_MANAGER,
                signal=SignalType.STRONG_SELL,
                confidence=0.9,  # High confidence triggers override
                reasoning="High volatility, position too large",
            )
        )

        result = fusion.fuse()

        # Risk manager should override
        assert result.final_signal == SignalType.SELL

    def test_consensus_calculation(self, fusion):
        """Test consensus level calculation."""
        fusion.start_fusion("AAPL")

        # All bullish = high consensus
        for i, role in enumerate([AgentRole.MARKET_ANALYST, AgentRole.FUNDAMENTAL_ANALYST]):
            fusion.add_opinion(
                AgentOpinion(
                    agent_id=f"agent-{i}",
                    role=role,
                    signal=SignalType.BUY,
                    confidence=0.8,
                    reasoning="Agreed",
                )
            )

        result = fusion.fuse()

        assert result.consensus_level > 0.5

    def test_low_confidence_filtering(self, fusion):
        """Test that low confidence opinions are filtered."""
        fusion.config.confidence_threshold = 0.5
        fusion.start_fusion("AAPL")

        # High confidence - included
        fusion.add_opinion(
            AgentOpinion(
                agent_id="high",
                role=AgentRole.MARKET_ANALYST,
                signal=SignalType.BUY,
                confidence=0.8,
                reasoning="Strong signal",
            )
        )

        # Low confidence - filtered out
        fusion.add_opinion(
            AgentOpinion(
                agent_id="low",
                role=AgentRole.SENTIMENT_ANALYST,
                signal=SignalType.SELL,
                confidence=0.2,
                reasoning="Weak signal",
            )
        )

        result = fusion.fuse()

        # Only high confidence opinion should be in votes
        assert len(result.weighted_votes) == 1

    def test_supporting_vs_opposing(self, fusion):
        """Test categorization of supporting vs opposing opinions."""
        fusion.start_fusion("AAPL")

        # Supporting buy
        fusion.add_opinion(
            AgentOpinion(
                agent_id="bull-1",
                role=AgentRole.BULL_RESEARCHER,
                signal=SignalType.BUY,
                confidence=0.8,
                reasoning="Growth",
            )
        )

        # Supporting buy
        fusion.add_opinion(
            AgentOpinion(
                agent_id="bull-2",
                role=AgentRole.MARKET_ANALYST,
                signal=SignalType.STRONG_BUY,
                confidence=0.9,
                reasoning="Breakout",
            )
        )

        # Opposing
        fusion.add_opinion(
            AgentOpinion(
                agent_id="bear-1",
                role=AgentRole.BEAR_RESEARCHER,
                signal=SignalType.SELL,
                confidence=0.6,
                reasoning="Overvalued",
            )
        )

        result = fusion.fuse()

        # Should have more supporting than opposing (bullish consensus)
        assert len(result.supporting_opinions) >= len(result.opposing_opinions)


class TestWeightedVote:
    """Test WeightedVote dataclass."""

    def test_weighted_vote_creation(self):
        """Test creating a weighted vote."""
        opinion = AgentOpinion(
            agent_id="test",
            role=AgentRole.MARKET_ANALYST,
            signal=SignalType.BUY,
            confidence=0.8,
            reasoning="Test",
        )

        vote = WeightedVote(
            opinion=opinion,
            weight=1.2,
            weighted_score=0.96,
        )

        assert vote.weight == 1.2
        assert vote.weighted_score == 0.96


class TestFusionResult:
    """Test FusionResult functionality."""

    def test_result_creation(self):
        """Test creating fusion result."""
        result = FusionResult(
            symbol="AAPL",
            final_signal=SignalType.BUY,
            final_confidence=0.75,
            consensus_level=0.8,
        )

        assert result.symbol == "AAPL"
        assert result.final_confidence == 0.75

    def test_get_recommendation_text(self):
        """Test human-readable recommendation."""
        result = FusionResult(
            symbol="AAPL",
            final_signal=SignalType.STRONG_BUY,
            final_confidence=0.9,
            consensus_level=0.8,
        )

        text = result.get_recommendation_text()
        assert "买入" in text or "Buy" in text

    def test_result_to_dict(self):
        """Test converting result to dictionary."""
        result = FusionResult(
            symbol="AAPL",
            final_signal=SignalType.BUY,
            final_confidence=0.75,
            consensus_level=0.8,
        )

        d = result.to_dict()
        assert d["symbol"] == "AAPL"
        assert d["final_signal"] == "BUY"
        assert "timestamp" in d


class TestDecisionFusionHistory:
    """Test fusion history tracking."""

    def test_get_fusion_history(self):
        """Test retrieving fusion history."""
        fusion = DecisionFusion()

        # Run multiple fusions
        for symbol in ["AAPL", "GOOGL"]:
            fusion.start_fusion(symbol)
            fusion.add_opinion(
                AgentOpinion(
                    agent_id="test",
                    role=AgentRole.MARKET_ANALYST,
                    signal=SignalType.BUY,
                    confidence=0.8,
                    reasoning="Test",
                )
            )
            fusion.fuse()

        history = fusion.get_fusion_history()
        assert len(history) == 2

    def test_get_latest_fusion(self):
        """Test retrieving latest fusion for symbol."""
        fusion = DecisionFusion()

        # Two fusions for AAPL
        for _ in range(2):
            fusion.start_fusion("AAPL")
            fusion.add_opinion(
                AgentOpinion(
                    agent_id="test",
                    role=AgentRole.MARKET_ANALYST,
                    signal=SignalType.BUY,
                    confidence=0.8,
                    reasoning="Test",
                )
            )
            fusion.fuse()

        latest = fusion.get_latest_fusion("AAPL")
        assert latest is not None
        assert latest.symbol == "AAPL"


class TestExecutionPlan:
    """Test execution plan generation."""

    def test_strong_signal_plan(self):
        """Test execution plan for strong signals."""
        fusion = DecisionFusion()
        fusion.start_fusion("AAPL")

        # Very high confidence and consensus
        fusion.add_opinion(
            AgentOpinion(
                agent_id="fund",
                role=AgentRole.FUNDAMENTAL_ANALYST,
                signal=SignalType.STRONG_BUY,
                confidence=0.95,
                reasoning="Exceptional",
            )
        )

        result = fusion.fuse()

        assert result.execution_plan["urgency"] == "high"
        assert result.execution_plan["position_size"] == "full"

    def test_weak_signal_plan(self):
        """Test execution plan for weak signals."""
        fusion = DecisionFusion()
        fusion.start_fusion("AAPL")

        # Conflicting signals with low confidence = reduced position
        fusion.add_opinion(
            AgentOpinion(
                agent_id="market",
                role=AgentRole.MARKET_ANALYST,
                signal=SignalType.BUY,
                confidence=0.4,
                reasoning="Weak buy signal",
            )
        )
        fusion.add_opinion(
            AgentOpinion(
                agent_id="sentiment",
                role=AgentRole.SENTIMENT_ANALYST,
                signal=SignalType.SELL,
                confidence=0.4,
                reasoning="Weak sell signal",
            )
        )

        result = fusion.fuse()

        # Conflicting low confidence signals should result in reduced position
        assert result.execution_plan["position_size"] in ["reduced", "standard"]