test_strategy.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试策略实现类 - 用于开发和测试的简单策略
"""

import logging
import pandas as pd
from typing import Dict, Any

from src.strategies.base_strategy import BaseStrategy

# 配置日志
logger = logging.getLogger(__name__)

class TestStrategy(BaseStrategy):
    """测试策略实现 - 简单的均线交叉策略"""
    
    def _get_description(self) -> str:
        """获取策略描述"""
        return "测试策略：简单的均线交叉策略，用于开发和测试"
    
    def _get_default_params(self) -> Dict[str, Any]:
        """获取策略默认参数"""
        return {
            "short_ma_period": 5,
            "long_ma_period": 20,
            "min_price": 10,
            "max_price": 100,
            "min_volume": 1000000,
            "min_turnover": 10000000
        }
    
    def check_stock(self, ts_code: str, df_stock: pd.DataFrame) -> bool:
        """
        检查单只股票是否符合测试策略条件
        
        参数:
            ts_code: 股票代码
            df_stock: 股票历史数据
            
        返回:
            bool: 是否符合条件
        """
        try:
            # 检查数据长度
            if len(df_stock) < max(self.params['short_ma_period'], self.params['long_ma_period']):
                return False
            
            # 检查价格范围
            latest_close = df_stock['close'].iloc[-1]
            if latest_close < self.params['min_price'] or latest_close > self.params['max_price']:
                return False
            
            # 检查成交量
            latest_volume = df_stock['vol'].iloc[-1]
            if latest_volume < self.params['min_volume']:
                return False
            
            # 检查成交额
            latest_turnover = df_stock['amount'].iloc[-1]
            if latest_turnover < self.params['min_turnover']:
                return False
            
            # 计算均线
            short_ma = df_stock['close'].rolling(window=self.params['short_ma_period']).mean()
            long_ma = df_stock['close'].rolling(window=self.params['long_ma_period']).mean()
            
            # 检查均线交叉
            # 短期均线上穿长期均线
            if short_ma.iloc[-1] > long_ma.iloc[-1] and short_ma.iloc[-2] <= long_ma.iloc[-2]:
                logger.debug(f"股票 {ts_code} 符合测试策略条件（均线交叉）")
                return True
            
            logger.debug(f"股票 {ts_code} 不符合测试策略条件")
            return False
            
        except Exception as e:
            logger.error(f"检查股票 {ts_code} 时出错: {e}")
            return False