cyb50-quant/cyb50-pro/tests/test_integration.py

"""
集成测试

端到端测试CYB50-Pro主控引擎
"""

import unittest
from datetime import datetime
import numpy as np
import pandas as pd

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

from core.engine import CYB50ProEngine


class TestCYB50ProEngine(unittest.TestCase):
    """测试CYB50-Pro主控引擎"""

    def setUp(self):
        self.engine = CYB50ProEngine()
        self.price_data = self._generate_test_data()

    def _generate_test_data(self, days=100):
        """生成测试数据"""
        np.random.seed(42)
        dates = pd.date_range(end=datetime.now(), periods=days, freq='D')
        returns = np.random.normal(0.001, 0.015, days)
        prices = 100 * np.exp(np.cumsum(returns))

        return pd.DataFrame({
            'open': prices * (1 + np.random.normal(0, 0.001, days)),
            'high': prices * (1 + abs(np.random.normal(0, 0.01, days))),
            'low': prices * (1 - abs(np.random.normal(0, 0.01, days))),
            'close': prices,
            'volume': np.random.lognormal(15, 0.5, days)
        }, index=dates)

    def test_engine_initialization(self):
        """测试引擎初始化"""
        self.assertIsNotNone(self.engine)
        self.assertIsNotNone(self.engine.ecosystem_fusion)
        self.assertIsNotNone(self.engine.signal_fusion)
        self.assertIsNotNone(self.engine.agent_router)
        self.assertIsNotNone(self.engine.risk_manager)
        self.assertEqual(len(self.engine.agents), 3)

    def test_run_cycle(self):
        """测试完整运行周期"""
        result = self.engine.run_cycle(
            price_data=self.price_data,
            account_value=1_000_000
        )

        self.assertIsInstance(result, dict)
        self.assertIn('timestamp', result)
        self.assertIn('ecosystem', result)
        self.assertIn('routing', result)
        self.assertIn('signals', result)

    def test_ecosystem_detection(self):
        """测试生态识别集成"""
        result = self.engine.run_cycle(
            price_data=self.price_data,
            account_value=1_000_000
        )

        ecosystem = result.get('ecosystem')
        self.assertIsNotNone(ecosystem)
        self.assertIn('macro', ecosystem)
        self.assertIn('meso', ecosystem)
        self.assertIn('micro', ecosystem)

    def test_routing_decision(self):
        """测试智能体路由集成"""
        result = self.engine.run_cycle(
            price_data=self.price_data,
            account_value=1_000_000
        )

        routing = result.get('routing')
        self.assertIsNotNone(routing)
        self.assertIn('active_agents', routing)
        self.assertIn('weights', routing)

    def test_risk_check(self):
        """测试风险管理集成"""
        result = self.engine.run_cycle(
            price_data=self.price_data,
            account_value=1_000_000
        )

        # 如果有信号，检查风险管理
        if result.get('executed'):
            risk_check = result.get('risk_check')
            self.assertIsNotNone(risk_check)
            self.assertIn('can_trade', risk_check)
            self.assertIn('risk_level', risk_check)

    def test_engine_start_stop(self):
        """测试引擎启动停止"""
        self.engine.start()
        self.assertTrue(self.engine.is_running)

        self.engine.stop()
        self.assertFalse(self.engine.is_running)

    def test_get_status(self):
        """测试获取引擎状态"""
        status = self.engine.get_status()

        self.assertIsInstance(status, dict)
        self.assertIn('is_running', status)
        self.assertIn('agents', status)
        self.assertIn('risk_summary', status)


if __name__ == '__main__':
    unittest.main()