etf/archive/framework/tests/test_integration.py

"""
集成测试

测试框架与定制组件的集成
"""

import pandas as pd
import numpy as np
import pytest

from framework.factors import FactorRegistry, FactorCombiner
from framework.signals import SignalGenerator
from framework.risk import CallbackHook, Position
from framework.strategy import StrategyBase
from framework.execution import Portfolio, BacktestExecutor

from strategies.shared.factors.momentum import MomentumFactor, VolatilityFactor
from strategies.shared.signals.selectors import TopNSelector
from strategies.shared.risk.controls import StopLossControl, premium_filter_callback
from strategies.rotation.strategy import RotationStrategy


class TestFactorIntegration:
    """测试因子集成"""
    
    def setup_method(self):
        """每个测试前清空注册表"""
        FactorRegistry.clear()
    
    def test_register_and_use_custom_factor(self):
        """测试注册并使用定制因子"""
        FactorRegistry.register(MomentumFactor)
        
        factor = FactorRegistry.get('momentum', n_days=25, crash_filter=True)
        
        dates = pd.date_range('2020-01-01', periods=100)
        data = pd.DataFrame({
            'close': np.random.randn(100).cumsum() + 100
        }, index=dates)
        
        values = factor.compute(data)
        
        assert len(values) == len(data)
    
    def test_combiner_with_custom_factors(self):
        """测试组合器使用定制因子"""
        FactorRegistry.register(MomentumFactor)
        FactorRegistry.register(VolatilityFactor)
        
        momentum = FactorRegistry.get('momentum', n_days=25)
        volatility = FactorRegistry.get('volatility', method='std', period=20)
        
        dates = pd.date_range('2020-01-01', periods=100)
        data = pd.DataFrame({
            'close': np.random.randn(100).cumsum() + 100,
            'high': np.random.randn(100).cumsum() + 105,
            'low': np.random.randn(100).cumsum() + 95
        }, index=dates)
        
        combiner = FactorCombiner([momentum, volatility], weights=[0.7, 0.3])
        result = combiner.compute(data)
        
        assert 'momentum' in result.columns
        assert 'volatility' in result.columns
        assert 'combined' in result.columns


class TestSignalIntegration:
    """测试信号生成器集成"""
    
    def test_custom_signal_generator_with_factors(self):
        """测试定制信号生成器与因子集成"""
        dates = pd.date_range('2020-01-01', periods=50)
        
        factor_data = pd.DataFrame({
            'momentum_A': np.random.randn(50),
            'momentum_B': np.random.randn(50),
            'momentum_C': np.random.randn(50),
        }, index=dates)
        
        selector = TopNSelector(select_num=2, min_score=0.0)
        result = selector.generate(factor_data)
        
        assert 'signal' in result.columns


class TestRiskIntegration:
    """测试风控组件集成"""
    
    def test_custom_risk_control(self):
        """测试定制风控组件"""
        control = StopLossControl(threshold=-0.05, trailing=True)
        
        position = Position(
            code='AAPL',
            entry_price=100.0,
            current_price=105.0,
            entry_time=pd.Timestamp.now()
        )
        
        # 首次检查，设置最高价
        assert control.check(position) == True
        
        # 价格下跌，触发跟踪止损
        position.current_price = 100.0
        assert control.check(position) == False
    
    def test_callback_hook_with_custom_callback(self):
        """测试回调钩子与定制回调集成"""
        hook = CallbackHook()
        
        # 注册定制回调
        callback = premium_filter_callback(threshold=0.10)
        hook.register('before_entry', callback)
        
        # 正常溢价通过
        result = hook.trigger('before_entry', 'AAPL', 100.0, premium=0.05)
        assert result == True
        
        # 高溢价拒绝
        result = hook.trigger('before_entry', 'AAPL', 100.0, premium=0.15)
        assert result == False


class TestStrategyIntegration:
    """测试策略集成"""
    
    def setup_method(self):
        """每个测试前清空注册表"""
        FactorRegistry.clear()
    
    def test_rotation_strategy_full_flow(self):
        """测试轮动策略完整流程"""
        strategy = RotationStrategy()
        
        # 生成测试数据
        dates = pd.date_range('2020-01-01', periods=100)
        data = pd.DataFrame({
            'close': np.random.randn(100).cumsum() + 100
        }, index=dates)
        
        # 运行策略
        result = strategy.run(data)
        
        assert 'signal' in result.columns
    
    def test_strategy_with_backtest_executor(self):
        """测试策略与回测执行器集成"""
        FactorRegistry.clear()
        strategy = RotationStrategy()
        
        dates = pd.date_range('2020-01-01', periods=100)
        data = pd.DataFrame({
            'close': np.random.randn(100).cumsum() + 100
        }, index=dates)
        
        signals = strategy.run(data)
        
        executor = BacktestExecutor(initial_capital=100000, trade_cost=0.001)
        portfolio = executor.execute(signals, data)
        
        assert isinstance(portfolio, Portfolio)


if __name__ == '__main__':
    pytest.main([__file__, '-v'])