""" 风控层测试 测试RiskControl、StopLossControl、PositionLimitControl、CallbackHook """ import pandas as pd import pytest from datetime import datetime, timedelta from framework.risk import ( RiskControl, StopLossControl, PositionLimitControl, PremiumControl, CallbackHook, Position, Trade, premium_filter_callback, crash_filter_callback, holding_time_stoploss_callback ) class TestPosition: """测试持仓信息""" def test_position_profit(self): """测试盈亏计算""" position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=110.0, current_date=datetime(2020, 1, 10), quantity=100, weight=0.33 ) assert position.profit_ratio == 0.10 assert position.is_profit == True assert position.holding_days == 9 def test_position_loss(self): """测试亏损计算""" position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=95.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.33 ) assert position.profit_ratio == -0.05 assert position.is_profit == False assert position.holding_days == 4 class TestStopLossControl: """测试止损控制""" def test_fixed_stoploss_check(self): """测试固定止损检查""" control = StopLossControl(threshold=-0.05) # 未触发止损 position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=96.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.33 ) assert control.check(position) == True # 触发止损 position.current_price = 94.0 assert control.check(position) == False def test_fixed_stoploss_apply(self): """测试固定止损应用""" control = StopLossControl(threshold=-0.05) position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=95.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.33 ) stop_price = control.apply(position) assert stop_price == 95.0 # 100 * (1 - 0.05) def test_trailing_stoploss(self): """测试跟踪止损""" control = StopLossControl(trailing=True, trailing_percent=0.03) position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=105.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.33 ) # 最高价更新为105 control.check(position) assert control._highest_price['code1'] == 105.0 # 当前价回撤到101（从105回撤4%），超过3%阈值 position.current_price = 101.0 assert control.check(position) == False # 止损价格应为 105 * (1 - 0.03) = 101.85 stop_price = control.apply(position) assert abs(stop_price - 101.85) < 0.01 class TestPositionLimitControl: """测试仓位限制控制""" def test_position_limit_check(self): """测试仓位限制检查""" control = PositionLimitControl(max_position=0.33) # 仓位未超限 position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=105.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.30 ) assert control.check(position) == True # 仓位超限 position.weight = 0.40 assert control.check(position) == False def test_position_limit_apply(self): """测试仓位限制应用""" control = PositionLimitControl(max_position=0.33) position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=105.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.50 ) suggested_weight = control.apply(position) assert suggested_weight == 0.33 class TestPremiumControl: """测试溢价控制""" def test_premium_filter(self): """测试溢价过滤""" control = PremiumControl(threshold=0.10, mode='filter') # 溢价未超限 assert control.check(None, premium=0.05) == True # 溢价超限 assert control.check(None, premium=0.15) == False def test_premium_penalize(self): """测试溢价降权""" control = PremiumControl(threshold=0.10, mode='penalize') # 降权模式下允许通过 assert control.check(None, premium=0.15) == True # 返回降权系数 position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=105.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.33 ) penalty = control.apply(position) assert penalty == 0.5 class TestCallbackHook: """测试回调钩子""" def test_register_hook(self): """测试注册回调""" hook = CallbackHook() def dummy_callback(code, price): return True hook.register('before_entry', dummy_callback) assert len(hook._hooks['before_entry']) == 1 def test_trigger_before_entry(self): """测试触发入场前回调""" hook = CallbackHook() # 注册溢价过滤回调 hook.register('before_entry', premium_filter_callback(threshold=0.10)) # 溢价正常，允许入场 result = hook.trigger('before_entry', 'code1', 100.0, premium=0.05) assert result == True # 溢价过高，拒绝入场 result = hook.trigger('before_entry', 'code1', 100.0, premium=0.15) assert result == False def test_trigger_dynamic_stoploss(self): """测试触发动态止损回调""" hook = CallbackHook() # 注册持仓时间止损回调 hook.register('dynamic_stoploss', holding_time_stoploss_callback()) # 持仓5天，止损-5% position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=95.0, current_date=datetime(2020, 1, 6), quantity=100, weight=0.33 ) stoploss = hook.trigger('dynamic_stoploss', position) # holding_days=5，返回-0.05 assert stoploss == -0.05 def test_trigger_custom_exit(self): """测试触发自定义出场回调""" hook = CallbackHook() def reversal_exit_callback(position): # 反转信号触发出场 return position.profit_ratio < -0.02 hook.register('custom_exit', reversal_exit_callback) # 未触发出场 position = Position( code='code1', entry_price=100.0, entry_date=datetime(2020, 1, 1), current_price=99.0, current_date=datetime(2020, 1, 5), quantity=100, weight=0.33 ) result = hook.trigger('custom_exit', position) assert result == False # 触发出场 position.current_price = 97.0 result = hook.trigger('custom_exit', position) assert result == True def test_multiple_callbacks(self): """测试多个回调组合""" hook = CallbackHook() # 注册多个入场前回调 hook.register('before_entry', premium_filter_callback(0.10)) hook.register('before_entry', lambda code, price, **kwargs: price > 50) # 溢价正常 + 价格>50，允许入场 result = hook.trigger('before_entry', 'code1', 100.0, premium=0.05) assert result == True # 溢价过高，拒绝入场（任一回调返回False） result = hook.trigger('before_entry', 'code1', 100.0, premium=0.15) assert result == False # 价格过低，拒绝入场 result = hook.trigger('before_entry', 'code1', 40.0, premium=0.05) assert result == False if __name__ == '__main__': pytest.main([__file__, '-v'])