""" 配置文件 - API密钥和交易参数配置 支持从数据库读取配置(优先),回退到环境变量和默认值 支持动态重载配置 """ import os from typing import Optional # 尝试从数据库加载配置 USE_DB_CONFIG = False _config_manager = None def _init_config_manager(): """初始化配置管理器""" global USE_DB_CONFIG, _config_manager if _config_manager is not None: return _config_manager try: import sys import logging from pathlib import Path logger = logging.getLogger(__name__) # 从trading_system目录向上两级到项目根目录,然后找backend project_root = Path(__file__).parent.parent backend_path = project_root / 'backend' logger.debug(f"尝试初始化配置管理器,项目根目录: {project_root}, backend路径: {backend_path}") if backend_path.exists(): sys.path.insert(0, str(backend_path)) try: from config_manager import config_manager _config_manager = config_manager USE_DB_CONFIG = True logger.info("✓ 配置管理器初始化成功,将从数据库读取配置") return config_manager except ImportError as e: logger.warning(f"无法导入config_manager: {e},将使用环境变量和默认配置") USE_DB_CONFIG = False return None except Exception as e: logger.warning(f"配置管理器初始化失败: {e},将使用环境变量和默认配置") USE_DB_CONFIG = False return None else: logger.warning(f"backend目录不存在: {backend_path},将使用环境变量和默认配置") USE_DB_CONFIG = False return None except Exception as e: import logging logger = logging.getLogger(__name__) logger.warning(f"配置管理器初始化异常: {e},将使用环境变量和默认配置") USE_DB_CONFIG = False return None # 初始化配置管理器 _init_config_manager() def _get_config_value(key, default=None): """获取配置值(支持动态重载)""" if _config_manager: try: _config_manager.reload() # 每次获取配置时重新加载 value = _config_manager.get(key, default) # 如果从数据库获取到值,直接返回(即使是空字符串也返回) if value is not None: return value except Exception as e: import logging logger = logging.getLogger(__name__) logger.debug(f"从配置管理器获取{key}失败: {e},尝试环境变量") # 回退到环境变量 env_value = os.getenv(key, default) return env_value if env_value is not None else default def _get_trading_config(): """获取交易配置(支持动态重载)""" if _config_manager: _config_manager.reload() # 每次获取配置时重新加载 return _config_manager.get_trading_config() # 回退到默认配置 return { 'MAX_POSITION_PERCENT': 0.05, 'MAX_TOTAL_POSITION_PERCENT': 0.30, 'MIN_POSITION_PERCENT': 0.01, 'MIN_CHANGE_PERCENT': 2.0, 'TOP_N_SYMBOLS': 10, 'STOP_LOSS_PERCENT': 0.03, 'TAKE_PROFIT_PERCENT': 0.05, 'SCAN_INTERVAL': 3600, 'KLINE_INTERVAL': '1h', 'PRIMARY_INTERVAL': '1h', 'CONFIRM_INTERVAL': '4h', 'ENTRY_INTERVAL': '15m', 'MIN_VOLUME_24H': 10000000, 'MIN_VOLATILITY': 0.02, 'MIN_SIGNAL_STRENGTH': 5, 'LEVERAGE': 10, 'USE_TRAILING_STOP': True, 'TRAILING_STOP_ACTIVATION': 0.01, 'TRAILING_STOP_PROTECT': 0.01, 'USE_UNICORN_WEBSOCKET': True, } # 币安API配置(优先从数据库,回退到环境变量和默认值) BINANCE_API_KEY: Optional[str] = _get_config_value('BINANCE_API_KEY', '') BINANCE_API_SECRET: Optional[str] = _get_config_value('BINANCE_API_SECRET', '') USE_TESTNET: bool = _get_config_value('USE_TESTNET', False) if _get_config_value('USE_TESTNET') is not None else os.getenv('USE_TESTNET', 'False').lower() == 'true' # 交易参数配置(优先从数据库读取,支持动态重载) TRADING_CONFIG = _get_trading_config() # 确保包含所有必要的默认值 defaults = { 'SCAN_INTERVAL': 3600, 'KLINE_INTERVAL': '1h', 'PRIMARY_INTERVAL': '1h', 'CONFIRM_INTERVAL': '4h', 'ENTRY_INTERVAL': '15m', } for key, value in defaults.items(): if key not in TRADING_CONFIG: TRADING_CONFIG[key] = value # 提供一个函数来重新加载配置 def reload_config(): """重新加载配置(供外部调用)""" global TRADING_CONFIG, BINANCE_API_KEY, BINANCE_API_SECRET, USE_TESTNET, _config_manager _init_config_manager() # 重新初始化配置管理器 if _config_manager: _config_manager.reload() BINANCE_API_KEY = _get_config_value('BINANCE_API_KEY', BINANCE_API_KEY) BINANCE_API_SECRET = _get_config_value('BINANCE_API_SECRET', BINANCE_API_SECRET) USE_TESTNET = _get_config_value('USE_TESTNET', False) if _get_config_value('USE_TESTNET') is not None else os.getenv('USE_TESTNET', 'False').lower() == 'true' TRADING_CONFIG = _get_trading_config() # 确保默认值 for key, value in defaults.items(): if key not in TRADING_CONFIG: TRADING_CONFIG[key] = value # 连接配置 CONNECTION_TIMEOUT = int(os.getenv('CONNECTION_TIMEOUT', '30')) # 连接超时时间(秒) CONNECTION_RETRIES = int(os.getenv('CONNECTION_RETRIES', '3')) # 连接重试次数 # 日志配置 LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO') LOG_FILE = 'trading_bot.log'