AlphaFutures/qihuo_analyzer/utils/config_manager.py

# 配置管理工具
import os
from dotenv import load_dotenv
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):
        """加载配置"""
        # 加载.env文件
        load_dotenv()
        
        # API配置
        self.openai_api_key = os.getenv('OPENAI_API_KEY', '')
        self.deepseek_api_key = os.getenv('DEEPSEEK_API_KEY', '')
        self.deepseek_api_url = os.getenv('DEEPSEEK_API_URL', 'https://api.deepseek.com/v1/chat/completions')
        
        # 数据库配置
        self.db_path = os.getenv('DB_PATH', './data/futures_analysis.db')
        
        # 天勤TQSDK配置
        self.tqserver_host = os.getenv('TQSERVER_HOST', 'api.shinnytech.com')
        self.tqserver_port = int(os.getenv('TQSERVER_PORT', '7777'))
        
        # 风险配置
        self.max_risk_percent = float(os.getenv('MAX_RISK_PERCENT', '0.02'))
        self.min_profit_loss_ratio = float(os.getenv('MIN_PROFIT_LOSS_RATIO', '1.5'))
        
        # 策略配置
        self.default_atr_multiplier = float(os.getenv('DEFAULT_ATR_MULTIPLIER', '2.0'))
        self.default_adx_threshold = float(os.getenv('DEFAULT_ADX_THRESHOLD', '20'))
        
        # 定时任务配置
        self.review_times = os.getenv('REVIEW_TIMES', '09:00,12:30,15:30').split(',')
    
    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()