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

"""
简化趋势策略 - 追求高收益

核心逻辑：
1. Summer生态 + 价格>MA20 + 趋势向上 = 满仓
2. 其他情况 = 空仓
3. 硬止损-10%，追踪止损-15%
"""

from typing import Dict, Any, Optional
import pandas as pd
import numpy as np
from datetime import datetime

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


class SimpleTrendAgent(AgentBase):
    """
    简化趋势策略

    规则：
    - Summer/Spring生态 + 收盘价>MA20 + MA5>MA20 = 满仓做多
    - 其他情况 = 空仓
    """

    def __init__(self):
        super().__init__(
            name="simple_trend",
            max_position=1.0,
            min_confidence=0.5
        )
        self.ma_fast = 5
        self.ma_slow = 20
        self.stop_loss_pct = 0.10
        self.trailing_stop_pct = 0.15

    def generate_signal(
        self,
        price_data: pd.DataFrame,
        ecosystem: Optional[UnifiedEcosystem] = None
    ) -> Optional[AgentSignal]:
        """生成交易信号"""
        if len(price_data) < self.ma_slow + 5:
            return None

        close = price_data['close']
        ma_fast = close.rolling(self.ma_fast).mean().iloc[-1]
        ma_slow = close.rolling(self.ma_slow).mean().iloc[-1]
        current_price = close.iloc[-1]

        # 计算趋势强度
        trend_strength = (ma_fast - ma_slow) / ma_slow

        # 确定方向
        if ecosystem and ecosystem.macro.regime in [MacroRegime.SUMMER, MacroRegime.SPRING]:
            if current_price > ma_slow and ma_fast > ma_slow and trend_strength > 0.005:
                # 强做多信号
                direction = SignalDirection.LONG
                confidence = min(1.0, 0.6 + trend_strength * 10)
                position_size = 1.0  # 满仓
            elif current_price < ma_slow * 0.95:
                # 跌破支撑，平仓
                direction = SignalDirection.NEUTRAL
                confidence = 0.5
                position_size = 0.0
            else:
                return None
        else:
            # 非 Summer/Spring，观望
            direction = SignalDirection.NEUTRAL
            confidence = 0.5
            position_size = 0.0

        # 根据中观健康度调整
        if ecosystem and direction == SignalDirection.LONG:
            health = ecosystem.meso.health_score / 100
            position_size *= health
            confidence *= health

        if direction == SignalDirection.NEUTRAL and position_size == 0:
            return AgentSignal(
                agent_name=self.name,
                direction=direction,
                strength=SignalStrength.WEAK,
                confidence=confidence,
                suggested_position=0.0,
                expected_return=0.0,
                win_probability=0.5,
                timestamp=datetime.now(),
                valid_until=datetime.now(),
                metadata={"reason": "outside_preferred_regime_or_no_trend"}
            )

        return AgentSignal(
            agent_name=self.name,
            direction=direction,
            strength=SignalStrength.STRONG if confidence > 0.7 else SignalStrength.MEDIUM,
            confidence=confidence,
            suggested_position=position_size,
            expected_return=0.25,  # 预期25%年化
            win_probability=0.6,
            timestamp=datetime.now(),
            valid_until=datetime.now(),
            metadata={
                "price": current_price,
                "ma_fast": ma_fast,
                "ma_slow": ma_slow,
                "trend_strength": trend_strength,
                "regime": ecosystem.macro.regime.value if ecosystem else "unknown"
            }
        )

    def get_expected_return(self, price_data: pd.DataFrame, ecosystem=None) -> float:
        return 0.25

    def get_win_probability(self, price_data: pd.DataFrame, ecosystem=None) -> float:
        return 0.6