import asyncio import logging from pathlib import Path from typing import Dict, Any from src.data.data_manager import DataManager from src.decision.decision_engine import DecisionEngine from src.execution.execution_engine import ExecutionEngine from config.config import Config class StockMonitorSystem: """三层交易系统主控制器""" def __init__(self, config_path: str = "config/config.yaml"): self.config = Config(config_path) self.logger = self._setup_logging() # 初始化三层组件 self.data_manager = DataManager(self.config) self.decision_engine = DecisionEngine(self.config) self.execution_engine = ExecutionEngine(self.config) # 系统状态 self.is_running = False def _setup_logging(self) -> logging.Logger: """设置日志配置""" from loguru import logger # 移除默认处理器 logger.remove() # 添加控制台输出 logger.add( sink=lambda msg: print(msg, end=""), format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}", level=self.config.get("logging.level", "INFO") ) # 添加文件输出 log_file = self.config.get("logging.file", "logs/stock_monitor.log") Path(log_file).parent.mkdir(parents=True, exist_ok=True) logger.add( sink=log_file, format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {name} | {message}", rotation=self.config.get("logging.rotation", "1 day"), retention=self.config.get("logging.retention", "30 days"), level=self.config.get("logging.level", "INFO") ) return logger async def start(self): """启动系统""" self.logger.info("🚀 启动三层交易系统...") try: # 1. 启动数据层 self.logger.info("📊 启动数据层...") await self.data_manager.start() # 2. 启动决策层 self.logger.info("🧠 启动决策层...") await self.decision_engine.start() # 3. 启动执行层 self.logger.info("🔄 启动执行层...") await self.execution_engine.start() self.is_running = True self.logger.info("✅ 系统启动完成") # 保持运行 await self._keep_running() except Exception as e: self.logger.error(f"❌ 系统启动失败: {e}") await self.stop() raise async def stop(self): """停止系统""" self.logger.info("🛑 停止三层交易系统...") self.is_running = False try: # 停止执行层 await self.execution_engine.stop() # 停止决策层 await self.decision_engine.stop() # 停止数据层 await self.data_manager.stop() self.logger.info("✅ 系统已停止") except Exception as e: self.logger.error(f"❌ 系统停止异常: {e}") async def _keep_running(self): """保持系统运行""" while self.is_running: try: await asyncio.sleep(1) except KeyboardInterrupt: self.logger.info("收到中断信号,正在停止...") await self.stop() break async def get_system_status(self) -> Dict[str, Any]: """获取系统状态""" return { "is_running": self.is_running, "data_manager": await self.data_manager.get_status(), "decision_engine": await self.decision_engine.get_status(), "execution_engine": await self.execution_engine.get_status(), "timestamp": asyncio.get_event_loop().time() } def main(): """主入口""" # 创建系统实例 system = StockMonitorSystem() # 启动系统 try: asyncio.run(system.start()) except KeyboardInterrupt: print("\n程序已终止") except Exception as e: print(f"程序异常: {e}") raise if __name__ == "__main__": main()