AlphaFuturesPro/backend/service_implementation/qihuo_analyzer/modules/risk_manager.py

# 风控管理模块
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'")