stock/config/config.py

"""
系统配置文件
配置股票分析系统的各项参数
"""

import os
from typing import Dict, Any, Optional


class Config:
	"""系统配置类"""
	
	# 数据库配置
	DATABASE_CONFIG = {
		"database_url": "sqlite:///stock_data.db",  # 默认使用SQLite
		"echo": False,  # 是否输出SQL语句
		"pool_size": 10,  # 连接池大小
		"max_overflow": 20,  # 最大溢出连接数
		"pool_timeout": 30,  # 连接池超时时间（秒）
		"pool_recycle": 3600,  # 连接回收时间（秒）
	}
	
	# 数据采集配置
	DATA_COLLECTION_CONFIG = {
		"akshare": {
			"base_url": "https://api.akshare.akfamily.xyz",
			"timeout": 30,  # 请求超时时间（秒）
			"retry_times": 3,  # 重试次数
			"retry_delay": 1,  # 重试延迟（秒）
		},
		"baostock": {
			"login_timeout": 10,  # 登录超时时间（秒）
			"query_timeout": 30,  # 查询超时时间（秒）
			"max_connections": 5,  # 最大连接数
		},
		"batch_size": 100,  # 批量处理大小
		"max_concurrent": 10,  # 最大并发数
	}
	
	# 定时任务配置
	SCHEDULER_CONFIG = {
		"daily_kline_update": {
			"enabled": True,  # 是否启用
			"time": "18:00",  # 执行时间（交易日收盘后）
			"timezone": "Asia/Shanghai",  # 时区
			"max_retries": 3,  # 最大重试次数
		},
		"weekly_financial_update": {
			"enabled": True,
			"day_of_week": "sat",  # 周六执行
			"time": "09:00",
			"timezone": "Asia/Shanghai",
			"max_retries": 3,
		},
		"monthly_basic_update": {
			"enabled": True,
			"day": 1,  # 每月1号
			"time": "10:00",
			"timezone": "Asia/Shanghai",
			"max_retries": 3,
		},
		"daily_health_check": {
			"enabled": True,
			"time": "08:00",
			"timezone": "Asia/Shanghai",
			"max_retries": 1,
		},
	}
	
	# 数据处理配置
	DATA_PROCESSING_CONFIG = {
		"validation": {
			"required_fields": ["code", "name"],  # 必需字段
			"numeric_fields": ["open", "high", "low", "close", "volume", "amount"],  # 数值字段
			"date_fields": ["date", "list_date"],  # 日期字段
		},
		"cleaning": {
			"remove_duplicates": True,  # 是否去重
			"fill_missing_values": True,  # 是否填充缺失值
			"standardize_formats": True,  # 是否标准化格式
		},
		"normalization": {
			"decimal_places": 2,  # 小数位数
			"date_format": "%Y-%m-%d",  # 日期格式
		},
	}
	
	# 日志配置
	LOGGING_CONFIG = {
		"level": "INFO",  # 日志级别
		"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
		"date_format": "%Y-%m-%d %H:%M:%S",
		"file": {
			"enabled": True,
			"filename": "logs/stock_system.log",
			"max_bytes": 10485760,  # 10MB
			"backup_count": 5,  # 备份文件数量
		},
		"console": {
			"enabled": True,
			"level": "INFO",
		},
	}
	
	# 性能配置
	PERFORMANCE_CONFIG = {
		"memory": {
			"max_memory_usage": 1024,  # 最大内存使用（MB）
			"gc_threshold": 512,  # 垃圾回收阈值（MB）
		},
		"database": {
			"query_timeout": 30,  # 查询超时时间（秒）
			"batch_size": 1000,  # 批量操作大小
			"max_connections": 50,  # 最大数据库连接数
		},
		"network": {
			"timeout": 30,  # 网络超时时间（秒）
			"retry_delay": 1,  # 重试延迟（秒）
		},
	}
	
	# 安全配置
	SECURITY_CONFIG = {
		"encryption": {
			"enabled": False,  # 是否启用数据加密
			"algorithm": "AES",  # 加密算法
		},
		"authentication": {
			"enabled": False,  # 是否启用认证
		},
		"backup": {
			"enabled": True,
			"interval": 24,  # 备份间隔（小时）
			"retention_days": 30,  # 保留天数
		},
	}
	
	# 监控配置
	MONITORING_CONFIG = {
		"enabled": True,
		"metrics": {
			"data_collection": True,  # 数据采集指标
			"database_performance": True,  # 数据库性能指标
			"system_resources": True,  # 系统资源指标
		},
		"alerts": {
			"enabled": True,
			"thresholds": {
				"memory_usage": 80,  # 内存使用率阈值（%）
				"cpu_usage": 80,  # CPU使用率阈值（%）
				"disk_usage": 90,  # 磁盘使用率阈值（%）
				"database_timeout": 10,  # 数据库超时次数阈值
			},
		},
	}
	
	# 开发配置
	DEVELOPMENT_CONFIG = {
		"debug": False,  # 调试模式
		"testing": False,  # 测试模式
		"profiling": False,  # 性能分析
		"logging_level": "DEBUG",  # 开发环境日志级别
	}
	
	@classmethod
	def get_database_url(cls) -> str:
		"""获取数据库URL"""
		# 优先使用环境变量
		database_url = os.getenv("DATABASE_URL")
		if database_url:
			return database_url
		
		return cls.DATABASE_CONFIG["database_url"]
	
	@classmethod
	def get_logging_config(cls) -> Dict[str, Any]:
		"""获取日志配置"""
		config = cls.LOGGING_CONFIG.copy()
		
		# 开发环境使用更详细的日志级别
		if cls.DEVELOPMENT_CONFIG["debug"]:
			config["level"] = cls.DEVELOPMENT_CONFIG["logging_level"]
			config["console"]["level"] = cls.DEVELOPMENT_CONFIG["logging_level"]
		
		return config
	
	@classmethod
	def get_data_collection_config(cls, source: str) -> Optional[Dict[str, Any]]:
		"""获取指定数据源的采集配置"""
		return cls.DATA_COLLECTION_CONFIG.get(source)
	
	@classmethod
	def get_scheduler_config(cls, task_name: str) -> Optional[Dict[str, Any]]:
		"""获取指定定时任务的配置"""
		return cls.SCHEDULER_CONFIG.get(task_name)
	
	@classmethod
	def update_from_environment(cls):
		"""从环境变量更新配置"""
		# 数据库配置
		if os.getenv("DATABASE_URL"):
			cls.DATABASE_CONFIG["database_url"] = os.getenv("DATABASE_URL")
		
		# 调试模式
		if os.getenv("DEBUG"):
			cls.DEVELOPMENT_CONFIG["debug"] = os.getenv("DEBUG").lower() == "true"
		
		# 日志级别
		if os.getenv("LOG_LEVEL"):
			cls.LOGGING_CONFIG["level"] = os.getenv("LOG_LEVEL")
			cls.LOGGING_CONFIG["console"]["level"] = os.getenv("LOG_LEVEL")
	
	@classmethod
	def validate_config(cls) -> bool:
		"""验证配置的有效性"""
		try:
			# 验证数据库配置
			database_url = cls.get_database_url()
			if not database_url:
				raise ValueError("数据库URL不能为空")
			
			# 验证数据采集配置
			for source in ["akshare", "baostock"]:
				source_config = cls.get_data_collection_config(source)
				if not source_config:
					raise ValueError(f"数据源 {source} 的配置不能为空")
			
			# 验证定时任务配置
			for task_name in cls.SCHEDULER_CONFIG.keys():
				task_config = cls.get_scheduler_config(task_name)
				if not task_config:
					raise ValueError(f"定时任务 {task_name} 的配置不能为空")
			
			return True
		except Exception as e:
			print(f"配置验证失败: {e}")
			return False


# 创建配置实例
config = Config()

# 从环境变量更新配置
config.update_from_environment()

# 验证配置
if not config.validate_config():
	raise RuntimeError("系统配置验证失败，请检查配置文件")