can you check in there, the scripts train_loop.py , {
    "./main.py": "# File: main.py\n\nimport os\nimport logging\nimport torch\nimport torch.multiprocessing as mp\nimport yaml\nimport argparse\n\nfrom training.runner import train_worker\nfrom training.config_validation import validate_and_cast_config\nfrom training.utils import (\n    setup_logging,\n    setup_seed,\n)\n\ndef main(config):\n    \"\"\"\n    Entry point for training. Initializes multiprocessing for distributed training.\n    \"\"\"\n    logger = logging.getLogger(\"Main\")\n    logger.info(\"Entering main() function.\")\n\n    # Set random seed\n    setup_seed(config[\"random_seed\"])\n\n    # Number of GPUs available\n    num_gpus = torch.cuda.device_count()\n    configured_world_size = config[\"distributed\"][\"world_size\"]\n    world_size = min(num_gpus, configured_world_size) if num_gpus >= 1 else 1\n    logger.info(f\"Number of GPUs available: {num_gpus}\")\n    logger.info(f\"Configured world_size: {configured_world_size}, using world_size: {world_size}\")\n\n    if world_size > 1:\n        # Use DistributedDataParallel (DDP)\n        logger.info(\"Using DistributedDataParallel for multi-GPU training.\")\n\n        # Set up environment variables for distributed training\n        os.environ[\"MASTER_ADDR\"] = config[\"distributed\"][\"master_addr\"]\n        os.environ[\"MASTER_PORT\"] = str(config[\"distributed\"][\"master_port\"])  # Ensure it's a string\n        logger.info(f\"MASTER_ADDR set to {config['distributed']['master_addr']}\")\n        logger.info(f\"MASTER_PORT set to {config['distributed']['master_port']}\")\n\n        try:\n            mp.spawn(\n                train_worker,\n                args=(config, world_size),\n                nprocs=world_size,\n                join=True\n            )\n        except Exception as e:\n            logger.error(f\"Training failed during mp.spawn: {e}\", exc_info=True)\n            raise\n    else:\n        # Single GPU or CPU\n        logger.info(\"Single GPU or CPU detected.\")\n        try:\n            train_worker(0, config, world_size)\n        except Exception as e:\n            logger.error(f\"Training failed: {e}\", exc_info=True)\n            raise\n\n\nif __name__ == \"__main__\":\n    # Parse command-line arguments for flexibility\n    parser = argparse.ArgumentParser(description=\"Distributed Training Script\")\n    parser.add_argument('--config_path', type=str, default=\"config/config.yaml\",\n                        help='Path to the configuration YAML file.')\n    args = parser.parse_args()\n\n    logger = setup_logging(\"logs/main.log\")\n    logger.info(\"Starting main execution.\")\n\n    config_path = args.config_path\n    if not os.path.exists(config_path):\n        logger.error(f\"Configuration file not found at path: {config_path}\")\n        raise FileNotFoundError(f\"Configuration file not found at path: {config_path}\")\n\n    try:\n        with open(config_path, \"r\") as f:\n            raw_cfg = yaml.safe_load(f)\n            logger.debug(f\"Loaded raw config: {raw_cfg}\")\n    except yaml.YAMLError as e:\n        logger.error(f\"Failed to parse the configuration file: {e}\")\n        raise\n\n    try:\n        config = validate_and_cast_config(raw_cfg)\n        logger.info(\"Configuration validated successfully.\")\n        logger.info(f\"Loaded configuration:\\n{yaml.dump(config)}\")  # Log the full configuration\n    except Exception as e:\n        logger.error(f\"Configuration validation failed: {e}\", exc_info=True)\n        raise\n\n    logger.info(\"Configuration loaded and validated successfully.\")\n\n    # Start training\n    try:\n        # Explicitly set multiprocessing start method for compatibility\n        if os.name == 'nt':\n            mp.set_start_method('spawn', force=True)\n        main(config)\n    except Exception as e:\n        logger.error(f\"Training failed with exception: {e}\", exc_info=True)\n        raise",
    "./collection_json.py": "# File: collection_json.py\n\nimport os\nimport json\nimport argparse\nimport logging\n\ndef setup_logging():\n    \"\"\"\n    Configure logging to output to both file and console for this script.\n    \"\"\"\n    logger = logging.getLogger(\"CollectionJSON\")\n    logger.setLevel(logging.DEBUG)\n\n    # Create console handler\n    ch = logging.StreamHandler()\n    ch.setLevel(logging.INFO)\n\n    # Create file handler\n    fh = logging.FileHandler(\"logs/collection_json.log\")\n    fh.setLevel(logging.DEBUG)\n\n    # Create formatter and add to handlers\n    formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(message)s\")\n    ch.setFormatter(formatter)\n    fh.setFormatter(formatter)\n\n    # Add handlers to the logger if none exist\n    if not logger.hasHandlers():\n        logger.addHandler(ch)\n        logger.addHandler(fh)\n\n    return logger\n\ndef collect_scripts_to_json(root_dir, output_file, logger):\n    \"\"\"\n    Traverse the given directory and its subdirectories to collect .py and .yaml files,\n    then save their contents into a JSON file.\n    \"\"\"\n    logger.info(\"Starting script collection.\")\n    scripts = {}\n\n    for dirpath, _, filenames in os.walk(root_dir):\n        for file in filenames:\n            if file.endswith(('.py', '.yaml')):\n                file_path = os.path.join(dirpath, file)\n                try:\n                    with open(file_path, 'r') as f:\n                        scripts[file_path] = f.read()\n                    logger.debug(f\"Collected script: {file_path}\")\n                except Exception as e:\n                    logger.error(f\"Failed to read {file_path}: {e}\")\n\n    try:\n        with open(output_file, 'w') as json_file:\n            json.dump(scripts, json_file, indent=4)\n        logger.info(f\"Scripts successfully collected and saved to {output_file}\")\n    except Exception as e:\n        logger.error(f\"Failed to write to {output_file}: {e}\")\n\ndef main():\n    parser = argparse.ArgumentParser(description=\"Collect .py and .yaml scripts into a JSON file.\")\n    parser.add_argument('--root_dir', type=str, default='./', help='Root directory to search for scripts.')\n    parser.add_argument('--output_file', type=str, default='scripts_collection.json', help='Output JSON file path.')\n\n    args = parser.parse_args()\n\n    logger = setup_logging()\n    logger.info(f\"Starting collection from root directory: {args.root_dir}\")\n\n    collect_scripts_to_json(args.root_dir, args.output_file, logger)\n\nif __name__ == \"__main__\":\n    main()",
    "./monitoring.py": "# File: monitoring.py\n\nimport os\nimport logging\nimport json\nfrom torch.utils.tensorboard import SummaryWriter\nfrom training.utils import log_memory_usage, log_gpu_memory\nimport gc\n\n# Initialize logger for Monitoring\nlogger = logging.getLogger(\"Monitoring\")\nlogger.setLevel(logging.INFO)\nif not logger.hasHandlers():\n    os.makedirs(\"logs\", exist_ok=True)\n    fh = logging.FileHandler(\"logs/monitoring.log\")\n    fh.setLevel(logging.INFO)\n    ch = logging.StreamHandler()\n    ch.setLevel(logging.INFO)\n    formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n    fh.setFormatter(formatter)\n    ch.setFormatter(formatter)\n    logger.addHandler(fh)\n    logger.addHandler(ch)\n\n\ndef interpret_metrics(avg_metrics, eval_metrics, logger: logging.Logger):\n    \"\"\"\n    Analyze and provide detailed insights based on training and evaluation metrics.\n\n    Args:\n        avg_metrics (dict): Dictionary containing average training metrics.\n        eval_metrics (dict): Dictionary containing evaluation metrics.\n        logger (logging.Logger): Logger instance for logging insights.\n    \"\"\"\n    try:\n        # Analyze training loss\n        actor_loss = avg_metrics.get(\"actor_loss\", None)\n        critic_loss = avg_metrics.get(\"critic_loss\", None)\n        entropy = avg_metrics.get(\"entropy\", None)\n        kl_divergence = avg_metrics.get(\"kl_divergence\", None)\n\n        if actor_loss is not None:\n            if actor_loss > 1.0:\n                logger.info(\"Actor loss is high, indicating the policy may be unstable. Consider lowering the learning rate or increasing the batch size.\")\n            elif actor_loss < 0.01:\n                logger.info(\"Actor loss is very low. The model might be converging well. However, monitor if the policy starts overfitting to the training data.\")\n            else:\n                logger.info(\"Actor loss is within an acceptable range. The policy is learning effectively.\")\n\n        if critic_loss is not None:\n            if critic_loss > 1.0:\n                logger.info(\"Critic loss is high. This could indicate the value function is not learning effectively. Consider adjusting the critic's learning rate or architecture.\")\n            elif critic_loss < 0.01:\n                logger.info(\"Critic loss is very low, suggesting the value function may have already converged. Monitor for underfitting or lack of variance.\")\n            else:\n                logger.info(\"Critic loss is within an acceptable range, suggesting the value function is learning effectively.\")\n\n        # Analyze entropy\n        if entropy is not None:\n            if entropy > 1.0:\n                logger.info(\"Entropy is high, suggesting the model is exploring more. This is beneficial during the early stages of training but may need reduction for better exploitation later.\")\n            elif entropy < 0.1:\n                logger.info(\"Entropy is low, indicating the model is exploiting learned policies. Ensure this doesn't lead to premature convergence or local optima.\")\n            else:\n                logger.info(\"Entropy is within a balanced range, maintaining a healthy exploration-exploitation trade-off.\")\n\n        # Analyze KL Divergence\n        if kl_divergence is not None:\n            if kl_divergence > 0.01:\n                logger.info(\"KL Divergence is high. Policy updates might be too aggressive. Consider reducing the learning rate or clipping thresholds.\")\n            elif kl_divergence < 0.001:\n                logger.info(\"KL Divergence is very low. Policy updates might be overly conservative. Consider increasing the learning rate or exploration strategies.\")\n            else:\n                logger.info(\"KL Divergence is within an acceptable range, indicating balanced policy updates.\")\n\n        # Analyze evaluation metrics\n        if eval_metrics:\n            average_reward = eval_metrics.get(\"average_reward\", 0)\n            sharpe_ratio = eval_metrics.get(\"sharpe_ratio\", 0)\n            max_drawdown = eval_metrics.get(\"max_drawdown\", 0)\n            average_trade_count = eval_metrics.get(\"average_trade_count\", 0)\n            average_win_rate = eval_metrics.get(\"average_win_rate\", 0)\n\n            if average_reward > 0:\n                logger.info(\"The model is achieving positive returns on evaluation, suggesting effective learning and policy optimization. Aim to maximize rewards by fine-tuning hyperparameters.\")\n            else:\n                logger.info(\"The model is achieving negative returns on evaluation, indicating suboptimal policy performance. Consider improving exploration strategies or increasing the training data diversity.\")\n\n            if sharpe_ratio > 1.0:\n                logger.info(\"The Sharpe Ratio is above 1.0, indicating a favorable risk-return trade-off. The model is performing well in terms of financial risk management.\")\n            elif 0 < sharpe_ratio <= 1.0:\n                logger.info(\"The Sharpe Ratio is moderate, suggesting a balanced but not exceptional risk-return trade-off. Further fine-tuning of the reward structure or trading strategy might be required.\")\n            else:\n                logger.info(\"The Sharpe Ratio is negative, indicating that the strategy's returns are not compensating for its risk. Consider revisiting the reward function or exploring alternative strategies.\")\n\n            if max_drawdown < 0.1:\n                logger.info(\"Max drawdown is low, indicating the model is maintaining controlled risk. This is a desirable attribute for a robust trading strategy.\")\n            else:\n                logger.info(\"Max drawdown is high, indicating significant risk exposure. Consider implementing risk management mechanisms or penalizing high drawdowns in the reward function.\")\n\n            logger.info(f\"Average Trade Count per Episode: {average_trade_count:.2f}\")\n            logger.info(f\"Average Win Rate: {average_win_rate:.2%}\")\n\n            if average_win_rate > 0.5:\n                logger.info(f\"The model has an average win rate of {average_win_rate:.2%}, showing that more than half of the trades are profitable. Aim to further improve profitability by refining the strategy.\")\n            else:\n                logger.info(f\"The model has an average win rate of {average_win_rate:.2%}. Explore strategies to improve precision and reduce losses in less favorable scenarios.\")\n\n    except Exception as e:\n        logger.error(f\"Exception in interpret_metrics: {e}\", exc_info=True)\n\n\ndef monitor_training(\n    epoch: int,\n    rank: int,\n    avg_metrics: dict,\n    eval_metrics: dict,\n    writer: SummaryWriter = None\n) -> None:\n    \"\"\"\n    Monitor and log training progress, metrics, and insights.\n\n    Args:\n        epoch (int): Current epoch number.\n        rank (int): Process rank (for compatibility).\n        avg_metrics (dict): Dictionary of average training metrics.\n        eval_metrics (dict): Dictionary of evaluation metrics.\n        writer (SummaryWriter, optional): TensorBoard SummaryWriter.\n    \"\"\"\n    logs_dir = \"logs\"\n    os.makedirs(logs_dir, exist_ok=True)\n\n    # Log metrics in console logs\n    logger.info(f\"Epoch {epoch + 1} - Training Metrics: {avg_metrics}\")\n    logger.info(f\"Epoch {epoch + 1} - Evaluation Metrics: {eval_metrics}\")\n\n    # Write metrics to a JSON file\n    metrics_file = os.path.join(logs_dir, \"metrics.json\")\n    try:\n        with open(metrics_file, \"a\", encoding='utf-8') as f:\n            json.dump(\n                {\n                    \"epoch\": epoch + 1,\n                    \"train_metrics\": avg_metrics,\n                    \"eval_metrics\": eval_metrics,\n                },\n                f,\n                ensure_ascii=False\n            )\n            f.write(\"\\n\")\n        logger.info(f\"Metrics appended to {metrics_file}\")\n    except Exception as e:\n        logger.error(f\"Failed to write to {metrics_file}: {e}\")\n\n    if writer:\n        # Write training metrics\n        for k, v in avg_metrics.items():\n            writer.add_scalar(f\"Train/{k}\", v, epoch)\n        # Write evaluation metrics\n        for k, v in eval_metrics.items():\n            writer.add_scalar(f\"Eval/{k}\", v, epoch)\n        writer.flush()\n\n    # Optionally log CPU/GPU usage\n    log_memory_usage(logger, phase=\"After Logging Metrics\")\n    log_gpu_memory(logger, phase=\"After Logging Metrics\")\n\n    # Interpret metrics\n    try:\n        interpret_metrics(avg_metrics, eval_metrics, logger)\n    except Exception as e:\n        logger.error(f\"Failed to interpret metrics: {e}\", exc_info=True)\n\n\ndef setup_tensorboard(rank: int, log_dir: str = \"runs\") -> SummaryWriter:\n    \"\"\"\n    Setup TensorBoard writer.\n\n    Args:\n        rank (int): Process rank.\n        log_dir (str): Directory for TensorBoard logs.\n\n    Returns:\n        SummaryWriter: TensorBoard SummaryWriter instance.\n    \"\"\"\n    writer = None\n    os.makedirs(log_dir, exist_ok=True)\n    try:\n        writer = SummaryWriter(log_dir=log_dir)\n        logger.info(f\"TensorBoard writer set up at {log_dir}.\")\n    except Exception as e:\n        logger.error(f\"Failed to set up TensorBoard writer: {e}\")\n    return writer",
    "./environment/trading_env.py": "# File: environment/trading_env.py\n\nimport gym\nfrom gym import spaces\nimport torch\nimport numpy as np\nimport os\nimport glob\nimport logging\n\n# Initialize logger for TradingEnv\ndef setup_trading_env_logging(log_dir=\"logs\", log_file=\"trading_env.log\", file_logging_enabled=False):\n    \"\"\"\n    Configure logging for the TradingEnv with an on/off switch for file logging.\n    \"\"\"\n    os.makedirs(log_dir, exist_ok=True)\n    logger = logging.getLogger(\"TradingEnv\")\n    logger.setLevel(logging.DEBUG)\n\n    # Create console handler\n    ch = logging.StreamHandler()\n    ch.setLevel(logging.INFO)\n\n    # Create formatter and add to handlers\n    formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n    ch.setFormatter(formatter)\n\n    # Add handlers to the logger if none exist\n    if not logger.hasHandlers():\n        logger.addHandler(ch)\n\n        # Add file handler based on the flag\n        if file_logging_enabled:\n            fh = logging.FileHandler(os.path.join(log_dir, log_file))\n            fh.setLevel(logging.DEBUG)\n            fh.setFormatter(formatter)\n            logger.addHandler(fh)\n        else:\n            # Add a NullHandler to effectively disable file logging\n            fh = logging.NullHandler()\n            logger.addHandler(fh)\n\n    return logger\n\nlogger = logging.getLogger(\"TradingEnv\")\nif not logger.hasHandlers():\n    logger = setup_trading_env_logging()\n\nclass TradingEnv(gym.Env):\n    \"\"\"\n    Custom Trading Environment that follows OpenAI Gym interface.\n    \"\"\"\n    metadata = {'render.modes': ['human']}\n\n    def __init__(self, config):\n        super(TradingEnv, self).__init__()\n\n        self.config = config\n\n        # Initialize attributes\n        self.feature_columns = config['data']['feature_columns']\n        self.n_steps = config['data']['n_steps']\n        self.max_steps = config['environment']['max_steps']\n\n        # Ensure feature_dim matches number of feature_columns\n        feature_dim = config['model']['feature_dim']\n        assert feature_dim == len(self.feature_columns), \\\n            f\"feature_dim ({feature_dim}) does not match the number of feature columns ({len(self.feature_columns)}).\"\n        logger.debug(f\"feature_dim ({feature_dim}) matches number of feature columns ({len(self.feature_columns)}).\")\n\n        # Load data from PT files\n        pt_data_dir = config['data'].get('pt_data_dir', 'data/processed/')\n        pt_data_prefix = config['data'].get('pt_data_prefix', 'processed_data_part')\n        pt_files = sorted(glob.glob(os.path.join(pt_data_dir, f\"{pt_data_prefix}_*.pt\")))\n\n        if not pt_files:\n            logger.error(f\"No processed data files found in {pt_data_dir}\")\n            raise FileNotFoundError(f\"No processed data files found in {pt_data_dir}\")\n\n        self.pt_files = pt_files\n        self.current_pt_file_idx = -1\n        self.data = None\n        self.close_prices = None  # Unnormalized close prices for current data chunk\n        self.next_close_prices = None  # Next close prices for supervised learning and synthetic reward\n\n        # Call load_next_data_chunk() after initializing attributes\n        self.load_next_data_chunk()\n\n        # Define action and observation space\n        self.action_dim = config['model']['action_dim']\n        self.action_space = spaces.Discrete(self.action_dim)\n        self.observation_space = spaces.Box(\n            low=-np.inf,\n            high=np.inf,\n            shape=(self.n_steps, len(self.feature_columns)),\n            dtype=np.float32\n        )\n        logger.debug(f\"Action space: {self.action_space}, Observation space: {self.observation_space}\")\n\n        # Initialize variables\n        self.current_step = 0\n        self.position = 0.0  # Number of shares held\n        self.initial_wallet_balance = config['environment'].get('initial_wallet_balance', 10000.0)\n        self.wallet_balance = self.initial_wallet_balance  # Cash available\n        self.total_profit = 0.0\n        self.total_trades = 0\n\n        # Action mappings for scaling positions (7 actions)\n        if self.action_dim != 7:\n            logger.warning(f\"Expected action_dim=7, but got action_dim={self.action_dim}. Action mappings may be incorrect.\")\n        self.action_mappings = {\n            0: ('sell', 1.0),   # Sell 100% of current position\n            1: ('sell', 0.5),   # Sell 50% of current position\n            2: ('sell', 0.25),  # Sell 25% of current position\n            3: ('hold', 0.0),   # Hold\n            4: ('buy', 0.25),   # Buy 25% of max possible position\n            5: ('buy', 0.5),    # Buy 50% of max possible position\n            6: ('buy', 1.0),    # Buy 100% of max possible position\n        }\n\n        # Initialize previous price\n        self.previous_price = None\n\n        # Feature indices mapping\n        self.feature_indices = {name: idx for idx, name in enumerate(self.feature_columns)}\n\n    def load_next_data_chunk(self):\n        self.current_pt_file_idx = (self.current_pt_file_idx + 1) % len(self.pt_files)\n        try:\n            loaded = torch.load(self.pt_files[self.current_pt_file_idx])\n            if 'data' not in loaded or 'close_prices' not in loaded or 'next_close_prices' not in loaded:\n                logger.error(f\"PT file {self.pt_files[self.current_pt_file_idx]} missing required keys.\")\n                raise KeyError(f\"PT file {self.pt_files[self.current_pt_file_idx]} missing required keys.\")\n            self.data = loaded['data']\n            self.close_prices = loaded['close_prices']\n            self.next_close_prices = loaded['next_close_prices']\n            logger.info(f\"Loaded data chunk from {self.pt_files[self.current_pt_file_idx]}\")\n            logger.debug(f\"Data chunk shape: {self.data.shape}, Close prices length: {len(self.close_prices)}\")\n        except Exception as e:\n            logger.error(f\"Failed to load data chunk from {self.pt_files[self.current_pt_file_idx]}: {e}\", exc_info=True)\n            raise\n\n        # Recalculate end_step based on new data\n        self.end_step = len(self.data)  # Corrected end_step calculation\n        self.current_step = 0  # Reset current_step when loading new data\n        logger.debug(f\"Data chunk loaded. Total steps: {self.end_step}\")\n\n    def reset(self):\n        if self.current_step >= self.end_step:\n            logger.debug(\"Current step exceeds end_step. Loading next data chunk.\")\n            self.load_next_data_chunk()\n\n        self.current_step = 0\n        self.position = 0.0\n        self.wallet_balance = self.initial_wallet_balance\n        self.total_profit = 0.0\n        self.total_trades = 0\n        self.entry_price = None\n        self.previous_price = self.close_prices[self.current_step]\n        logger.debug(f\"Environment reset. Starting at step {self.current_step}\")\n        return self._next_observation()\n\n    def _next_observation(self):\n        obs = self.data[self.current_step]\n        if torch.isnan(obs).any():\n            logger.error(f\"Observation contains NaN values at step {self.current_step}\")\n            raise ValueError(f\"Observation contains NaN values at step {self.current_step}\")\n        logger.debug(f\"Next observation shape: {obs.shape}\")\n        return obs.numpy().astype(np.float32)\n\n    def step(self, action):\n        try:\n            logger.debug(f\"Step {self.current_step}: Received action {action}\")\n            if action not in self.action_mappings:\n                logger.error(f\"Invalid action: {action}\")\n                raise ValueError(f\"Invalid action: {action}\")\n\n            action_type, amount = self.action_mappings[action]\n\n            idx = self.current_step  # Corrected idx calculation\n            if idx >= len(self.data) - 1:\n                done = True\n                obs = np.zeros_like(self.observation_space.sample())\n                info = {\n                    'total_profit': self.total_profit,\n                    'wallet_balance': self.wallet_balance,\n                    'position': self.position\n                }\n                logger.debug(f\"Reached end of data at step {self.current_step}. Ending episode.\")\n                return obs, 0.0, done, info\n\n            # Get current and next prices as floats\n            current_price = self.close_prices[idx].item()\n            next_price = self.next_close_prices[idx].item()\n            logger.debug(f\"Current price at step {self.current_step}: {current_price}\")\n\n            # Transaction cost\n            fee_rate = self.config['environment'].get('transaction_fee', 0.001)  # 0.1%\n            transaction_cost = 0.0\n\n            # Calculate max position based on wallet balance\n            max_possible_position = self.wallet_balance / current_price\n\n            # Execute action\n            if action_type == 'buy':\n                # Determine how much to buy\n                buy_amount = amount * max_possible_position\n                cost = buy_amount * current_price\n                transaction_cost = fee_rate * cost\n                total_cost = cost + transaction_cost\n\n                if total_cost > self.wallet_balance:\n                    # Can't afford this purchase\n                    logger.debug(f\"Cannot afford to buy {buy_amount} units at {current_price} each. Holding instead.\")\n                    buy_amount = 0.0\n                    total_cost = 0.0\n                    transaction_cost = 0.0\n\n                self.position += buy_amount\n                self.wallet_balance -= total_cost\n                self.total_trades += 1\n                logger.debug(f\"Action: Buy {buy_amount} units. New position: {self.position}, Wallet balance: {self.wallet_balance}\")\n            elif action_type == 'sell':\n                # Determine how much to sell\n                sell_amount = amount * self.position\n                revenue = sell_amount * current_price\n                transaction_cost = fee_rate * revenue\n                total_revenue = revenue - transaction_cost\n\n                self.position -= sell_amount\n                self.wallet_balance += total_revenue\n                self.total_trades += 1\n                logger.debug(f\"Action: Sell {sell_amount} units. New position: {self.position}, Wallet balance: {self.wallet_balance}\")\n            else:\n                # Hold action\n                logger.debug(\"Action: Hold. No change in position.\")\n\n            # Prevent negative position or wallet balance\n            if self.position < 0:\n                logger.error(\"Negative position encountered.\")\n                self.position = 0.0\n\n            if self.wallet_balance < 0:\n                logger.error(\"Negative wallet balance encountered.\")\n                # Apply penalty or end the episode\n                pass  # Depending on how you want to handle negative balance\n\n            # Update total profit (unrealized profit)\n            market_value = self.position * current_price\n            total_assets = self.wallet_balance + market_value\n            self.total_profit = total_assets - self.initial_wallet_balance\n\n            # Calculate reward\n            reward = self._calculate_reward()\n\n            self.previous_price = current_price\n            self.current_step += 1\n            done = self.current_step >= self.end_step or self.current_step >= self.max_steps\n\n            obs = self._next_observation() if not done else np.zeros_like(self.observation_space.sample())\n            info = {\n                'total_profit': self.total_profit,\n                'wallet_balance': self.wallet_balance,\n                'position': self.position\n            }\n\n            logger.debug(f\"Step {self.current_step}: Done={done}, Total profit: {self.total_profit}, Wallet balance: {self.wallet_balance}, Position: {self.position}\")\n\n            return obs, reward, done, info\n        except Exception as e:\n            logger.error(f\"Exception in step method: {e}\", exc_info=True)\n            done = True\n            obs = np.zeros_like(self.observation_space.sample())\n            reward = -1.0  # Assign negative reward on exception\n            info = {\n                'total_profit': self.total_profit,\n                'wallet_balance': self.wallet_balance,\n                'position': self.position,\n                'error': str(e)\n            }\n            return obs, reward, done, info\n\n    def _calculate_reward(self):\n        \"\"\"\n        Reward based on the increase or decrease in total assets (wallet balance + position value).\n        \"\"\"\n        # Calculate current total assets\n        current_price = self.close_prices[self.current_step].item()\n        market_value = self.position * current_price\n        total_assets = self.wallet_balance + market_value\n\n        # Calculate profit since last step\n        profit = total_assets - (self.wallet_balance + self.position * self.previous_price)\n\n        # Penalize negative wallet balance\n        if self.wallet_balance < 0:\n            penalty = self.config['environment'].get('negative_balance_penalty', 10.0)\n            profit -= penalty\n\n        # Apply reward scaling from config\n        reward_scale = self.config['data'].get('reward_scale', 1.0)\n        reward = profit * reward_scale\n\n        return reward\n\n    def render(self, mode='human'):\n        # Implement rendering if needed\n        print(f\"Step: {self.current_step}\")\n        print(f\"Position: {self.position}\")\n        print(f\"Wallet Balance: {self.wallet_balance}\")\n        print(f\"Total Profit: {self.total_profit}\")",
    "./training/runner.py": "# training/runner.py\n\nimport os\nimport torch\nimport torch.distributed as dist\nfrom torch.nn.parallel import DistributedDataParallel as DDP\nfrom training.train_loop import train_model\nfrom training.config_validation import validate_and_cast_config\nfrom training.utils import setup_logging, setup_seed, load_checkpoint\nfrom environment.trading_env import TradingEnv\nfrom models.models import PPOAgent\nfrom data_processing.preprocess import export_and_save_preprocessed_data\nimport logging\nimport sys\n\n\ndef setup_runner_logging(rank, log_dir=\"logs\"):\n    \"\"\"\n    Configure logging for the runner.\n    Each rank will have its own log file.\n    \"\"\"\n    os.makedirs(log_dir, exist_ok=True)\n    logger = logging.getLogger(f\"RunnerRank{rank}\")\n    logger.setLevel(logging.DEBUG)\n\n    # Create handlers only if they haven't been created yet\n    if not logger.hasHandlers():\n        # Console handler for INFO level\n        ch = logging.StreamHandler()\n        ch.setLevel(logging.INFO)\n        # File handler for DEBUG level\n        fh = logging.FileHandler(os.path.join(log_dir, f\"runner_rank{rank}.log\"))\n        fh.setLevel(logging.DEBUG)\n\n        # Create formatter and add to handlers\n        formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n        ch.setFormatter(formatter)\n        fh.setFormatter(formatter)\n\n        # Add handlers to the logger\n        logger.addHandler(ch)\n        logger.addHandler(fh)\n\n    return logger\n\n\ndef train_worker(rank, config, world_size):\n    \"\"\"\n    Worker function for distributed training.\n    Initializes the environment, model, and starts the training loop.\n    \"\"\"\n    logger = setup_runner_logging(rank)\n    logger.info(f\"Rank {rank}: Starting training worker.\")\n\n    try:\n        # Set random seed for reproducibility\n        setup_seed(config.get(\"random_seed\", 42) + rank)\n        logger.debug(f\"Rank {rank}: Random seed set.\")\n\n        # Initialize the process group\n        backend = \"nccl\" if torch.cuda.is_available() else \"gloo\"\n        logger.debug(f\"Rank {rank}: Initializing process group with backend={backend}\")\n        dist.init_process_group(\n            backend=backend,\n            init_method=\"env://\",\n            world_size=world_size,\n            rank=rank\n        )\n        logger.info(f\"Rank {rank}: Process group initialized.\")\n\n        # Set device for each rank\n        if torch.cuda.is_available():\n            if rank >= torch.cuda.device_count():\n                logger.error(f\"Rank {rank}: No GPU available for this rank.\")\n                raise ValueError(f\"Rank {rank} exceeds available GPUs.\")\n            torch.cuda.set_device(rank)\n            device = torch.device(f\"cuda:{rank}\")\n            logger.debug(f\"Rank {rank}: Device set to {device}\")\n        else:\n            device = torch.device(\"cpu\")\n            logger.debug(f\"Rank {rank}: Device set to CPU\")\n\n        # Only rank 0 performs data preprocessing\n        if rank == 0 and config.get(\"preprocessing\", {}).get(\"enable\", True):\n            logger.info(\"Rank 0: Starting data preprocessing.\")\n            export_and_save_preprocessed_data(config, logger)\n        elif config.get(\"preprocessing\", {}).get(\"enable\", True):\n            logger.info(f\"Rank {rank}: Waiting for data preprocessing to complete.\")\n        else:\n            logger.info(\"Data preprocessing is disabled in the configuration.\")\n\n        # Ensure all processes wait for data preprocessing to complete\n        logger.debug(f\"Rank {rank}: Waiting at barrier after preprocessing.\")\n        dist.barrier()\n        logger.debug(f\"Rank {rank}: Passed barrier after preprocessing.\")\n\n        # Initialize environment\n        try:\n            env = TradingEnv(config)\n            logger.info(f\"Rank {rank}: Environment initialized.\")\n        except Exception as e:\n            logger.error(f\"Rank {rank}: Failed to initialize environment: {e}\", exc_info=True)\n            raise\n\n        # Initialize model\n        try:\n            model = PPOAgent(\n                feature_dim=config[\"model\"][\"feature_dim\"],\n                hidden_dim=config[\"model\"][\"ppo_hidden_dim\"],\n                action_dim=config[\"model\"][\"action_dim\"],\n                n_heads=config[\"model\"][\"ppo_n_heads\"],\n                n_layers=config[\"model\"][\"ppo_n_layers\"],\n                dropout=config[\"model\"][\"ppo_dropout\"],\n                temperature=config[\"model\"].get('temperature', 1.0),\n                layer_norm=config[\"model\"].get('layer_norm', False)\n            ).to(device)\n            logger.info(f\"Rank {rank}: PPOAgent initialized.\")\n        except Exception as e:\n            logger.error(f\"Rank {rank}: Failed to initialize PPOAgent: {e}\", exc_info=True)\n            raise\n\n        # Load checkpoint if required\n        start_epoch = 0\n        best_reward = float(\"-inf\")\n        resume_ckpt = config[\"checkpoint\"].get(\"resume_checkpoint\", \"\")\n        if resume_ckpt:\n            logger.info(f\"Rank {rank}: Resuming from checkpoint {resume_ckpt}\")\n            if os.path.exists(resume_ckpt):\n                try:\n                    epoch_loaded, best_reward_loaded = load_checkpoint(\n                        resume_ckpt,\n                        model,\n                        None,  # If you want to also load optimizer states, pass them here\n                        None,\n                        device,\n                    )\n                    start_epoch = epoch_loaded + 1\n                    best_reward = best_reward_loaded\n                    logger.info(f\"Rank {rank}: Resumed from epoch {start_epoch}, best_reward={best_reward}\")\n                except Exception as e:\n                    logger.error(f\"Rank {rank}: Failed to load checkpoint: {e}\", exc_info=True)\n                    raise\n            else:\n                logger.warning(f\"Rank {rank}: Checkpoint {resume_ckpt} not found. Starting fresh.\")\n        else:\n            logger.info(f\"Rank {rank}: No checkpoint specified. Starting from epoch 0.\")\n\n        # Wrap model in DDP\n        try:\n            if world_size > 1:\n                model = DDP(model, device_ids=[rank] if torch.cuda.is_available() else None)\n                logger.debug(f\"Rank {rank}: Model wrapped in DDP.\")\n            else:\n                logger.debug(f\"Rank {rank}: Single GPU mode. DDP not applied.\")\n        except Exception as e:\n            logger.error(f\"Rank {rank}: Failed to wrap model in DDP: {e}\", exc_info=True)\n            raise\n\n        # Start training loop\n        try:\n            logger.info(f\"Rank {rank}: Starting training loop from epoch {start_epoch + 1} to {config['training']['epochs']}\")\n            train_model(\n                env=env,\n                config=config,\n                device=device,\n                model=model,\n                rank=rank,\n                world_size=world_size,\n                start_epoch=start_epoch,\n                num_epochs=config[\"training\"][\"epochs\"],\n                best_reward=best_reward\n            )\n        except Exception as e:\n            logger.error(f\"Rank {rank}: Training failed: {e}\", exc_info=True)\n            raise\n    finally:\n        if dist.is_initialized():\n            logger.info(f\"Rank {rank}: Destroying process group.\")\n            dist.destroy_process_group()\n\ndef main():\n    import argparse\n    import yaml\n\n    # Parse command-line arguments\n    parser = argparse.ArgumentParser(description=\"Distributed PPO Training\")\n    parser.add_argument(\"--config_path\", type=str, required=True, help=\"Path to the config.yaml file\")\n    args = parser.parse_args()\n\n    # Load configuration\n    try:\n        with open(args.config_path, \"r\") as f:\n            config = yaml.safe_load(f)\n        # Validate and cast configuration\n        config = validate_and_cast_config(config)\n    except Exception as e:\n        print(f\"Failed to load or validate configuration: {e}\")\n        sys.exit(1)\n\n    # Setup logging (main logger)\n    main_logger = setup_logging(log_dir=\"logs\", log_file=\"main.log\")\n    main_logger.info(\"Main: Configuration loaded and validated.\")\n\n    # Set random seed\n    try:\n        setup_seed(config.get(\"random_seed\", 42))\n        main_logger.info(\"Main: Random seed set for reproducibility.\")\n    except Exception as e:\n        main_logger.error(f\"Main: Failed to set random seed: {e}\", exc_info=True)\n        sys.exit(1)\n\n    # Distributed setup\n    world_size = config[\"distributed\"][\"world_size\"]\n    master_addr = config[\"distributed\"][\"master_addr\"]\n    master_port = config[\"distributed\"][\"master_port\"]\n\n    os.environ['MASTER_ADDR'] = master_addr\n    os.environ['MASTER_PORT'] = master_port\n\n    if world_size > 1:\n        main_logger.info(f\"Main: Initializing Distributed Data Parallel with world_size={world_size}\")\n        try:\n            torch.multiprocessing.spawn(\n                train_worker,\n                args=(config, world_size),\n                nprocs=world_size,\n                join=True\n            )\n        except Exception as e:\n            main_logger.error(f\"Main: DDP training failed to spawn processes: {e}\", exc_info=True)\n            sys.exit(1)\n    else:\n        main_logger.info(\"Main: Running training on a single GPU without DDP.\")\n        try:\n            train_worker(0, config, world_size)\n        except Exception as e:\n            main_logger.error(f\"Main: Single GPU training failed: {e}\", exc_info=True)\n            sys.exit(1)\n\n\nif __name__ == \"__main__\":\n    main()",
    "./training/train_loop.py": "# File: training/train_loop.py\n\nimport os\nimport logging\nimport torch\nimport numpy as np\nimport torch.distributed as dist  # Distributed training\nfrom torch.utils.data import DataLoader, TensorDataset, DistributedSampler\nimport torch.nn.functional as F\nimport sys\nimport gc\nimport copy  # For copying model weights\n\nfrom training.utils import (\n    compute_advantages_and_returns,\n    save_checkpoint,\n    log_memory_usage,\n    log_gpu_memory,\n    setup_logger,\n)\nfrom monitoring import monitor_training, setup_tensorboard\nfrom training.evaluation import evaluate_model\nfrom training.collect_trajectories import collect_trajectories\nfrom training.metric_analysis import analyze_metrics\nfrom training.optimizer_setup import (\n    setup_optimizers,\n    setup_schedulers,\n    update_entropy_coefficient,\n)\nfrom environment.trading_env import TradingEnv\n\n# Import autocast and GradScaler from torch.cuda.amp\nfrom torch.cuda.amp import autocast, GradScaler\n\ndef train_model(\n    env,\n    config,\n    device,\n    model,\n    rank,\n    world_size,\n    start_epoch,\n    num_epochs,\n    best_reward=-float(\"inf\"),\n):\n    \"\"\"\n    Main training loop for PPO with DDP.\n    \"\"\"\n    logger = setup_logger(f\"TrainLoopRank{rank}\", \"train_loop\")\n\n    # Initialize previous_weights for logging weight updates\n    previous_weights = None\n\n    try:\n        logger.info(f\"Starting training from epoch {start_epoch + 1} to {num_epochs}\")\n        logger.info(f\"Rank {rank}: Using batch size {config['training']['batch_size']}\")\n\n        # Optimizer and Scheduler setup\n        optimizer_actor, optimizer_critic, optimizer_supervised = setup_optimizers(model, config)\n        scheduler_actor, scheduler_critic, scheduler_supervised = setup_schedulers(\n            optimizer_actor, optimizer_critic, optimizer_supervised, config, num_epochs\n        )\n\n        max_grad_norm = config[\"training\"].get(\"max_grad_norm\", 0.5)\n        batch_size = config[\"training\"][\"batch_size\"]\n        ppo_epochs = config[\"training\"][\"ppo_epochs\"]\n        supervised_loss_weight = config[\"training\"].get(\"supervised_loss_weight\", 1.0)\n\n        # Setup TensorBoard writer (only on rank 0)\n        writer = None\n        if config[\"logging\"][\"use_tensorboard\"] and rank == 0:\n            writer = setup_tensorboard(rank, log_dir=config[\"logging\"][\"log_dir\"])\n            logger.debug(\"TensorBoard writer initialized.\")\n\n        # Initialize AMP scaler\n        scaler = GradScaler()\n        logger.debug(\"AMP GradScaler initialized.\")\n\n        # Early stopping parameters\n        early_stopping = config[\"training\"].get(\"early_stopping\", {})\n        early_stopping_enabled = early_stopping.get(\"enabled\", False)\n        patience = early_stopping.get(\"patience\", 10)\n        stopping_metric = early_stopping.get(\"metric\", \"average_reward\")\n        stopping_mode = early_stopping.get(\"mode\", \"max\")\n        no_improvement_epochs = 0\n        best_metric_value = -float(\"inf\") if stopping_mode == \"max\" else float(\"inf\")\n\n        for epoch in range(start_epoch, num_epochs):\n            logger.info(f\"Rank {rank}, Epoch {epoch + 1}/{num_epochs}\")\n            log_memory_usage(logger, phase=f\"Epoch {epoch + 1} Start\")\n            log_gpu_memory(logger, phase=f\"Epoch {epoch + 1} Start\")\n\n            # Update entropy coefficient based on schedule\n            entropy_coeff = update_entropy_coefficient(config, epoch, logger)\n\n            # Reset environment for new epoch\n            logger.debug(f\"Rank {rank}: Resetting environment for new epoch.\")\n            state = env.reset()\n\n            # Collect trajectories\n            trajectories, action_counts = collect_trajectories(\n                env, model, device, config, num_episodes=10, logger=logger\n            )\n            logger.info(\n                f\"Rank {rank}: Collected {len(trajectories['rewards'])} steps.\"\n            )\n            logger.info(f\"Rank {rank}: Action distribution during data collection: {action_counts}\")\n            logger.debug(\"After collect_trajectories\")\n\n            # Convert collected data to tensors\n            try:\n                states = torch.stack(\n                    [torch.from_numpy(np.array(s)) for s in trajectories[\"states\"]]\n                ).float()\n                actions = torch.tensor(\n                    trajectories[\"actions\"], dtype=torch.long\n                )\n                rewards = torch.tensor(\n                    trajectories[\"rewards\"], dtype=torch.float32\n                )\n                dones = torch.tensor(\n                    trajectories[\"dones\"], dtype=torch.float32\n                )\n                values = torch.tensor(\n                    trajectories[\"values\"], dtype=torch.float32\n                )\n                log_probs_old = torch.tensor(\n                    trajectories[\"log_probs\"], dtype=torch.float32\n                )\n                next_close_prices = torch.tensor(\n                    trajectories[\"next_close_prices\"], dtype=torch.float32\n                )\n\n                # Move tensors to device\n                states = states.to(device)\n                actions = actions.to(device)\n                rewards = rewards.to(device)\n                dones = dones.to(device)\n                values = values.to(device)\n                log_probs_old = log_probs_old.to(device)\n                next_close_prices = next_close_prices.to(device)\n\n                logger.debug(\"Converted collected trajectories to tensors.\")\n            except Exception as e:\n                logger.error(\n                    f\"Exception during tensor conversion: {e}\", exc_info=True\n                )\n                raise\n\n            # Compute advantages and returns\n            try:\n                advantages, returns = compute_advantages_and_returns(\n                    rewards=rewards,\n                    values=values,\n                    dones=dones,\n                    gamma=config[\"training\"][\"gamma\"],\n                    gae_lambda=config[\"training\"][\"gae_lambda\"],\n                )\n                # Normalize advantages\n                advantages_mean = advantages.mean().item()\n                advantages_std = advantages.std().item()\n                advantages = (advantages - advantages_mean) / (\n                    advantages_std + 1e-8\n                )\n                logger.debug(f\"Computed advantages and returns. Advantages mean: {advantages_mean}, std: {advantages_std}\")\n            except Exception as e:\n                logger.error(\n                    f\"Exception during advantage and return computation: {e}\",\n                    exc_info=True,\n                )\n                raise\n\n            # Create a TensorDataset from the collected trajectories\n            dataset = TensorDataset(\n                states, actions, returns, advantages, log_probs_old, next_close_prices\n            )\n\n            # Create a DistributedSampler to distribute the data across processes\n            sampler = DistributedSampler(\n                dataset, num_replicas=world_size, rank=rank, shuffle=True\n            )\n\n            # Create a DataLoader to handle batching and shuffling\n            dataloader = DataLoader(\n                dataset, batch_size=batch_size, sampler=sampler\n            )\n\n            # PPO Update\n            for ppo_epoch in range(ppo_epochs):\n                try:\n                    logger.debug(f\"Rank {rank}: Starting PPO Epoch {ppo_epoch + 1}\")\n\n                    for batch_idx, (\n                        batch_states,\n                        batch_actions,\n                        batch_returns,\n                        batch_advantages,\n                        batch_log_probs_old,\n                        batch_next_close_prices,\n                    ) in enumerate(dataloader):\n                        optimizer_actor.zero_grad()\n                        optimizer_critic.zero_grad()\n                        optimizer_supervised.zero_grad()\n\n                        if device.type == \"cuda\":\n                            with autocast():\n                                logits, values_pred, predicted_close = model(batch_states)\n                                if torch.isnan(logits).any() or torch.isinf(\n                                    logits\n                                ).any():\n                                    logger.error(\n                                        \"Logits contain NaNs or Infs during PPO update.\"\n                                    )\n                                    raise ValueError(\n                                        \"Invalid logits detected during PPO update.\"\n                                    )\n                                action_dist = torch.distributions.Categorical(\n                                    logits=logits\n                                )\n                                log_probs = action_dist.log_prob(batch_actions)\n                                entropy = action_dist.entropy().mean()\n\n                                # Calculate ratios\n                                ratios = torch.exp(\n                                    log_probs - batch_log_probs_old\n                                )\n\n                                # Surrogate loss\n                                surr1 = ratios * batch_advantages\n                                surr2 = torch.clamp(\n                                    ratios,\n                                    1.0 - config[\"training\"][\"clip_epsilon\"],\n                                    1.0 + config[\"training\"][\"clip_epsilon\"],\n                                ) * batch_advantages\n                                actor_loss = (\n                                    -torch.min(surr1, surr2).mean()\n                                    - entropy_coeff * entropy\n                                )\n\n                                # Critic loss\n                                critic_loss = F.mse_loss(\n                                    values_pred.squeeze(-1), batch_returns\n                                )\n\n                                # Supervised loss\n                                supervised_loss = F.mse_loss(\n                                    predicted_close.squeeze(-1), batch_next_close_prices\n                                )\n\n                                # Total loss\n                                total_loss = actor_loss + critic_loss + supervised_loss_weight * supervised_loss\n\n                                # KL Divergence\n                                with torch.no_grad():\n                                    old_action_dist = torch.distributions.Categorical(\n                                        logits=logits.detach()\n                                    )\n                                    kl_divergence = torch.distributions.kl.kl_divergence(\n                                        old_action_dist, action_dist\n                                    ).mean()\n                            logger.debug(\"Forward pass with autocast completed.\")\n\n                            # Backward pass with AMP\n                            scaler.scale(total_loss).backward()\n                            scaler.unscale_(optimizer_actor)\n                            scaler.unscale_(optimizer_critic)\n                            scaler.unscale_(optimizer_supervised)\n\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.actor_head.parameters(), max_grad_norm\n                            )\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.critic_head.parameters(), max_grad_norm\n                            )\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.supervised_head.parameters(), max_grad_norm\n                            )\n\n                            scaler.step(optimizer_actor)\n                            scaler.step(optimizer_critic)\n                            scaler.step(optimizer_supervised)\n\n                            scaler.update()\n                            logger.debug(\n                                \"Backward pass and optimizer steps completed with AMP.\"\n                            )\n                        else:\n                            # Forward pass without autocast\n                            logits, values_pred, predicted_close = model(batch_states)\n                            if torch.isnan(logits).any() or torch.isinf(\n                                logits\n                            ).any():\n                                logger.error(\n                                    \"Logits contain NaNs or Infs during PPO update.\"\n                                )\n                                raise ValueError(\n                                    \"Invalid logits detected during PPO update.\"\n                                )\n                            action_dist = torch.distributions.Categorical(\n                                logits=logits\n                            )\n                            log_probs = action_dist.log_prob(batch_actions)\n                            entropy = action_dist.entropy().mean()\n\n                            # Calculate ratios\n                            ratios = torch.exp(\n                                log_probs - batch_log_probs_old\n                            )\n\n                            # Surrogate loss\n                            surr1 = ratios * batch_advantages\n                            surr2 = torch.clamp(\n                                ratios,\n                                1.0 - config[\"training\"][\"clip_epsilon\"],\n                                1.0 + config[\"training\"][\"clip_epsilon\"],\n                            ) * batch_advantages\n                            actor_loss = (\n                                -torch.min(surr1, surr2).mean()\n                                - entropy_coeff * entropy\n                            )\n\n                            # Critic loss\n                            critic_loss = F.mse_loss(\n                                values_pred.squeeze(-1), batch_returns\n                            )\n\n                            # Supervised loss\n                            supervised_loss = F.mse_loss(\n                                predicted_close.squeeze(-1), batch_next_close_prices\n                            )\n\n                            # Total loss\n                            total_loss = actor_loss + critic_loss + supervised_loss_weight * supervised_loss\n\n                            # KL Divergence\n                            with torch.no_grad():\n                                old_action_dist = torch.distributions.Categorical(\n                                    logits=logits.detach()\n                                )\n                                kl_divergence = torch.distributions.kl.kl_divergence(\n                                    old_action_dist, action_dist\n                                ).mean()\n\n                            # Backward pass\n                            total_loss.backward()\n\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.actor_head.parameters(), max_grad_norm\n                            )\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.critic_head.parameters(), max_grad_norm\n                            )\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.supervised_head.parameters(), max_grad_norm\n                            )\n\n                            optimizer_actor.step()\n                            optimizer_critic.step()\n                            optimizer_supervised.step()\n                            logger.debug(\n                                \"Backward pass and optimizer steps completed without AMP.\"\n                            )\n\n                        # Logging\n                        logger.info(\n                            f\"Rank {rank}, Epoch {epoch + 1}, PPO Epoch {ppo_epoch + 1}, \"\n                            f\"Actor Loss: {actor_loss.item():.6f}, Critic Loss: {critic_loss.item():.6f}, \"\n                            f\"Supervised Loss: {supervised_loss.item():.6f}, Entropy: {entropy.item():.6f}, KL Divergence: {kl_divergence.item():.6f}\"\n                        )\n\n                        # Store metrics for logging\n                        avg_metrics = {\n                            \"actor_loss\": actor_loss.item(),\n                            \"critic_loss\": critic_loss.item(),\n                            \"supervised_loss\": supervised_loss.item(),\n                            \"entropy\": entropy.item(),\n                            \"kl_divergence\": kl_divergence.item(),\n                            \"advantages_mean\": advantages_mean,\n                            \"advantages_std\": advantages_std,\n                        }\n\n                except Exception as e:\n                    logger.error(\n                        f\"Exception during PPO epoch {ppo_epoch + 1}: {e}\",\n                        exc_info=True,\n                    )\n                    raise\n\n            # Update learning rate schedulers\n            if scheduler_actor and scheduler_critic and scheduler_supervised:\n                scheduler_actor.step()\n                scheduler_critic.step()\n                scheduler_supervised.step()\n                logger.debug(\n                    f\"Learning rate updated: Actor LR={scheduler_actor.get_last_lr()[0]}, \"\n                    f\"Critic LR={scheduler_critic.get_last_lr()[0]}, Supervised LR={scheduler_supervised.get_last_lr()[0]}\"\n                )\n\n            # Reduce metrics across all processes\n            try:\n                avg_reward = rewards.mean()\n                dist.all_reduce(avg_reward, op=dist.ReduceOp.SUM)\n                avg_reward = avg_reward.item() / world_size\n                logger.info(\n                    f\"Rank {rank}, Epoch {epoch + 1} - Average Reward: {avg_reward:.6f}\"\n                )\n            except Exception as e:\n                logger.error(\n                    f\"Exception during metric reduction: {e}\", exc_info=True\n                )\n                raise\n\n            # Only rank 0 performs evaluation and checkpointing\n            if rank == 0:\n                try:\n                    # Ensure all processes reach this point but only rank 0 runs this block\n                    logger.debug(\n                        \"Rank 0: Waiting at barrier before evaluation and checkpointing.\"\n                    )\n                    dist.barrier()\n                    logger.debug(\n                        \"Rank 0: Passed barrier before evaluation and checkpointing.\"\n                    )\n\n                    # Evaluation\n                    if (epoch + 1) % config[\"evaluation\"][\"interval_eval\"] == 0:\n                        eval_env = TradingEnv(\n                            config\n                        )  # Create separate environment for evaluation\n                        logger.info(\"Rank 0: Starting model evaluation.\")\n                        eval_metrics = evaluate_model(\n                            model.module, eval_env, config, device  # Pass full model\n                        )\n                        logger.info(\"Rank 0: Model evaluation completed.\")\n                    else:\n                        eval_metrics = {}\n\n                    # Monitoring\n                    monitor_training(\n                        epoch, rank, avg_metrics, eval_metrics, writer\n                    )\n                    logger.debug(\"Monitoring metrics completed.\")\n\n                    # Analyze metrics and provide insights\n                    analyze_metrics(avg_metrics, eval_metrics, logger)\n\n                    # Checkpointing\n                    current_metric_value = eval_metrics.get(\n                        stopping_metric, None\n                    )\n                    improved = False\n                    if current_metric_value is not None:\n                        if (\n                            stopping_mode == \"max\"\n                            and current_metric_value > best_metric_value\n                        ) or (\n                            stopping_mode == \"min\"\n                            and current_metric_value < best_metric_value\n                        ):\n                            best_metric_value = current_metric_value\n                            no_improvement_epochs = 0\n                            improved = True\n                        else:\n                            no_improvement_epochs += 1\n                    else:\n                        logger.warning(\n                            f\"Evaluation metric '{stopping_metric}' not found in eval_metrics.\"\n                        )\n\n                    if avg_reward > best_reward:\n                        best_reward = avg_reward\n                        checkpoint_path = config[\"checkpoint\"][\"best_model_path\"]\n                        save_checkpoint(\n                            checkpoint_path,\n                            model.module,\n                            optimizer_actor,\n                            optimizer_critic,\n                            epoch,\n                            best_reward,\n                        )\n                        logger.info(\n                            f\"New best model saved at {checkpoint_path} with reward {best_reward:.6f}\"\n                        )\n\n                    # Early stopping check\n                    if (\n                        early_stopping_enabled\n                        and no_improvement_epochs >= patience\n                    ):\n                        logger.info(\n                            f\"Early stopping triggered after {patience} epochs with no improvement.\"\n                        )\n                        break  # Exit training loop\n\n                except Exception as e:\n                    logger.error(\n                        f\"Exception during evaluation or checkpointing: {e}\",\n                        exc_info=True,\n                    )\n                    raise\n            else:\n                logger.debug(\n                    f\"Rank {rank}: Waiting for evaluation and checkpointing.\"\n                )\n                dist.barrier()\n                logger.debug(\n                    f\"Rank {rank}: Finished waiting for evaluation and checkpointing.\"\n                )\n\n            # Log resource usage at the end of the epoch\n            try:\n                log_memory_usage(logger, phase=f\"Epoch {epoch + 1} End\")\n                log_gpu_memory(logger, phase=f\"Epoch {epoch + 1} End\")\n                logger.debug(f\"Rank {rank}: Resource usage logged at epoch end.\")\n            except Exception as e:\n                logger.error(\n                    f\"Exception during resource logging: {e}\", exc_info=True\n                )\n\n            # Only rank 0 logs weight norms and weight updates\n            if rank == 0:\n                # Log weight norms\n                log_weight_norms(model.module, logger, epoch, writer)\n\n                # Log weight updates\n                previous_weights = log_weight_updates(model.module, previous_weights, logger, epoch, writer)\n\n                # Optionally, log weight histograms to TensorBoard\n                if writer:\n                    for name, param in model.module.named_parameters():\n                        if param.requires_grad:\n                            writer.add_histogram(f\"Weights/{name}\", param.data.cpu().numpy(), epoch)\n\n            # Synchronize processes at the end of each epoch\n            try:\n                # Adding logs around the barrier\n                logger.debug(\n                    f\"Rank {rank}: About to enter barrier (end of epoch)\"\n                )\n                dist.barrier()\n                logger.debug(f\"Rank {rank}: Passed barrier (end of epoch).\")\n            except Exception as e:\n                logger.error(\n                    f\"Exception during end-of-epoch barrier: {e}\", exc_info=True\n                )\n                raise\n\n            # Resource cleanup\n            try:\n                if torch.cuda.is_available():\n                    torch.cuda.empty_cache()\n                    logger.debug(\"GPU cache emptied.\")\n                gc.collect()\n                logger.debug(\"Garbage collection completed.\")\n            except Exception as e:\n                logger.error(\n                    f\"Exception during resource cleanup: {e}\", exc_info=True\n                )\n\n        if writer:\n            writer.close()\n            logger.debug(\"TensorBoard writer closed.\")\n        logger.info(f\"Rank {rank}: Training completed.\")\n\n    except Exception as e:\n        logger.error(\n            f\"Training failed on Rank {rank} with exception: {e}\", exc_info=True\n        )\n        # Ensure all processes exit if one fails\n        if dist.is_initialized():\n            dist.destroy_process_group()\n            logger.debug(\"Process group destroyed due to training failure.\")\n        sys.exit(1)\n\ndef log_weight_norms(model, logger, epoch, writer=None):\n    \"\"\"\n    Logs the L2 norm of each parameter tensor in the model.\n\n    Args:\n        model: The model whose weights are to be logged.\n        logger: The logger instance for logging.\n        epoch: Current epoch number.\n        writer: TensorBoard SummaryWriter (optional).\n    \"\"\"\n    total_norm = 0.0\n    for name, param in model.named_parameters():\n        if param.requires_grad:\n            param_norm = param.data.norm(2).item()\n            total_norm += param_norm ** 2\n            logger.debug(f\"Epoch {epoch + 1}: Norm of {name}: {param_norm:.6f}\")\n            if writer:\n                writer.add_scalar(f\"Weights/{name}_norm\", param_norm, epoch)\n    total_norm = total_norm ** 0.5\n    logger.info(f\"Epoch {epoch + 1}: Total model L2 norm: {total_norm:.6f}\")\n    if writer:\n        writer.add_scalar(\"Weights/Total_L2_norm\", total_norm, epoch)\n\ndef log_weight_updates(model, previous_weights, logger, epoch, writer=None):\n    \"\"\"\n    Computes the difference between current weights and previous weights, logs the norm.\n\n    Args:\n        model: The current model.\n        previous_weights: The state dict of the previous model weights.\n        logger: The logger instance for logging.\n        epoch: Current epoch number.\n        writer: TensorBoard SummaryWriter (optional).\n    Returns:\n        The current weights state dict for use in the next epoch.\n    \"\"\"\n    weight_update_norm = 0.0\n    current_weights = {}\n    for name, param in model.named_parameters():\n        if param.requires_grad:\n            current_weights[name] = param.data.clone().detach()\n            if previous_weights is not None:\n                diff = param.data - previous_weights[name]\n                param_update_norm = diff.norm(2).item()\n                weight_update_norm += param_update_norm ** 2\n                logger.debug(f\"Epoch {epoch + 1}: Weight update norm of {name}: {param_update_norm:.6f}\")\n                if writer:\n                    writer.add_scalar(f\"WeightUpdates/{name}_update_norm\", param_update_norm, epoch)\n    weight_update_norm = weight_update_norm ** 0.5\n    if previous_weights is not None:\n        logger.info(f\"Epoch {epoch + 1}: Total weight update L2 norm: {weight_update_norm:.6f}\")\n        if writer:\n            writer.add_scalar(\"WeightUpdates/Total_update_L2_norm\", weight_update_norm, epoch)\n    return current_weights",
    "./training/metric_analysis.py": "# training/metric_analysis.py\n\nimport logging\n\ndef analyze_metrics(avg_metrics, eval_metrics, logger):\n    \"\"\"\n    Analyze metrics and provide insights.\n    \"\"\"\n    try:\n        # Analyze training loss\n        if 'actor_loss' in avg_metrics:\n            if avg_metrics['actor_loss'] > 1.0:\n                logger.info(\"Actor loss is high. Consider lowering the learning rate.\")\n            elif avg_metrics['actor_loss'] < 0.01:\n                logger.info(\"Actor loss is very low. Model might be converging.\")\n\n        if 'critic_loss' in avg_metrics:\n            if avg_metrics['critic_loss'] > 1.0:\n                logger.info(\"Critic loss is high. Monitor for potential overfitting.\")\n            elif avg_metrics['critic_loss'] < 0.01:\n                logger.info(\"Critic loss is very low. Model might be underfitting.\")\n\n        # Analyze entropy\n        if 'entropy' in avg_metrics:\n            if avg_metrics['entropy'] > 1.0:\n                logger.info(\"Entropy is high. Model is exploring.\")\n            elif avg_metrics['entropy'] < 0.1:\n                logger.info(\"Entropy is low. Model is exploiting.\")\n\n        # Analyze evaluation metrics\n        if eval_metrics.get('average_reward', 0) > 0:\n            logger.info(\"Model is making positive returns on evaluation.\")\n        else:\n            logger.info(\"Model is making negative returns on evaluation.\")\n\n        if 'sharpe_ratio' in eval_metrics:\n            logger.info(f\"Sharpe Ratio: {eval_metrics['sharpe_ratio']:.6f}\")\n        if 'max_drawdown' in eval_metrics:\n            logger.info(f\"Max Drawdown: {eval_metrics['max_drawdown']:.6f}\")\n        if 'average_win_rate' in eval_metrics:\n            logger.info(f\"Average Win Rate: {eval_metrics['average_win_rate']:.2%}\")\n    except Exception as e:\n        logger.error(f\"Exception during metric analysis: {e}\", exc_info=True)",
    "./training/collect_trajectories.py": "# File: training/collect_trajectories.py\n\nimport torch\nimport numpy as np\nimport logging\nfrom torch.cuda.amp import autocast\nfrom collections import Counter\n\ndef collect_trajectories(env, model, device, config, num_episodes=10, logger=None):\n    \"\"\"\n    Collect trajectories by interacting with the environment.\n    \"\"\"\n    if logger is None:\n        logger = logging.getLogger(\"CollectTrajectories\")\n\n    model.eval()\n    trajectories = {\n        'states': [],\n        'actions': [],\n        'rewards': [],\n        'dones': [],\n        'values': [],\n        'log_probs': [],\n        'next_close_prices': []\n    }\n    max_steps_per_episode = config[\"environment\"][\"max_steps\"]\n\n    try:\n        all_action_counts = Counter()\n        for episode in range(num_episodes):\n            logger.debug(f\"Collecting trajectory for Episode {episode + 1}\")\n            state = env.reset()\n            done = False\n            steps = 0\n            action_counts = Counter()\n\n            while not done and steps < max_steps_per_episode:\n                logger.debug(f\"Episode {episode + 1}, Step {steps + 1}: Collecting action.\")\n                state_tensor = torch.tensor(state, dtype=torch.float32).unsqueeze(0).to(device)\n                next_close_price = env.next_close_prices[env.current_step]  # Get the next close price\n\n                with torch.no_grad():\n                    if device.type == 'cuda':\n                        with autocast():\n                            logits, value, predicted_close = model(state_tensor)\n                    else:\n                        logits, value, predicted_close = model(state_tensor)\n\n                    if torch.isnan(logits).any() or torch.isinf(logits).any():\n                        logger.error(\"Logits contain NaNs or Infs during trajectory collection.\")\n                        raise ValueError(\"Invalid logits detected.\")\n\n                    action_dist = torch.distributions.Categorical(logits=logits)\n                    action = action_dist.sample()\n                    log_prob = action_dist.log_prob(action)\n                    value = value.item()\n\n                next_state, reward, done, _ = env.step(action.item())\n\n                # Logging actions\n                logger.debug(f\"Action taken: {action.item()}, Log Prob: {log_prob.item()}, Value: {value}\")\n\n                # Store experience\n                trajectories['states'].append(state)\n                trajectories['actions'].append(action.item())\n                trajectories['rewards'].append(reward)\n                trajectories['dones'].append(done)\n                trajectories['values'].append(value)\n                trajectories['log_probs'].append(log_prob.item())\n                trajectories['next_close_prices'].append(next_close_price)  # Add next close price\n\n                # Map action index to count\n                action_counts[action.item()] += 1\n\n                state = next_state\n                steps += 1\n                logger.debug(f\"Episode {episode + 1}, Step {steps}: State updated.\")\n\n            all_action_counts.update(action_counts)\n            logger.info(f\"Episode {episode + 1}: Action counts: {action_counts}\")\n\n    except Exception as e:\n        logger.error(f\"Exception during trajectory collection: {e}\", exc_info=True)\n        raise\n    finally:\n        model.train()\n\n    # Logging action distribution\n    logger.info(f\"Action distribution during data collection: {all_action_counts}\")\n\n    logger.info(f\"Collected {num_episodes} episodes with total steps: {len(trajectories['rewards'])}\")\n    return trajectories, all_action_counts",
    "./training/optimizer_setup.py": "# training/optimizer_setup.py\n\nimport torch\nimport logging\n\ndef setup_optimizers(model, config):\n    training_config = config[\"training\"]\n\n    optimizer_actor = torch.optim.AdamW(\n        model.module.actor_head.parameters(),\n        lr=training_config[\"actor_lr\"],\n        betas=tuple(training_config[\"optimizer\"].get(\"betas\", (0.9, 0.999))),\n        eps=float(training_config[\"optimizer\"].get(\"eps\", 1e-8)),\n        weight_decay=float(training_config[\"optimizer\"].get(\"weight_decay\", 1e-5))\n    )\n\n    optimizer_critic = torch.optim.AdamW(\n        model.module.critic_head.parameters(),\n        lr=training_config[\"critic_lr\"],\n        betas=tuple(training_config[\"optimizer\"].get(\"betas\", (0.9, 0.999))),\n        eps=float(training_config[\"optimizer\"].get(\"eps\", 1e-8)),\n        weight_decay=float(training_config[\"optimizer\"].get(\"weight_decay\", 1e-5))\n    )\n\n    optimizer_supervised = torch.optim.AdamW(\n        model.module.supervised_head.parameters(),\n        lr=training_config[\"supervised_lr\"],  # Ensure supervised_lr is included\n        betas=tuple(training_config[\"optimizer\"].get(\"betas\", (0.9, 0.999))),\n        eps=float(training_config[\"optimizer\"].get(\"eps\", 1e-8)),\n        weight_decay=float(training_config[\"optimizer\"].get(\"weight_decay\", 1e-5))\n    )\n\n    return optimizer_actor, optimizer_critic, optimizer_supervised\n\ndef setup_schedulers(optimizer_actor, optimizer_critic, optimizer_supervised, config, num_epochs):\n    scheduler_config = config[\"training\"].get(\"lr_scheduler\", {})\n    eta_min = float(scheduler_config.get(\"eta_min\", 1e-6))\n    t_max = int(scheduler_config.get(\"T_max\", num_epochs))\n\n    scheduler_actor = torch.optim.lr_scheduler.CosineAnnealingLR(\n        optimizer_actor, T_max=t_max, eta_min=eta_min\n    )\n    scheduler_critic = torch.optim.lr_scheduler.CosineAnnealingLR(\n        optimizer_critic, T_max=t_max, eta_min=eta_min\n    )\n    scheduler_supervised = torch.optim.lr_scheduler.CosineAnnealingLR(\n        optimizer_supervised, T_max=t_max, eta_min=eta_min\n    )\n\n    return scheduler_actor, scheduler_critic, scheduler_supervised\n\ndef update_entropy_coefficient(config, epoch, logger):\n    entropy_coeff = config[\"training\"][\"entropy_coeff\"]\n    entropy_schedule = config[\"training\"].get(\"entropy_schedule\", None)\n    if entropy_schedule and entropy_schedule.get(\"type\", \"\") == \"linear\":\n        start = entropy_schedule.get(\"start\", entropy_coeff)\n        end = entropy_schedule.get(\"end\", 0.0)\n        decay_epochs = entropy_schedule.get(\"decay_epochs\", config[\"training\"][\"epochs\"])\n\n        # Linear decay\n        entropy_coeff = max(\n            end,\n            start - ((start - end) / decay_epochs) * epoch\n        )\n        logger.debug(f\"Entropy coefficient updated to {entropy_coeff}\")\n\n    return entropy_coeff",
    "./training/utils.py": "# training/utils.py\n\nimport os\nimport torch\nimport logging\nimport numpy as np\nimport random\nimport gc\nimport psutil\n\ndef setup_logging(log_file=\"utils.log\"):\n    \"\"\"\n    Configure logging for utilities.\n    \"\"\"\n    logger = logging.getLogger(\"Utils\")\n    logger.setLevel(logging.DEBUG)\n\n    if not logger.hasHandlers():\n        os.makedirs(\"logs\", exist_ok=True)\n        # File handler\n        fh = logging.FileHandler(log_file)\n        fh.setLevel(logging.DEBUG)\n        # Console handler\n        ch = logging.StreamHandler()\n        ch.setLevel(logging.INFO)\n        # Formatter\n        formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n        fh.setFormatter(formatter)\n        ch.setFormatter(formatter)\n        # Add handlers to logger\n        logger.addHandler(fh)\n        logger.addHandler(ch)\n\n    return logger\n\ndef setup_logger(name, log_file=None):\n    \"\"\"\n    Configure logging.\n    \"\"\"\n    logger = logging.getLogger(name)\n    logger.setLevel(logging.DEBUG)\n    if not logger.hasHandlers():\n        os.makedirs(\"logs\", exist_ok=True)\n        fh = logging.FileHandler(f\"logs/{log_file}.log\" if log_file else f\"logs/{name}.log\")\n        fh.setLevel(logging.DEBUG)\n        ch = logging.StreamHandler()\n        ch.setLevel(logging.INFO)\n        formatter = logging.Formatter(\n            \"%(asctime)s - %(levelname)s - %(name)s - %(message)s\"\n        )\n        fh.setFormatter(formatter)\n        ch.setFormatter(formatter)\n        logger.addHandler(fh)\n        logger.addHandler(ch)\n    return logger\n\ndef setup_seed(seed):\n    \"\"\"\n    Set the random seed for reproducibility.\n    \"\"\"\n    torch.manual_seed(seed)\n    np.random.seed(seed)\n    random.seed(seed)\n    os.environ[\"PYTHONHASHSEED\"] = str(seed)\n    if torch.cuda.is_available():\n        torch.cuda.manual_seed_all(seed)\n        torch.backends.cudnn.deterministic = True\n        torch.backends.cudnn.benchmark = False\n    logger = setup_logger(\"Utils\")\n    logger.info(f\"Random seed set to {seed}\")\n\ndef load_checkpoint(checkpoint_path, model, optimizer_actor, optimizer_critic, device):\n    logger = setup_logger(\"Utils\")\n    logger.info(f\"Attempting to load checkpoint from {checkpoint_path}\")\n\n    if os.path.isfile(checkpoint_path):\n        try:\n            checkpoint = torch.load(checkpoint_path, map_location=device)\n            model.load_state_dict(checkpoint[\"model_state_dict\"])\n            logger.info(\"Model state dict loaded successfully.\")\n            # Verify that all parameters are loaded\n            for name, param in model.named_parameters():\n                if param.requires_grad:\n                    logger.debug(f\"Parameter {name} loaded with shape {param.shape}\")\n            if optimizer_actor and optimizer_critic:\n                optimizer_actor.load_state_dict(checkpoint[\"optimizer_actor_state_dict\"])\n                optimizer_critic.load_state_dict(checkpoint[\"optimizer_critic_state_dict\"])\n                logger.info(\"Optimizer state dicts loaded successfully.\")\n            epoch = checkpoint[\"epoch\"]\n            best_reward = checkpoint.get(\"best_reward\", float(\"-inf\"))\n            logger.info(f\"Checkpoint loaded successfully from {checkpoint_path}, resuming at epoch {epoch}.\")\n            return epoch, best_reward\n        except Exception as e:\n            logger.error(f\"Error loading checkpoint '{checkpoint_path}': {e}\", exc_info=True)\n            raise\n    else:\n        logger.warning(f\"No checkpoint found at {checkpoint_path}.\")\n        return 0, float(\"-inf\")\n\ndef save_checkpoint(checkpoint_path, model, optimizer_actor, optimizer_critic, epoch, best_reward):\n    \"\"\"\n    Saves a checkpoint of the model and optimizer states.\n    \"\"\"\n    logger = setup_logger(\"Utils\")\n    try:\n        os.makedirs(os.path.dirname(checkpoint_path), exist_ok=True)\n        torch.save(\n            {\n                \"model_state_dict\": model.state_dict(),\n                \"optimizer_actor_state_dict\": optimizer_actor.state_dict() if optimizer_actor else None,\n                \"optimizer_critic_state_dict\": optimizer_critic.state_dict() if optimizer_critic else None,\n                \"epoch\": epoch,\n                \"best_reward\": best_reward,\n            },\n            checkpoint_path,\n        )\n        logger.info(f\"Checkpoint saved successfully at {checkpoint_path}.\")\n    except Exception as e:\n        logger.error(f\"Failed to save checkpoint: {e}\", exc_info=True)\n\ndef compute_advantages_and_returns(rewards, values, dones, gamma, gae_lambda, reward_scale=1.0):\n    \"\"\"\n    Compute Generalized Advantage Estimation (GAE) and returns.\n    \"\"\"\n    rewards = rewards * reward_scale\n    advantages = torch.zeros_like(rewards)\n    returns = torch.zeros_like(rewards)\n    gae = 0\n    next_value = 0\n\n    for t in reversed(range(len(rewards))):\n        mask = 1.0 - dones[t]\n        delta = rewards[t] + gamma * next_value * mask - values[t]\n        gae = delta + gamma * gae_lambda * mask * gae\n        advantages[t] = gae\n        next_value = values[t]\n\n    returns = advantages + values\n    return advantages, returns\n\ndef log_memory_usage(logger, phase=\"\"):\n    \"\"\"\n    Logs the current CPU and memory usage.\n    \"\"\"\n    try:\n        process = psutil.Process(os.getpid())\n        mem_info = process.memory_info()\n        mem_usage_mb = mem_info.rss / (1024 ** 2)  # Convert bytes to MB\n        cpu_percent = psutil.cpu_percent(interval=1)\n        logger.info(f\"Resource Usage [{phase}]: CPU: {cpu_percent}% | Memory: {mem_usage_mb:.2f} MB\")\n    except Exception as e:\n        logger.error(f\"Failed to log memory usage: {e}\", exc_info=True)\n\ndef log_gpu_memory(logger, phase=\"\"):\n    \"\"\"\n    Logs the current GPU memory usage.\n    \"\"\"\n    try:\n        if torch.cuda.is_available():\n            allocated = torch.cuda.memory_allocated() / (1024 ** 2)  # in MB\n            reserved = torch.cuda.memory_reserved() / (1024 ** 2)     # in MB\n            logger.info(f\"GPU Memory Usage [{phase}]: Allocated: {allocated:.2f} MB | Reserved: {reserved:.2f} MB\")\n        else:\n            logger.info(f\"GPU Memory Usage [{phase}]: GPU not available.\")\n    except Exception as e:\n        logger.error(f\"Failed to log GPU memory usage: {e}\", exc_info=True)",
    "./training/__init__.py": "",
    "./training/config_validation.py": "# training/config_validation.py\n\nfrom pydantic import BaseModel, ValidationError\nfrom typing import List, Dict, Optional\nimport logging\nfrom training.utils import setup_logger\n\n# Define Nested Pydantic Models\nclass DataConfig(BaseModel):\n    raw_data_path: str\n    feature_columns: List[str]\n    n_steps: int\n    reward_scale: float\n    processed_data_path: str = \"data/processed/processed_data.csv\"\n    normalize_features: bool = True\n    pt_data_dir: str\n    num_splits: int\n    normalization: Optional[Dict[str, str]] = None\n    window_stride: int = 1\n    max_windows_per_file: int = 10000\n\nclass ModelConfig(BaseModel):\n    feature_dim: int\n    ppo_hidden_dim: int\n    action_dim: int\n    ppo_n_heads: int\n    ppo_n_layers: int\n    ppo_dropout: float\n    temperature: float = 1.0\n    layer_norm: bool = False\n\nclass TrainingConfig(BaseModel):\n    epochs: int\n    batch_size: int\n    actor_lr: float\n    critic_lr: float\n    supervised_lr: float  # Added supervised_lr\n    ppo_epochs: int\n    clip_epsilon: float\n    entropy_coeff: float\n    entropy_schedule: Optional[Dict] = None\n    critic_coeff: float\n    gamma: float\n    gae_lambda: float\n    weight_decay: float = 0.0\n    max_grad_norm: float = 0.5\n    lr_scheduler: Optional[Dict] = None\n    early_stopping: Optional[Dict] = None\n    optimizer: Optional[Dict] = None\n    supervised_loss_weight: float = 1.0  # Added supervised_loss_weight\n\nclass CheckpointConfig(BaseModel):\n    best_model_path: str\n    resume_checkpoint: str = \"\"\n    save_best_only: bool = False\n\nclass EvaluationConfig(BaseModel):\n    interval_eval: int\n    episodes: int\n    baseline_reward: float\n    gamma: float = 0.99\n    additional_metrics: List[str] = []\n\nclass LoggingConfig(BaseModel):\n    log_dir: str\n    use_tensorboard: bool\n\nclass DistributedConfig(BaseModel):\n    master_addr: str\n    master_port: str\n    world_size: int = 1\n\nclass PreprocessingConfig(BaseModel):\n    enable: bool\n    normalization_type: Optional[str] = None\n    handle_missing: Optional[str] = None\n    remove_outliers: bool = False\n    feature_engineering: List[str] = []\n\nclass EnvironmentConfig(BaseModel):\n    max_steps: int\n    transaction_fee: float = 0.001\n\nclass ReproducibilityConfig(BaseModel):\n    deterministic: bool = True\n    benchmark: bool = False\n\nclass ConfigModel(BaseModel):\n    random_seed: int\n    data: DataConfig\n    model: ModelConfig\n    training: TrainingConfig\n    checkpoint: CheckpointConfig\n    evaluation: EvaluationConfig\n    logging: LoggingConfig\n    distributed: DistributedConfig\n    preprocessing: PreprocessingConfig\n    environment: EnvironmentConfig\n    reproducibility: ReproducibilityConfig\n\ndef validate_and_cast_config(config):\n    \"\"\"\n    Validate and cast the configuration dictionary to ensure all required keys\n    are present and of correct types.\n    \"\"\"\n    logger = setup_logger(\"ConfigValidation\")\n    logger.info(\"Validating configuration parameters.\")\n\n    try:\n        validated_config = ConfigModel(**config)\n        logger.info(\"Configuration validated successfully.\")\n        return validated_config.dict()\n    except ValidationError as e:\n        logger.error(f\"Configuration validation failed: {e}\", exc_info=True)\n        raise",
    "./training/evaluation.py": "# File: training/evaluation.py\n\nimport torch\nimport numpy as np\nimport logging\nfrom collections import Counter\nfrom training.utils import log_memory_usage, log_gpu_memory\nimport gc\nimport os\nimport glob\n\ndef setup_logging(log_dir=\"logs\", log_file=\"evaluation.log\"):\n    \"\"\"\n    Configure logging for the evaluation script.\n    \"\"\"\n    os.makedirs(log_dir, exist_ok=True)\n    logger = logging.getLogger(\"Evaluation\")\n    logger.setLevel(logging.DEBUG)\n\n    # Create console handler\n    ch = logging.StreamHandler()\n    ch.setLevel(logging.INFO)\n\n    # Create file handler\n    fh = logging.FileHandler(os.path.join(log_dir, log_file))\n    fh.setLevel(logging.DEBUG)\n\n    # Create formatter and add to handlers\n    formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n    ch.setFormatter(formatter)\n    fh.setFormatter(formatter)\n\n    # Add handlers to the logger if none exist\n    if not logger.hasHandlers():\n        logger.addHandler(ch)\n        logger.addHandler(fh)\n\n    return logger\n\ndef evaluate_model(model, environment, config, device):\n    \"\"\"\n    Evaluate the model with comprehensive financial metrics.\n\n    Args:\n        model: The full model, including shared_encoder and actor_head.\n        environment: Trading environment.\n        config: Full configuration dictionary.\n        device: Computation device (e.g., 'cpu' or 'cuda').\n\n    Returns:\n        Dictionary containing evaluation metrics.\n    \"\"\"\n    logger = logging.getLogger(\"Evaluation\")\n    if not logger.hasHandlers():\n        logger = setup_logging()\n\n    logger.info(\"Starting model evaluation.\")\n    log_memory_usage(logger, phase=\"Before Evaluation\")\n    log_gpu_memory(logger, phase=\"Before Evaluation\")\n\n    model.eval()  # Set model to evaluation mode\n    total_rewards = []\n    sharpe_ratios = []\n    max_drawdowns = []\n    trade_counts = []\n    win_rates = []\n    action_entropies = []\n\n    num_episodes = config[\"evaluation\"][\"episodes\"]\n    gamma = config[\"training\"].get(\"gamma\", 0.99)  # Discount factor used in returns\n    reward_scale = config['data'].get('reward_scale', 1.0)\n\n    # Ensure 'close' is in feature columns\n    if 'close' not in config['data']['feature_columns']:\n        logger.error(\"'close' column is missing from feature_columns.\")\n        raise ValueError(\"'close' column is required in feature_columns for evaluation.\")\n\n    # Define action labels matching the environment's action mappings\n    action_labels = {\n        0: 'Sell 100%',\n        1: 'Sell 50%',\n        2: 'Sell 25%',\n        3: 'Hold',\n        4: 'Buy 25%',\n        5: 'Buy 50%',\n        6: 'Buy 100%',\n    }\n\n    successful_episodes = 0\n\n    for episode in range(num_episodes):\n        try:\n            logger.debug(f\"Starting Evaluation Episode {episode + 1}\")\n            state = environment.reset()\n            episode_reward = 0.0\n            cumulative_returns = 0.0\n            peak = 0.0\n            done = False\n            steps = 0\n            max_steps_per_episode = config[\"environment\"][\"max_steps\"]\n            action_counts = Counter()\n            episode_returns = []\n            returns = []\n\n            while not done and steps < max_steps_per_episode:\n                logger.debug(f\"Evaluation Episode {episode + 1}, Step {steps + 1}: Selecting action.\")\n                state_tensor = torch.tensor(state, dtype=torch.float32).unsqueeze(0).to(device)\n                with torch.no_grad():\n                    # Use the full model for forward pass\n                    logits, _, _ = model(state_tensor)\n                    action_dist = torch.distributions.Categorical(logits=logits)\n                    action = action_dist.sample().item()\n\n                logger.debug(f\"Evaluation Episode {episode + 1}, Step {steps + 1}: Action selected: {action}\")\n\n                step_result = environment.step(action)\n                next_state, reward, done, info = step_result\n\n                scaled_reward = reward * reward_scale\n                cumulative_returns += scaled_reward\n                episode_reward += scaled_reward\n                peak = max(peak, cumulative_returns)\n                episode_returns.append(scaled_reward)\n\n                # Discounted return\n                G = scaled_reward * (gamma ** steps)\n                returns.append(G)\n\n                # Map action index to label\n                action_label = action_labels.get(action, \"Unknown\")\n                action_counts[action_label] += 1\n\n                logger.debug(f\"Evaluation Episode {episode + 1}, Step {steps + 1}: Reward: {reward}, Scaled Reward: {scaled_reward}\")\n                steps += 1\n                state = next_state\n\n            total_rewards.append(episode_reward)\n            successful_episodes += 1\n\n            # Compute Sharpe ratio\n            avg_return = np.mean(episode_returns) if len(episode_returns) > 0 else 0\n            std_dev_return = np.std(episode_returns) + 1e-8\n            # Annualized Sharpe ratio (assuming 252 trading days)\n            sharpe_ratio = (avg_return / std_dev_return) * np.sqrt(252)\n            sharpe_ratios.append(sharpe_ratio)\n\n            # Compute max drawdown\n            cumulative_returns_array = np.cumsum(episode_returns)\n            peak_array = np.maximum.accumulate(cumulative_returns_array)\n            drawdowns = (peak_array - cumulative_returns_array)\n            max_drawdown = np.max(drawdowns) if len(drawdowns) > 0 else 0\n            max_drawdowns.append(max_drawdown)\n\n            # Compute win rate\n            wins = sum(1 for r in episode_returns if r > 0)\n            total_trades = len(episode_returns)\n            win_rate = wins / total_trades if total_trades > 0 else 0\n            win_rates.append(win_rate)\n\n            trade_counts.append(total_trades)\n            logger.info(\n                f\"Evaluation Episode {episode + 1}: Reward={episode_reward:.6f}, Trades={total_trades}, \"\n                f\"Sharpe Ratio={sharpe_ratio:.6f}, Max Drawdown={max_drawdown:.6f}, Win Rate={win_rate:.2%}\"\n            )\n            logger.info(f\"Evaluation Episode {episode + 1}: Action counts: {action_counts}\")\n\n            # Compute action entropy\n            total_actions = sum(action_counts.values())\n            action_probs = np.array(list(action_counts.values())) / total_actions\n            action_entropy = -np.sum(action_probs * np.log(action_probs + 1e-8))\n            logger.info(f\"Evaluation Episode {episode + 1}: Action Entropy: {action_entropy:.6f}\")\n            action_entropies.append(action_entropy)\n\n            # Resource cleanup\n            del state_tensor\n            if torch.cuda.is_available():\n                torch.cuda.empty_cache()\n            gc.collect()\n\n        except Exception as e:\n            logger.error(f\"Exception during episode {episode + 1} evaluation: {e}\", exc_info=True)\n            continue  # Proceed to next episode\n\n    # Aggregate metrics\n    try:\n        if successful_episodes == 0:\n            logger.warning(\"No successful evaluation episodes completed. Metrics cannot be calculated.\")\n            metrics = {}\n        else:\n            avg_reward = np.mean(total_rewards) if total_rewards else 0\n            avg_sharpe_ratio = np.mean(sharpe_ratios) if sharpe_ratios else 0\n            avg_max_drawdown = np.mean(max_drawdowns) if max_drawdowns else 0\n            avg_trade_count = np.mean(trade_counts) if trade_counts else 0\n            avg_win_rate = np.mean(win_rates) if win_rates else 0\n            avg_action_entropy = np.mean(action_entropies) if action_entropies else 0\n\n            metrics = {\n                \"average_reward\": avg_reward,\n                \"sharpe_ratio\": avg_sharpe_ratio,\n                \"max_drawdown\": avg_max_drawdown,\n                \"average_trade_count\": avg_trade_count,\n                \"average_win_rate\": avg_win_rate,\n                \"average_action_entropy\": avg_action_entropy\n            }\n\n            logger.info(f\"Evaluation Metrics: {metrics}\")\n\n        log_memory_usage(logger, phase=\"After Evaluation\")\n        log_gpu_memory(logger, phase=\"After Evaluation\")\n        return metrics\n\n    except Exception as e:\n        logger.error(f\"Exception during metric aggregation: {e}\", exc_info=True)\n        raise",
    "./config/config.yaml": "# File: config/config.yaml\n\n# =====================\n# Random Seed for Reproducibility\n# =====================\nrandom_seed: 42\n\n# =====================\n# Logging Configuration\n# =====================\nlogging:\n  log_dir: \"./logs\"\n  use_tensorboard: true\n\n# =====================\n# Distributed Training Configuration\n# =====================\ndistributed:\n  master_addr: \"localhost\"\n  master_port: \"29500\"\n  world_size: 2\n\n# =====================\n# Data Configuration\n# =====================\ndata:\n  raw_data_path: \"./data/dataset.csv\"\n  feature_columns:\n    - open\n    - high\n    - low\n    - close\n    - volume\n    - elapsed_time\n    - rsi\n    - adx\n    - roc\n    - stochastic_k\n    - stochastic_d\n    - macd\n    - macd_signal\n    - macd_hist\n    - mfi\n    - bb_upper\n    - bb_middle\n    - bb_lower\n    - kc_upper\n    - kc_lower\n    - dc_upper\n    - dc_lower\n    - atr\n    - pivot\n    - fibonacci_23\n    - fibonacci_38\n    - fibonacci_50\n    - fibonacci_61\n    - vpt\n    - cmf\n    - obv\n    - supertrend\n  n_steps: 40\n  reward_scale: 1.0\n  pt_data_dir: \"./data/processed/\"\n  num_splits: 5\n  normalize_features: false\n  normalization:\n    open: \"standard\"\n    high: \"standard\"\n    low: \"standard\"\n    close: \"standard\"\n    volume: \"min-max\"\n  window_stride: 1\n\n# =====================\n# Model Configuration\n# =====================\nmodel:\n  feature_dim: 32\n  ppo_hidden_dim: 512\n  action_dim: 7\n  ppo_n_heads: 16\n  ppo_n_layers: 8\n  ppo_dropout: 0.1\n  temperature: 1.0\n  layer_norm: true\n\n# =====================\n# Training Configuration\n# =====================\ntraining:\n  epochs: 500\n  batch_size: 128\n  actor_lr: 1e-4\n  critic_lr: 1e-4\n  supervised_lr: 1e-4  # Added supervised_lr\n  ppo_epochs: 5\n  clip_epsilon: 0.2\n  entropy_coeff: 0.01\n  entropy_schedule:\n    type: linear\n    start: 0.01\n    end: 0.005  # Adjusted to keep entropy higher for longer\n    decay_epochs: 200  # Extended decay period\n  critic_coeff: 1.0\n  gamma: 0.99\n  gae_lambda: 0.95\n  weight_decay: 1e-5\n  max_grad_norm: 0.5\n  lr_scheduler:\n    type: cosine\n    eta_min: 1e-6\n    T_max: 500\n  early_stopping:\n    enabled: true\n    patience: 30\n    metric: \"average_reward\"\n    mode: \"max\"\n  optimizer:\n    type: AdamW\n    betas: [0.9, 0.999]\n    eps: 1e-8\n    weight_decay: 1e-5\n  supervised_loss_weight: 1.0  # Added supervised_loss_weight\n\n# =====================\n# Checkpointing Configuration\n# =====================\ncheckpoint:\n  best_model_path: \"./checkpoints/best_model.pth\"\n  resume_checkpoint: \"\"\n  save_best_only: true\n\n# =====================\n# Evaluation Configuration\n# =====================\nevaluation:\n  interval_eval: 5\n  episodes: 20\n  baseline_reward: 0.0\n  gamma: 0.99\n  additional_metrics:\n    - kl_divergence\n    - sharpe_ratio\n    - max_drawdown\n    - average_win_rate\n\n# =====================\n# Environment Configuration\n# =====================\nenvironment:\n  max_steps: 288\n  transaction_fee: 0.0005\n  time_penalty: 0.0        # Penalty per time step to encourage faster decision making\n  hold_reward: 0.01        # Small reward for holding position\n  action_penalty: 0.01     # Penalty for trading to prevent excessive trading\n\n# =====================\n# Preprocessing Configuration\n# =====================\npreprocessing:\n  enable: true\n  normalization_type: \"standard\"\n  handle_missing: \"impute_mean\"\n  remove_outliers: true\n  feature_engineering:\n    - create_moving_averages\n    - create_volatility_indicators\n    - create_momentum_indicators\n\n# =====================\n# Reproducibility Configuration\n# =====================\nreproducibility:\n  deterministic: true\n  benchmark: false",
    "./models/models.py": "# File: models/models.py\n\nimport os\nimport torch\nimport torch.nn as nn\nfrom models.transformer import TransformerEncoder\nimport logging\nimport torch.nn.init as init\n\ndef setup_models_logging(log_dir=\"logs\", log_file=\"models.log\", file_logging_enabled=False):\n    \"\"\"\n    Configure logging for the models module with an on/off switch for file logging.\n    \"\"\"\n    os.makedirs(log_dir, exist_ok=True)\n    logger = logging.getLogger(\"Models\")\n    logger.setLevel(logging.DEBUG)\n\n    # Create console handler\n    ch = logging.StreamHandler()\n    ch.setLevel(logging.INFO)\n\n    # Create formatter and add to handlers\n    formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n    ch.setFormatter(formatter)\n\n    # Add handlers to the logger if none exist\n    if not logger.hasHandlers():\n        logger.addHandler(ch)\n\n        # Add file handler based on the flag\n        if file_logging_enabled:\n            fh = logging.FileHandler(os.path.join(log_dir, log_file))\n            fh.setLevel(logging.DEBUG)\n            fh.setFormatter(formatter)\n            logger.addHandler(fh)\n        else:\n            # Add a NullHandler to effectively disable file logging\n            fh = logging.NullHandler()\n            logger.addHandler(fh)\n\n    return logger\n\n# Initialize logger for Models\ndef initialize_models_logging():\n    logger = logging.getLogger(\"Models\")\n    if not logger.hasHandlers():\n        logger = setup_models_logging()\n    return logger\n\nlogger = initialize_models_logging()\n\ndef initialize_weights(m):\n    \"\"\"\n    Initialize weights of the model.\n    Uses He (Kaiming) initialization for Linear layers assuming they are followed by LeakyReLU activations.\n    Uses default initialization for LayerNorm layers.\n    \"\"\"\n    if isinstance(m, nn.Linear):\n        # Initialize weights using Kaiming Uniform for layers followed by LeakyReLU\n        init.kaiming_uniform_(m.weight, nonlinearity='leaky_relu')\n        if m.bias is not None:\n            init.constant_(m.bias, 0)\n        logger.debug(f\"Initialized Linear layer with Kaiming Uniform.\")\n    elif isinstance(m, nn.LayerNorm):\n        # Initialize LayerNorm weights to 1 and biases to 0\n        init.constant_(m.weight, 1)\n        init.constant_(m.bias, 0)\n        logger.debug(f\"Initialized LayerNorm layer with constant weights and biases.\")\n\n    # Check for NaN/Inf in weights\n    if hasattr(m, 'weight') and (torch.isnan(m.weight).any() or torch.isinf(m.weight).any()):\n        logger.error(f\"NaN/Inf detected in weights of {type(m).__name__}\")\n        raise ValueError(f\"NaN/Inf detected in weights of {type(m).__name__}\")\n    # Check for NaN/Inf in biases\n    if hasattr(m, 'bias') and m.bias is not None and (torch.isnan(m.bias).any() or torch.isinf(m.bias).any()):\n        logger.error(f\"NaN/Inf detected in bias of {type(m).__name__}\")\n        raise ValueError(f\"NaN/Inf detected in bias of {type(m).__name__}\")\n\nclass PPOAgent(nn.Module):\n    def __init__(self, feature_dim, hidden_dim, action_dim, n_heads, n_layers,\n                 dropout=0.1, temperature=1.0, layer_norm=False):\n        super().__init__()\n        self.actor_encoder = TransformerEncoder(feature_dim, hidden_dim, n_heads, n_layers, dropout=dropout)\n        self.critic_encoder = TransformerEncoder(feature_dim, hidden_dim, n_heads, n_layers, dropout=dropout)\n        self.supervised_encoder = TransformerEncoder(feature_dim, hidden_dim, n_heads, n_layers, dropout=dropout)\n\n        self.actor_head = nn.Linear(hidden_dim, action_dim)\n        self.critic_head = nn.Linear(hidden_dim, 1)\n        self.supervised_head = nn.Linear(hidden_dim, 1)  # Predict next close price\n\n        self.temperature = temperature\n        self.layer_norm_flag = layer_norm\n        if self.layer_norm_flag:\n            self.actor_layer_norm = nn.LayerNorm(hidden_dim)\n            self.critic_layer_norm = nn.LayerNorm(hidden_dim)\n            self.supervised_layer_norm = nn.LayerNorm(hidden_dim)\n            logger.debug(\"PPOAgent: LayerNorm initialized for actor, critic, and supervised heads.\")\n\n        self.apply(initialize_weights)\n        logger.info(\"PPOAgent weights initialized.\")\n\n    def forward(self, x):\n        logger.debug(f\"PPOAgent forward pass: input shape {x.shape}\")\n\n        # Actor path\n        actor_x = self.actor_encoder(x)\n        actor_x = actor_x[:, -1]\n        if self.layer_norm_flag:\n            actor_x = self.actor_layer_norm(actor_x)\n        logits = self.actor_head(actor_x)\n        if self.temperature != 1.0:\n            logits = logits / self.temperature\n\n        # Critic path\n        critic_x = self.critic_encoder(x)\n        critic_x = critic_x[:, -1]\n        if self.layer_norm_flag:\n            critic_x = self.critic_layer_norm(critic_x)\n        value = self.critic_head(critic_x)\n\n        # Supervised path\n        supervised_x = self.supervised_encoder(x)\n        supervised_x = supervised_x[:, -1]\n        if self.layer_norm_flag:\n            supervised_x = self.supervised_layer_norm(supervised_x)\n        predicted_close = self.supervised_head(supervised_x)\n\n        return logits, value, predicted_close.squeeze(-1)",
    "./models/transformer.py": "# models/transformer.py\n\nimport os\nimport torch\nimport torch.nn as nn\nimport math\nimport logging\n\ndef setup_transformer_logging(log_dir=\"logs\", log_file=\"transformer.log\", file_logging_enabled=False):\n    \"\"\"\n    Configure logging for the transformer module with an on/off switch for file logging.\n\n    Args:\n        log_dir (str): Directory to store log files.\n        log_file (str): Log file name.\n        file_logging_enabled (bool): Flag to enable/disable file logging.\n    \"\"\"\n    os.makedirs(log_dir, exist_ok=True)\n    logger = logging.getLogger(\"Transformer\")\n    logger.setLevel(logging.DEBUG)\n\n    # Create console handler\n    ch = logging.StreamHandler()\n    ch.setLevel(logging.INFO)\n\n    # Create formatter and add to handlers\n    formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n    ch.setFormatter(formatter)\n\n    # Add handlers to the logger if none exist\n    if not logger.hasHandlers():\n        logger.addHandler(ch)\n\n        # Add file handler based on the flag\n        if file_logging_enabled:\n            fh = logging.FileHandler(os.path.join(log_dir, log_file))\n            fh.setLevel(logging.DEBUG)\n            fh.setFormatter(formatter)\n            logger.addHandler(fh)\n        else:\n            # Add a NullHandler to effectively disable file logging\n            fh = logging.NullHandler()\n            logger.addHandler(fh)\n\n    return logger\n\n# Initialize logger for Transformer\ndef initialize_transformer_logging():\n    logger = logging.getLogger(\"Transformer\")\n    if not logger.hasHandlers():\n        logger = setup_transformer_logging()\n    return logger\n\nlogger = initialize_transformer_logging()\n\nclass PositionalEncoding(nn.Module):\n    def __init__(self, d_model, max_len=5000):\n        super().__init__()\n        self.encoding = torch.zeros(max_len, d_model)\n        position = torch.arange(0, max_len).unsqueeze(1).float()\n        div_term = torch.exp(torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model))\n        self.encoding[:, 0::2] = torch.sin(position * div_term)\n        self.encoding[:, 1::2] = torch.cos(position * div_term)\n        self.encoding = self.encoding.unsqueeze(0)  # Shape: (1, max_len, d_model)\n        logger.debug(f\"PositionalEncoding initialized with d_model={d_model}, max_len={max_len}\")\n\n    def forward(self, x):\n        \"\"\"\n        Add positional encoding to input tensor.\n\n        Args:\n            x (Tensor): Input tensor of shape (batch_size, seq_len, d_model).\n\n        Returns:\n            Tensor: Tensor with positional encoding added.\n        \"\"\"\n        try:\n            device = x.device\n            seq_len = x.size(1)\n            logger.debug(f\"PositionalEncoding forward: input shape {x.shape}, seq_len={seq_len}\")\n            if seq_len > self.encoding.size(1):\n                logger.error(f\"Sequence length {seq_len} exceeds maximum length {self.encoding.size(1)}.\")\n                raise ValueError(f\"Sequence length {seq_len} exceeds maximum length {self.encoding.size(1)}.\")\n            pos_enc = self.encoding[:, :seq_len, :].to(device)\n            logger.debug(f\"PositionalEncoding forward: position encoding shape {pos_enc.shape}\")\n            return x + pos_enc\n        except Exception as e:\n            logger.error(f\"Exception in PositionalEncoding forward: {e}\", exc_info=True)\n            raise\n\nclass TransformerEncoder(nn.Module):\n    def __init__(self, input_dim, hidden_dim, n_heads, n_layers, dropout=0.1):\n        super().__init__()\n        self.embedding = nn.Linear(input_dim, hidden_dim)\n        logger.debug(f\"TransformerEncoder: Embedding layer initialized with input_dim={input_dim}, hidden_dim={hidden_dim}\")\n        self.pos_enc = PositionalEncoding(hidden_dim)\n        self.transformer_encoder = nn.TransformerEncoder(\n            nn.TransformerEncoderLayer(\n                d_model=hidden_dim,\n                nhead=n_heads,\n                dropout=dropout,\n                batch_first=True\n            ),\n            num_layers=n_layers\n        )\n        logger.debug(f\"TransformerEncoder: TransformerEncoder initialized with n_layers={n_layers}, n_heads={n_heads}, dropout={dropout}\")\n        self.dropout = nn.Dropout(dropout)\n\n    def forward(self, x):\n        \"\"\"\n        Forward pass for the Transformer encoder.\n\n        Args:\n            x (Tensor): Input tensor of shape (batch_size, seq_len, input_dim).\n\n        Returns:\n            Tensor: Output tensor of shape (batch_size, seq_len, hidden_dim).\n        \"\"\"\n        try:\n            logger.debug(f\"TransformerEncoder forward: input shape {x.shape}\")\n            x = self.embedding(x)\n            logger.debug(f\"TransformerEncoder forward: after embedding, shape {x.shape}\")\n            x = self.pos_enc(x)\n            logger.debug(f\"TransformerEncoder forward: after positional encoding, shape {x.shape}\")\n            x = self.dropout(x)\n            logger.debug(f\"TransformerEncoder forward: after dropout, shape {x.shape}\")\n            x = self.transformer_encoder(x)\n            logger.debug(f\"TransformerEncoder forward: after transformer encoder, shape {x.shape}\")\n            return x\n        except Exception as e:\n            logger.error(f\"Exception in TransformerEncoder forward: {e}\", exc_info=True)\n            raise",
    "./data_processing/__init__.py": "",
    "./data_processing/preprocess.py": "# File: data_processing/preprocess.py\n\nimport os\nimport logging\nimport pandas as pd\nimport numpy as np\nimport joblib\nimport torch\nfrom sklearn.preprocessing import MinMaxScaler\nfrom typing import Dict\n\ndef setup_logging(log_dir=\"logs\", log_file=\"preprocess.log\"):\n    \"\"\"\n    Configure logging for the preprocessing script.\n\n    Args:\n        log_dir (str): Directory to store log files.\n        log_file (str): Log file name.\n    \n    Returns:\n        logging.Logger: Configured logger instance.\n    \"\"\"\n    os.makedirs(log_dir, exist_ok=True)\n    logger = logging.getLogger(\"Preprocessing\")\n    logger.setLevel(logging.INFO)\n\n    # Create console handler\n    ch = logging.StreamHandler()\n    ch.setLevel(logging.INFO)\n\n    # Create file handler\n    fh = logging.FileHandler(os.path.join(log_dir, log_file))\n    fh.setLevel(logging.DEBUG)\n\n    # Create formatter and add to handlers\n    formatter = logging.Formatter(\"%(asctime)s - %(levelname)s - %(name)s - %(message)s\")\n    ch.setFormatter(formatter)\n    fh.setFormatter(formatter)\n\n    # Add handlers to the logger if none exist\n    if not logger.hasHandlers():\n        logger.addHandler(ch)\n        logger.addHandler(fh)\n\n    return logger\n\n# Initialize logger\nlogger = logging.getLogger(\"Preprocessing\")\nif not logger.hasHandlers():\n    logger = setup_logging()\n\ndef validate_feature_columns(df, feature_columns, logger):\n    \"\"\"\n    Validate that all specified feature columns exist in the dataset.\n    \"\"\"\n    missing_columns = [col for col in feature_columns if col not in df.columns]\n    if missing_columns:\n        logger.error(f\"Missing feature columns: {missing_columns}\")\n        raise KeyError(f\"Missing feature columns: {missing_columns}\")\n\ndef normalize_feature_columns(df: pd.DataFrame, feat_cols: list, logger: logging.Logger) -> pd.DataFrame:\n    \"\"\"\n    Normalize the feature columns using MinMaxScaler, including 'close'.\n    \"\"\"\n    validate_feature_columns(df, feat_cols, logger)\n    available_features = [col for col in feat_cols if col in df.columns]\n\n    if len(available_features) != len(feat_cols):\n        logger.warning(f\"Missing columns in dataset: {set(feat_cols) - set(available_features)}\")\n\n    scaler_path = os.path.join('scalers', 'minmax_scaler.joblib')\n    os.makedirs('scalers', exist_ok=True)\n\n    if os.path.exists(scaler_path):\n        logger.info(f\"Loading scaler from {scaler_path}\")\n        scaler = joblib.load(scaler_path)\n        if hasattr(scaler, 'feature_names_in_') and not np.array_equal(scaler.feature_names_in_, available_features):\n            logger.warning(\"Scaler feature names do not match features to normalize. Re-fitting the scaler.\")\n            scaler = MinMaxScaler()\n            scaler.fit(df[available_features])\n            joblib.dump(scaler, scaler_path)\n            logger.info(f\"Scaler re-fitted and saved to {scaler_path}\")\n    else:\n        logger.info(f\"No scaler found at {scaler_path}. Fitting a new scaler.\")\n        scaler = MinMaxScaler()\n        scaler.fit(df[available_features])\n        joblib.dump(scaler, scaler_path)\n        logger.info(f\"Scaler fitted and saved to {scaler_path}\")\n    \n    df[available_features] = scaler.transform(df[available_features])\n    return df\n\ndef create_sliding_windows(data, window_size, stride=1):\n    \"\"\"\n    Create sliding windows from data using numpy strides.\n    \"\"\"\n    num_samples = data.shape[0]\n    num_windows = (num_samples - window_size) // stride + 1\n    if num_windows <= 0:\n        return np.array([])  # Return empty array if not enough data\n    # Calculate shape and strides\n    shape = (num_windows, window_size, data.shape[1])\n    strides = (stride * data.strides[0], data.strides[0], data.strides[1])\n    return np.lib.stride_tricks.as_strided(data, shape=shape, strides=strides).copy()\n\ndef export_and_save_preprocessed_data(config: Dict, logger: logging.Logger) -> None:\n    \"\"\"\n    Preprocess the raw data and save it efficiently, handling overlapping windows without memory issues.\n    \"\"\"\n    raw_data_path = config['data']['raw_data_path']\n    pt_data_dir = config['data'].get('pt_data_dir', 'data/processed/')\n    pt_data_prefix = config.get('data', {}).get('pt_data_prefix', 'processed_data_part')\n    os.makedirs(pt_data_dir, exist_ok=True)\n\n    # Check if already preprocessed\n    existing_pt_files = [fname for fname in os.listdir(pt_data_dir) if fname.startswith(pt_data_prefix)]\n    if existing_pt_files:\n        logger.info(f\"Processed data files already exist in '{pt_data_dir}'. Skipping preprocessing.\")\n        return\n\n    try:\n        logger.info(f\"Reading raw data from: {raw_data_path}\")\n        if not os.path.exists(raw_data_path):\n            logger.error(f\"Raw data file not found at: {raw_data_path}\")\n            raise FileNotFoundError(f\"Raw data file not found at: {raw_data_path}\")\n        \n        df = pd.read_csv(raw_data_path)\n        logger.info(f\"Initial data shape: {df.shape}\")\n\n        # Ensure 'close' is in feature columns\n        feat_cols = config['data']['feature_columns']\n        validate_feature_columns(df, feat_cols, logger)\n\n        # Replace infinite values with NaN\n        df.replace([np.inf, -np.inf], np.nan, inplace=True)\n\n        # Drop rows with NaN or infinite values\n        initial_length = len(df)\n        df.dropna(inplace=True)\n        cleaned_length = len(df)\n        logger.info(f\"Removed {initial_length - cleaned_length} rows containing NaN or infinite values.\")\n\n        # Check if dataframe is empty after cleaning\n        if df.empty:\n            logger.error(\"Preprocessed dataframe is empty after removing NaNs and infinite values.\")\n            raise ValueError(\"No data left after preprocessing.\")\n\n        # Save unnormalized 'close' prices separately before normalization\n        unnormalized_close_prices = df['close'].values.copy()  # Copy before normalization\n\n        # Prepare next_close_prices for supervised learning and synthetic reward\n        next_close_prices = unnormalized_close_prices[1:]\n        unnormalized_close_prices = unnormalized_close_prices[:-1]  # Align with features\n\n        # Remove the last row in df to align with next_close_prices\n        df = df.iloc[:-1, :].reset_index(drop=True)\n\n        # Ensure lengths match\n        if len(df) != len(next_close_prices):\n            logger.error(\"Mismatch between data length and next_close_prices length.\")\n            raise ValueError(\"Mismatch between data length and next_close_prices length.\")\n\n        # Normalize features if required\n        normalize_features = config['data'].get('normalize_features', True)\n        if normalize_features:\n            logger.info(\"Normalizing feature columns, including 'close'.\")\n            df = normalize_feature_columns(df, feat_cols, logger)\n\n        # Parameters for windowing\n        window_stride = config['data'].get('window_stride', 1)  # Stride of 1 for full overlap\n        n_steps = config['data']['n_steps']\n        max_windows_per_file = config['data'].get('max_windows_per_file', 10000)\n\n        data_values = df[feat_cols].values  # shape (num_samples, num_features)\n        total_windows = (len(data_values) - n_steps) // window_stride + 1\n        if total_windows <= 0:\n            logger.error(\"Not enough data to create any windows with the specified n_steps and window_stride.\")\n            raise ValueError(\"Insufficient data for preprocessing.\")\n\n        logger.info(f\"Total windows: {total_windows}\")\n\n        # Create sliding windows for data and close prices\n        windows_data = create_sliding_windows(data_values, n_steps, stride=window_stride)  # shape (num_windows, n_steps, num_features)\n        last_indices = np.arange(n_steps - 1, n_steps - 1 + total_windows * window_stride, window_stride)\n        windows_close_prices = unnormalized_close_prices[last_indices]  # Shape: (num_windows,)\n        windows_next_close_prices = next_close_prices[last_indices]  # Shape: (num_windows,)\n\n        if windows_data.shape[0] != windows_close_prices.shape[0]:\n            logger.error(\"Mismatch between number of windows and number of close prices.\")\n            raise ValueError(\"Mismatch between number of windows and number of close prices.\")\n\n        num_files = (windows_data.shape[0] + max_windows_per_file -1) // max_windows_per_file\n        logger.info(f\"Number of files to be created: {num_files}\")\n\n        for file_idx in range(num_files):\n            start_idx = file_idx * max_windows_per_file\n            end_idx = min(start_idx + max_windows_per_file, windows_data.shape[0])\n            batch_window_data = windows_data[start_idx:end_idx]\n            batch_close_prices = windows_close_prices[start_idx:end_idx]\n            batch_next_close_prices = windows_next_close_prices[start_idx:end_idx]\n\n            data_tensor = torch.from_numpy(batch_window_data).float()\n            close_price_tensor = torch.from_numpy(batch_close_prices).float()\n            next_close_price_tensor = torch.from_numpy(batch_next_close_prices).float()\n            pt_file_path = os.path.join(pt_data_dir, f\"{pt_data_prefix}_{file_idx}.pt\")\n            torch.save({\n                'data': data_tensor,\n                'close_prices': close_price_tensor,\n                'next_close_prices': next_close_price_tensor\n            }, pt_file_path)\n            logger.info(f\"Saved file {file_idx} to {pt_file_path} with {data_tensor.shape[0]} windows.\")\n\n    except Exception as e:\n        logger.error(f\"Failed to preprocess data: {e}\", exc_info=True)\n        raise"
} we have made some change to : # File: training/train_loop.py

import os
import logging
import torch
import numpy as np
import torch.distributed as dist
from torch.utils.data import DataLoader, TensorDataset, DistributedSampler
import torch.nn.functional as F
import sys
import gc

from training.utils import (
    compute_advantages_and_returns,
    save_checkpoint,
    log_memory_usage,
    log_gpu_memory,
    setup_logger,
)
from monitoring import monitor_training, setup_tensorboard
from training.evaluation import evaluate_model
from training.collect_trajectories import collect_trajectories
from training.metric_analysis import analyze_metrics
from training.optimizer_setup import (
    setup_optimizers,
    setup_schedulers,
    update_entropy_coefficient,
)
from environment.trading_env import TradingEnv

def train_model(
    env,
    config,
    device,
    model,
    rank,
    world_size,
    start_epoch,
    num_epochs,
    best_reward=-float("inf"),
):
    """
    Main training loop for PPO with DDP.
    """
    logger = setup_logger(f"TrainLoopRank{rank}", "train_loop")

    try:
        logger.info(f"Starting training from epoch {start_epoch + 1} to {num_epochs}")
        logger.info(f"Rank {rank}: Using batch size {config['training']['batch_size']}")

        # Optimizer and Scheduler setup
        optimizer_actor, optimizer_critic, optimizer_supervised = setup_optimizers(model, config)
        scheduler_actor, scheduler_critic, scheduler_supervised = setup_schedulers(
            optimizer_actor, optimizer_critic, optimizer_supervised, config, num_epochs
        )

        max_grad_norm = config["training"].get("max_grad_norm", 0.5)
        batch_size = config["training"]["batch_size"]
        ppo_epochs = config["training"]["ppo_epochs"]
        supervised_loss_weight = config["training"].get("supervised_loss_weight", 1.0)
        accumulation_steps = config["training"].get("accumulation_steps", 1)  # Start with no accumulation

        # Setup TensorBoard writer (only on rank 0)
        writer = None
        if config["logging"]["use_tensorboard"] and rank == 0:
            writer = setup_tensorboard(rank, log_dir=config["logging"]["log_dir"])
            logger.debug("TensorBoard writer initialized.")

        # Early stopping parameters
        early_stopping = config["training"].get("early_stopping", {})
        early_stopping_enabled = early_stopping.get("enabled", False)
        patience = early_stopping.get("patience", 10)
        stopping_metric = early_stopping.get("metric", "average_reward")
        stopping_mode = early_stopping.get("mode", "max")
        no_improvement_epochs = 0
        best_metric_value = -float("inf") if stopping_mode == "max" else float("inf")

        for epoch in range(start_epoch, num_epochs):
            logger.info(f"Rank {rank}, Epoch {epoch + 1}/{num_epochs}")
            log_memory_usage(logger, phase=f"Epoch {epoch + 1} Start")
            log_gpu_memory(logger, phase=f"Epoch {epoch + 1} Start")

            # Update entropy coefficient based on schedule
            entropy_coeff = update_entropy_coefficient(config, epoch, logger)

            # Reset environment for new epoch
            logger.debug(f"Rank {rank}: Resetting environment for new epoch.")
            state = env.reset()

            # Collect trajectories
            trajectories, action_counts = collect_trajectories(
                env, model, device, config, num_episodes=10, logger=logger
            )
            logger.info(
                f"Rank {rank}: Collected {len(trajectories['rewards'])} steps."
            )
            logger.info(f"Rank {rank}: Action distribution during data collection: {action_counts}")
            logger.debug("After collect_trajectories")

            # Convert collected data to tensors
            try:
                states = torch.stack(
                    [torch.from_numpy(np.array(s)) for s in trajectories["states"]]
                ).float()
                actions = torch.tensor(
                    trajectories["actions"], dtype=torch.long
                )
                rewards = torch.tensor(
                    trajectories["rewards"], dtype=torch.float32
                )
                dones = torch.tensor(
                    trajectories["dones"], dtype=torch.float32
                )
                values = torch.tensor(
                    trajectories["values"], dtype=torch.float32
                )
                log_probs_old = torch.tensor(
                    trajectories["log_probs"], dtype=torch.float32
                )
                next_close_prices = torch.tensor(
                    trajectories["next_close_prices"], dtype=torch.float32
                )

                # Move tensors to device
                states = states.to(device)
                actions = actions.to(device)
                rewards = rewards.to(device)
                dones = dones.to(device)
                values = values.to(device)
                log_probs_old = log_probs_old.to(device)
                next_close_prices = next_close_prices.to(device)

                logger.debug("Converted collected trajectories to tensors.")
            except Exception as e:
                logger.error(
                    f"Exception during tensor conversion: {e}", exc_info=True
                )
                raise

            # Compute advantages and returns
            try:
                advantages, returns = compute_advantages_and_returns(
                    rewards=rewards,
                    values=values,
                    dones=dones,
                    gamma=config["training"]["gamma"],
                    gae_lambda=config["training"]["gae_lambda"],
                )
                # Normalize advantages
                advantages_mean = advantages.mean().item()
                advantages_std = advantages.std().item()
                advantages = (advantages - advantages_mean) / (
                    advantages_std + 1e-8
                )
                logger.debug(f"Computed advantages and returns. Advantages mean: {advantages_mean}, std: {advantages_std}")
            except Exception as e:
                logger.error(
                    f"Exception during advantage and return computation: {e}",
                    exc_info=True,
                )
                raise

            # Create a TensorDataset from the collected trajectories
            dataset = TensorDataset(
                states, actions, returns, advantages, log_probs_old, next_close_prices
            )

            # Create a DistributedSampler to distribute the data across processes
            sampler = DistributedSampler(
                dataset, num_replicas=world_size, rank=rank, shuffle=True
            )

            # Create a DataLoader to handle batching and shuffling
            dataloader = DataLoader(
                dataset, batch_size=batch_size, sampler=sampler
            )

            # Initialize accumulators for metrics
            total_actor_loss = 0.0
            total_critic_loss = 0.0
            total_supervised_loss = 0.0
            total_entropy = 0.0
            total_kl_divergence = 0.0
            num_updates = 0

            # Optimizers zero gradients before starting epoch
            optimizer_actor.zero_grad()
            optimizer_critic.zero_grad()
            optimizer_supervised.zero_grad()

            # PPO Update
            for ppo_epoch in range(ppo_epochs):
                logger.debug(f"Rank {rank}: Starting PPO Epoch {ppo_epoch + 1}")

                for batch_idx, (
                    batch_states,
                    batch_actions,
                    batch_returns,
                    batch_advantages,
                    batch_log_probs_old,
                    batch_next_close_prices,
                ) in enumerate(dataloader):
                    # Move batch to device if not already
                    batch_states = batch_states.to(device)
                    batch_actions = batch_actions.to(device)
                    batch_returns = batch_returns.to(device)
                    batch_advantages = batch_advantages.to(device)
                    batch_log_probs_old = batch_log_probs_old.to(device)
                    batch_next_close_prices = batch_next_close_prices.to(device)

                    # Forward pass
                    logits, values_pred, predicted_close = model(batch_states)
                    action_dist = torch.distributions.Categorical(logits=logits)
                    log_probs = action_dist.log_prob(batch_actions)
                    entropy = action_dist.entropy().mean()

                    # Calculate ratios
                    ratios = torch.exp(log_probs - batch_log_probs_old)

                    # Surrogate loss
                    surr1 = ratios * batch_advantages
                    surr2 = torch.clamp(
                        ratios,
                        1.0 - config["training"]["clip_epsilon"],
                        1.0 + config["training"]["clip_epsilon"],
                    ) * batch_advantages
                    actor_loss = (
                        -torch.min(surr1, surr2).mean()
                        - entropy_coeff * entropy
                    )

                    # Critic loss
                    critic_loss = F.mse_loss(
                        values_pred.squeeze(-1), batch_returns
                    )

                    # Supervised loss
                    supervised_loss = F.mse_loss(
                        predicted_close.squeeze(-1), batch_next_close_prices
                    )

                    # Total loss
                    total_loss = actor_loss + critic_loss + supervised_loss_weight * supervised_loss

                    # KL Divergence
                    with torch.no_grad():
                        old_action_dist = torch.distributions.Categorical(
                            logits=logits.detach()
                        )
                        kl_divergence = torch.distributions.kl.kl_divergence(
                            old_action_dist, action_dist
                        ).mean()

                    # Backward pass
                    total_loss.backward()

                    # Clip gradients
                    torch.nn.utils.clip_grad_norm_(
                        model.parameters(), max_grad_norm
                    )

                    # Optimizer steps
                    optimizer_actor.step()
                    optimizer_critic.step()
                    optimizer_supervised.step()

                    # Zero gradients
                    optimizer_actor.zero_grad()
                    optimizer_critic.zero_grad()
                    optimizer_supervised.zero_grad()

                    # Accumulate metrics
                    total_actor_loss += actor_loss.item()
                    total_critic_loss += critic_loss.item()
                    total_supervised_loss += supervised_loss.item()
                    total_entropy += entropy.item()
                    total_kl_divergence += kl_divergence.item()
                    num_updates += 1

                    # Logging
                    if (batch_idx + 1) % 10 == 0:
                        logger.info(
                            f"Rank {rank}, Epoch {epoch + 1}, PPO Epoch {ppo_epoch + 1}, Batch {batch_idx + 1}, "
                            f"Actor Loss: {actor_loss.item():.6f}, Critic Loss: {critic_loss.item():.6f}, "
                            f"Supervised Loss: {supervised_loss.item():.6f}, Entropy: {entropy.item():.6f}, KL Divergence: {kl_divergence.item():.6f}"
                        )

                # Update learning rate schedulers at the end of each PPO epoch
                scheduler_actor.step()
                scheduler_critic.step()
                scheduler_supervised.step()

            # Compute average metrics over the updates
            if num_updates > 0:
                avg_metrics = {
                    "actor_loss": total_actor_loss / num_updates,
                    "critic_loss": total_critic_loss / num_updates,
                    "supervised_loss": total_supervised_loss / num_updates,
                    "entropy": total_entropy / num_updates,
                    "kl_divergence": total_kl_divergence / num_updates,
                    "advantages_mean": advantages_mean,
                    "advantages_std": advantages_std,
                }
            else:
                avg_metrics = {
                    "actor_loss": 0.0,
                    "critic_loss": 0.0,
                    "supervised_loss": 0.0,
                    "entropy": 0.0,
                    "kl_divergence": 0.0,
                    "advantages_mean": advantages_mean,
                    "advantages_std": advantages_std,
                }

            # Reduce metrics across all processes
            try:
                avg_reward = rewards.mean()
                dist.all_reduce(avg_reward, op=dist.ReduceOp.SUM)
                avg_reward = avg_reward.item() / world_size
                logger.info(
                    f"Rank {rank}, Epoch {epoch + 1} - Average Reward: {avg_reward:.6f}"
                )
            except Exception as e:
                logger.error(
                    f"Exception during metric reduction: {e}", exc_info=True
                )
                raise

            # Only rank 0 performs evaluation and checkpointing
            if rank == 0:
                try:
                    # Ensure all processes reach this point but only rank 0 runs this block
                    dist.barrier()

                    # Evaluation
                    if (epoch + 1) % config["evaluation"]["interval_eval"] == 0:
                        eval_env = TradingEnv(
                            config
                        )  # Create separate environment for evaluation
                        logger.info("Rank 0: Starting model evaluation.")
                        eval_metrics = evaluate_model(
                            model.module, eval_env, config, device  # Pass full model
                        )
                        logger.info("Rank 0: Model evaluation completed.")
                    else:
                        eval_metrics = {}

                    # Monitoring
                    monitor_training(
                        epoch, rank, avg_metrics, eval_metrics, writer
                    )

                    # Analyze metrics and provide insights
                    analyze_metrics(avg_metrics, eval_metrics, logger)

                    # Checkpointing
                    current_metric_value = eval_metrics.get(
                        stopping_metric, None
                    )
                    improved = False
                    if current_metric_value is not None:
                        if (
                            stopping_mode == "max"
                            and current_metric_value > best_metric_value
                        ) or (
                            stopping_mode == "min"
                            and current_metric_value < best_metric_value
                        ):
                            best_metric_value = current_metric_value
                            no_improvement_epochs = 0
                            improved = True
                        else:
                            no_improvement_epochs += 1
                    else:
                        logger.warning(
                            f"Evaluation metric '{stopping_metric}' not found in eval_metrics."
                        )

                    if avg_reward > best_reward:
                        best_reward = avg_reward
                        checkpoint_path = config["checkpoint"]["best_model_path"]
                        save_checkpoint(
                            checkpoint_path,
                            model.module,
                            optimizer_actor,
                            optimizer_critic,
                            epoch,
                            best_reward,
                        )
                        logger.info(
                            f"New best model saved at {checkpoint_path} with reward {best_reward:.6f}"
                        )

                    # Early stopping check
                    if (
                        early_stopping_enabled
                        and no_improvement_epochs >= patience
                    ):
                        logger.info(
                            f"Early stopping triggered after {patience} epochs with no improvement."
                        )
                        break  # Exit training loop

                except Exception as e:
                    logger.error(
                        f"Exception during evaluation or checkpointing: {e}",
                        exc_info=True,
                    )
                    raise
            else:
                dist.barrier()

            # Log resource usage at the end of the epoch
            try:
                log_memory_usage(logger, phase=f"Epoch {epoch + 1} End")
                log_gpu_memory(logger, phase=f"Epoch {epoch + 1} End")
            except Exception as e:
                logger.error(
                    f"Exception during resource logging: {e}", exc_info=True
                )

            # Synchronize processes at the end of each epoch
            try:
                dist.barrier()
            except Exception as e:
                logger.error(
                    f"Exception during end-of-epoch barrier: {e}", exc_info=True
                )
                raise

            # Resource cleanup
            try:
                if torch.cuda.is_available():
                    torch.cuda.empty_cache()
                gc.collect()
            except Exception as e:
                logger.error(
                    f"Exception during resource cleanup: {e}", exc_info=True
                )

        if writer:
            writer.close()
            logger.debug("TensorBoard writer closed.")
        logger.info(f"Rank {rank}: Training completed.")

    except Exception as e:
        logger.error(
            f"Training failed on Rank {rank} with exception: {e}", exc_info=True
        )
        # Ensure all processes exit if one fails
        if dist.is_initialized():
            dist.destroy_process_group()
            logger.debug("Process group destroyed due to training failure.")
        sys.exit(1) but someone  some fonctionnalities was omit and the routine for L2 metrics is gone , the training epochs only batch 10 is processed , batch 1 to 9 are skipped somehow.. check all the routines,  also make sure the weight policy routine get its update process correctly