87 lines
3.6 KiB
Python
87 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test COB Integration Status in Enhanced Orchestrator
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.append(str(Path('.').absolute()))
|
|
|
|
from core.enhanced_orchestrator import EnhancedTradingOrchestrator
|
|
from core.data_provider import DataProvider
|
|
|
|
async def test_cob_integration():
|
|
print("=" * 60)
|
|
print("COB INTEGRATION AUDIT")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
data_provider = DataProvider()
|
|
orchestrator = EnhancedTradingOrchestrator(
|
|
data_provider=data_provider,
|
|
symbols=['ETH/USDT', 'BTC/USDT'],
|
|
enhanced_rl_training=True
|
|
)
|
|
|
|
print(f"✓ Enhanced Orchestrator created")
|
|
print(f"Has COB integration attribute: {hasattr(orchestrator, 'cob_integration')}")
|
|
print(f"COB integration value: {orchestrator.cob_integration}")
|
|
print(f"COB integration type: {type(orchestrator.cob_integration)}")
|
|
print(f"COB integration active: {getattr(orchestrator, 'cob_integration_active', 'Not set')}")
|
|
|
|
if orchestrator.cob_integration:
|
|
print("\n--- COB Integration Details ---")
|
|
print(f"COB Integration class: {orchestrator.cob_integration.__class__.__name__}")
|
|
|
|
# Check if it has the expected methods
|
|
methods_to_check = ['get_statistics', 'get_cob_snapshot', 'add_dashboard_callback', 'start', 'stop']
|
|
for method in methods_to_check:
|
|
has_method = hasattr(orchestrator.cob_integration, method)
|
|
print(f"Has {method}: {has_method}")
|
|
|
|
# Try to get statistics
|
|
if hasattr(orchestrator.cob_integration, 'get_statistics'):
|
|
try:
|
|
stats = orchestrator.cob_integration.get_statistics()
|
|
print(f"COB statistics: {stats}")
|
|
except Exception as e:
|
|
print(f"Error getting COB statistics: {e}")
|
|
|
|
# Try to get a snapshot
|
|
if hasattr(orchestrator.cob_integration, 'get_cob_snapshot'):
|
|
try:
|
|
snapshot = orchestrator.cob_integration.get_cob_snapshot('ETH/USDT')
|
|
print(f"ETH/USDT snapshot: {snapshot}")
|
|
except Exception as e:
|
|
print(f"Error getting COB snapshot: {e}")
|
|
|
|
# Check if COB integration needs to be started
|
|
print(f"\n--- Starting COB Integration ---")
|
|
try:
|
|
await orchestrator.start_cob_integration()
|
|
print("✓ COB integration started successfully")
|
|
|
|
# Wait a moment and check statistics again
|
|
await asyncio.sleep(3)
|
|
if hasattr(orchestrator.cob_integration, 'get_statistics'):
|
|
stats = orchestrator.cob_integration.get_statistics()
|
|
print(f"COB statistics after start: {stats}")
|
|
|
|
except Exception as e:
|
|
print(f"Error starting COB integration: {e}")
|
|
else:
|
|
print("\n❌ COB integration is None - this explains the dashboard issues")
|
|
print("The Enhanced Orchestrator failed to initialize COB integration")
|
|
|
|
# Check the error flag
|
|
if hasattr(orchestrator, '_cob_integration_failed'):
|
|
print(f"COB integration failed flag: {orchestrator._cob_integration_failed}")
|
|
|
|
except Exception as e:
|
|
print(f"Error in COB audit: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_cob_integration()) |