# 风控管理模块 import pandas as pd from typing import Dict, Optional, Tuple from qihuo_analyzer.utils.technical_analysis import calculate_atr from qihuo_analyzer.core.models import StrategyConfig, RiskParams class RiskManager: """风险管理器""" def __init__(self, config: Optional[StrategyConfig] = None): self.config = config or StrategyConfig() def calculate_stop_loss(self, data: pd.DataFrame, entry_price: float, direction: str, atr_multiplier: Optional[float] = None) -> float: """计算止损位""" atr_multiplier = atr_multiplier or self.config.atr_multiplier # 计算ATR atr = calculate_atr(data).iloc[-1] # 根据方向计算止损位 if direction == 'long': stop_loss = entry_price - (atr * atr_multiplier) elif direction == 'short': stop_loss = entry_price + (atr * atr_multiplier) else: raise ValueError("Direction must be 'long' or 'short'") return stop_loss def calculate_position_size(self, account_balance: float, data: pd.DataFrame, direction: str, entry_price: float, contract_multiplier: float = 10, margin_rate: float = 0.1) -> Dict: """计算仓位大小""" # 计算ATR atr = calculate_atr(data).iloc[-1] # 计算每手风险 if direction == 'long': risk_per_unit = atr * self.config.atr_multiplier * contract_multiplier elif direction == 'short': risk_per_unit = atr * self.config.atr_multiplier * contract_multiplier else: raise ValueError("Direction must be 'long' or 'short'") # 计算最大风险金额 max_risk_amount = account_balance * self.config.max_risk_percent # 计算建议手数 suggested_units = max_risk_amount / risk_per_unit suggested_units = max(1, int(suggested_units)) # 至少1手 # 计算保证金需求 margin_per_unit = entry_price * contract_multiplier * margin_rate total_margin = suggested_units * margin_per_unit # 计算实际风险比例 actual_risk_percent = (risk_per_unit * suggested_units) / account_balance # 计算杠杆比例 leverage = (suggested_units * entry_price * contract_multiplier) / account_balance return { 'suggested_units': suggested_units, 'risk_per_unit': risk_per_unit, 'max_risk_amount': max_risk_amount, 'margin_per_unit': margin_per_unit, 'total_margin': total_margin, 'actual_risk_percent': actual_risk_percent, 'leverage': leverage, 'atr': atr } def calculate_profit_loss_ratio(self, entry_price: float, stop_loss: float, target_price: float, direction: str) -> float: """计算盈亏比""" if direction == 'long': profit = target_price - entry_price loss = entry_price - stop_loss elif direction == 'short': profit = entry_price - target_price loss = stop_loss - entry_price else: raise ValueError("Direction must be 'long' or 'short'") if loss == 0: return float('inf') return profit / loss def validate_trade(self, account_balance: float, data: pd.DataFrame, direction: str, entry_price: float, target_price: float, contract_multiplier: float = 10, margin_rate: float = 0.1) -> Dict: """验证交易是否符合风控要求""" # 计算止损位 stop_loss = self.calculate_stop_loss(data, entry_price, direction) # 计算盈亏比 pl_ratio = self.calculate_profit_loss_ratio(entry_price, stop_loss, target_price, direction) # 计算仓位大小 position_info = self.calculate_position_size(account_balance, data, direction, entry_price, contract_multiplier, margin_rate) # 检查各项风控指标 checks = { 'profit_loss_ratio': { 'value': pl_ratio, 'required': self.config.min_profit_loss_ratio, 'pass': pl_ratio >= self.config.min_profit_loss_ratio }, 'risk_percent': { 'value': position_info['actual_risk_percent'] * 100, 'required': self.config.max_risk_percent * 100, 'pass': position_info['actual_risk_percent'] <= self.config.max_risk_percent }, 'leverage': { 'value': position_info['leverage'], 'required': 5, # 最大杠杆 'pass': position_info['leverage'] <= 5 }, 'margin_utilization': { 'value': (position_info['total_margin'] / account_balance) * 100, 'required': 30, # 最大保证金使用率 'pass': (position_info['total_margin'] / account_balance) <= 0.3 } } # 综合判断 all_passed = all(check['pass'] for check in checks.values()) return { 'valid': all_passed, 'checks': checks, 'position_info': position_info, 'stop_loss': stop_loss, 'profit_loss_ratio': pl_ratio } def generate_risk_report(self, account_balance: float, data: pd.DataFrame, direction: str, entry_price: float, target_price: float, contract_multiplier: float = 10, margin_rate: float = 0.1) -> Dict: """生成风险报告""" # 验证交易 validation_result = self.validate_trade(account_balance, data, direction, entry_price, target_price, contract_multiplier, margin_rate) # 生成风险建议 suggestions = [] if not validation_result['checks']['profit_loss_ratio']['pass']: suggestions.append(f"盈亏比不足，建议调整目标价至{self._calculate_adjusted_target(entry_price, validation_result['stop_loss'], direction):.2f}") if not validation_result['checks']['risk_percent']['pass']: suggestions.append(f"风险比例过高，建议减少仓位至{int(validation_result['position_info']['suggested_units'] * 0.8)}手") if not validation_result['checks']['leverage']['pass']: suggestions.append("杠杆比例过高，建议降低仓位") if not validation_result['checks']['margin_utilization']['pass']: suggestions.append("保证金使用率过高，建议减少仓位") # 计算风险回报比 risk_return_ratio = self._calculate_risk_return_ratio(validation_result['profit_loss_ratio'], validation_result['position_info']['actual_risk_percent']) report = { 'account_balance': account_balance, 'direction': direction, 'entry_price': entry_price, 'stop_loss': validation_result['stop_loss'], 'target_price': target_price, 'profit_loss_ratio': validation_result['profit_loss_ratio'], 'position_info': validation_result['position_info'], 'risk_metrics': { 'risk_return_ratio': risk_return_ratio, 'max_drawdown_estimate': self._estimate_max_drawdown(account_balance, validation_result['position_info']), 'recovery_factor': self._calculate_recovery_factor(risk_return_ratio) }, 'suggestions': suggestions, 'validation_result': validation_result } return report def _calculate_adjusted_target(self, entry_price: float, stop_loss: float, direction: str) -> float: """计算调整后的目标价""" if direction == 'long': loss = entry_price - stop_loss required_profit = loss * self.config.min_profit_loss_ratio return entry_price + required_profit elif direction == 'short': loss = stop_loss - entry_price required_profit = loss * self.config.min_profit_loss_ratio return entry_price - required_profit else: raise ValueError("Direction must be 'long' or 'short'") def _calculate_risk_return_ratio(self, pl_ratio: float, risk_percent: float) -> float: """计算风险回报比""" return pl_ratio * (1 - risk_percent) def _estimate_max_drawdown(self, account_balance: float, position_info: Dict) -> float: """估算最大回撤""" max_loss = position_info['risk_per_unit'] * position_info['suggested_units'] return (max_loss / account_balance) * 100 def _calculate_recovery_factor(self, risk_return_ratio: float) -> float: """计算恢复因子""" if risk_return_ratio <= 0: return 0 return risk_return_ratio * 0.8 def monitor_position_risk(self, current_price: float, entry_price: float, stop_loss: float, target_price: float, direction: str, units: int, contract_multiplier: float = 10) -> Dict: """监控持仓风险""" # 计算当前盈亏 if direction == 'long': current_profit = (current_price - entry_price) * units * contract_multiplier distance_to_stop = entry_price - current_price distance_to_target = target_price - current_price elif direction == 'short': current_profit = (entry_price - current_price) * units * contract_multiplier distance_to_stop = current_price - entry_price distance_to_target = entry_price - current_price else: raise ValueError("Direction must be 'long' or 'short'") # 计算浮盈比例 unrealized_pnl_percent = (current_profit / (entry_price * units * contract_multiplier)) * 100 # 计算止损触发距离 stop_percent = (distance_to_stop / entry_price) * 100 # 计算目标达成距离 target_percent = (distance_to_target / entry_price) * 100 # 风险状态评估 risk_status = self._assess_risk_status(current_price, stop_loss, target_price, direction) return { 'current_price': current_price, 'entry_price': entry_price, 'stop_loss': stop_loss, 'target_price': target_price, 'current_profit': current_profit, 'unrealized_pnl_percent': unrealized_pnl_percent, 'distance_to_stop': distance_to_stop, 'distance_to_target': distance_to_target, 'stop_percent': stop_percent, 'target_percent': target_percent, 'risk_status': risk_status } def _assess_risk_status(self, current_price: float, stop_loss: float, target_price: float, direction: str) -> str: """评估风险状态""" if direction == 'long': if current_price <= stop_loss: return 'stop_loss_triggered' elif current_price >= target_price: return 'target_reached' elif current_price > stop_loss * 1.05: return 'low_risk' else: return 'medium_risk' elif direction == 'short': if current_price >= stop_loss: return 'stop_loss_triggered' elif current_price <= target_price: return 'target_reached' elif current_price < stop_loss * 0.95: return 'low_risk' else: return 'medium_risk' else: raise ValueError("Direction must be 'long' or 'short'")