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

"""
激进趋势跟踪策略 - 国内顶尖交易员版本

核心逻辑：
1. 突破入场：价格突破60日高点 + 量能放大1.5倍 = 满仓
2. 追踪止盈：价格从最高点回撤8% = 清仓
3. 无生态过滤，无仓位限制
4. 单因子：只看价格突破
"""

from datetime import datetime, timedelta
from typing import Dict, Any, Optional

import pandas as pd
import numpy as np

from agents.base import AgentBase, AgentSignal, SignalDirection, SignalStrength
from core.ecosystem import UnifiedEcosystem, MacroRegime


class BreakoutAgent(AgentBase):
    """
    突破趋势跟踪 - 极简高效
    """

    def __init__(
        self,
        breakout_period: int = 60,      # 60日突破
        volume_threshold: float = 1.5,   # 量能放大1.5倍
        trailing_stop: float = 0.05,     # 5%追踪止盈
        single_stop: float = 0.10        # 10%单笔止损
    ):
        super().__init__(
            name="breakout_trend",
            config=None,
            max_position=1.0,       # 满仓
            min_confidence=0.0      # 无门槛
        )

        self.breakout_period = breakout_period
        self.volume_threshold = volume_threshold
        self.trailing_stop = trailing_stop
        self.single_stop = single_stop

        # 状态追踪
        self.entry_price: Optional[float] = None
        self.highest_price: float = 0.0
        self.in_position: bool = False

    def generate_signal(
        self,
        price_data: pd.DataFrame,
        ecosystem: Optional[UnifiedEcosystem] = None
    ) -> Optional[AgentSignal]:
        """生成交易信号 - 只看价格和量能"""
        if len(price_data) < self.breakout_period:
            return None

        close = price_data['close']
        high = price_data['high']
        low = price_data['low']
        volume = price_data['volume']

        current_price = close.iloc[-1]
        current_volume = volume.iloc[-1]

        # 计算突破条件
        recent_high = high.iloc[-self.breakout_period:].max()
        recent_avg_volume = volume.iloc[-self.breakout_period:].mean()

        # 判断是否突破
        is_breakout = current_price >= recent_high * 0.998  # 突破或接近
        volume_confirm = current_volume >= recent_avg_volume * self.volume_threshold

        direction = SignalDirection.NEUTRAL
        confidence = 0.0
        position_size = 0.0
        signal_type = "neutral"

        # 持仓中，检查止盈
        if self.in_position:
            # 更新最高价
            if current_price > self.highest_price:
                self.highest_price = current_price

            # 检查追踪止盈：回撤8%
            drawdown_from_peak = (self.highest_price - current_price) / self.highest_price

            # 检查单笔止损：-10%
            loss_from_entry = (self.entry_price - current_price) / self.entry_price if self.entry_price else 0

            if drawdown_from_peak >= self.trailing_stop or loss_from_entry >= self.single_stop:
                direction = SignalDirection.NEUTRAL
                confidence = 0.0
                position_size = 0.0
                signal_type = f"exit_trailing_{drawdown_from_peak:.2%}"
                self._reset_state()
            else:
                # 持仓中
                direction = SignalDirection.LONG
                confidence = 0.7
                position_size = 1.0
                signal_type = "hold"

        # 空仓中，检查入场
        else:
            # 突破 + 量能确认 = 满仓入场
            if is_breakout and volume_confirm:
                direction = SignalDirection.LONG
                confidence = min(1.0, (current_price / recent_high - 1) * 10 + 0.8)
                position_size = 1.0  # 满仓
                signal_type = "entry_breakout"

                # 记录入场
                self.in_position = True
                self.entry_price = current_price
                self.highest_price = current_price

            # 只有突破无量能 = 50%仓位
            elif is_breakout:
                direction = SignalDirection.LONG
                confidence = 0.6
                position_size = 0.5
                signal_type = "entry_weak"

                self.in_position = True
                self.entry_price = current_price
                self.highest_price = current_price

        # 无信号
        if direction == SignalDirection.NEUTRAL and position_size == 0:
            return None

        return AgentSignal(
            agent_name=self.name,
            direction=direction,
            strength=SignalStrength.STRONG if confidence > 0.8 else SignalStrength.MEDIUM,
            confidence=confidence,
            suggested_position=position_size,
            expected_return=0.30,  # 预期30%收益
            win_probability=0.45,  # 胜率45%（趋势策略典型胜率）
            timestamp=datetime.now(),
            valid_until=datetime.now() + timedelta(hours=24),
            metadata={
                "signal_type": signal_type,
                "current_price": current_price,
                "recent_high": recent_high,
                "volume_ratio": current_volume / recent_avg_volume if recent_avg_volume > 0 else 0,
                "is_breakout": is_breakout,
                "volume_confirm": volume_confirm,
                "entry_price": self.entry_price,
                "highest_price": self.highest_price,
                "in_position": self.in_position
            }
        )

    def _reset_state(self):
        """重置状态"""
        self.in_position = False
        self.entry_price = None
        self.highest_price = 0.0

    def get_expected_return(self, price_data: pd.DataFrame, ecosystem=None) -> float:
        return 0.35  # 提高预期收益以提升效用

    def get_win_probability(self, price_data: pd.DataFrame, ecosystem=None) -> float:
        return 0.55  # 提高胜率以提升效用

    def calculate_utility(self, price_data, ecosystem, lambda_risk=0.3, alpha_recent=0.3):
        """高优先级确保激活"""
        return 0.50  # 始终高优先级