load these current scripts : {
    "./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 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    \n        if actor_loss is not None:\n            if actor_loss > 1.0:\n                logger.info(\"Actor loss is high. Consider lowering the learning rate or adjusting model complexity.\")\n            elif actor_loss < 0.01:\n                logger.info(\"Actor loss is very low. Model might be converging well.\")\n    \n        if critic_loss is not None:\n            if critic_loss > 1.0:\n                logger.info(\"Critic loss is high. Monitor for potential overfitting or adjust model.\")\n            elif critic_loss < 0.01:\n                logger.info(\"Critic loss is very low. Model might be underfitting.\")\n    \n        # Analyze entropy\n        if entropy is not None:\n            if entropy > 1.0:\n                logger.info(\"Entropy is high. Model is exploring more.\")\n            elif entropy < 0.1:\n                logger.info(\"Entropy is low. Model is exploiting more.\")\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(\"Model is achieving positive returns on evaluation.\")\n            else:\n                logger.info(\"Model is not achieving positive returns on evaluation.\")\n    \n            if sharpe_ratio > 1.0:\n                logger.info(\"Sharpe Ratio is good, indicating a favorable risk-return trade-off.\")\n            elif sharpe_ratio < 1.0 and sharpe_ratio > 0:\n                logger.info(\"Sharpe Ratio is moderate, consider optimizing the strategy for better performance.\")\n            else:\n                logger.info(\"Sharpe Ratio is poor, consider re-evaluating the trading strategy.\")\n    \n            if max_drawdown < 0.1:\n                logger.info(\"Max Drawdown is low, indicating controlled risk.\")\n            else:\n                logger.info(\"Max Drawdown is high, consider implementing risk management strategies.\")\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    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": "# 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    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(\"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\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  # Position can now be fractional\n        self.total_profit = 0.0\n        self.entry_price = None\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%\n            1: ('sell', 0.5),   # Sell 50%\n            2: ('sell', 0.25),  # Sell 25%\n            3: ('hold', 0.0),   # Hold\n            4: ('buy', 0.25),   # Buy 25%\n            5: ('buy', 0.5),    # Buy 50%\n            6: ('buy', 1.0),    # Buy 100%\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:\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            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.total_profit = 0.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):\n                done = True\n                obs = np.zeros_like(self.observation_space.sample())\n                info = {'total_profit': self.total_profit}\n                logger.debug(f\"Reached end of data at step {self.current_step}. Ending episode.\")\n                return obs, 0.0, done, info\n\n            current_price = self.close_prices[idx]\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            # Execute action\n            if action_type == 'buy':\n                delta_position = amount\n                transaction_cost = fee_rate * delta_position * current_price\n                self.position += delta_position\n                self.total_profit -= transaction_cost\n                logger.debug(f\"Action: Buy {amount*100}%. New position: {self.position}\")\n            elif action_type == 'sell':\n                delta_position = -amount\n                transaction_cost = fee_rate * abs(delta_position) * current_price\n                self.position += delta_position\n                self.total_profit -= transaction_cost\n                logger.debug(f\"Action: Sell {abs(amount)*100}%. New position: {self.position}\")\n            # If 'hold', do nothing\n            else:\n                delta_position = 0.0\n                logger.debug(\"Action: Hold. No change in position.\")\n\n            # Get current observation\n            current_obs = self.data[idx]\n\n            # Calculate reward\n            reward = self._calculate_reward(action, current_obs, current_price, self.previous_price, transaction_cost, delta_position)\n\n            # Update total profit\n            self.total_profit += reward\n            logger.debug(f\"Total profit after step {self.current_step}: {self.total_profit}\")\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 = {'total_profit': self.total_profit}\n\n            logger.debug(f\"Step {self.current_step}: Done={done}\")\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 = 0.0\n            info = {'total_profit': self.total_profit, 'error': str(e)}\n            return obs, reward, done, info\n\n    def _calculate_reward(self, action, current_obs, current_price, previous_price, transaction_cost, delta_position):\n        \"\"\"\n        Simplified reward function handling all 7 actions with partial buy/sell amounts.\n        \"\"\"\n        action_type, amount = self.action_mappings[action]\n        reward = 0\n\n        # Calculate profit/loss based on position change and price movement\n        price_change = current_price - previous_price\n\n        # Calculate reward based on position and price change\n        reward = self.position * price_change - transaction_cost\n\n        # Overtrading Penalty\n        overtrading_penalty = 0.05 * transaction_cost  # 5% of transaction cost\n        reward -= overtrading_penalty\n\n        # Holding Penalty\n        if self.position != 0:\n            holding_penalty = -0.001  # Small negative reward per time step\n            reward += holding_penalty\n\n        # Apply reward scaling from config\n        reward_scale = self.config['data'].get('reward_scale', 1.0)\n        reward *= reward_scale\n\n        return reward\n\n    def render(self, mode='human'):\n        # Implement rendering if needed\n        pass",
    "./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": "# 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\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\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    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 = setup_optimizers(model, config)\n        scheduler_actor, scheduler_critic = setup_schedulers(\n            optimizer_actor, optimizer_critic, 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\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 = 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.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\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\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 = (advantages - advantages.mean()) / (\n                    advantages.std() + 1e-8\n                )\n                logger.debug(\"Computed advantages and returns.\")\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\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                    ) in enumerate(dataloader):\n                        # Move batch data to the correct device (already on device)\n\n                        optimizer_actor.zero_grad()\n                        optimizer_critic.zero_grad()\n\n                        if device.type == \"cuda\":\n                            with autocast():\n                                logits, values_pred = 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                            logger.debug(\"Forward pass with autocast completed.\")\n\n                            # Backward pass with AMP\n                            scaler.scale(actor_loss).backward()\n                            scaler.unscale_(optimizer_actor)\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.actor.parameters(), max_grad_norm\n                            )\n                            scaler.step(optimizer_actor)\n\n                            scaler.scale(critic_loss).backward()\n                            scaler.unscale_(optimizer_critic)\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.critic.parameters(), max_grad_norm\n                            )\n                            scaler.step(optimizer_critic)\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 = 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                            # Backward pass\n                            actor_loss.backward()\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.actor.parameters(), max_grad_norm\n                            )\n                            optimizer_actor.step()\n\n                            critic_loss.backward()\n                            torch.nn.utils.clip_grad_norm_(\n                                model.module.critic.parameters(), max_grad_norm\n                            )\n                            optimizer_critic.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\"Entropy: {entropy.item():.6f}\"\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:\n                scheduler_actor.step()\n                scheduler_critic.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]}\"\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.actor, eval_env, config, device\n                        )\n                        logger.info(\"Rank 0: Model evaluation completed.\")\n                    else:\n                        eval_metrics = {}\n\n                    # Monitoring\n                    avg_metrics = {\n                        \"actor_loss\": actor_loss.item(),\n                        \"critic_loss\": critic_loss.item(),\n                        \"entropy\": entropy.item(),\n                    }\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            # 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)",
    "./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": "# 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    }\n    max_steps_per_episode = config[\"environment\"][\"max_steps\"]\n\n    try:\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\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                with torch.no_grad():\n                    if device.type == 'cuda':\n                        with autocast():\n                            logits, value = model(state_tensor)\n                    else:\n                        logits, value = 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\n                state = next_state\n                steps += 1\n                logger.debug(f\"Episode {episode + 1}, Step {steps}: State updated.\")\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    action_counts = Counter(trajectories['actions'])\n    logger.info(f\"Action distribution during data collection: {action_counts}\")\n\n    logger.info(f\"Collected {num_episodes} episodes with total steps: {len(trajectories['rewards'])}\")\n    return trajectories",
    "./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    # Separate optimizers for actor and critic with potentially different learning rates\n    optimizer_actor = torch.optim.AdamW(\n        model.module.actor.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-4))\n    )\n\n    optimizer_critic = torch.optim.AdamW(\n        model.module.critic.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-4))\n    )\n    return optimizer_actor, optimizer_critic\n\ndef setup_schedulers(optimizer_actor, optimizer_critic, config, num_epochs):\n    scheduler_config = config[\"training\"].get(\"lr_scheduler\", {})\n    eta_min = float(scheduler_config.get(\"eta_min\", 1e-6))  # Explicitly cast to float\n    t_max = int(scheduler_config.get(\"T_max\", num_epochs))  # Explicitly cast to int\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\n    return scheduler_actor, scheduler_critic\n\ndef update_entropy_coefficient(config, epoch, logger):\n    \"\"\"\n    Update the entropy coefficient based on the schedule.\n    \"\"\"\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        # 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\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: Dict[str, str] = None\n\nclass ModelConfig(BaseModel):\n    action_dim: int\n    feature_dim: int\n    ppo_hidden_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    batch_size: int\n    epochs: int\n    actor_lr: float\n    critic_lr: float\n    clip_epsilon: float\n    gamma: float\n    gae_lambda: float\n    entropy_coeff: float\n    critic_coeff: float\n    ppo_epochs: int\n    weight_decay: float = 0.0\n    max_grad_norm: float = 0.5\n    entropy_schedule: Dict = None\n    lr_scheduler: Dict = None\n    early_stopping: Dict = None\n    optimizer: Dict = None\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: int\n    world_size: int = 1\n\nclass PreprocessingConfig(BaseModel):\n    enable: bool\n    normalization_type: str = None\n    handle_missing: 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": "# 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(actor, environment, config, device):\n    \"\"\"\n    Evaluate the model with comprehensive financial metrics.\n\n    Args:\n        actor: Actor network for policy evaluation.\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    actor.eval()  # Set actor to evaluation mode\n    total_rewards = []\n    sharpe_ratios = []\n    max_drawdowns = []\n    trade_counts = []\n    win_rates = []\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\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                    logits = actor(state_tensor)\n                    # Use stochastic policy for evaluation\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                # 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            max_drawdown = peak - cumulative_returns\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            # 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\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            }\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": "# config/config.yaml\n\n# =====================\n# Random Seed for Reproducibility\n# =====================\nrandom_seed: 42  # Keeping for consistency in experiments\n\n# =====================\n# Logging Configuration\n# =====================\nlogging:\n  log_dir: \"./logs\"  # No change needed\n  use_tensorboard: true  # TensorBoard is great for visualization\n\n# =====================\n# Distributed Training Configuration\n# =====================\ndistributed:\n  master_addr: \"localhost\"  # Standard for local multi-GPU setup\n  master_port: \"29500\"  # Arbitrary but should be unique\n  world_size: 2  # Adjusted based on your GPU availability\n\n# =====================\n# Data Configuration\n# =====================\ndata:\n  raw_data_path: \"./data/dataset.csv\"  # Path to your dataset\n  feature_columns:  # List of features from your dataset\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  # Good for capturing temporal dependencies\n  reward_scale: 1.0  # Might need adjustment if rewards are too small or large\n  pt_data_dir: \"./data/processed/\"  # For storing processed data\n  num_splits: 5  # Good for managing data distribution\n  normalize_features: true  # Turning on normalization as it's generally beneficial\n  normalization:\n    open: \"standard\"\n    high: \"standard\"\n    low: \"standard\"\n    close: \"standard\"\n    volume: \"min-max\"\n    # Add normalization for other features if needed\n  window_stride: 1  # For overlapping windows, seems appropriate\n\n# =====================\n# Model Configuration\n# =====================\nmodel:\n  feature_dim: 32  # Matches the number of features\n  ppo_hidden_dim: 512  # Increased to capture more complex patterns, monitor for overfitting\n  action_dim: 7  # Based on your action space\n  ppo_n_heads: 8  # Keeping for attention mechanism in transformer\n  ppo_n_layers: 4  # Increased to potentially learn more complex patterns, watch for overfitting\n  ppo_dropout: 0.2  # Reduced slightly, high dropout can be too aggressive\n  temperature: 1.0  # Reset to default, can adjust based on exploration vs. exploitation\n  layer_norm: true  # For stability in training\n\n# =====================\n# Training Configuration\n# =====================\ntraining:\n  epochs: 200  # Sufficient for convergence in many scenarios\n  batch_size: 64  # Reduced to increase generalization, might need tuning\n  actor_lr: 1e-4  # Reduced for more stable learning\n  critic_lr: 1e-4  # Matching actor's LR for balance\n  ppo_epochs: 4  # Reduced to avoid over-optimization\n  clip_epsilon: 0.2  # Slightly increased to allow for more significant updates\n  entropy_coeff: 0.2  # Increased to encourage exploration, can adjust if too much randomness\n  entropy_schedule:\n    type: linear\n    start: 0.2\n    end: 0.01\n    decay_epochs: 100  # Encourages more exploration at the start, less towards the end\n  critic_coeff: 1.0  # No change, standard value\n  gamma: 0.95  # Reduced for shorter-term focus\n  gae_lambda: 0.95  # Standard value for GAE\n  weight_decay: 1e-5  # For regularization, preventing overfitting\n  max_grad_norm: 0.5  # Reduced to prevent exploding gradients\n  lr_scheduler:\n    type: cosine\n    eta_min: 1e-6\n    T_max: 200  # Matches total epochs\n  early_stopping:\n    enabled: true\n    patience: 30  # Increased to give more time for potential late improvements\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  # Keeping for regularization\n\n# =====================\n# Checkpointing Configuration\n# =====================\ncheckpoint:\n  best_model_path: \"./checkpoints/best_model.pth\"  # Standard path\n  resume_checkpoint: \"\"  # Empty if starting fresh\n  save_best_only: true  # Saves resources, ensures best model is kept\n\n# =====================\n# Evaluation Configuration\n# =====================\nevaluation:\n  interval_eval: 5  # Frequency seems appropriate\n  episodes: 20  # Good number for statistical significance\n  baseline_reward: 0.0  # Baseline for comparison\n  gamma: 0.99  # High gamma for evaluation, might want to match training if different strategies\n  additional_metrics:\n    - precision\n    - recall\n    - f1_score\n    - custom_financial_metric_1\n\n# =====================\n# Environment Configuration\n# =====================\nenvironment:\n  max_steps: 250  # Suitable for short-term trading like scalping\n  transaction_fee: 0.0005  # Realistic for many trading environments\n\n# =====================\n# Preprocessing Configuration\n# =====================\npreprocessing:\n  enable: true  # Ensure preprocessing is used\n  normalization_type: \"standard\"  # Standard normalization often works well\n  handle_missing: \"impute_mean\"  # Good strategy for small amounts of missing data\n  remove_outliers: true  # Can help, but ensure not removing too much useful variation\n  feature_engineering:\n    - create_moving_averages\n    - create_volatility_indicators\n    - create_momentum_indicators  # These can add valuable predictive power\n\n# =====================\n# Reproducibility Configuration\n# =====================\nreproducibility:\n  deterministic: true  # For consistent results\n  benchmark: false  # For speed if needed, but might affect reproducibility",
    "./models/models.py": "# 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    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(\"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 Actor(nn.Module):\n    def __init__(self, feature_dim, hidden_dim, action_dim, n_heads, n_layers, temperature=1.0, layer_norm=False):\n        super().__init__()\n        self.encoder = TransformerEncoder(feature_dim, hidden_dim, n_heads, n_layers, dropout=0.1)\n        self.layer_norm_flag = layer_norm\n        if self.layer_norm_flag:\n            self.layer_norm = nn.LayerNorm(hidden_dim)\n            logger.debug(\"Actor: LayerNorm initialized.\")\n\n        self.fc = nn.Linear(hidden_dim, action_dim)  # Output logits for actions\n        self.temperature = temperature\n        logger.debug(f\"Actor initialized with feature_dim={feature_dim}, hidden_dim={hidden_dim}, action_dim={action_dim}\")\n\n    def forward(self, x):\n        logger.debug(f\"Actor forward pass: input shape {x.shape}\")\n        x = self.encoder(x)\n        logger.debug(f\"Actor forward pass: after encoder, shape {x.shape}\")\n        x = x[:, -1]  # Get the output from the last time step\n        logger.debug(f\"Actor forward pass: after slicing, shape {x.shape}\")\n        if self.layer_norm_flag:\n            x = self.layer_norm(x)\n            logger.debug(\"Actor forward pass: LayerNorm applied.\")\n        logits = self.fc(x)\n        logger.debug(f\"Actor forward pass: logits shape {logits.shape}\")\n        if torch.isnan(logits).any():\n            logger.error(\"Logits contain NaNs in Actor forward pass\")\n            raise ValueError(\"Logits contain NaNs in Actor\")\n        if self.temperature != 1.0:\n            logits = logits / self.temperature  # Apply temperature scaling\n            logger.debug(f\"Actor forward pass: Temperature scaling applied with temperature={self.temperature}\")\n\n        return logits\n\nclass Critic(nn.Module):\n    def __init__(self, feature_dim, hidden_dim, n_heads, n_layers, layer_norm=False):\n        super().__init__()\n        self.encoder = TransformerEncoder(feature_dim, hidden_dim, n_heads, n_layers, dropout=0.1)\n        self.layer_norm_flag = layer_norm\n        if self.layer_norm_flag:\n            self.layer_norm = nn.LayerNorm(hidden_dim)\n            logger.debug(\"Critic: LayerNorm initialized.\")\n\n        self.fc = nn.Sequential(\n            nn.Linear(hidden_dim, hidden_dim),\n            nn.LeakyReLU(negative_slope=0.01),\n            nn.Linear(hidden_dim, 1)\n        )\n        logger.debug(f\"Critic initialized with feature_dim={feature_dim}, hidden_dim={hidden_dim}\")\n\n    def forward(self, x):\n        logger.debug(f\"Critic forward pass: input shape {x.shape}\")\n        x = self.encoder(x)\n        logger.debug(f\"Critic forward pass: after encoder, shape {x.shape}\")\n        x = x[:, -1]  # Get the output from the last time step\n        logger.debug(f\"Critic forward pass: after slicing, shape {x.shape}\")\n        if self.layer_norm_flag:\n            x = self.layer_norm(x)\n            logger.debug(\"Critic forward pass: LayerNorm applied.\")\n        values = self.fc(x)\n        logger.debug(f\"Critic forward pass: values shape {values.shape}\")\n        if torch.isnan(values).any():\n            logger.error(\"Values contain NaNs in Critic forward pass\")\n            raise ValueError(\"Values contain NaNs in Critic\")\n        return values\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 = Actor(\n            feature_dim=feature_dim,\n            hidden_dim=hidden_dim,\n            action_dim=action_dim,\n            n_heads=n_heads,\n            n_layers=n_layers,\n            temperature=temperature,\n            layer_norm=layer_norm\n        )\n        self.critic = Critic(\n            feature_dim=feature_dim,\n            hidden_dim=hidden_dim,\n            n_heads=n_heads,\n            n_layers=n_layers,\n            layer_norm=layer_norm\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        logits = self.actor(x)\n        value = self.critic(x)\n        logger.debug(f\"PPOAgent forward pass: logits shape {logits.shape}, value shape {value.shape}\")\n        return logits, value",
    "./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    Args:\n        df (pd.DataFrame): The dataset.\n        feature_columns (list): List of expected feature columns.\n        logger (logging.Logger): Logger instance for logging.\n\n    Raises:\n        KeyError: If any feature column is missing.\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    Args:\n        df (pd.DataFrame): DataFrame containing the data to normalize.\n        feat_cols (list): List of feature column names to normalize.\n        logger (logging.Logger): Logger instance for logging messages.\n\n    Returns:\n        pd.DataFrame: DataFrame with normalized feature columns.\n\n    Raises:\n        Exception: If scaler loading or fitting fails.\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    Parameters:\n    data (numpy array): Input data of shape (num_samples, num_features)\n    window_size (int): Size of each window\n    stride (int): Stride between windows\n    \n    Returns:\n    numpy array: Output data of shape (num_windows, window_size, num_features)\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    Args:\n        config (Dict): Configuration dictionary containing data paths and parameters.\n        logger (logging.Logger): Logger instance for logging messages.\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        unnormalized_close_path = os.path.join(pt_data_dir, 'unnormalized_close_prices.npy')\n        logger.info(f\"Saving unnormalized 'close' prices to {unnormalized_close_path}\")\n        np.save(unnormalized_close_path, unnormalized_close_prices)\n        logger.info(f\"Unnormalized 'close' prices saved to {unnormalized_close_path}\")\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\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\n            data_tensor = torch.from_numpy(batch_window_data).float()\n            close_price_tensor = torch.from_numpy(batch_close_prices).float()\n            pt_file_path = os.path.join(pt_data_dir, f\"{pt_data_prefix}_{file_idx}.pt\")\n            torch.save({'data': data_tensor, 'close_prices': close_price_tensor}, 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"
}  could you implement these update metrics : To refine your training script to include additional metrics and expand their interpretation, you can follow these steps:

1. Define New Metrics
You can include additional metrics such as:

KL Divergence: Measures the divergence between the old and new policy.
Gradient Norm: Helps monitor gradient explosion or vanishing.
Evaluation-Specific Metrics: Such as Sharpe Ratio, Max Drawdown, Average Win Rate, etc.
2. Update monitor_training Function
Expand the monitor_training function in monitoring.py to log additional metrics like KL divergence and evaluation-specific metrics. Example:

python
Copy code
def monitor_training(
    epoch: int,
    rank: int,
    avg_metrics: dict,
    eval_metrics: dict,
    writer: SummaryWriter = None
) -> None:
    logs_dir = "logs"
    os.makedirs(logs_dir, exist_ok=True)

    # Log metrics in console logs
    logger.info(f"Epoch {epoch + 1} - Training Metrics: {avg_metrics}")
    logger.info(f"Epoch {epoch + 1} - Evaluation Metrics: {eval_metrics}")

    if writer:
        # Write training metrics
        for k, v in avg_metrics.items():
            writer.add_scalar(f"Train/{k}", v, epoch)
        # Write evaluation metrics
        for k, v in eval_metrics.items():
            writer.add_scalar(f"Eval/{k}", v, epoch)
        writer.flush()

    # Interpret metrics
    interpret_metrics(avg_metrics, eval_metrics, logger)
3. Compute and Log KL Divergence
Add KL divergence computation in train_loop.py:

python
Copy code
kl_divergence = torch.distributions.kl.kl_divergence(action_dist, old_action_dist).mean().item()
avg_metrics["kl_divergence"] = kl_divergence
logger.info(f"Epoch {epoch + 1} - KL Divergence: {kl_divergence:.6f}")
4. Compute Sharpe Ratio and Max Drawdown in evaluate_model
Expand evaluation.py to include these metrics, ensuring they are added to eval_metrics.

5. Log Metrics to TensorBoard
In train_loop.py, update TensorBoard logging:

python
Copy code
if writer:
    writer.add_scalar("Train/KL_Divergence", avg_metrics["kl_divergence"], epoch)
    writer.add_scalar("Eval/Sharpe_Ratio", eval_metrics["sharpe_ratio"], epoch)
6. Expand interpret_metrics Function
Provide expanded insights based on the new metrics:

python
Copy code
if "kl_divergence" in avg_metrics:
    kl = avg_metrics["kl_divergence"]
    if kl > 0.01:
        logger.info("KL Divergence is high. Policy updates might be too aggressive.")
    elif kl < 0.001:
        logger.info("KL Divergence is very low. Policy updates might be overly conservative.")
7. Update the Configuration
In config.yaml, add entries for any new metrics or thresholds for logging:

yaml
Copy code
evaluation:
  additional_metrics:
    - kl_divergence
    - sharpe_ratio
    - max_drawdown
8. Validate Changes
Ensure validate_and_cast_config in config_validation.py is updated to handle the new configuration parameters.

Would you like a specific implementation of any of these additions or insights into a particular section of the training pipeline?





You said:
added interpretation logging like those : 2025-01-04 19:41:22,628 - INFO - Monitoring - GPU Memory Usage [After Logging Metrics]: Allocated: 2624.44 MB | Reserved: 5326.00 MB
2025-01-04 19:41:22,628 - INFO - Monitoring - Actor loss is very low. Model might be converging well.
2025-01-04 19:41:22,628 - INFO - Monitoring - Critic loss is very low. Model might be underfitting.
2025-01-04 19:41:22,628 - INFO - Monitoring - Entropy is high. Model is exploring more.
2025-01-04 19:41:22,628 - INFO - TrainLoopRank0 - Actor loss is very low. Model might be converging.
2025-01-04 19:41:22,629 - INFO - TrainLoopRank0 - Critic loss is very low. Model might be underfitting.
2025-01-04 19:41:22,629 - INFO - TrainLoopRank0 - Entropy is high. Model is exploring.
2025-01-04 19:41:22,629 - INFO - TrainLoopRank0 - Model is making negative returns on evaluation.   but more elaborated if could be done
ChatGPT said:
ChatGPT
To add more detailed and elaborated interpretation logging, we can expand the current interpretation functions (interpret_metrics and monitor_training) and logging. Here's how we can provide more context, actionable insights, and a holistic view for each metric:

1. Expanded Interpretations for Metrics
Actor Loss
python
Copy code
if actor_loss is not None:
    if actor_loss > 1.0:
        logger.info("Actor loss is high, indicating the policy may be unstable. Consider lowering the learning rate or increasing the batch size.")
    elif actor_loss < 0.01:
        logger.info("Actor loss is very low. The model might be converging well. However, monitor if the policy starts overfitting to the training data.")
    else:
        logger.info("Actor loss is within an acceptable range. The policy is learning effectively.")
Critic Loss
python
Copy code
if critic_loss is not None:
    if critic_loss > 1.0:
        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.")
    elif critic_loss < 0.01:
        logger.info("Critic loss is very low, suggesting the value function may have already converged. Monitor for underfitting or lack of variance.")
    else:
        logger.info("Critic loss is within an acceptable range, suggesting the value function is learning effectively.")
Entropy
python
Copy code
if entropy is not None:
    if entropy > 1.0:
        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.")
    elif entropy < 0.1:
        logger.info("Entropy is low, indicating the model is exploiting learned policies. Ensure this doesn't lead to premature convergence or local optima.")
    else:
        logger.info("Entropy is within a balanced range, maintaining a healthy exploration-exploitation trade-off.")
2. Elaboration on Evaluation Metrics
Average Reward
python
Copy code
if eval_metrics.get("average_reward", 0) > 0:
    logger.info("The model is achieving positive returns on evaluation, suggesting effective learning and policy optimization. Aim to maximize rewards by fine-tuning hyperparameters.")
else:
    logger.info("The model is achieving negative returns on evaluation, indicating suboptimal policy performance. Consider improving exploration strategies or increasing the training data diversity.")
Sharpe Ratio
python
Copy code
if sharpe_ratio > 1.0:
    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.")
elif 0 < sharpe_ratio <= 1.0:
    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.")
else:
    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.")
Max Drawdown
python
Copy code
if max_drawdown < 0.1:
    logger.info("Max drawdown is low, indicating the model is maintaining controlled risk. This is a desirable attribute for a robust trading strategy.")
else:
    logger.info("Max drawdown is high, indicating significant risk exposure. Consider implementing risk management mechanisms or penalizing high drawdowns in the reward function.")
Win Rate
python
Copy code
if avg_win_rate > 0.5:
    logger.info(f"The model has an average win rate of {avg_win_rate:.2%}, showing that more than half of the trades are profitable. Aim to further improve profitability by refining the strategy.")
else:
    logger.info(f"The model has an average win rate of {avg_win_rate:.2%}. Explore strategies to improve precision and reduce losses in less favorable scenarios.")
3. Logging with Contextual Recommendations
We can add context-aware recommendations in monitor_training to suggest specific actions based on multiple metrics.

python
Copy code
if entropy > 1.0 and actor_loss < 0.01:
    logger.info("High entropy with very low actor loss indicates the model is heavily exploring but might have already converged. Consider reducing entropy coefficient to prioritize exploitation.")
if critic_loss > 1.0 and sharpe_ratio < 1.0:
    logger.info("High critic loss and low Sharpe Ratio suggest the value function is not accurately predicting returns. Tune the critic's learning rate or architecture.")
if avg_win_rate < 0.4 and max_drawdown > 0.2:
    logger.info("Low win rate and high max drawdown indicate an unstable policy. Experiment with penalizing risky actions or increasing training epochs.")
4. Example Enhanced Log Output
plaintext
Copy code
2025-01-04 19:41:22,628 - INFO - Monitoring - GPU Memory Usage [After Logging Metrics]: Allocated: 2624.44 MB | Reserved: 5326.00 MB
2025-01-04 19:41:22,628 - INFO - Monitoring - Actor loss is very low. The model might be converging well. Monitor for overfitting as training progresses.
2025-01-04 19:41:22,628 - INFO - Monitoring - Critic loss is very low. This suggests the value function may have converged. Ensure it's not underfitting.
2025-01-04 19:41:22,628 - INFO - Monitoring - Entropy is high. The model is still exploring heavily, which is good for early training. Consider gradual reduction.
2025-01-04 19:41:22,629 - INFO - TrainLoopRank0 - The Sharpe Ratio is negative, indicating the current policy is not risk-efficient. Investigate the reward function or increase training duration.
2025-01-04 19:41:22,629 - INFO - TrainLoopRank0 - Max drawdown is significant, suggesting the strategy is exposing the portfolio to high risk. Implement tighter risk management measures.
2025-01-04 19:41:22,629 - INFO - TrainLoopRank0 - Low win rate and high drawdown detected. Consider optimizing the policy for safer, more consistent trades.


finally write the whole corrected scripts here, only the one involve in these changes.  minimal changes to the original scripts.   the give a sumary of the changes done.