simple_strategies.py

"""
QTrading 简单策略模块
"""

import pandas as pd
import numpy as np
from strategy.base_strategy import BaseStrategy
from utils.common import normalize_score

class MA20Strategy(BaseStrategy):
    """MA20 策略（简单均线策略）"""

    name: str = "MA20Strategy"
    description: str = "基于 20 日均线的简单策略"

    def __init__(self, adjust_method: str = "qfq"):
        super().__init__(adjust_method)
        self.params = {
            'ma_period': 20,
            'trend_threshold': 0.02,
            'volatility_threshold': 0.03
        }

    def score(self, data: pd.DataFrame) -> float:
        """计算股票得分（0~100）"""
        if data.empty:
            return 0.0

        # 计算 20 日均线
        ma20 = self.calculate_ma(data, period=self.params['ma_period'])
        if len(ma20) < self.params['ma_period']:
            return 0.0

        # 获取最新价格和均线值
        current_price = data['close'].iloc[-1]
        latest_ma = ma20.iloc[-1]

        # 趋势评分（价格在均线上方得分高）
        trend_score = 100 if current_price > latest_ma else 0
        if abs(current_price - latest_ma) / latest_ma < self.params['trend_threshold']:
            trend_score = 50

        # 波动率评分（波动率越低得分越高）
        volatility = self.calculate_price_volatility(data, period=20)
        volatility_score = max(0, 100 - volatility * 10)

        # 成交量评分
        volume_momentum = self.calculate_volume_momentum(data, period=20)
        volume_score = normalize_score(volume_momentum, -100, 100)

        # 综合评分
        total_score = (trend_score * 0.6 + volatility_score * 0.2 + volume_score * 0.2)
        return max(0, min(100, round(total_score, 2)))

class RSIStrategy(BaseStrategy):
    """RSI 策略（相对强弱指标策略）"""

    name: str = "RSIStrategy"
    description: str = "基于 RSI 指标的策略"

    def __init__(self, adjust_method: str = "qfq"):
        super().__init__(adjust_method)
        self.params = {
            'rsi_period': 14,
            'oversold_threshold': 30,
            'overbought_threshold': 70,
            'trend_period': 20
        }

    def score(self, data: pd.DataFrame) -> float:
        """计算股票得分（0~100）"""
        if data.empty:
            return 0.0

        # 计算 RSI
        rsi = self.calculate_rsi(data, period=self.params['rsi_period'])
        if len(rsi) < self.params['rsi_period'] + 1:
            return 0.0

        current_rsi = rsi.iloc[-1]

        # RSI 评分（中间区域得分高）
        if current_rsi < self.params['oversold_threshold']:
            rsi_score = 30  # 超卖区域得分较低
        elif current_rsi > self.params['overbought_threshold']:
            rsi_score = 40  # 超买区域得分较低
        else:
            # 中间区域得分最高
            distance = min(abs(current_rsi - 50), 20)
            rsi_score = 100 - (distance / 20) * 60

        # 趋势评分
        ma20 = self.calculate_ma(data, period=self.params['trend_period'])
        current_price = data['close'].iloc[-1]
        latest_ma = ma20.iloc[-1]
        trend_score = 100 if current_price > latest_ma else 20

        # 波动率评分
        volatility = self.calculate_price_volatility(data, period=20)
        volatility_score = max(0, 100 - volatility * 10)

        # 综合评分
        total_score = (rsi_score * 0.5 + trend_score * 0.3 + volatility_score * 0.2)
        return max(0, min(100, round(total_score, 2)))

class MACDStrategy(BaseStrategy):
    """MACD 策略"""

    name: str = "MACDStrategy"
    description: str = "基于 MACD 指标的策略"

    def __init__(self, adjust_method: str = "qfq"):
        super().__init__(adjust_method)
        self.params = {
            'fast_period': 12,
            'slow_period': 26,
            'signal_period': 9,
            'trend_period': 20
        }

    def score(self, data: pd.DataFrame) -> float:
        """计算股票得分（0~100）"""
        if data.empty:
            return 0.0

        # 计算 MACD
        macd_result = self.calculate_macd(
            data,
            self.params['fast_period'],
            self.params['slow_period'],
            self.params['signal_period']
        )
        if len(macd_result['macd']) < self.params['slow_period'] + 1:
            return 0.0

        current_diff = macd_result['diff'].iloc[-1]
        current_dea = macd_result['dea'].iloc[-1]
        current_macd = macd_result['macd'].iloc[-1]

        # MACD 评分
        macd_score = 0
        if current_diff > current_dea and current_macd > 0:
            macd_score = 100  # 金叉且MACD柱为正
        elif current_diff > current_dea and current_macd < 0:
            macd_score = 70  # 金叉但MACD柱为负
        elif current_diff < current_dea and current_macd > 0:
            macd_score = 30  # 死叉但MACD柱为正
        else:
            macd_score = 0  # 死叉且MACD柱为负

        # 趋势评分
        ma20 = self.calculate_ma(data, period=self.params['trend_period'])
        current_price = data['close'].iloc[-1]
        latest_ma = ma20.iloc[-1]
        trend_score = 100 if current_price > latest_ma else 20

        # 成交量评分
        volume_momentum = self.calculate_volume_momentum(data, period=20)
        volume_score = normalize_score(volume_momentum, -100, 100)

        # 综合评分
        total_score = (macd_score * 0.5 + trend_score * 0.3 + volume_score * 0.2)
        return max(0, min(100, round(total_score, 2)))

class BollingerBandStrategy(BaseStrategy):
    """布林带策略"""

    name: str = "BollingerBandStrategy"
    description: str = "基于布林带指标的策略"

    def __init__(self, adjust_method: str = "qfq"):
        super().__init__(adjust_method)
        self.params = {
            'bb_period': 20,
            'bb_std': 2,
            'trend_period': 50
        }

    def score(self, data: pd.DataFrame) -> float:
        """计算股票得分（0~100）"""
        if data.empty:
            return 0.0

        # 计算布林带
        bb_result = self.calculate_bollinger_bands(
            data,
            period=self.params['bb_period'],
            num_std=self.params['bb_std']
        )
        if len(bb_result['middle']) < self.params['bb_period']:
            return 0.0

        current_price = data['close'].iloc[-1]
        upper_band = bb_result['upper'].iloc[-1]
        middle_band = bb_result['middle'].iloc[-1]
        lower_band = bb_result['lower'].iloc[-1]

        # 布林带位置评分
        if current_price > upper_band:
            bb_score = 40  # 上轨上方（超买）
        elif current_price < lower_band:
            bb_score = 60  # 下轨下方（超卖）
        elif current_price > middle_band:
            bb_score = 80  # 中轨上方（上升趋势）
        else:
            bb_score = 20  # 中轨下方（下降趋势）

        # 布林带宽度评分（波动率）
        bb_width = (upper_band - lower_band) / middle_band
        width_score = max(0, 100 - bb_width * 1000)

        # 趋势评分
        ma50 = self.calculate_ma(data, period=self.params['trend_period'])
        latest_ma50 = ma50.iloc[-1]
        trend_score = 100 if current_price > latest_ma50 else 40

        # 综合评分
        total_score = (bb_score * 0.4 + width_score * 0.3 + trend_score * 0.3)
        return max(0, min(100, round(total_score, 2)))

class MomentumStrategy(BaseStrategy):
    """动量策略"""

    name: str = "MomentumStrategy"
    description: str = "基于价格动量的策略"

    def __init__(self, adjust_method: str = "qfq"):
        super().__init__(adjust_method)
        self.params = {
            'momentum_period': 20,
            'volatility_period': 20,
            'correlation_period': 20
        }

    def score(self, data: pd.DataFrame) -> float:
        """计算股票得分（0~100）"""
        if data.empty:
            return 0.0

        # 价格动量评分
        price_momentum = self.calculate_price_momentum(
            data, period=self.params['momentum_period']
        )
        momentum_score = normalize_score(price_momentum, -50, 50)

        # 波动率评分（波动率越低得分越高）
        volatility = self.calculate_price_volatility(
            data, period=self.params['volatility_period']
        )
        volatility_score = max(0, 100 - volatility * 10)

        # 价格-成交量相关性评分
        correlation = self.calculate_price_volume_correlation(
            data, period=self.params['correlation_period']
        )
        correlation_score = normalize_score(correlation, -1, 1)

        # 综合评分
        total_score = (momentum_score * 0.5 + volatility_score * 0.3 + correlation_score * 0.2)
        return max(0, min(100, round(total_score, 2)))