DeFi Yield Farming Strategies: Risk Assessment and Portfolio Optimization

// Yield composition analysis
const yieldSources = {
    tradingFees: {
        sustainability: "high",
        volatility: "low",
        description: "Generated from actual trading activity"
    },
    tokenIncentives: {
        sustainability: "medium",
        volatility: "high", 
        description: "Protocol tokens distributed to liquidity providers"
    },
    borrowingInterest: {
        sustainability: "high",
        volatility: "medium",
        description: "Interest paid by borrowers"
    },
    liquidationFees: {
        sustainability: "medium",
        volatility: "high",
        description: "Fees from liquidating undercollateralized positions"
    }
}

# risk_assessment.py
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple
 
class DeFiRiskAssessor:
    def __init__(self):
        self.risk_weights = {
            'smart_contract': 0.25,
            'impermanent_loss': 0.20,
            'liquidation': 0.20,
            'regulatory': 0.15,
            'counterparty': 0.10,
            'market': 0.10
        }
    
    def calculate_impermanent_loss(self, price_ratio: float) -> float:
        """
        Calculate impermanent loss for a 50/50 liquidity pool.
        
        Args:
            price_ratio: Current price / Initial price of one asset relative to the other
            
        Returns:
            Impermanent loss as a percentage
        """
        if price_ratio <= 0:
            return 1.0  # 100% loss
        
        # Formula: IL = (2 * sqrt(price_ratio)) / (1 + price_ratio) - 1
        il = (2 * np.sqrt(price_ratio)) / (1 + price_ratio) - 1
        return abs(il)
    
    def assess_protocol_risk(self, protocol_data: Dict) -> Dict:
        """
        Assess overall protocol risk based on multiple factors.
        
        Args:
            protocol_data: Dictionary containing protocol metrics
            
        Returns:
            Risk assessment with scores and recommendations
        """
        risk_score = 0
        risk_factors = {}
        
        # Smart contract risk assessment
        audit_score = self._assess_audit_quality(protocol_data.get('audits', []))
        risk_factors['smart_contract'] = audit_score
        risk_score += audit_score * self.risk_weights['smart_contract']
        
        # TVL and liquidity assessment
        tvl_score = self._assess_tvl_stability(protocol_data.get('tvl_history', []))
        risk_factors['liquidity'] = tvl_score
        
        # Token distribution analysis
        token_score = self._assess_token_distribution(protocol_data.get('token_distribution', {}))
        risk_factors['tokenomics'] = token_score
        
        # Calculate overall risk score (0-100, lower is better)
        overall_risk = min(100, risk_score * 100)
        
        return {
            'overall_risk_score': overall_risk,
            'risk_factors': risk_factors,
            'risk_level': self._categorize_risk(overall_risk),
            'recommendations': self._generate_recommendations(overall_risk, risk_factors)
        }
    
    def _assess_audit_quality(self, audits: List[Dict]) -> float:
        """Assess smart contract audit quality."""
        if not audits:
            return 0.8  # High risk if no audits
        
        audit_score = 0
        for audit in audits:
            # Score based on auditor reputation and findings
            auditor_weight = {
                'Trail of Bits': 0.9,
                'ConsenSys Diligence': 0.85,
                'OpenZeppelin': 0.8,
                'Quantstamp': 0.75
            }.get(audit.get('auditor'), 0.5)
            
            # Penalty for critical findings
            critical_findings = audit.get('critical_findings', 0)
            finding_penalty = min(0.4, critical_findings * 0.1)
            
            audit_score += auditor_weight - finding_penalty
        
        return min(1.0, audit_score / len(audits))
    
    def _assess_tvl_stability(self, tvl_history: List[float]) -> float:
        """Assess TVL stability and trend."""
        if len(tvl_history) < 30:  # Need at least 30 data points
            return 0.6
        
        # Calculate volatility
        returns = np.diff(tvl_history) / tvl_history[:-1]
        volatility = np.std(returns)
        
        # Lower volatility = lower risk
        stability_score = max(0.1, 1.0 - (volatility * 10))
        return stability_score
    
    def _assess_token_distribution(self, distribution: Dict) -> float:
        """Assess token distribution centralization."""
        if not distribution:
            return 0.7  # Medium risk if unknown
        
        # Check concentration in top holders
        top_10_percentage = distribution.get('top_10_holders_percentage', 50)
        
        # Higher concentration = higher risk
        if top_10_percentage > 70:
            return 0.8  # High risk
        elif top_10_percentage > 50:
            return 0.6  # Medium risk
        else:
            return 0.3  # Low risk
    
    def _categorize_risk(self, risk_score: float) -> str:
        """Categorize risk level."""
        if risk_score < 30:
            return "Low"
        elif risk_score < 60:
            return "Medium"
        else:
            return "High"
    
    def _generate_recommendations(self, risk_score: float, factors: Dict) -> List[str]:
        """Generate risk-based recommendations."""
        recommendations = []
        
        if risk_score > 70:
            recommendations.append("Consider reducing position size or avoiding this protocol")
        
        if factors.get('smart_contract', 0) > 0.7:
            recommendations.append("Wait for additional audits before investing")
        
        if factors.get('liquidity', 0) > 0.6:
            recommendations.append("Monitor TVL stability closely")
        
        recommendations.append("Implement stop-loss mechanisms")
        recommendations.append("Diversify across multiple protocols")
        
        return recommendations
 
# Example usage
if __name__ == "__main__":
    assessor = DeFiRiskAssessor()
    
    # Example protocol data
    protocol_data = {
        'audits': [
            {'auditor': 'Trail of Bits', 'critical_findings': 0},
            {'auditor': 'ConsenSys Diligence', 'critical_findings': 1}
        ],
        'tvl_history': np.random.normal(100000000, 5000000, 60).tolist(),  # Mock TVL data
        'token_distribution': {'top_10_holders_percentage': 45}
    }
    
    risk_assessment = assessor.assess_protocol_risk(protocol_data)
    print(f"Risk Assessment: {risk_assessment}")

# portfolio_optimizer.py
import numpy as np
import pandas as pd
from scipy.optimize import minimize
import matplotlib.pyplot as plt
from typing import Dict, List, Tuple
 
class DeFiPortfolioOptimizer:
    def __init__(self):
        self.risk_free_rate = 0.02  # 2% risk-free rate assumption
    
    def calculate_portfolio_metrics(self, weights: np.ndarray, 
                                  expected_returns: np.ndarray, 
                                  cov_matrix: np.ndarray) -> Tuple[float, float, float]:
        """
        Calculate portfolio return, risk, and Sharpe ratio.
        
        Args:
            weights: Portfolio weights
            expected_returns: Expected returns for each asset
            cov_matrix: Covariance matrix of returns
            
        Returns:
            Tuple of (expected_return, volatility, sharpe_ratio)
        """
        portfolio_return = np.sum(weights * expected_returns)
        portfolio_variance = np.dot(weights.T, np.dot(cov_matrix, weights))
        portfolio_volatility = np.sqrt(portfolio_variance)
        
        sharpe_ratio = (portfolio_return - self.risk_free_rate) / portfolio_volatility
        
        return portfolio_return, portfolio_volatility, sharpe_ratio
    
    def optimize_portfolio(self, expected_returns: np.ndarray, 
                          cov_matrix: np.ndarray,
                          target_return: float = None) -> Dict:
        """
        Optimize portfolio allocation using mean-variance optimization.
        
        Args:
            expected_returns: Expected annual returns for each protocol
            cov_matrix: Covariance matrix of returns
            target_return: Target portfolio return (optional)
            
        Returns:
            Optimization results with weights and metrics
        """
        n_assets = len(expected_returns)
        
        # Constraints
        constraints = [
            {'type': 'eq', 'fun': lambda x: np.sum(x) - 1}  # Weights sum to 1
        ]
        
        if target_return:
            constraints.append({
                'type': 'eq',
                'fun': lambda x: np.sum(x * expected_returns) - target_return
            })
        
        # Bounds (0% to 40% per protocol to ensure diversification)
        bounds = tuple((0, 0.4) for _ in range(n_assets))
        
        # Initial guess (equal weights)
        initial_guess = np.array([1.0 / n_assets] * n_assets)
        
        # Objective function (minimize portfolio variance)
        def objective(weights):
            return np.dot(weights.T, np.dot(cov_matrix, weights))
        
        # Optimize
        result = minimize(
            objective,
            initial_guess,
            method='SLSQP',
            bounds=bounds,
            constraints=constraints
        )
        
        if result.success:
            optimal_weights = result.x
            port_return, port_vol, sharpe = self.calculate_portfolio_metrics(
                optimal_weights, expected_returns, cov_matrix
            )
            
            return {
                'weights': optimal_weights,
                'expected_return': port_return,
                'volatility': port_vol,
                'sharpe_ratio': sharpe,
                'optimization_success': True
            }
        else:
            return {'optimization_success': False, 'error': result.message}
    
    def generate_efficient_frontier(self, expected_returns: np.ndarray,
                                  cov_matrix: np.ndarray,
                                  num_portfolios: int = 100) -> pd.DataFrame:
        """Generate efficient frontier for portfolio visualization."""
        
        min_return = np.min(expected_returns)
        max_return = np.max(expected_returns)
        target_returns = np.linspace(min_return, max_return, num_portfolios)
        
        efficient_portfolios = []
        
        for target in target_returns:
            result = self.optimize_portfolio(expected_returns, cov_matrix, target)
            
            if result['optimization_success']:
                efficient_portfolios.append({
                    'return': result['expected_return'],
                    'volatility': result['volatility'],
                    'sharpe_ratio': result['sharpe_ratio'],
                    'weights': result['weights']
                })
        
        return pd.DataFrame(efficient_portfolios)
 
# Example DeFi protocols analysis
def analyze_defi_protocols():
    """Analyze historical performance of major DeFi protocols."""
    
    # Sample protocol data (replace with real historical data)
    protocols = {
        'Uniswap V3 ETH/USDC': {
            'historical_apy': [15.2, 18.7, 12.4, 22.1, 16.8],
            'impermanent_loss_risk': 'medium',
            'smart_contract_risk': 'low',
            'liquidity': 'high'
        },
        'Compound USDC': {
            'historical_apy': [4.2, 5.1, 3.8, 4.7, 4.5],
            'impermanent_loss_risk': 'none',
            'smart_contract_risk': 'low',
            'liquidity': 'high'
        },
        'Curve 3Pool': {
            'historical_apy': [8.5, 9.2, 7.8, 10.1, 8.9],
            'impermanent_loss_risk': 'low',
            'smart_contract_risk': 'low',
            'liquidity': 'very_high'
        },
        'Yearn Finance vaults': {
            'historical_apy': [12.8, 15.4, 10.2, 18.7, 14.1],
            'impermanent_loss_risk': 'varies',
            'smart_contract_risk': 'medium',
            'liquidity': 'medium'
        }
    }
    
    # Calculate expected returns and covariance
    returns_data = []
    protocol_names = []
    
    for name, data in protocols.items():
        returns_data.append(data['historical_apy'])
        protocol_names.append(name)
    
    returns_df = pd.DataFrame(returns_data, index=protocol_names).T
    expected_returns = returns_df.mean().values / 100  # Convert to decimal
    cov_matrix = returns_df.cov().values / 10000  # Scale covariance
    
    return expected_returns, cov_matrix, protocol_names
 
# Example optimization
if __name__ == "__main__":
    expected_returns, cov_matrix, protocol_names = analyze_defi_protocols()
    
    optimizer = DeFiPortfolioOptimizer()
    result = optimizer.optimize_portfolio(expected_returns, cov_matrix)
    
    if result['optimization_success']:
        print("Optimal Portfolio Allocation:")
        for i, protocol in enumerate(protocol_names):
            print(f"{protocol}: {result['weights'][i]:.2%}")
        
        print(f"\nExpected Return: {result['expected_return']:.2%}")
        print(f"Volatility: {result['volatility']:.2%}")
        print(f"Sharpe Ratio: {result['sharpe_ratio']:.2f}")

// Example: Leveraged farming with Aave and Compound
contract LeveragedYieldFarmer {
    using SafeERC20 for IERC20;
    
    struct Position {
        address asset;
        uint256 collateralAmount;
        uint256 borrowedAmount;
        uint256 farmingAmount;
        uint256 leverageRatio;
    }
    
    mapping(address => Position) public positions;
    
    function openLeveragedPosition(
        address asset,
        uint256 initialAmount,
        uint256 targetLeverage
    ) external {
        require(targetLeverage <= 3e18, "Max 3x leverage");
        
        // 1. Deposit initial collateral to Aave
        IERC20(asset).safeTransferFrom(msg.sender, address(this), initialAmount);
        aavePool.supply(asset, initialAmount, address(this), 0);
        
        // 2. Calculate borrowing amount for target leverage
        uint256 borrowAmount = (initialAmount * (targetLeverage - 1e18)) / 1e18;
        
        // 3. Borrow additional assets
        aavePool.borrow(asset, borrowAmount, 2, 0, address(this));
        
        // 4. Deploy total amount to yield farming
        uint256 totalFarmingAmount = initialAmount + borrowAmount;
        yieldProtocol.deposit(asset, totalFarmingAmount);
        
        // 5. Record position
        positions[msg.sender] = Position({
            asset: asset,
            collateralAmount: initialAmount,
            borrowedAmount: borrowAmount,
            farmingAmount: totalFarmingAmount,
            leverageRatio: targetLeverage
        });
        
        emit PositionOpened(msg.sender, asset, totalFarmingAmount, targetLeverage);
    }
    
    function monitorPosition(address user) external view returns (uint256 healthFactor) {
        Position memory pos = positions[user];
        
        // Get current collateral value
        uint256 collateralValue = aavePool.getUserAccountData(address(this)).totalCollateralETH;
        uint256 debtValue = aavePool.getUserAccountData(address(this)).totalDebtETH;
        
        // Calculate health factor
        healthFactor = (collateralValue * 8500) / (debtValue * 10000); // 85% LTV
        
        return healthFactor;
    }
    
    function autoRebalance(address user) external {
        uint256 healthFactor = monitorPosition(user);
        
        // Trigger rebalancing if health factor drops below 1.2
        if (healthFactor < 1.2e18) {
            _reducePosition(user, 20); // Reduce by 20%
        }
    }
}

# cross_chain_arbitrage.py
import asyncio
import aiohttp
from web3 import Web3
from typing import Dict, List
 
class CrossChainYieldArbitrage:
    def __init__(self):
        self.chains = {
            'ethereum': {
                'rpc': 'https://eth-mainnet.alchemyapi.io/v2/YOUR_KEY',
                'protocols': ['compound', 'aave', 'uniswap']
            },
            'polygon': {
                'rpc': 'https://polygon-mainnet.alchemyapi.io/v2/YOUR_KEY', 
                'protocols': ['aave', 'quickswap', 'curve']
            },
            'arbitrum': {
                'rpc': 'https://arb-mainnet.alchemyapi.io/v2/YOUR_KEY',
                'protocols': ['gmx', 'radiant', 'camelot']
            }
        }
        
        self.bridge_costs = {
            ('ethereum', 'polygon'): 0.002,  # 0.2% bridge cost
            ('ethereum', 'arbitrum'): 0.001,  # 0.1% bridge cost
            ('polygon', 'arbitrum'): 0.0015,  # 0.15% bridge cost
        }
    
    async def fetch_yields(self) -> Dict:
        """Fetch current yields across all chains and protocols."""
        yields = {}
        
        async with aiohttp.ClientSession() as session:
            for chain, config in self.chains.items():
                yields[chain] = {}
                
                for protocol in config['protocols']:
                    # Mock API calls (replace with real protocol APIs)
                    try:
                        url = f"https://api.{protocol}.com/yields"
                        async with session.get(url) as response:
                            if response.status == 200:
                                data = await response.json()
                                yields[chain][protocol] = data.get('apy', 0)
                            else:
                                yields[chain][protocol] = 0
                    except:
                        yields[chain][protocol] = 0
        
        return yields
    
    def calculate_arbitrage_opportunity(self, yields: Dict, 
                                     asset: str, 
                                     amount: float) -> List[Dict]:
        """Calculate profitable arbitrage opportunities."""
        opportunities = []
        
        for source_chain in yields:
            for source_protocol in yields[source_chain]:
                source_yield = yields[source_chain][source_protocol]
                
                for target_chain in yields:
                    if source_chain == target_chain:
                        continue
                    
                    for target_protocol in yields[target_chain]:
                        target_yield = yields[target_chain][target_protocol]
                        
                        # Calculate bridge cost
                        bridge_key = (source_chain, target_chain)
                        bridge_cost = self.bridge_costs.get(bridge_key, 0.005)  # Default 0.5%
                        
                        # Calculate net arbitrage profit
                        yield_diff = target_yield - source_yield
                        net_profit = yield_diff - (bridge_cost * 2)  # Round trip cost
                        
                        if net_profit > 0.01:  # Minimum 1% profit threshold
                            opportunities.append({
                                'source': f"{source_chain}/{source_protocol}",
                                'target': f"{target_chain}/{target_protocol}",
                                'source_yield': source_yield,
                                'target_yield': target_yield,
                                'bridge_cost': bridge_cost * 2,
                                'net_profit': net_profit,
                                'profit_amount': amount * net_profit
                            })
        
        # Sort by profitability
        return sorted(opportunities, key=lambda x: x['net_profit'], reverse=True)
    
    async def monitor_arbitrage(self, asset: str, amount: float, 
                              min_profit_threshold: float = 0.02):
        """Continuously monitor for arbitrage opportunities."""
        while True:
            try:
                yields = await self.fetch_yields()
                opportunities = self.calculate_arbitrage_opportunity(yields, asset, amount)
                
                profitable_ops = [op for op in opportunities if op['net_profit'] > min_profit_threshold]
                
                if profitable_ops:
                    print(f"Found {len(profitable_ops)} arbitrage opportunities:")
                    for op in profitable_ops[:3]:  # Top 3
                        print(f"  {op['source']} → {op['target']}: {op['net_profit']:.2%} profit")
                
                # Wait 5 minutes before next check
                await asyncio.sleep(300)
                
            except Exception as e:
                print(f"Error monitoring arbitrage: {e}")
                await asyncio.sleep(60)  # Wait 1 minute on error
 
# Example usage
async def main():
    arbitrage = CrossChainYieldArbitrage()
    await arbitrage.monitor_arbitrage("USDC", 10000, min_profit_threshold=0.02)
 
if __name__ == "__main__":
    asyncio.run(main())

# yield_monitor.py
import asyncio
import logging
from dataclasses import dataclass
from typing import Dict, List
import json
from datetime import datetime, timedelta
 
@dataclass
class YieldPosition:
    protocol: str
    asset: str
    amount: float
    entry_apy: float
    current_apy: float
    duration_days: int
    total_earned: float
    impermanent_loss: float
 
class YieldFarmMonitor:
    def __init__(self, alert_thresholds: Dict = None):
        self.positions: List[YieldPosition] = []
        self.alert_thresholds = alert_thresholds or {
            'apy_drop_threshold': 0.5,  # Alert if APY drops 50%
            'impermanent_loss_threshold': 0.05,  # Alert if IL > 5%
            'health_factor_threshold': 1.3  # Alert if health factor < 1.3
        }
        
        # Setup logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def add_position(self, position: YieldPosition):
        """Add a new yield farming position to monitor."""
        self.positions.append(position)
        self.logger.info(f"Added position: {position.protocol} - {position.asset}")
    
    async def check_position_health(self, position: YieldPosition) -> Dict:
        """Check the health of a specific position."""
        alerts = []
        
        # Check APY degradation
        apy_change = (position.current_apy - position.entry_apy) / position.entry_apy
        if apy_change < -self.alert_thresholds['apy_drop_threshold']:
            alerts.append({
                'type': 'apy_drop',
                'severity': 'high',
                'message': f"APY dropped {abs(apy_change):.1%} from entry"
            })
        
        # Check impermanent loss
        if position.impermanent_loss > self.alert_thresholds['impermanent_loss_threshold']:
            alerts.append({
                'type': 'impermanent_loss',
                'severity': 'medium',
                'message': f"Impermanent loss: {position.impermanent_loss:.2%}"
            })
        
        # Calculate position ROI
        roi = (position.total_earned - (position.amount * position.impermanent_loss)) / position.amount
        
        return {
            'position': position,
            'roi': roi,
            'alerts': alerts,
            'health_score': self._calculate_health_score(position, alerts)
        }
    
    def _calculate_health_score(self, position: YieldPosition, alerts: List[Dict]) -> float:
        """Calculate overall health score for a position (0-100)."""
        base_score = 100
        
        # Penalize based on alerts
        for alert in alerts:
            if alert['severity'] == 'high':
                base_score -= 30
            elif alert['severity'] == 'medium':
                base_score -= 15
            else:
                base_score -= 5
        
        # Adjust based on performance
        if position.current_apy > position.entry_apy:
            base_score += 10  # Bonus for outperforming
        
        return max(0, min(100, base_score))
    
    async def generate_performance_report(self) -> Dict:
        """Generate comprehensive performance report."""
        total_invested = sum(pos.amount for pos in self.positions)
        total_earned = sum(pos.total_earned for pos in self.positions)
        total_il = sum(pos.amount * pos.impermanent_loss for pos in self.positions)
        
        net_profit = total_earned - total_il
        overall_roi = net_profit / total_invested if total_invested > 0 else 0
        
        # Category breakdown
        category_performance = {}
        for pos in self.positions:
            category = self._categorize_protocol(pos.protocol)
            if category not in category_performance:
                category_performance[category] = {
                    'invested': 0,
                    'earned': 0,
                    'count': 0
                }
            
            category_performance[category]['invested'] += pos.amount
            category_performance[category]['earned'] += pos.total_earned
            category_performance[category]['count'] += 1
        
        return {
            'timestamp': datetime.now().isoformat(),
            'overall_performance': {
                'total_invested': total_invested,
                'total_earned': total_earned,
                'total_impermanent_loss': total_il,
                'net_profit': net_profit,
                'roi': overall_roi
            },
            'category_breakdown': category_performance,
            'position_count': len(self.positions),
            'avg_apy': np.mean([pos.current_apy for pos in self.positions])
        }
    
    def _categorize_protocol(self, protocol: str) -> str:
        """Categorize protocol by type."""
        if any(dex in protocol.lower() for dex in ['uniswap', 'sushiswap', 'curve']):
            return 'DEX_LP'
        elif any(lending in protocol.lower() for lending in ['compound', 'aave', 'venus']):
            return 'LENDING'
        elif any(yield_agg in protocol.lower() for yield_agg in ['yearn', 'harvest', 'autofarm']):
            return 'YIELD_AGGREGATOR'
        else:
            return 'OTHER'
 
# Example monitoring setup
async def setup_monitoring():
    monitor = YieldFarmMonitor()
    
    # Add sample positions
    positions = [
        YieldPosition(
            protocol="Uniswap V3 ETH/USDC",
            asset="ETH-USDC",
            amount=10000,
            entry_apy=0.15,
            current_apy=0.12,
            duration_days=30,
            total_earned=123.45,
            impermanent_loss=0.02
        ),
        YieldPosition(
            protocol="Compound USDC",
            asset="USDC",
            amount=5000,
            entry_apy=0.045,
            current_apy=0.042,
            duration_days=45,
            total_earned=28.75,
            impermanent_loss=0.0
        )
    ]
    
    for pos in positions:
        monitor.add_position(pos)
    
    # Generate performance report
    report = await monitor.generate_performance_report()
    print(json.dumps(report, indent=2))
 
if __name__ == "__main__":
    asyncio.run(setup_monitoring())

# stop_loss_manager.py
import asyncio
from web3 import Web3
from typing import Dict, List
 
class DeFiStopLossManager:
    def __init__(self, web3_provider: str):
        self.w3 = Web3(Web3.HTTPProvider(web3_provider))
        self.monitored_positions = {}
        self.stop_loss_rules = {}
    
    def set_stop_loss(self, position_id: str, 
                     stop_loss_percentage: float,
                     trailing_stop: bool = False):
        """Set stop-loss rules for a position."""
        self.stop_loss_rules[position_id] = {
            'stop_loss_percentage': stop_loss_percentage,
            'trailing_stop': trailing_stop,
            'highest_value': None,
            'triggered': False
        }
    
    async def monitor_positions(self):
        """Continuously monitor positions for stop-loss triggers."""
        while True:
            for position_id, position in self.monitored_positions.items():
                if position_id not in self.stop_loss_rules:
                    continue
                
                rule = self.stop_loss_rules[position_id]
                if rule['triggered']:
                    continue
                
                current_value = await self._get_position_value(position)
                entry_value = position['entry_value']
                
                # Update highest value for trailing stop
                if rule['trailing_stop']:
                    if rule['highest_value'] is None or current_value > rule['highest_value']:
                        rule['highest_value'] = current_value
                
                # Check stop-loss trigger
                reference_value = rule['highest_value'] if rule['trailing_stop'] else entry_value
                loss_percentage = (reference_value - current_value) / reference_value
                
                if loss_percentage >= rule['stop_loss_percentage']:
                    await self._execute_stop_loss(position_id, position)
                    rule['triggered'] = True
            
            await asyncio.sleep(60)  # Check every minute
    
    async def _execute_stop_loss(self, position_id: str, position: Dict):
        """Execute stop-loss by closing the position."""
        self.logger.warning(f"Executing stop-loss for position {position_id}")
        
        try:
            # Implementation depends on specific protocol
            # This is a simplified example
            
            if position['protocol'] == 'uniswap_v3':
                await self._close_uniswap_position(position)
            elif position['protocol'] == 'compound':
                await self._close_compound_position(position)
            
            self.logger.info(f"Stop-loss executed successfully for {position_id}")
            
        except Exception as e:
            self.logger.error(f"Failed to execute stop-loss for {position_id}: {e}")
    
    async def _get_position_value(self, position: Dict) -> float:
        """Get current USD value of a position."""
        # Implementation varies by protocol
        # This would typically involve calling protocol contracts
        # and fetching current token prices
        pass
    
    async def _close_uniswap_position(self, position: Dict):
        """Close Uniswap V3 liquidity position."""
        # Implement Uniswap V3 position closing logic
        pass
    
    async def _close_compound_position(self, position: Dict):
        """Close Compound lending position."""
        # Implement Compound position closing logic
        pass
 
# Example usage
async def main():
    stop_loss_manager = DeFiStopLossManager("https://eth-mainnet.alchemyapi.io/v2/YOUR_KEY")
    
    # Add position monitoring
    stop_loss_manager.monitored_positions['pos_1'] = {
        'protocol': 'uniswap_v3',
        'token_pair': 'ETH/USDC',
        'entry_value': 10000,
        'position_id': '12345'
    }
    
    # Set 10% stop-loss with trailing stop
    stop_loss_manager.set_stop_loss('pos_1', 0.10, trailing_stop=True)
    
    # Start monitoring
    await stop_loss_manager.monitor_positions()
 
if __name__ == "__main__":
    asyncio.run(main())