""" FinRL Agent for Algorithmic Trading This module provides a FinRL-based reinforcement learning agent that can be integrated with the existing algorithmic trading system. It supports various RL algorithms including PPO, A2C, DDPG, and TD3. """ import numpy as np import pandas as pd import gymnasium as gym from gymnasium import spaces from stable_baselines3 import PPO, A2C, DDPG, TD3 from stable_baselines3.common.vec_env import DummyVecEnv from stable_baselines3.common.callbacks import EvalCallback import torch import logging from typing import Dict, List, Tuple, Optional, Any from dataclasses import dataclass import yaml logger = logging.getLogger(__name__) @dataclass class FinRLConfig: """Configuration for FinRL agent""" algorithm: str = "PPO" # PPO, A2C, DDPG, TD3 learning_rate: float = 0.0003 batch_size: int = 64 buffer_size: int = 1000000 learning_starts: int = 100 gamma: float = 0.99 tau: float = 0.005 train_freq: int = 1 gradient_steps: int = 1 target_update_interval: int = 1 exploration_fraction: float = 0.1 exploration_initial_eps: float = 1.0 exploration_final_eps: float = 0.05 max_grad_norm: float = 10.0 verbose: int = 1 tensorboard_log: str = "logs/finrl_tensorboard" class TradingEnvironment(gym.Env): """ Custom trading environment for FinRL This environment simulates a trading scenario where the agent can: - Buy, sell, or hold positions - Use technical indicators for decision making - Manage portfolio value and risk """ def __init__(self, data: pd.DataFrame, initial_balance: float = 100000, transaction_fee: float = 0.001, max_position: int = 100): super().__init__() self.data = data self.initial_balance = initial_balance self.transaction_fee = transaction_fee self.max_position = max_position # Reset state self.reset() # Define action space: [-1, 0, 1] for sell, hold, buy self.action_space = spaces.Discrete(3) # Define observation space # Features: OHLCV + technical indicators + portfolio state n_features = len(self._get_features(self.data.iloc[0])) self.observation_space = spaces.Box( low=-np.inf, high=np.inf, shape=(n_features,), dtype=np.float32 ) def _get_features(self, row: pd.Series) -> np.ndarray: """Extract features from market data row""" features = [] # Price features features.extend([ row['open'], row['high'], row['low'], row['close'], row['volume'] ]) # Technical indicators (if available) for indicator in ['sma_20', 'sma_50', 'rsi', 'bb_upper', 'bb_lower', 'macd']: if indicator in row.index: features.append(row[indicator]) else: features.append(0.0) # Portfolio state features.extend([ self.balance, self.position, self.portfolio_value, self.total_return ]) return np.array(features, dtype=np.float32) def _calculate_portfolio_value(self) -> float: """Calculate current portfolio value""" current_price = self.data.iloc[self.current_step]['close'] return self.balance + (self.position * current_price) def _calculate_reward(self) -> float: """Calculate reward based on portfolio performance""" current_value = self._calculate_portfolio_value() previous_value = self.previous_portfolio_value # Calculate return if previous_value > 0: return (current_value - previous_value) / previous_value else: return 0.0 def step(self, action: int) -> Tuple[np.ndarray, float, bool, bool, Dict]: """Execute one step in the environment""" # Get current market data current_data = self.data.iloc[self.current_step] current_price = current_data['close'] # Execute action if action == 0: # Sell if self.position > 0: shares_to_sell = min(self.position, self.max_position) sell_value = shares_to_sell * current_price * (1 - self.transaction_fee) self.balance += sell_value self.position -= shares_to_sell elif action == 2: # Buy if self.balance > 0: max_shares = min( int(self.balance / current_price), self.max_position - self.position ) if max_shares > 0: buy_value = max_shares * current_price * (1 + self.transaction_fee) self.balance -= buy_value self.position += max_shares # Update portfolio value self.previous_portfolio_value = self.portfolio_value self.portfolio_value = self._calculate_portfolio_value() self.total_return = (self.portfolio_value - self.initial_balance) / self.initial_balance # Calculate reward reward = self._calculate_reward() # Move to next step self.current_step += 1 # Check if episode is done done = self.current_step >= len(self.data) - 1 # Get observation if not done: observation = self._get_features(self.data.iloc[self.current_step]) else: # Use last available data for final observation observation = self._get_features(self.data.iloc[-1]) info = { 'balance': self.balance, 'position': self.position, 'portfolio_value': self.portfolio_value, 'total_return': self.total_return, 'current_price': current_price } return observation, reward, done, False, info def reset(self, seed: Optional[int] = None) -> Tuple[np.ndarray, Dict]: """Reset the environment""" super().reset(seed=seed) self.current_step = 0 self.balance = self.initial_balance self.position = 0 self.portfolio_value = self.initial_balance self.previous_portfolio_value = self.initial_balance self.total_return = 0.0 observation = self._get_features(self.data.iloc[self.current_step]) info = { 'balance': self.balance, 'position': self.position, 'portfolio_value': self.portfolio_value, 'total_return': self.total_return } return observation, info class FinRLAgent: """ FinRL-based reinforcement learning agent for algorithmic trading """ def __init__(self, config: FinRLConfig): self.config = config self.model = None self.env = None self.eval_env = None self.callback = None logger.info(f"Initializing FinRL agent with algorithm: {config.algorithm}") def create_environment(self, data: pd.DataFrame, initial_balance: float = 100000) -> TradingEnvironment: """Create trading environment from market data""" return TradingEnvironment( data=data, initial_balance=initial_balance, transaction_fee=0.001, max_position=100 ) def prepare_data(self, data: pd.DataFrame) -> pd.DataFrame: """Prepare data with technical indicators for FinRL""" df = data.copy() # Add technical indicators if not present if 'sma_20' not in df.columns: df['sma_20'] = df['close'].rolling(window=20).mean() if 'sma_50' not in df.columns: df['sma_50'] = df['close'].rolling(window=50).mean() if 'rsi' not in df.columns: df['rsi'] = self._calculate_rsi(df['close']) if 'bb_upper' not in df.columns or 'bb_lower' not in df.columns: bb_upper, bb_lower = self._calculate_bollinger_bands(df['close']) df['bb_upper'] = bb_upper df['bb_lower'] = bb_lower if 'macd' not in df.columns: df['macd'] = self._calculate_macd(df['close']) # Fill NaN values df = df.fillna(method='bfill').fillna(0) return df def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> pd.Series: """Calculate RSI indicator""" delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi def _calculate_bollinger_bands(self, prices: pd.Series, period: int = 20, std_dev: int = 2) -> Tuple[pd.Series, pd.Series]: """Calculate Bollinger Bands""" sma = prices.rolling(window=period).mean() std = prices.rolling(window=period).std() upper_band = sma + (std * std_dev) lower_band = sma - (std * std_dev) return upper_band, lower_band def _calculate_macd(self, prices: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.Series: """Calculate MACD indicator""" ema_fast = prices.ewm(span=fast).mean() ema_slow = prices.ewm(span=slow).mean() macd_line = ema_fast - ema_slow return macd_line def train(self, data: pd.DataFrame, total_timesteps: int = 100000, eval_freq: int = 10000, eval_data: Optional[pd.DataFrame] = None) -> Dict[str, Any]: """Train the FinRL agent""" logger.info("Starting FinRL agent training") # Prepare data train_data = self.prepare_data(data) # Create training environment self.env = DummyVecEnv([lambda: self.create_environment(train_data)]) # Create evaluation environment if provided if eval_data is not None: eval_data = self.prepare_data(eval_data) self.eval_env = DummyVecEnv([lambda: self.create_environment(eval_data)]) self.callback = EvalCallback( self.eval_env, best_model_save_path="models/finrl_best/", log_path="logs/finrl_eval/", eval_freq=eval_freq, deterministic=True, render=False ) # Initialize model based on algorithm if self.config.algorithm == "PPO": self.model = PPO( "MlpPolicy", self.env, learning_rate=self.config.learning_rate, batch_size=self.config.batch_size, gamma=self.config.gamma, verbose=self.config.verbose, tensorboard_log=self.config.tensorboard_log ) elif self.config.algorithm == "A2C": self.model = A2C( "MlpPolicy", self.env, learning_rate=self.config.learning_rate, gamma=self.config.gamma, verbose=self.config.verbose, tensorboard_log=self.config.tensorboard_log ) elif self.config.algorithm == "DDPG": self.model = DDPG( "MlpPolicy", self.env, learning_rate=self.config.learning_rate, buffer_size=self.config.buffer_size, learning_starts=self.config.learning_starts, gamma=self.config.gamma, tau=self.config.tau, train_freq=self.config.train_freq, gradient_steps=self.config.gradient_steps, verbose=self.config.verbose, tensorboard_log=self.config.tensorboard_log ) elif self.config.algorithm == "TD3": self.model = TD3( "MlpPolicy", self.env, learning_rate=self.config.learning_rate, buffer_size=self.config.buffer_size, learning_starts=self.config.learning_starts, gamma=self.config.gamma, tau=self.config.tau, train_freq=self.config.train_freq, gradient_steps=self.config.gradient_steps, target_update_interval=self.config.target_update_interval, verbose=self.config.verbose, tensorboard_log=self.config.tensorboard_log ) else: raise ValueError(f"Unsupported algorithm: {self.config.algorithm}") # Train the model callbacks = [self.callback] if self.callback else None self.model.learn( total_timesteps=total_timesteps, callback=callbacks ) logger.info("FinRL agent training completed") return { 'algorithm': self.config.algorithm, 'total_timesteps': total_timesteps, 'model_path': f"models/finrl_{self.config.algorithm.lower()}" } def predict(self, data: pd.DataFrame) -> List[int]: """Generate trading predictions using the trained model""" if self.model is None: raise ValueError("Model not trained. Call train() first.") # Prepare data test_data = self.prepare_data(data) # Create test environment test_env = self.create_environment(test_data) predictions = [] obs, _ = test_env.reset() done = False while not done: action, _ = self.model.predict(obs, deterministic=True) predictions.append(action) obs, _, done, _, _ = test_env.step(action) return predictions def evaluate(self, data: pd.DataFrame) -> Dict[str, float]: """Evaluate the trained model on test data""" if self.model is None: raise ValueError("Model not trained. Call train() first.") # Prepare data test_data = self.prepare_data(data) # Create test environment test_env = self.create_environment(test_data) obs, _ = test_env.reset() done = False total_reward = 0 steps = 0 while not done: action, _ = self.model.predict(obs, deterministic=True) obs, reward, done, _, info = test_env.step(action) total_reward += reward steps += 1 # Calculate metrics final_portfolio_value = info['portfolio_value'] initial_balance = test_env.initial_balance total_return = (final_portfolio_value - initial_balance) / initial_balance return { 'total_reward': total_reward, 'total_return': total_return, 'final_portfolio_value': final_portfolio_value, 'steps': steps, 'sharpe_ratio': total_reward / steps if steps > 0 else 0 } def save_model(self, path: str): """Save the trained model""" if self.model is None: raise ValueError("No model to save. Train the model first.") self.model.save(path) logger.info(f"Model saved to {path}") def load_model(self, path: str): """Load a trained model""" if self.config.algorithm == "PPO": self.model = PPO.load(path) elif self.config.algorithm == "A2C": self.model = A2C.load(path) elif self.config.algorithm == "DDPG": self.model = DDPG.load(path) elif self.config.algorithm == "TD3": self.model = TD3.load(path) else: raise ValueError(f"Unsupported algorithm: {self.config.algorithm}") logger.info(f"Model loaded from {path}") def create_finrl_agent_from_config(config_path: str) -> FinRLAgent: """Create FinRL agent from configuration file""" with open(config_path, 'r') as file: config_data = yaml.safe_load(file) finrl_config = FinRLConfig(**config_data.get('finrl', {})) return FinRLAgent(finrl_config)