cyb50-quant/cyb50-pro/agents/coordinator.py

"""
智能体协同机制

管理多智能体协同，处理信号冲突和叠加
"""

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Any
from datetime import datetime
from enum import Enum
import uuid

import pandas as pd

from agents.base import AgentBase, AgentSignal, SignalDirection


class ConflictResolutionMethod(Enum):
    """冲突解决方式"""
    CONFIDENCE_WEIGHTED = "confidence_weighted"
    HIGHER_WINS = "higher_wins"
    CANCEL = "cancel"


@dataclass
class CoordinatedSignal:
    """协同后的信号"""
    original_signals: List[AgentSignal]
    final_direction: SignalDirection
    final_strength: float  # 0-1
    final_position: float  # 0-1
    coordination_type: str  # "conflict_resolved", "reinforced", "single"
    reasoning: str
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ReinforcementResult:
    """信号叠加增强结果"""
    is_reinforced: bool
    direction: SignalDirection
    boost_factor: float
    original_count: int
    avg_confidence: float


class AgentCoordinator:
    """
    智能体协同器

    功能：
    1. 信号冲突处理：当智能体生成反向信号时，比较置信度处理
    2. 信号叠加增强：当多个智能体生成同向信号时，提升仓位上限
    """

    def __init__(
        self,
        conflict_resolution: ConflictResolutionMethod = ConflictResolutionMethod.CONFIDENCE_WEIGHTED,
        min_confidence_diff: float = 0.2,
        reinforcement_threshold: int = 3,
        position_boost: float = 0.2,
        stop_tighten: float = 0.5,
        max_position_boost: float = 0.3
    ):
        self.conflict_resolution = conflict_resolution
        self.min_confidence_diff = min_confidence_diff
        self.reinforcement_threshold = reinforcement_threshold
        self.position_boost = position_boost
        self.stop_tighten = stop_tighten
        self.max_position_boost = max_position_boost

        self.coordination_history: List[CoordinatedSignal] = []

    def coordinate(
        self,
        signals: Dict[str, AgentSignal],
        weights: Dict[str, float]
    ) -> Optional[CoordinatedSignal]:
        """
        协同多个智能体的信号

        Args:
            signals: 各智能体的信号
            weights: 各智能体的权重

        Returns:
            CoordinatedSignal: 协同后的信号
        """
        if not signals:
            return None

        # 单信号直接返回
        if len(signals) == 1:
            signal = list(signals.values())[0]
            return CoordinatedSignal(
                original_signals=[signal],
                final_direction=signal.direction,
                final_strength=signal.confidence,
                final_position=signal.suggested_position,
                coordination_type="single",
                reasoning="单信号，无需协同"
            )

        # 分组：多头、空头、中性
        long_signals = []
        short_signals = []
        neutral_signals = []

        for name, signal in signals.items():
            if signal.direction == SignalDirection.LONG:
                long_signals.append((name, signal))
            elif signal.direction == SignalDirection.SHORT:
                short_signals.append((name, signal))
            else:
                neutral_signals.append((name, signal))

        # 情况1: 只有同向信号 - 叠加增强
        if long_signals and not short_signals:
            return self._reinforce_signals(long_signals, weights, SignalDirection.LONG)

        if short_signals and not long_signals:
            return self._reinforce_signals(short_signals, weights, SignalDirection.SHORT)

        # 情况2: 有反向信号 - 冲突处理
        if long_signals and short_signals:
            return self._resolve_conflict(long_signals, short_signals, weights)

        # 情况3: 只有中性信号
        if neutral_signals:
            avg_confidence = sum(s.confidence for _, s in neutral_signals) / len(neutral_signals)
            return CoordinatedSignal(
                original_signals=[s for _, s in neutral_signals],
                final_direction=SignalDirection.NEUTRAL,
                final_strength=avg_confidence,
                final_position=0.0,
                coordination_type="conflict_resolved",
                reasoning="所有智能体均输出中性信号，观望"
            )

        return None

    def _reinforce_signals(
        self,
        signals: List[Tuple[str, AgentSignal]],
        weights: Dict[str, float],
        direction: SignalDirection
    ) -> CoordinatedSignal:
        """
        信号叠加增强

        当多个智能体生成同向信号时：
        1. 提升该方向仓位上限
        2. 收紧止损
        """
        if len(signals) < self.reinforcement_threshold:
            # 未达到增强阈值，正常加权
            weighted_position = sum(
                signal.suggested_position * weights.get(name, 1/len(signals))
                for name, signal in signals
            )
            avg_confidence = sum(s.confidence for _, s in signals) / len(signals)

            coordinated = CoordinatedSignal(
                original_signals=[s for _, s in signals],
                final_direction=direction,
                final_strength=avg_confidence,
                final_position=min(1.0, weighted_position),
                coordination_type="reinforced",
                reasoning=f"{len(signals)}个智能体同向，未达增强阈值"
            )
        else:
            # 达到增强阈值，提升仓位
            base_position = sum(
                signal.suggested_position * weights.get(name, 1/len(signals))
                for name, signal in signals
            )

            # 计算增强因子
            boost = min(
                self.max_position_boost,
                self.position_boost * (len(signals) - self.reinforcement_threshold + 1)
            )

            enhanced_position = min(1.0, base_position * (1 + boost))
            avg_confidence = sum(s.confidence for _, s in signals) / len(signals)

            coordinated = CoordinatedSignal(
                original_signals=[s for _, s in signals],
                final_direction=direction,
                final_strength=min(1.0, avg_confidence * 1.1),  # 置信度小幅提升
                final_position=enhanced_position,
                coordination_type="reinforced",
                reasoning=f"{len(signals)}个智能体同向，仓位提升{boost:.1%}，止损收紧{self.stop_tighten:.1%}"
            )

        self.coordination_history.append(coordinated)
        return coordinated

    def _resolve_conflict(
        self,
        long_signals: List[Tuple[str, AgentSignal]],
        short_signals: List[Tuple[str, AgentSignal]],
        weights: Dict[str, float]
    ) -> CoordinatedSignal:
        """
        信号冲突处理

        当存在反向信号时：
        1. 比较双方加权置信度
        2. 根据配置的策略处理冲突
        """
        # 计算各方加权力量和
        long_strength = sum(
            signal.confidence * weights.get(name, 0.5)
            for name, signal in long_signals
        )

        short_strength = sum(
            signal.confidence * weights.get(name, 0.5)
            for name, signal in short_signals
        )

        # 计算数量
        long_count = len(long_signals)
        short_count = len(short_signals)

        all_signals = [s for _, s in long_signals] + [s for _, s in short_signals]

        # 冲突解决策略
        if self.conflict_resolution == ConflictResolutionMethod.CANCEL:
            # 取消所有信号，观望
            return CoordinatedSignal(
                original_signals=all_signals,
                final_direction=SignalDirection.NEUTRAL,
                final_strength=0.0,
                final_position=0.0,
                coordination_type="conflict_resolved",
                reasoning=f"信号冲突（多:{long_count} vs 空:{short_count}），取消所有信号观望"
            )

        elif self.conflict_resolution == ConflictResolutionMethod.HIGHER_WINS:
            # 较强方获胜
            if long_strength > short_strength:
                winner_signals = long_signals
                winner_direction = SignalDirection.LONG
            else:
                winner_signals = short_signals
                winner_direction = SignalDirection.SHORT

            avg_confidence = sum(s.confidence for _, s in winner_signals) / len(winner_signals)
            weighted_position = sum(
                signal.suggested_position * weights.get(name, 1/len(winner_signals))
                for name, signal in winner_signals
            )

            return CoordinatedSignal(
                original_signals=all_signals,
                final_direction=winner_direction,
                final_strength=avg_confidence,
                final_position=weighted_position,
                coordination_type="conflict_resolved",
                reasoning=f"信号冲突，{winner_direction.value}方获胜（力量比{long_strength:.2f}:{short_strength:.2f}）"
            )

        else:  # CONFIDENCE_WEIGHTED
            # 按置信度加权执行双方
            total_strength = long_strength + short_strength

            if total_strength == 0:
                return CoordinatedSignal(
                    original_signals=all_signals,
                    final_direction=SignalDirection.NEUTRAL,
                    final_strength=0.0,
                    final_position=0.0,
                    coordination_type="conflict_resolved",
                    reasoning="信号冲突，双方力量均为0，观望"
                )

            long_ratio = long_strength / total_strength
            short_ratio = short_strength / total_strength

            # 净仓位
            net_position = long_ratio - short_ratio

            if abs(net_position) < self.min_confidence_diff:
                return CoordinatedSignal(
                    original_signals=all_signals,
                    final_direction=SignalDirection.NEUTRAL,
                    final_strength=abs(net_position),
                    final_position=0.0,
                    coordination_type="conflict_resolved",
                    reasoning=f"信号冲突，净仓位{net_position:.2f}低于阈值，观望"
                )

            # 确定方向和仓位
            if net_position > 0:
                direction = SignalDirection.LONG
                position = min(1.0, net_position)
            else:
                direction = SignalDirection.SHORT
                position = min(1.0, abs(net_position))

            # 综合置信度
            avg_confidence = sum(s.confidence for s in all_signals) / len(all_signals)

            return CoordinatedSignal(
                original_signals=all_signals,
                final_direction=direction,
                final_strength=avg_confidence,
                final_position=position,
                coordination_type="conflict_resolved",
                reasoning=f"信号冲突，置信度加权结果：{direction.value}，仓位{position:.2f}（多{long_ratio:.2f} vs 空{short_ratio:.2f}）"
            )

    def get_coordination_summary(self) -> Dict[str, Any]:
        """获取协同历史摘要"""
        if not self.coordination_history:
            return {}

        total = len(self.coordination_history)

        type_counts = {}
        for coord in self.coordination_history:
            type_counts[coord.coordination_type] = type_counts.get(coord.coordination_type, 0) + 1

        return {
            "total_coordinations": total,
            "type_distribution": type_counts,
            "avg_final_position": sum(c.final_position for c in self.coordination_history) / total,
            "last_direction": self.coordination_history[-1].final_direction.value if self.coordination_history else None
        }