Spaces:

chen470
/

drl-trading-bot-dev2

Paused

drl-trading-bot-dev2 / src /env /advanced_env.py

DRL Trading Bot

Feature: HTF Agent integration — live trading, API endpoints, UI tab

fc115d5 8 months ago

14.8 kB

"""
Advanced Trading Environment
Uses sophisticated features and reward function for competitive performance.
"""

import gymnasium as gym
from gymnasium import spaces
import numpy as np
import pandas as pd
from typing import Optional, Tuple, Dict, Any, List
from enum import IntEnum
import logging

from .advanced_features import AdvancedFeatureEngine
from .advanced_rewards import AdvancedRewardCalculator

logger = logging.getLogger(__name__)


class Actions(IntEnum):
    HOLD = 0
    BUY = 1
    SELL = 2


class Positions(IntEnum):
    SHORT = -1
    FLAT = 0
    LONG = 1


class AdvancedTradingEnv(gym.Env):
    """
    Advanced trading environment with sophisticated features.
    
    Key innovations:
    1. 50+ advanced features (multi-timeframe, regime, patterns)
    2. Profit-factor optimized reward function
    3. Market feature context for smart rewards
    4. Dynamic position sizing based on confidence
    """
    
    metadata = {'render_modes': ['human', 'ansi']}
    
    def __init__(
        self,
        df: pd.DataFrame,
        initial_balance: float = 10000.0,
        lookback_window: int = 48,  # 48 hours of context
        trading_fee: float = 0.0004,  # 0.04% (Binance maker fee)
        position_size: float = 0.25,  # 25% of portfolio per trade
        max_position: int = 1,
        render_mode: Optional[str] = None,
    ):
        super().__init__()
        
        self.render_mode = render_mode
        
        # Data and features
        self.raw_df = df.copy()
        self.feature_engine = AdvancedFeatureEngine()
        self._prepare_data()
        
        # Environment parameters
        self.initial_balance = initial_balance
        self.lookback_window = lookback_window
        self.trading_fee = trading_fee
        self.position_size = position_size
        self.max_position = max_position
        
        # Advanced reward calculator
        self.reward_calculator = AdvancedRewardCalculator(initial_balance=initial_balance)
        
        # Feature dimensions
        self.feature_columns = self.feature_engine.get_feature_columns()
        available_features = [f for f in self.feature_columns if f in self.df.columns]
        self.feature_columns = available_features
        self.n_features = len(self.feature_columns)
        self.n_ohlcv = 5
        self.n_agent_state = 8  # Extended agent state
        
        # Observation space
        obs_dim = self.lookback_window * (self.n_features + self.n_ohlcv) + self.n_agent_state
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf, shape=(obs_dim,), dtype=np.float32
        )
        
        # Action space
        self.action_space = spaces.Discrete(3)
        
        self._reset_state()
        
    def _prepare_data(self):
        """Compute all advanced features."""
        logger.info("Computing advanced features...")
        self.df = self.feature_engine.compute_all(self.raw_df)
        self.df = self.df.replace([np.inf, -np.inf], np.nan)
        self.df = self.df.fillna(method='ffill').fillna(0)
        
        # Ensure enough data after feature computation
        min_lookback = 250  # Need 250 bars for features like EMA 200
        self.df = self.df.iloc[min_lookback:]
        
        self.prices = self.df['close'].values
        self.highs = self.df['high'].values
        self.lows = self.df['low'].values
        
        logger.info(f"Prepared {len(self.df)} candles with {self.n_features if hasattr(self, 'n_features') else 'N/A'} features")
        
    def _reset_state(self):
        """Reset all state variables."""
        self.balance = self.initial_balance
        self.position = Positions.FLAT
        self.position_price = 0.0
        self.position_size_units = 0.0
        self.unrealized_pnl = 0.0
        self.realized_pnl = 0.0
        self.peak_balance = self.initial_balance
        self.current_drawdown = 0.0
        self.trade_count = 0
        self.current_step = self.lookback_window
        self.done = False
        self.trades: List[Dict] = []
        self.steps_since_trade = 0
        
    def _get_market_features(self) -> Dict:
        """Get current market features for reward calculation."""
        if self.current_step >= len(self.df):
            return {}
            
        row = self.df.iloc[self.current_step]
        
        features = {}
        for col in ['trend_strength', 'position_in_range', 'vol_ratio_24h', 
                    'adx', 'rsi_14', 'trending_regime', 'zscore_100']:
            if col in row.index:
                features[col] = float(row[col])
                
        return features
        
    def _get_observation(self) -> np.ndarray:
        """Construct observation with advanced features."""
        start_idx = self.current_step - self.lookback_window
        end_idx = self.current_step
        
        window_data = self.df.iloc[start_idx:end_idx]
        
        # OHLCV (normalized)
        ohlcv_cols = ['open', 'high', 'low', 'close', 'volume']
        ohlcv = window_data[ohlcv_cols].values.copy()
        
        # Normalize prices by first close
        price_norm = ohlcv[0, 3]
        if price_norm > 0:
            ohlcv[:, :4] = ohlcv[:, :4] / price_norm - 1
        
        # Normalize volume
        vol_mean = ohlcv[:, 4].mean()
        if vol_mean > 0:
            ohlcv[:, 4] = ohlcv[:, 4] / vol_mean - 1
            
        # Advanced features (already normalized/scaled)
        features = window_data[self.feature_columns].values.copy()
        features = np.nan_to_num(features, nan=0.0, posinf=1.0, neginf=-1.0)
        
        # Clip extreme values
        features = np.clip(features, -5, 5)
        
        # Combine
        combined = np.concatenate([ohlcv, features], axis=1)
        flat_history = combined.flatten()
        
        # Enhanced agent state
        portfolio_value = self._get_portfolio_value()
        market_features = self._get_market_features()
        
        agent_state = np.array([
            float(self.position),  # Position: -1, 0, 1
            self.unrealized_pnl / self.initial_balance,  # Unrealized P&L ratio
            self.balance / self.initial_balance,  # Balance ratio
            self.current_drawdown,  # Current drawdown
            min(self.trade_count / 100, 1.0),  # Normalized trade count
            min(self.steps_since_trade / 48, 1.0),  # Time since last trade (normalized)
            market_features.get('trend_strength', 0.5),  # Current trend
            market_features.get('trending_regime', 0.5),  # Trend regime
        ], dtype=np.float32)
        
        observation = np.concatenate([flat_history, agent_state]).astype(np.float32)
        
        return observation
        
    def _get_portfolio_value(self) -> float:
        """Calculate total portfolio value."""
        return self.balance + self.unrealized_pnl
        
    def _update_unrealized_pnl(self):
        """Update unrealized P&L based on current position."""
        if self.position == Positions.FLAT:
            self.unrealized_pnl = 0.0
            return
            
        current_price = self.prices[self.current_step]
        
        if self.position == Positions.LONG:
            self.unrealized_pnl = (current_price - self.position_price) * self.position_size_units
        elif self.position == Positions.SHORT:
            self.unrealized_pnl = (self.position_price - current_price) * self.position_size_units
            
    def _update_drawdown(self):
        """Update peak balance and current drawdown."""
        portfolio_value = self._get_portfolio_value()
        if portfolio_value > self.peak_balance:
            self.peak_balance = portfolio_value
        self.current_drawdown = (self.peak_balance - portfolio_value) / self.peak_balance
        
    def _execute_trade(self, action: int) -> Tuple[float, Optional[float]]:
        """Execute trading action."""
        current_price = self.prices[self.current_step]
        prev_price = self.prices[self.current_step - 1]
        trade_pnl = None
        
        # Calculate step return
        if self.position == Positions.LONG:
            step_return = (current_price - prev_price) / prev_price
        elif self.position == Positions.SHORT:
            step_return = (prev_price - current_price) / prev_price
        else:
            step_return = 0.0
            
        # Execute action
        if action == Actions.BUY and self.position != Positions.LONG:
            if self.position == Positions.SHORT:
                trade_pnl = self._close_position(current_price)
            self._open_position(current_price, Positions.LONG)
            self.steps_since_trade = 0
            
        elif action == Actions.SELL and self.position != Positions.SHORT:
            if self.position == Positions.LONG:
                trade_pnl = self._close_position(current_price)
            self._open_position(current_price, Positions.SHORT)
            self.steps_since_trade = 0
            
        else:
            self.steps_since_trade += 1
            
        return step_return, trade_pnl
        
    def _open_position(self, price: float, position_type: Positions):
        """Open new position."""
        trade_amount = self.balance * self.position_size
        fee = trade_amount * self.trading_fee
        
        self.position = position_type
        self.position_price = price
        self.position_size_units = (trade_amount - fee) / price
        self.balance -= fee
        self.trade_count += 1
        
    def _close_position(self, price: float) -> float:
        """Close current position."""
        if self.position == Positions.LONG:
            pnl = (price - self.position_price) * self.position_size_units
        elif self.position == Positions.SHORT:
            pnl = (self.position_price - price) * self.position_size_units
        else:
            return 0.0
            
        # Apply fee
        fee = abs(pnl) * self.trading_fee
        pnl -= fee
        
        # Record trade
        self.trades.append({
            'entry_price': self.position_price,
            'exit_price': price,
            'position': int(self.position),
            'pnl': pnl,
            'step': self.current_step,
        })
        
        # Update balance
        self.balance += pnl
        self.realized_pnl += pnl
        
        # Reset position
        self.position = Positions.FLAT
        self.position_price = 0.0
        self.position_size_units = 0.0
        
        return pnl
        
    def step(self, action: int) -> Tuple[np.ndarray, float, bool, bool, Dict]:
        """Execute one step."""
        if self.done:
            raise RuntimeError("Episode done, call reset()")
            
        # Get market features for reward calculation
        market_features = self._get_market_features()
        
        # Execute trade
        step_return, trade_pnl = self._execute_trade(action)
        
        # Update state
        self._update_unrealized_pnl()
        self._update_drawdown()
        
        # Calculate advanced reward
        portfolio_value = self._get_portfolio_value()
        reward = self.reward_calculator.calculate_reward(
            step_return=step_return,
            portfolio_value=portfolio_value,
            position=int(self.position),
            action_taken=action,
            trade_pnl=trade_pnl,
            market_features=market_features,
        )
        
        # Move to next step
        self.current_step += 1
        
        # Check termination
        terminated = False
        truncated = False
        
        if self.current_step >= len(self.df) - 1:
            truncated = True
            self.done = True
            
        if self.balance <= 0 or self.current_drawdown > 0.25:  # Stop at 25% drawdown
            terminated = True
            self.done = True
            reward -= 5.0  # Heavy penalty for blowing up
            
        observation = self._get_observation()
        
        info = {
            'balance': self.balance,
            'portfolio_value': portfolio_value,
            'position': int(self.position),
            'unrealized_pnl': self.unrealized_pnl,
            'realized_pnl': self.realized_pnl,
            'drawdown': self.current_drawdown,
            'trade_count': self.trade_count,
            'step': self.current_step,
            'price': self.prices[min(self.current_step, len(self.prices)-1)],
        }
        
        return observation, reward, terminated, truncated, info
        
    def reset(self, seed: Optional[int] = None, options: Optional[Dict] = None) -> Tuple[np.ndarray, Dict]:
        """Reset environment."""
        super().reset(seed=seed)
        
        self._reset_state()
        self.reward_calculator.reset(self.initial_balance)
        
        # Random start for training diversity
        if options and options.get('random_start', True):
            max_start = len(self.df) - self.lookback_window - 500
            if max_start > self.lookback_window:
                self.current_step = self.np_random.integers(self.lookback_window, max_start)
                
        observation = self._get_observation()
        info = {'balance': self.balance, 'portfolio_value': self.initial_balance}
        
        return observation, info
        
    def render(self):
        """Render environment."""
        if self.render_mode == 'human':
            print(self._render_ansi())
        elif self.render_mode == 'ansi':
            return self._render_ansi()
            
    def _render_ansi(self) -> str:
        """String representation."""
        pos_str = {-1: 'SHORT', 0: 'FLAT', 1: 'LONG'}[int(self.position)]
        return (
            f"Step: {self.current_step} | "
            f"Price: ${self.prices[self.current_step]:.2f} | "
            f"Position: {pos_str} | "
            f"Portfolio: ${self._get_portfolio_value():.2f} | "
            f"Trades: {self.trade_count} | "
            f"DD: {self.current_drawdown:.2%}"
        )
        
    def get_episode_metrics(self) -> Dict:
        """Get episode metrics."""
        metrics = self.reward_calculator.get_episode_metrics()
        metrics['trades'] = self.trades
        metrics['final_balance'] = self.balance
        metrics['final_portfolio_value'] = self._get_portfolio_value()
        return metrics


def create_advanced_env(df: pd.DataFrame, config: Optional[Dict] = None) -> AdvancedTradingEnv:
    """Factory function to create advanced environment."""
    config = config or {}
    return AdvancedTradingEnv(
        df=df,
        initial_balance=config.get('initial_balance', 10000.0),
        lookback_window=config.get('lookback_window', 48),
        trading_fee=config.get('trading_fee', 0.0004),
        position_size=config.get('position_size', 0.25),
    )

	"""
	Advanced Trading Environment
	Uses sophisticated features and reward function for competitive performance.
	"""

	import gymnasium as gym
	from gymnasium import spaces
	import numpy as np
	import pandas as pd
	from typing import Optional, Tuple, Dict, Any, List
	from enum import IntEnum
	import logging

	from .advanced_features import AdvancedFeatureEngine
	from .advanced_rewards import AdvancedRewardCalculator

	logger = logging.getLogger(__name__)


	class Actions(IntEnum):
	HOLD = 0
	BUY = 1
	SELL = 2


	class Positions(IntEnum):
	SHORT = -1
	FLAT = 0
	LONG = 1


	class AdvancedTradingEnv(gym.Env):
	"""
	Advanced trading environment with sophisticated features.

	Key innovations:
	1. 50+ advanced features (multi-timeframe, regime, patterns)
	2. Profit-factor optimized reward function
	3. Market feature context for smart rewards
	4. Dynamic position sizing based on confidence
	"""

	metadata = {'render_modes': ['human', 'ansi']}

	def __init__(
	self,
	df: pd.DataFrame,
	initial_balance: float = 10000.0,
	lookback_window: int = 48, # 48 hours of context
	trading_fee: float = 0.0004, # 0.04% (Binance maker fee)
	position_size: float = 0.25, # 25% of portfolio per trade
	max_position: int = 1,
	render_mode: Optional[str] = None,
	):
	super().__init__()

	self.render_mode = render_mode

	# Data and features
	self.raw_df = df.copy()
	self.feature_engine = AdvancedFeatureEngine()
	self._prepare_data()

	# Environment parameters
	self.initial_balance = initial_balance
	self.lookback_window = lookback_window
	self.trading_fee = trading_fee
	self.position_size = position_size
	self.max_position = max_position

	# Advanced reward calculator
	self.reward_calculator = AdvancedRewardCalculator(initial_balance=initial_balance)

	# Feature dimensions
	self.feature_columns = self.feature_engine.get_feature_columns()
	available_features = [f for f in self.feature_columns if f in self.df.columns]
	self.feature_columns = available_features
	self.n_features = len(self.feature_columns)
	self.n_ohlcv = 5
	self.n_agent_state = 8 # Extended agent state

	# Observation space
	obs_dim = self.lookback_window * (self.n_features + self.n_ohlcv) + self.n_agent_state
	self.observation_space = spaces.Box(
	low=-np.inf, high=np.inf, shape=(obs_dim,), dtype=np.float32
	)

	# Action space
	self.action_space = spaces.Discrete(3)

	self._reset_state()

	def _prepare_data(self):
	"""Compute all advanced features."""
	logger.info("Computing advanced features...")
	self.df = self.feature_engine.compute_all(self.raw_df)
	self.df = self.df.replace([np.inf, -np.inf], np.nan)
	self.df = self.df.fillna(method='ffill').fillna(0)

	# Ensure enough data after feature computation
	min_lookback = 250 # Need 250 bars for features like EMA 200
	self.df = self.df.iloc[min_lookback:]

	self.prices = self.df['close'].values
	self.highs = self.df['high'].values
	self.lows = self.df['low'].values

	logger.info(f"Prepared {len(self.df)} candles with {self.n_features if hasattr(self, 'n_features') else 'N/A'} features")

	def _reset_state(self):
	"""Reset all state variables."""
	self.balance = self.initial_balance
	self.position = Positions.FLAT
	self.position_price = 0.0
	self.position_size_units = 0.0
	self.unrealized_pnl = 0.0
	self.realized_pnl = 0.0
	self.peak_balance = self.initial_balance
	self.current_drawdown = 0.0
	self.trade_count = 0
	self.current_step = self.lookback_window
	self.done = False
	self.trades: List[Dict] = []
	self.steps_since_trade = 0

	def _get_market_features(self) -> Dict:
	"""Get current market features for reward calculation."""
	if self.current_step >= len(self.df):
	return {}

	row = self.df.iloc[self.current_step]

	features = {}
	for col in ['trend_strength', 'position_in_range', 'vol_ratio_24h',
	'adx', 'rsi_14', 'trending_regime', 'zscore_100']:
	if col in row.index:
	features[col] = float(row[col])

	return features

	def _get_observation(self) -> np.ndarray:
	"""Construct observation with advanced features."""
	start_idx = self.current_step - self.lookback_window
	end_idx = self.current_step

	window_data = self.df.iloc[start_idx:end_idx]

	# OHLCV (normalized)
	ohlcv_cols = ['open', 'high', 'low', 'close', 'volume']
	ohlcv = window_data[ohlcv_cols].values.copy()

	# Normalize prices by first close
	price_norm = ohlcv[0, 3]
	if price_norm > 0:
	ohlcv[:, :4] = ohlcv[:, :4] / price_norm - 1

	# Normalize volume
	vol_mean = ohlcv[:, 4].mean()
	if vol_mean > 0:
	ohlcv[:, 4] = ohlcv[:, 4] / vol_mean - 1

	# Advanced features (already normalized/scaled)
	features = window_data[self.feature_columns].values.copy()
	features = np.nan_to_num(features, nan=0.0, posinf=1.0, neginf=-1.0)

	# Clip extreme values
	features = np.clip(features, -5, 5)

	# Combine
	combined = np.concatenate([ohlcv, features], axis=1)
	flat_history = combined.flatten()

	# Enhanced agent state
	portfolio_value = self._get_portfolio_value()
	market_features = self._get_market_features()

	agent_state = np.array([
	float(self.position), # Position: -1, 0, 1
	self.unrealized_pnl / self.initial_balance, # Unrealized P&L ratio
	self.balance / self.initial_balance, # Balance ratio
	self.current_drawdown, # Current drawdown
	min(self.trade_count / 100, 1.0), # Normalized trade count
	min(self.steps_since_trade / 48, 1.0), # Time since last trade (normalized)
	market_features.get('trend_strength', 0.5), # Current trend
	market_features.get('trending_regime', 0.5), # Trend regime
	], dtype=np.float32)

	observation = np.concatenate([flat_history, agent_state]).astype(np.float32)

	return observation

	def _get_portfolio_value(self) -> float:
	"""Calculate total portfolio value."""
	return self.balance + self.unrealized_pnl

	def _update_unrealized_pnl(self):
	"""Update unrealized P&L based on current position."""
	if self.position == Positions.FLAT:
	self.unrealized_pnl = 0.0
	return

	current_price = self.prices[self.current_step]

	if self.position == Positions.LONG:
	self.unrealized_pnl = (current_price - self.position_price) * self.position_size_units
	elif self.position == Positions.SHORT:
	self.unrealized_pnl = (self.position_price - current_price) * self.position_size_units

	def _update_drawdown(self):
	"""Update peak balance and current drawdown."""
	portfolio_value = self._get_portfolio_value()
	if portfolio_value > self.peak_balance:
	self.peak_balance = portfolio_value
	self.current_drawdown = (self.peak_balance - portfolio_value) / self.peak_balance

	def _execute_trade(self, action: int) -> Tuple[float, Optional[float]]:
	"""Execute trading action."""
	current_price = self.prices[self.current_step]
	prev_price = self.prices[self.current_step - 1]
	trade_pnl = None

	# Calculate step return
	if self.position == Positions.LONG:
	step_return = (current_price - prev_price) / prev_price
	elif self.position == Positions.SHORT:
	step_return = (prev_price - current_price) / prev_price
	else:
	step_return = 0.0

	# Execute action
	if action == Actions.BUY and self.position != Positions.LONG:
	if self.position == Positions.SHORT:
	trade_pnl = self._close_position(current_price)
	self._open_position(current_price, Positions.LONG)
	self.steps_since_trade = 0

	elif action == Actions.SELL and self.position != Positions.SHORT:
	if self.position == Positions.LONG:
	trade_pnl = self._close_position(current_price)
	self._open_position(current_price, Positions.SHORT)
	self.steps_since_trade = 0

	else:
	self.steps_since_trade += 1

	return step_return, trade_pnl

	def _open_position(self, price: float, position_type: Positions):
	"""Open new position."""
	trade_amount = self.balance * self.position_size
	fee = trade_amount * self.trading_fee

	self.position = position_type
	self.position_price = price
	self.position_size_units = (trade_amount - fee) / price
	self.balance -= fee
	self.trade_count += 1

	def _close_position(self, price: float) -> float:
	"""Close current position."""
	if self.position == Positions.LONG:
	pnl = (price - self.position_price) * self.position_size_units
	elif self.position == Positions.SHORT:
	pnl = (self.position_price - price) * self.position_size_units
	else:
	return 0.0

	# Apply fee
	fee = abs(pnl) * self.trading_fee
	pnl -= fee

	# Record trade
	self.trades.append({
	'entry_price': self.position_price,
	'exit_price': price,
	'position': int(self.position),
	'pnl': pnl,
	'step': self.current_step,
	})

	# Update balance
	self.balance += pnl
	self.realized_pnl += pnl

	# Reset position
	self.position = Positions.FLAT
	self.position_price = 0.0
	self.position_size_units = 0.0

	return pnl

	def step(self, action: int) -> Tuple[np.ndarray, float, bool, bool, Dict]:
	"""Execute one step."""
	if self.done:
	raise RuntimeError("Episode done, call reset()")

	# Get market features for reward calculation
	market_features = self._get_market_features()

	# Execute trade
	step_return, trade_pnl = self._execute_trade(action)

	# Update state
	self._update_unrealized_pnl()
	self._update_drawdown()

	# Calculate advanced reward
	portfolio_value = self._get_portfolio_value()
	reward = self.reward_calculator.calculate_reward(
	step_return=step_return,
	portfolio_value=portfolio_value,
	position=int(self.position),
	action_taken=action,
	trade_pnl=trade_pnl,
	market_features=market_features,
	)

	# Move to next step
	self.current_step += 1

	# Check termination
	terminated = False
	truncated = False

	if self.current_step >= len(self.df) - 1:
	truncated = True
	self.done = True

	if self.balance <= 0 or self.current_drawdown > 0.25: # Stop at 25% drawdown
	terminated = True
	self.done = True
	reward -= 5.0 # Heavy penalty for blowing up

	observation = self._get_observation()

	info = {
	'balance': self.balance,
	'portfolio_value': portfolio_value,
	'position': int(self.position),
	'unrealized_pnl': self.unrealized_pnl,
	'realized_pnl': self.realized_pnl,
	'drawdown': self.current_drawdown,
	'trade_count': self.trade_count,
	'step': self.current_step,
	'price': self.prices[min(self.current_step, len(self.prices)-1)],
	}

	return observation, reward, terminated, truncated, info

	def reset(self, seed: Optional[int] = None, options: Optional[Dict] = None) -> Tuple[np.ndarray, Dict]:
	"""Reset environment."""
	super().reset(seed=seed)

	self._reset_state()
	self.reward_calculator.reset(self.initial_balance)

	# Random start for training diversity
	if options and options.get('random_start', True):
	max_start = len(self.df) - self.lookback_window - 500
	if max_start > self.lookback_window:
	self.current_step = self.np_random.integers(self.lookback_window, max_start)

	observation = self._get_observation()
	info = {'balance': self.balance, 'portfolio_value': self.initial_balance}

	return observation, info

	def render(self):
	"""Render environment."""
	if self.render_mode == 'human':
	print(self._render_ansi())
	elif self.render_mode == 'ansi':
	return self._render_ansi()

	def _render_ansi(self) -> str:
	"""String representation."""
	pos_str = {-1: 'SHORT', 0: 'FLAT', 1: 'LONG'}[int(self.position)]
	return (
	f"Step: {self.current_step} \| "
	f"Price: ${self.prices[self.current_step]:.2f} \| "
	f"Position: {pos_str} \| "
	f"Portfolio: ${self._get_portfolio_value():.2f} \| "
	f"Trades: {self.trade_count} \| "
	f"DD: {self.current_drawdown:.2%}"
	)

	def get_episode_metrics(self) -> Dict:
	"""Get episode metrics."""
	metrics = self.reward_calculator.get_episode_metrics()
	metrics['trades'] = self.trades
	metrics['final_balance'] = self.balance
	metrics['final_portfolio_value'] = self._get_portfolio_value()
	return metrics


	def create_advanced_env(df: pd.DataFrame, config: Optional[Dict] = None) -> AdvancedTradingEnv:
	"""Factory function to create advanced environment."""
	config = config or {}
	return AdvancedTradingEnv(
	df=df,
	initial_balance=config.get('initial_balance', 10000.0),
	lookback_window=config.get('lookback_window', 48),
	trading_fee=config.get('trading_fee', 0.0004),
	position_size=config.get('position_size', 0.25),
	)