""" 配置文件 - 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 # 使用基础日志(因为可能还没有配置好日志系统) import sys from pathlib import Path try: # 从trading_system目录向上两级到项目根目录,然后找backend project_root = Path(__file__).parent.parent backend_path = project_root / 'backend' print(f"[配置管理器] 尝试初始化,项目根目录: {project_root}") print(f"[配置管理器] backend路径: {backend_path}") print(f"[配置管理器] backend目录存在: {backend_path.exists()}") if not backend_path.exists(): print(f"[配置管理器] ❌ backend目录不存在: {backend_path}") USE_DB_CONFIG = False return None # 检查backend目录内容 backend_files = list(backend_path.iterdir()) if backend_path.exists() else [] print(f"[配置管理器] backend目录内容: {[f.name for f in backend_files[:10]]}") # 检查config_manager.py是否存在 config_manager_file = backend_path / 'config_manager.py' print(f"[配置管理器] config_manager.py存在: {config_manager_file.exists()}") if not config_manager_file.exists(): print(f"[配置管理器] ❌ config_manager.py不存在: {config_manager_file}") USE_DB_CONFIG = False return None # 添加到sys.path backend_str = str(backend_path) if backend_str not in sys.path: sys.path.insert(0, backend_str) print(f"[配置管理器] 已添加路径到sys.path: {backend_str}") # 尝试导入 try: print(f"[配置管理器] 尝试导入config_manager...") from config_manager import config_manager print(f"[配置管理器] ✓ 导入成功") # 测试数据库连接 try: print(f"[配置管理器] 测试数据库连接...") config_manager.reload() print(f"[配置管理器] ✓ 数据库连接成功,已加载 {len(config_manager._cache)} 个配置项") # 检查API密钥 api_key = config_manager.get('BINANCE_API_KEY') api_secret = config_manager.get('BINANCE_API_SECRET') print(f"[配置管理器] API密钥检查: KEY存在={bool(api_key)}, SECRET存在={bool(api_secret)}") _config_manager = config_manager USE_DB_CONFIG = True print(f"[配置管理器] ✓ 配置管理器初始化成功,将从数据库读取配置") return config_manager except Exception as db_error: print(f"[配置管理器] ⚠ 数据库连接失败: {db_error}") print(f"[配置管理器] 错误类型: {type(db_error).__name__}") import traceback print(f"[配置管理器] 错误详情:\n{traceback.format_exc()}") USE_DB_CONFIG = False return None except ImportError as e: print(f"[配置管理器] ❌ 无法导入config_manager: {e}") import traceback print(f"[配置管理器] 导入错误详情:\n{traceback.format_exc()}") USE_DB_CONFIG = False return None except Exception as e: print(f"[配置管理器] ❌ 配置管理器初始化失败: {e}") import traceback print(f"[配置管理器] 错误详情:\n{traceback.format_exc()}") USE_DB_CONFIG = False return None except Exception as e: print(f"[配置管理器] ❌ 初始化异常: {e}") import traceback print(f"[配置管理器] 异常详情:\n{traceback.format_exc()}") USE_DB_CONFIG = False return None # 初始化配置管理器(在模块加载时执行) # 注意:此时日志系统可能还没初始化,所以使用print输出 try: _init_config_manager() except Exception as e: print(f"[config.py] 配置管理器初始化异常: {e}") import traceback print(traceback.format_exc()) 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, } # 币安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'