model_strategy.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
大模型选股策略实现类 - 基于AI模型的选股策略
"""

import logging
import pandas as pd
import json
from pathlib import Path
from typing import Dict, Any

from src.strategies.base_strategy import BaseStrategy
from src.config_manager import config_manager

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

class ModelStrategy(BaseStrategy):
    """大模型选股策略实现"""
    
    def _get_description(self) -> str:
        """获取策略描述"""
        return "大模型选股策略：基于AI模型分析股票数据"
    
    def _get_default_params(self) -> Dict[str, Any]:
        """获取策略默认参数"""
        return {
            "learning_window": "90日",
            "profit_threshold": "3%",
            "stock_count": "10只",
            "volume_ratio_threshold": "1.5倍"
        }
    
    def check_stock(self, ts_code: str, df_stock: pd.DataFrame) -> bool:
        """
        检查单只股票是否符合大模型选股策略条件
        
        参数:
            ts_code: 股票代码
            df_stock: 股票历史数据
            
        返回:
            bool: 是否符合条件
        """
        try:
            # 加载最佳参数
            best_params = self._load_best_parameters()
            
            # 计算选股因素
            factors = self._get_selection_factors(df_stock, best_params)
            
            # 使用大模型分析股票
            is_selected = self._analyze_with_model(ts_code, df_stock, factors, best_params)
            
            logger.debug(f"股票 {ts_code} 符合大模型策略条件: {is_selected}")
            return is_selected
            
        except Exception as e:
            logger.error(f"检查股票 {ts_code} 时出错: {e}")
            return False
    
    def _load_best_parameters(self) -> Dict[str, Any]:
        """加载最优参数"""
        params_file = "results/optimization/best_parameters.json"
        
        if Path(params_file).exists():
            try:
                with open(params_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                logger.error(f"加载最优参数失败: {e}")
        
        return self.params
    
    def _get_selection_factors(self, df_stock: pd.DataFrame, params: Dict[str, Any]) -> Dict[str, Any]:
        """获取选股因素"""
        factors = {}
        
        try:
            # 计算基本指标
            latest_close = df_stock['close'].iloc[-1]
            latest_volume = df_stock['vol'].iloc[-1]
            latest_turnover = df_stock['amount'].iloc[-1]
            
            # 计算移动平均线
            ma5 = df_stock['close'].rolling(window=5).mean().iloc[-1]
            ma10 = df_stock['close'].rolling(window=10).mean().iloc[-1]
            ma20 = df_stock['close'].rolling(window=20).mean().iloc[-1]
            
            # 计算成交量相关指标
            avg_volume = df_stock['vol'].rolling(window=20).mean().iloc[-1]
            volume_ratio = latest_volume / avg_volume if avg_volume != 0 else 0
            
            factors.update({
                "价格": f"{latest_close:.2f}",
                "成交量": f"{latest_volume:,}",
                "量比": f"{volume_ratio:.1f}",
                "MA5": f"{ma5:.2f}",
                "MA10": f"{ma10:.2f}",
                "MA20": f"{ma20:.2f}"
            })
            
        except Exception as e:
            logger.error(f"获取选股因素失败: {e}")
        
        return factors
    
    def _analyze_with_model(self, ts_code: str, df_stock: pd.DataFrame, factors: Dict[str, Any], params: Dict[str, Any]) -> bool:
        """使用大模型分析股票"""
        try:
            # 导入模型API（避免循环导入）
            from src.model_api import ModelAPI
            model_api = ModelAPI()
            
            # 准备分析请求
            analysis_request = {
                "stock_code": ts_code,
                "factors": factors,
                "params": params,
                "data": df_stock.tail(30).to_dict('records')
            }
            
            # 调用模型分析
            analysis_result = model_api.analyze_stock(analysis_request)
            
            # 解析结果
            if analysis_result and analysis_result.get('score', 0) > 0:
                return True
            
            return False
            
        except Exception as e:
            logger.error(f"大模型分析失败: {e}")
            return False