aszerW
512b73ac04
核心组件： - RiskControl: 风控抽象基类 - StopLossControl: 止损控制（固定止损/跟踪止损） - PositionLimitControl: 仓位限制控制 - PremiumControl: 溢价控制（filter/penalize模式） 回调钩子机制： - CallbackHook: 回调管理器（注册/触发） - 5个核心回调：before_entry, after_entry, before_exit, after_exit, dynamic_stoploss, custom_exit 便捷回调函数： - premium_filter_callback: 溢价过滤回调 - crash_filter_callback: 崩盘检测回调 - holding_time_stoploss_callback: 持仓时间动态止损 测试覆盖：13个测试全部通过
293 lines
8.8 KiB
Python
293 lines
8.8 KiB
Python
"""
|
||
风控层测试
|
||
|
||
测试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']) |