mirror of
https://github.com/B-Wear/QuantumEdge.git
synced 2026-07-27 15:37:46 +00:00
B-Wear
322 lines
12 KiB
Python
322 lines
12 KiB
Python
import logging
|
|
import threading
|
|
import queue
|
|
import json
|
|
import os
|
|
from typing import Dict, List, Optional, Any
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
import time
|
|
from .trading_system import TradingSystem, create_trading_system
|
|
from .code_guardian import CodeGuardian, create_guardian
|
|
from .system_monitor import SystemMonitor
|
|
from .dashboard import create_dashboard
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('logs/ai_agent_manager.log'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@dataclass
|
|
class BotInstance:
|
|
"""Represents a trading bot instance"""
|
|
bot_id: str
|
|
trading_system: TradingSystem
|
|
config: Dict
|
|
status: str
|
|
last_update: datetime
|
|
performance_metrics: Dict
|
|
message_queue: queue.Queue
|
|
|
|
class AIAgentManager:
|
|
def __init__(self, project_root: str):
|
|
self.project_root = project_root
|
|
self.bots: Dict[str, BotInstance] = {}
|
|
self.code_guardian = create_guardian(project_root)
|
|
self.system_monitor = SystemMonitor(project_root)
|
|
self.message_queue = queue.Queue()
|
|
self.running = True
|
|
|
|
# Start management threads
|
|
self.threads = {
|
|
'monitor': threading.Thread(target=self._monitor_bots),
|
|
'message_handler': threading.Thread(target=self._handle_messages),
|
|
'code_repair': threading.Thread(target=self._continuous_code_repair)
|
|
}
|
|
|
|
for thread in self.threads.values():
|
|
thread.daemon = True
|
|
thread.start()
|
|
|
|
# Initialize dashboard
|
|
self.dashboard = create_dashboard()
|
|
self.dashboard_thread = threading.Thread(target=self._run_dashboard)
|
|
self.dashboard_thread.daemon = True
|
|
self.dashboard_thread.start()
|
|
|
|
def create_bot(self, config: Dict) -> str:
|
|
"""Create a new trading bot instance"""
|
|
try:
|
|
bot_id = f"bot_{len(self.bots) + 1}"
|
|
|
|
# Save bot-specific config
|
|
config_path = os.path.join(self.project_root, 'config', f'{bot_id}_config.json')
|
|
with open(config_path, 'w') as f:
|
|
json.dump(config, f, indent=2)
|
|
|
|
# Create trading system
|
|
trading_system = create_trading_system(config_path)
|
|
|
|
# Create bot instance
|
|
bot = BotInstance(
|
|
bot_id=bot_id,
|
|
trading_system=trading_system,
|
|
config=config,
|
|
status='initialized',
|
|
last_update=datetime.now(),
|
|
performance_metrics={},
|
|
message_queue=queue.Queue()
|
|
)
|
|
|
|
self.bots[bot_id] = bot
|
|
logger.info(f"Created new bot instance: {bot_id}")
|
|
|
|
return bot_id
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error creating bot: {str(e)}")
|
|
raise
|
|
|
|
def start_bot(self, bot_id: str):
|
|
"""Start a trading bot"""
|
|
if bot_id in self.bots:
|
|
try:
|
|
bot = self.bots[bot_id]
|
|
bot.status = 'running'
|
|
bot.last_update = datetime.now()
|
|
logger.info(f"Started bot: {bot_id}")
|
|
|
|
# Send start message to bot
|
|
self._send_message_to_bot(bot_id, {
|
|
'type': 'command',
|
|
'action': 'start',
|
|
'timestamp': datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error starting bot {bot_id}: {str(e)}")
|
|
self.bots[bot_id].status = 'error'
|
|
|
|
def stop_bot(self, bot_id: str):
|
|
"""Stop a trading bot"""
|
|
if bot_id in self.bots:
|
|
try:
|
|
bot = self.bots[bot_id]
|
|
bot.status = 'stopped'
|
|
bot.last_update = datetime.now()
|
|
bot.trading_system.stop()
|
|
logger.info(f"Stopped bot: {bot_id}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error stopping bot {bot_id}: {str(e)}")
|
|
|
|
def send_message_to_bot(self, bot_id: str, message: Dict):
|
|
"""Send a message to a specific bot"""
|
|
if bot_id in self.bots:
|
|
self._send_message_to_bot(bot_id, message)
|
|
logger.info(f"Sent message to bot {bot_id}: {message}")
|
|
|
|
def broadcast_message(self, message: Dict):
|
|
"""Send a message to all bots"""
|
|
for bot_id in self.bots:
|
|
self._send_message_to_bot(bot_id, message)
|
|
logger.info(f"Broadcast message to all bots: {message}")
|
|
|
|
def get_bot_status(self, bot_id: str) -> Optional[Dict]:
|
|
"""Get status of a specific bot"""
|
|
if bot_id in self.bots:
|
|
bot = self.bots[bot_id]
|
|
return {
|
|
'bot_id': bot.bot_id,
|
|
'status': bot.status,
|
|
'last_update': bot.last_update.isoformat(),
|
|
'performance_metrics': bot.performance_metrics,
|
|
'trading_system_status': bot.trading_system.get_status()
|
|
}
|
|
return None
|
|
|
|
def get_system_status(self) -> Dict:
|
|
"""Get overall system status"""
|
|
return {
|
|
'total_bots': len(self.bots),
|
|
'active_bots': len([b for b in self.bots.values() if b.status == 'running']),
|
|
'code_health': self.code_guardian.get_status_report(),
|
|
'system_health': self.system_monitor.get_system_status(),
|
|
'bots': {bot_id: self.get_bot_status(bot_id) for bot_id in self.bots}
|
|
}
|
|
|
|
def _send_message_to_bot(self, bot_id: str, message: Dict):
|
|
"""Internal method to send message to bot"""
|
|
if bot_id in self.bots:
|
|
self.bots[bot_id].message_queue.put(message)
|
|
|
|
def _monitor_bots(self):
|
|
"""Continuously monitor bot performance and health"""
|
|
while self.running:
|
|
try:
|
|
for bot_id, bot in self.bots.items():
|
|
if bot.status == 'running':
|
|
# Update bot metrics
|
|
status = bot.trading_system.get_status()
|
|
bot.performance_metrics = status['metrics']
|
|
bot.last_update = datetime.now()
|
|
|
|
# Check for issues
|
|
if status['system_health']['alerts']['unacknowledged'] > 0:
|
|
self._handle_bot_issue(bot_id, status)
|
|
|
|
time.sleep(60) # Check every minute
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in bot monitoring: {str(e)}")
|
|
time.sleep(300)
|
|
|
|
def _handle_messages(self):
|
|
"""Handle incoming messages from bots"""
|
|
while self.running:
|
|
try:
|
|
for bot_id, bot in self.bots.items():
|
|
while not bot.message_queue.empty():
|
|
message = bot.message_queue.get_nowait()
|
|
self._process_bot_message(bot_id, message)
|
|
|
|
time.sleep(1)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error handling messages: {str(e)}")
|
|
time.sleep(60)
|
|
|
|
def _continuous_code_repair(self):
|
|
"""Continuously monitor and repair code issues"""
|
|
while self.running:
|
|
try:
|
|
# Get code issues
|
|
status = self.code_guardian.get_status_report()
|
|
|
|
# Attempt to fix issues
|
|
for issue in self.code_guardian.issues:
|
|
if not issue.fixed and issue.suggested_fix:
|
|
if self.code_guardian.fix_issue(issue):
|
|
logger.info(f"Fixed code issue: {issue.description}")
|
|
|
|
time.sleep(300) # Check every 5 minutes
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in code repair: {str(e)}")
|
|
time.sleep(600)
|
|
|
|
def _handle_bot_issue(self, bot_id: str, status: Dict):
|
|
"""Handle issues with a specific bot"""
|
|
try:
|
|
# Create alert
|
|
alert = {
|
|
'type': 'alert',
|
|
'bot_id': bot_id,
|
|
'timestamp': datetime.now().isoformat(),
|
|
'status': status
|
|
}
|
|
|
|
# Log alert
|
|
logger.warning(f"Bot issue detected: {alert}")
|
|
|
|
# Add to message queue
|
|
self.message_queue.put(alert)
|
|
|
|
# Take action based on issue severity
|
|
if status['system_health']['alerts']['by_level'].get('error', 0) > 0:
|
|
self.stop_bot(bot_id)
|
|
logger.error(f"Stopped bot {bot_id} due to critical issues")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error handling bot issue: {str(e)}")
|
|
|
|
def _process_bot_message(self, bot_id: str, message: Dict):
|
|
"""Process message from a bot"""
|
|
try:
|
|
if message['type'] == 'status_update':
|
|
self.bots[bot_id].status = message['status']
|
|
self.bots[bot_id].last_update = datetime.now()
|
|
|
|
elif message['type'] == 'alert':
|
|
self._handle_bot_issue(bot_id, message)
|
|
|
|
elif message['type'] == 'performance_update':
|
|
self.bots[bot_id].performance_metrics.update(message['metrics'])
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing bot message: {str(e)}")
|
|
|
|
def _run_dashboard(self):
|
|
"""Run the dashboard"""
|
|
try:
|
|
self.dashboard.run_server(debug=True, port=8050)
|
|
except Exception as e:
|
|
logger.error(f"Error running dashboard: {str(e)}")
|
|
|
|
def stop(self):
|
|
"""Stop the AI Agent Manager"""
|
|
self.running = False
|
|
|
|
# Stop all bots
|
|
for bot_id in self.bots:
|
|
self.stop_bot(bot_id)
|
|
|
|
# Stop components
|
|
self.code_guardian.stop()
|
|
self.system_monitor.stop()
|
|
|
|
# Wait for threads
|
|
for thread in self.threads.values():
|
|
if thread.is_alive():
|
|
thread.join()
|
|
|
|
def create_agent_manager(project_root: str) -> AIAgentManager:
|
|
"""Create and initialize an AI Agent Manager"""
|
|
return AIAgentManager(project_root)
|
|
|
|
if __name__ == "__main__":
|
|
# Example usage
|
|
manager = create_agent_manager(".")
|
|
|
|
try:
|
|
# Create some test bots
|
|
config1 = {
|
|
"trading": {"initial_capital": 100000},
|
|
"strategy": {"name": "trend_following"},
|
|
"risk_management": {"max_positions": 5}
|
|
}
|
|
|
|
config2 = {
|
|
"trading": {"initial_capital": 50000},
|
|
"strategy": {"name": "mean_reversion"},
|
|
"risk_management": {"max_positions": 3}
|
|
}
|
|
|
|
# Create and start bots
|
|
bot1_id = manager.create_bot(config1)
|
|
bot2_id = manager.create_bot(config2)
|
|
|
|
manager.start_bot(bot1_id)
|
|
manager.start_bot(bot2_id)
|
|
|
|
# Monitor for a while
|
|
time.sleep(3600) # Run for 1 hour
|
|
|
|
finally:
|
|
manager.stop() |