AlphaFuturesPro/backend/service_implementation/qihuo_analyzer/utils/config_manager.py

# 配置管理工具
import os
import json
from typing import Dict, Optional


class ConfigManager:
    """配置管理类"""
    
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(ConfigManager, cls).__new__(cls)
            cls._instance._load_config()
        return cls._instance
    
    def _load_config(self):
        """加载配置"""
        # 配置文件路径
        config_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))), 'config.json')
        print(f"[ConfigManager]尝试加载配置文件: {config_path}")
        # 默认配置
        default_config = {
            'openai_api_key': '',
            'deepseek_api_key': '',
            'deepseek_api_url': 'https://api.deepseek.com/v1/chat/completions',
            'db_path': './data/futures_analysis.db',
            'tqserver_host': 'api.shinnytech.com',
            'tqserver_port': 7777,
            'max_risk_percent': 0.02,
            'min_profit_loss_ratio': 1.5,
            'default_atr_multiplier': 2.0,
            'default_adx_threshold': 20,
            'review_times': '09:00,12:30,15:30'.split(',')
        }
        
        # 从config.json加载配置
        if os.path.exists(config_path):
            try:
                with open(config_path, 'r', encoding='utf-8') as f:
                    config_data = json.load(f)
                
                # 从aiModel部分获取API配置
                if 'aiModel' in config_data and 'models' in config_data['aiModel']:
                    # 查找启用的模型
                    enabled_models = [model for model in config_data['aiModel']['models'] if model.get('enabled', False)]
                    
                    # 优先使用DeepSeek模型的配置
                    deepseek_model = next((model for model in config_data['aiModel']['models'] if model.get('name') == 'DeepSeek'), None)
                    if deepseek_model:
                        default_config['deepseek_api_key'] = deepseek_model.get('apiKey', '')
                        default_config['deepseek_api_url'] = deepseek_model.get('apiBaseUrl', 'https://api.deepseek.com/v1/chat/completions')
                    
                    # 查找GPT-4模型的配置
                    gpt4_model = next((model for model in config_data['aiModel']['models'] if model.get('name') == 'GPT-4'), None)
                    if gpt4_model:
                        default_config['openai_api_key'] = gpt4_model.get('apiKey', '')
        
            except Exception as e:
                print(f"加载config.json失败: {e}")
                # 使用默认配置
        
        # 设置配置值
        self.openai_api_key = default_config['openai_api_key']
        self.deepseek_api_key = default_config['deepseek_api_key']
        self.deepseek_api_url = default_config['deepseek_api_url']
        self.db_path = default_config['db_path']
        self.tqserver_host = default_config['tqserver_host']
        self.tqserver_port = default_config['tqserver_port']
        self.max_risk_percent = default_config['max_risk_percent']
        self.min_profit_loss_ratio = default_config['min_profit_loss_ratio']
        self.default_atr_multiplier = default_config['default_atr_multiplier']
        self.default_adx_threshold = default_config['default_adx_threshold']
        self.review_times = default_config['review_times']
    
    def get_config(self) -> Dict:
        """获取所有配置"""
        return {
            'openai_api_key': self.openai_api_key,
            'deepseek_api_key': self.deepseek_api_key,
            'deepseek_api_url': self.deepseek_api_url,
            'db_path': self.db_path,
            'tqserver_host': self.tqserver_host,
            'tqserver_port': self.tqserver_port,
            'max_risk_percent': self.max_risk_percent,
            'min_profit_loss_ratio': self.min_profit_loss_ratio,
            'default_atr_multiplier': self.default_atr_multiplier,
            'default_adx_threshold': self.default_adx_threshold,
            'review_times': self.review_times
        }
    
    def update_config(self, config: Dict):
        """更新配置"""
        for key, value in config.items():
            if hasattr(self, key):
                setattr(self, key, value)


# 全局配置实例
config_manager = ConfigManager()