Reinforcement Learning DeFi: AI Agent Yield Optimization Tutorial

# Core dependencies
npm install ethers hardhat @openzeppelin/contracts
pip install gym stable-baselines3 pandas numpy web3 ccxt

# DeFi protocol interfaces
npm install @uniswap/v3-sdk @uniswap/sdk-core
pip install uniswap-python compound-protocol-python

// config.js
module.exports = {
  ETHEREUM_RPC_URL: process.env.ETHEREUM_RPC_URL,
  PRIVATE_KEY: process.env.PRIVATE_KEY,
  UNISWAP_V3_ROUTER: '0xE592427A0AEce92De3Edee1F18E0157C05861564',
  COMPOUND_COMPTROLLER: '0x3d9819210A31b4961b30EF54bE2aeD79B9c9Cd3B'
};

import gym
from gym import spaces
import numpy as np
from web3 import Web3
import pandas as pd

class DeFiYieldEnvironment(gym.Env):
    def __init__(self, initial_balance=10000):
        super().__init__()
        
        # Define observation space (portfolio state + market data)
        self.observation_space = spaces.Box(
            low=0, high=np.inf, 
            shape=(20,), dtype=np.float32
        )
        
        # Define action space (allocation percentages across pools)
        self.action_space = spaces.Box(
            low=0, high=1, 
            shape=(5,), dtype=np.float32
        )
        
        self.initial_balance = initial_balance
        self.current_balance = initial_balance
        self.positions = {}
        self.step_count = 0
        
        # Initialize Web3 connection
        self.w3 = Web3(Web3.HTTPProvider(RPC_URL))
        self.pool_contracts = self._load_pool_contracts()
        
    def step(self, action):
        """Execute trading action and return new state, reward, done, info"""
        
        # Normalize action to ensure sum equals 1
        action = action / np.sum(action)
        
        # Calculate target allocation based on action
        target_allocation = {
            'uniswap_eth_usdc': action[0],
            'compound_dai': action[1], 
            'aave_usdt': action[2],
            'curve_3pool': action[3],
            'yearn_yfi': action[4]
        }
        
        # Execute rebalancing trades
        gas_costs = self._rebalance_portfolio(target_allocation)
        
        # Calculate yield earned this step
        yield_earned = self._calculate_yield_earned()
        
        # Calculate impermanent loss
        impermanent_loss = self._calculate_impermanent_loss()
        
        # Update portfolio value
        self.current_balance = self._get_portfolio_value()
        
        # Calculate reward (yield - gas - impermanent loss)
        reward = yield_earned - gas_costs - impermanent_loss
        
        # Get new observation
        observation = self._get_observation()
        
        # Check if episode is done (24 hours or significant loss)
        done = self.step_count >= 288 or self.current_balance < 0.5 * self.initial_balance
        
        info = {
            'balance': self.current_balance,
            'yield': yield_earned,
            'gas_costs': gas_costs,
            'impermanent_loss': impermanent_loss
        }
        
        self.step_count += 1
        return observation, reward, done, info
    
    def _get_observation(self):
        """Build observation vector from current state"""
        
        # Portfolio allocation percentages
        allocation = [self.positions.get(pool, 0) for pool in self.pool_contracts.keys()]
        
        # Current APY rates from each pool
        apy_rates = [self._get_pool_apy(pool) for pool in self.pool_contracts.keys()]
        
        # Token price ratios and volatility
        price_data = self._get_price_data()
        
        # Gas price and network congestion
        network_stats = [
            self.w3.eth.gas_price / 1e9,  # Gas price in Gwei
            self._get_network_congestion()
        ]
        
        # Time-based features
        time_features = [
            self.step_count / 288,  # Progress through day
            self._get_market_hours_indicator()
        ]
        
        observation = np.array(
            allocation + apy_rates + price_data + network_stats + time_features,
            dtype=np.float32
        )
        
        return observation
    
    def _rebalance_portfolio(self, target_allocation):
        """Execute trades to match target allocation"""
        gas_costs = 0
        
        for pool_name, target_weight in target_allocation.items():
            current_weight = self.positions.get(pool_name, 0)
            weight_diff = target_weight - current_weight
            
            if abs(weight_diff) > 0.01:  # Minimum rebalancing threshold
                trade_amount = weight_diff * self.current_balance
                
                if trade_amount > 0:
                    # Enter position
                    gas_costs += self._enter_position(pool_name, abs(trade_amount))
                else:
                    # Exit position
                    gas_costs += self._exit_position(pool_name, abs(trade_amount))
                
                self.positions[pool_name] = target_weight
        
        return gas_costs
    
    def reset(self):
        """Reset environment to initial state"""
        self.current_balance = self.initial_balance
        self.positions = {}
        self.step_count = 0
        return self._get_observation()

from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv
from stable_baselines3.common.callbacks import EvalCallback

def create_training_environment():
    """Create vectorized training environment"""
    
    def make_env():
        return DeFiYieldEnvironment(initial_balance=10000)
    
    env = DummyVecEnv([make_env])
    return env

def train_yield_agent():
    """Train RL agent for yield optimization"""
    
    # Create training environment
    env = create_training_environment()
    
    # Configure PPO agent with DeFi-specific parameters
    model = PPO(
        "MlpPolicy", 
        env,
        learning_rate=3e-4,
        n_steps=2048,
        batch_size=64,
        n_epochs=10,
        gamma=0.99,  # Discount factor for future rewards
        gae_lambda=0.95,  # Generalized Advantage Estimation
        clip_range=0.2,  # PPO clipping parameter
        verbose=1
    )
    
    # Create evaluation callback
    eval_env = create_training_environment()
    eval_callback = EvalCallback(
        eval_env,
        best_model_save_path='./models/',
        log_path='./logs/',
        eval_freq=5000,
        deterministic=True,
        render=False
    )
    
    # Train agent
    print("Starting RL training for DeFi yield optimization...")
    model.learn(
        total_timesteps=100000,
        callback=eval_callback
    )
    
    # Save trained model
    model.save("defi_yield_agent")
    print("Training completed. Model saved to defi_yield_agent.zip")
    
    return model

# Train the agent
trained_agent = train_yield_agent()

class RiskManager:
    def __init__(self, max_drawdown=0.2, max_single_position=0.4):
        self.max_drawdown = max_drawdown
        self.max_single_position = max_single_position
        self.peak_balance = 0
        
    def validate_action(self, action, current_state, balance_history):
        """Validate and potentially modify actions based on risk parameters"""
        
        # Update peak balance tracking
        current_balance = balance_history[-1]
        self.peak_balance = max(self.peak_balance, current_balance)
        
        # Calculate current drawdown
        drawdown = (self.peak_balance - current_balance) / self.peak_balance
        
        # Emergency stop if maximum drawdown exceeded
        if drawdown > self.max_drawdown:
            print(f"EMERGENCY STOP: Drawdown {drawdown:.2%} exceeds limit")
            return self._emergency_exit_action()
        
        # Normalize and cap position sizes
        normalized_action = action / np.sum(action)
        capped_action = np.minimum(normalized_action, self.max_single_position)
        
        # Renormalize after capping
        final_action = capped_action / np.sum(capped_action)
        
        return final_action
    
    def _emergency_exit_action(self):
        """Return action that exits all positions"""
        return np.array([0, 0, 0, 0, 0])  # Exit all positions

# Integrate risk management into trading loop
def safe_trading_step(agent, observation, risk_manager, balance_history):
    """Execute trading step with risk management overlay"""
    
    # Get agent's raw action
    raw_action, _ = agent.predict(observation)
    
    # Apply risk management
    safe_action = risk_manager.validate_action(
        raw_action, observation, balance_history
    )
    
    return safe_action

import ccxt
import asyncio
from web3.auto import w3

class MarketDataFeed:
    def __init__(self):
        self.exchanges = {
            'binance': ccxt.binance({'enableRateLimit': True}),
            'uniswap': self._init_uniswap_connection()
        }
        self.price_cache = {}
        
    async def get_live_prices(self, symbols):
        """Fetch current prices from multiple sources"""
        
        prices = {}
        
        # Get CEX prices
        for exchange_name, exchange in self.exchanges.items():
            if exchange_name != 'uniswap':
                try:
                    tickers = await exchange.fetch_tickers(symbols)
                    for symbol, ticker in tickers.items():
                        if symbol not in prices:
                            prices[symbol] = []
                        prices[symbol].append(ticker['last'])
                except Exception as e:
                    print(f"Error fetching from {exchange_name}: {e}")
        
        # Get DEX prices from Uniswap
        uniswap_prices = await self._get_uniswap_prices(symbols)
        for symbol, price in uniswap_prices.items():
            if symbol not in prices:
                prices[symbol] = []
            prices[symbol].append(price)
        
        # Calculate average prices
        avg_prices = {}
        for symbol, price_list in prices.items():
            avg_prices[symbol] = np.mean(price_list)
        
        self.price_cache = avg_prices
        return avg_prices
    
    async def get_pool_metrics(self, pool_addresses):
        """Fetch current APY and TVL for DeFi pools"""
        
        metrics = {}
        
        for pool_addr in pool_addresses:
            try:
                # Query pool contract for current metrics
                pool_contract = w3.eth.contract(
                    address=pool_addr,
                    abi=POOL_ABI
                )
                
                tvl = pool_contract.functions.getTotalValueLocked().call()
                apy = pool_contract.functions.getCurrentAPY().call()
                
                metrics[pool_addr] = {
                    'tvl': tvl / 1e18,  # Convert from wei
                    'apy': apy / 100    # Convert from basis points
                }
                
            except Exception as e:
                print(f"Error fetching metrics for {pool_addr}: {e}")
                
        return metrics

# Usage in main trading loop
async def main_trading_loop():
    """Main execution loop with live data"""
    
    # Initialize components
    env = DeFiYieldEnvironment()
    agent = PPO.load("defi_yield_agent")
    risk_manager = RiskManager()
    data_feed = MarketDataFeed()
    
    balance_history = [env.initial_balance]
    
    while True:
        # Get current market data
        prices = await data_feed.get_live_prices(['ETH/USD', 'BTC/USD'])
        pool_metrics = await data_feed.get_pool_metrics(MONITORED_POOLS)
        
        # Update environment with fresh data
        observation = env._get_observation()
        
        # Get safe action from agent
        action = safe_trading_step(
            agent, observation, risk_manager, balance_history
        )
        
        # Execute step
        obs, reward, done, info = env.step(action)
        balance_history.append(info['balance'])
        
        print(f"Step reward: {reward:.4f}, Balance: ${info['balance']:.2f}")
        
        if done:
            print("Episode completed")
            env.reset()
            balance_history = [env.initial_balance]
        
        # Wait 5 minutes before next step
        await asyncio.sleep(300)

# Run the trading system
if __name__ == "__main__":
    asyncio.run(main_trading_loop())

import matplotlib.pyplot as plt
import pandas as pd
from datetime import datetime

class PerformanceTracker:
    def __init__(self):
        self.trade_log = []
        self.balance_history = []
        self.reward_history = []
        
    def log_trade(self, timestamp, action, reward, balance, gas_cost):
        """Record trading step details"""
        
        self.trade_log.append({
            'timestamp': timestamp,
            'action': action.tolist(),
            'reward': reward,
            'balance': balance,
            'gas_cost': gas_cost
        })
        
        self.balance_history.append(balance)
        self.reward_history.append(reward)
    
    def calculate_metrics(self):
        """Calculate performance statistics"""
        
        if len(self.balance_history) < 2:
            return {}
        
        # Convert to pandas for easier analysis
        df = pd.DataFrame(self.trade_log)
        
        # Calculate returns
        initial_balance = self.balance_history[0]
        final_balance = self.balance_history[-1]
        total_return = (final_balance - initial_balance) / initial_balance
        
        # Calculate Sharpe ratio
        returns = np.diff(self.balance_history) / self.balance_history[:-1]
        sharpe_ratio = np.mean(returns) / np.std(returns) * np.sqrt(365 * 24 * 12)  # Annualized
        
        # Maximum drawdown
        peak_balance = np.maximum.accumulate(self.balance_history)
        drawdowns = (peak_balance - self.balance_history) / peak_balance
        max_drawdown = np.max(drawdowns)
        
        # Win rate
        positive_rewards = sum(1 for r in self.reward_history if r > 0)
        win_rate = positive_rewards / len(self.reward_history)
        
        # Gas efficiency
        total_gas = sum(trade.get('gas_cost', 0) for trade in self.trade_log)
        gas_percentage = total_gas / initial_balance
        
        metrics = {
            'total_return': total_return,
            'annualized_return': total_return * (365 * 24 * 12 / len(self.balance_history)),
            'sharpe_ratio': sharpe_ratio,
            'max_drawdown': max_drawdown,
            'win_rate': win_rate,
            'total_gas_cost': total_gas,
            'gas_percentage': gas_percentage,
            'total_trades': len(self.trade_log)
        }
        
        return metrics
    
    def generate_report(self):
        """Generate performance visualization"""
        
        metrics = self.calculate_metrics()
        
        # Create subplot figure
        fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 10))
        
        # Balance over time
        ax1.plot(self.balance_history)
        ax1.set_title('Portfolio Balance Over Time')
        ax1.set_ylabel('Balance ($)')
        ax1.grid(True)
        
        # Cumulative rewards
        cumulative_rewards = np.cumsum(self.reward_history)
        ax2.plot(cumulative_rewards)
        ax2.set_title('Cumulative Rewards')
        ax2.set_ylabel('Cumulative Reward')
        ax2.grid(True)
        
        # Reward distribution
        ax3.hist(self.reward_history, bins=50, alpha=0.7)
        ax3.set_title('Reward Distribution')
        ax3.set_xlabel('Reward')
        ax3.set_ylabel('Frequency')
        
        # Drawdown chart
        peak_balance = np.maximum.accumulate(self.balance_history)
        drawdowns = (peak_balance - self.balance_history) / peak_balance * 100
        ax4.fill_between(range(len(drawdowns)), 0, -drawdowns, alpha=0.5, color='red')
        ax4.set_title('Drawdown (%)')
        ax4.set_ylabel('Drawdown %')
        ax4.grid(True)
        
        plt.tight_layout()
        plt.savefig(f'performance_report_{datetime.now().strftime("%Y%m%d_%H%M")}.png')
        
        # Print metrics summary
        print("\n" + "="*50)
        print("PERFORMANCE SUMMARY")
        print("="*50)
        for metric, value in metrics.items():
            if isinstance(value, float):
                print(f"{metric.replace('_', ' ').title()}: {value:.4f}")
            else:
                print(f"{metric.replace('_', ' ').title()}: {value}")
        print("="*50)
        
        return metrics

# Integrate tracking into trading loop
tracker = PerformanceTracker()

# Modified trading step with logging
def tracked_trading_step(env, agent, risk_manager, tracker):
    observation = env._get_observation()
    action = safe_trading_step(agent, observation, risk_manager, tracker.balance_history)
    obs, reward, done, info = env.step(action)
    
    # Log performance data
    tracker.log_trade(
        timestamp=datetime.now(),
        action=action,
        reward=reward,
        balance=info['balance'],
        gas_cost=info.get('gas_costs', 0)
    )
    
    return obs, reward, done, info

# Generate daily reports
def daily_report_job():
    """Generate performance report and save results"""
    metrics = tracker.generate_report()
    
    # Save metrics to CSV for long-term tracking
    metrics_df = pd.DataFrame([metrics])
    metrics_df['date'] = datetime.now().date()
    
    # Append to historical metrics file
    try:
        historical = pd.read_csv('historical_metrics.csv')
        combined = pd.concat([historical, metrics_df])
    except FileNotFoundError:
        combined = metrics_df
    
    combined.to_csv('historical_metrics.csv', index=False)
    
    return metrics

# Schedule daily reporting
import schedule
schedule.every().day.at("00:00").do(daily_report_job)

import logging
import smtplib
from email.mime.text import MIMEText
import json

class ProductionDeployment:
    def __init__(self, config_file='production_config.json'):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        
        # Setup logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('defi_agent.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
        # Initialize alert system
        self.alert_email = self.config['alerts']['email']
        self.smtp_config = self.config['alerts']['smtp']
        
    def deploy_agent(self, model_path):
        """Deploy agent with production safeguards"""
        
        try:
            # Load trained model
            agent = PPO.load(model_path)
            self.logger.info(f"Loaded model from {model_path}")
            
            # Initialize production environment
            env = DeFiYieldEnvironment(
                initial_balance=self.config['trading']['initial_balance']
            )
            
            # Setup enhanced risk management
            risk_manager = RiskManager(
                max_drawdown=self.config['risk']['max_drawdown'],
                max_single_position=self.config['risk']['max_position_size']
            )
            
            # Start trading loop with error handling
            self._run_trading_loop(agent, env, risk_manager)
            
        except Exception as e:
            self.logger.error(f"Deployment failed: {str(e)}")
            self.send_alert(f"CRITICAL: Agent deployment failed - {str(e)}")
            raise
    
    def _run_trading_loop(self, agent, env, risk_manager):
        """Production trading loop with monitoring"""
        
        consecutive_errors = 0
        max_consecutive_errors = 5
        
        while True:
            try:
                # Execute trading step
                observation = env._get_observation()
                action = safe_trading_step(agent, observation, risk_manager, [])
                obs, reward, done, info = env.step(action)
                
                # Log successful step
                self.logger.info(
                    f"Trading step completed - Reward: {reward:.4f}, "
                    f"Balance: ${info['balance']:.2f}"
                )
                
                # Reset error counter
                consecutive_errors = 0
                
                # Check for alert conditions
                self._check_alert_conditions(info)
                
                if done:
                    self.logger.info("Episode completed, resetting environment")
                    env.reset()
                
                # Wait for next step
                time.sleep(self.config['trading']['step_interval'])
                
            except Exception as e:
                consecutive_errors += 1
                self.logger.error(f"Trading step error ({consecutive_errors}/{max_consecutive_errors}): {str(e)}")
                
                if consecutive_errors >= max_consecutive_errors:
                    self.send_alert(f"CRITICAL: {consecutive_errors} consecutive errors - {str(e)}")
                    break
                
                # Exponential backoff
                time.sleep(min(60 * (2 ** consecutive_errors), 300))
    
    def _check_alert_conditions(self, step_info):
        """Check if any alert conditions are met"""
        
        current_balance = step_info['balance']
        initial_balance = self.config['trading']['initial_balance']
        
        # Significant loss alert
        loss_percentage = (initial_balance - current_balance) / initial_balance
        if loss_percentage > self.config['alerts']['loss_threshold']:
            self.send_alert(
                f"ALERT: Portfolio loss of {loss_percentage:.2%} detected. "
                f"Current balance: ${current_balance:.2f}"
            )
        
        # High gas costs alert
        if step_info.get('gas_costs', 0) > self.config['alerts']['gas_threshold']:
            self.send_alert(
                f"ALERT: High gas costs detected: ${step_info['gas_costs']:.2f}"
            )
        
        # Unusual reward alert
        if step_info.get('reward', 0) < -self.config['alerts']['negative_reward_threshold']:
            self.send_alert(
                f"ALERT: Large negative reward: {step_info['reward']:.4f}"
            )
    
    def send_alert(self, message):
        """Send email alert"""
        
        try:
            msg = MIMEText(message)
            msg['Subject'] = 'DeFi Trading Agent Alert'
            msg['From'] = self.smtp_config['username']
            msg['To'] = self.alert_email
            
            server = smtplib.SMTP(self.smtp_config['host'], self.smtp_config['port'])
            server.starttls()
            server.login(self.smtp_config['username'], self.smtp_config['password'])
            server.sendmail(self.smtp_config['username'], [self.alert_email], msg.as_string())
            server.quit()
            
            self.logger.info(f"Alert sent: {message}")
            
        except Exception as e:
            self.logger.error(f"Failed to send alert: {str(e)}")

# Production configuration
production_config = {
    "trading": {
        "initial_balance": 50000,
        "step_interval": 300
    },
    "risk": {
        "max_drawdown": 0.15,
        "max_position_size": 0.3
    },
    "alerts": {
        "email": "admin@yourcompany.com",
        "smtp": {
            "host": "smtp.gmail.com",
            "port": 587,
            "username": "alerts@yourcompany.com",
            "password": "app_password"
        },
        "loss_threshold": 0.1,
        "gas_threshold": 100,
        "negative_reward_threshold": 50
    }
}

# Save configuration
with open('production_config.json', 'w') as f:
    json.dump(production_config, f, indent=2)

# Deploy to production
if __name__ == "__main__":
    deployment = ProductionDeployment()
    deployment.deploy_agent("defi_yield_agent.zip")