Coverage for core/models/risk_config.py: 72.41%

87 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-10-13 18:58 +0000

1""" 

2风险管理配置模型 

3""" 

4 

5from decimal import Decimal 

6from typing import Any, Dict, List, Optional 

7 

8from pydantic import BaseModel, Field, field_validator 

9 

10 

11class RiskConfigModel(BaseModel): 

12 """风险管理配置模型""" 

13 

14 # 基础风险参数 

15 max_position_ratio: float = Field( 

16 1.0, ge=0.01, le=1.0, description="单只股票最大持仓比例" 

17 ) 

18 stop_loss_ratio: float = Field(0.3, ge=0.01, le=0.5, description="止损比例") 

19 max_drawdown: float = Field(0.3, ge=0.01, le=0.5, description="最大回撤") 

20 allowed_symbols: List[str] = Field( 

21 default_factory=list, description="允许交易的股票代码" 

22 ) 

23 

24 # 高级风险参数 

25 max_single_order_ratio: float = Field( 

26 0.05, ge=0.01, le=0.5, description="单笔订单最大比例" 

27 ) 

28 max_daily_loss_ratio: float = Field( 

29 0.02, ge=0.01, le=0.1, description="日最大亏损比例" 

30 ) 

31 min_cash_ratio: float = Field(0.1, ge=0.01, le=0.5, description="最小现金比例") 

32 max_leverage: float = Field(1.0, ge=1.0, le=5.0, description="最大杠杆比例") 

33 

34 # 风险监控参数 

35 risk_alert_threshold: float = Field(0.8, ge=0.5, le=1.0, description="风险告警阈值") 

36 enable_auto_stop_loss: bool = Field(True, description="启用自动止损") 

37 enable_position_limit: bool = Field(True, description="启用持仓限制") 

38 enable_turnover_limit: bool = Field(True, description="启用交易量限制") 

39 

40 # 风险事件记录 

41 max_risk_events: int = Field( 

42 1000, ge=100, le=10000, description="最大风险事件记录数" 

43 ) 

44 enable_risk_logging: bool = Field(True, description="启用风险日志记录") 

45 

46 @field_validator("max_position_ratio") 

47 @classmethod 

48 def validate_max_position_ratio(cls, v): 

49 if v <= 0 or v > 1: 

50 raise ValueError("最大持仓比例必须在0-1之间") 

51 return v 

52 

53 @field_validator("stop_loss_ratio") 

54 @classmethod 

55 def validate_stop_loss_ratio(cls, v): 

56 if v <= 0 or v > 0.5: 

57 raise ValueError("止损比例必须在0-0.5之间") 

58 return v 

59 

60 @field_validator("max_drawdown") 

61 @classmethod 

62 def validate_max_drawdown(cls, v): 

63 if v <= 0 or v > 0.5: 

64 raise ValueError("最大回撤必须在0-0.5之间") 

65 return v 

66 

67 @field_validator("max_single_order_ratio") 

68 @classmethod 

69 def validate_max_single_order_ratio(cls, v): 

70 if v <= 0 or v > 0.5: 

71 raise ValueError("单笔订单最大比例必须在0-0.5之间") 

72 return v 

73 

74 @field_validator("max_daily_loss_ratio") 

75 @classmethod 

76 def validate_max_daily_loss_ratio(cls, v): 

77 if v <= 0 or v > 0.1: 

78 raise ValueError("日最大亏损比例必须在0-0.1之间") 

79 return v 

80 

81 @field_validator("min_cash_ratio") 

82 @classmethod 

83 def validate_min_cash_ratio(cls, v): 

84 if v <= 0 or v > 0.5: 

85 raise ValueError("最小现金比例必须在0-0.5之间") 

86 return v 

87 

88 @field_validator("max_leverage") 

89 @classmethod 

90 def validate_max_leverage(cls, v): 

91 if v < 1.0 or v > 5.0: 

92 raise ValueError("最大杠杆比例必须在1.0-5.0之间") 

93 return v 

94 

95 def to_dict(self) -> Dict[str, Any]: 

96 """转换为字典""" 

97 return self.model_dump() 

98 

99 @classmethod 

100 def from_dict(cls, data: Dict[str, Any]) -> "RiskConfigModel": 

101 """从字典创建""" 

102 return cls(**data) 

103 

104 def get_default_config() -> "RiskConfigModel": 

105 """获取默认配置""" 

106 return RiskConfigModel() 

107 

108 def get_conservative_config() -> "RiskConfigModel": 

109 """获取保守配置""" 

110 return RiskConfigModel( 

111 max_position_ratio=0.05, 

112 stop_loss_ratio=0.03, 

113 max_drawdown=0.1, 

114 max_single_order_ratio=0.02, 

115 max_daily_loss_ratio=0.01, 

116 min_cash_ratio=0.2, 

117 max_leverage=1.0, 

118 ) 

119 

120 def get_aggressive_config() -> "RiskConfigModel": 

121 """获取激进配置""" 

122 return RiskConfigModel( 

123 max_position_ratio=0.2, 

124 stop_loss_ratio=0.08, 

125 max_drawdown=0.25, 

126 max_single_order_ratio=0.1, 

127 max_daily_loss_ratio=0.05, 

128 min_cash_ratio=0.05, 

129 max_leverage=2.0, 

130 ) 

131 

132 

133class RiskConfigTemplate(BaseModel): 

134 """风险配置模板""" 

135 

136 name: str = Field(..., description="模板名称") 

137 description: str = Field(..., description="模板描述") 

138 config: RiskConfigModel = Field(..., description="风险配置") 

139 is_default: bool = Field(False, description="是否为默认模板") 

140 

141 

142class RiskConfigPreset: 

143 """风险配置预设""" 

144 

145 @staticmethod 

146 def get_presets() -> List[RiskConfigTemplate]: 

147 """获取预设配置""" 

148 return [ 

149 RiskConfigTemplate( 

150 name="保守型", 

151 description="适合稳健投资者,风险控制严格", 

152 config=RiskConfigModel.get_conservative_config(), 

153 is_default=False, 

154 ), 

155 RiskConfigTemplate( 

156 name="平衡型", 

157 description="适合一般投资者,风险收益平衡", 

158 config=RiskConfigModel.get_default_config(), 

159 is_default=True, 

160 ), 

161 RiskConfigTemplate( 

162 name="激进型", 

163 description="适合激进投资者,追求高收益", 

164 config=RiskConfigModel.get_aggressive_config(), 

165 is_default=False, 

166 ), 

167 ] 

168 

169 @staticmethod 

170 def get_preset_by_name(name: str) -> Optional[RiskConfigTemplate]: 

171 """根据名称获取预设配置""" 

172 presets = RiskConfigPreset.get_presets() 

173 for preset in presets: 

174 if preset.name == name: 

175 return preset 

176 return None