something is wrong with this new version of preprocess.py,  the routine hang on saving the unscalled file for close.....  but in the previous version it was working just fine....  here is the new non working scripts : # File: data_processing/preprocess.py

import os
import logging
import pandas as pd
import numpy as np
import joblib
import torch
from sklearn.preprocessing import MinMaxScaler
from typing import Dict

def setup_logging(log_dir="logs", log_file="preprocess.log"):
    """
    Configure logging for the preprocessing script.

    Args:
        log_dir (str): Directory to store log files.
        log_file (str): Log file name.
    
    Returns:
        logging.Logger: Configured logger instance.
    """
    os.makedirs(log_dir, exist_ok=True)
    logger = logging.getLogger("Preprocessing")
    logger.setLevel(logging.INFO)

    # Create console handler
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)

    # Create file handler
    fh = logging.FileHandler(os.path.join(log_dir, log_file))
    fh.setLevel(logging.DEBUG)

    # Create formatter and add to handlers
    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s")
    ch.setFormatter(formatter)
    fh.setFormatter(formatter)

    # Add handlers to the logger if none exist
    if not logger.hasHandlers():
        logger.addHandler(ch)
        logger.addHandler(fh)

    return logger

# Initialize logger
logger = logging.getLogger("Preprocessing")
if not logger.hasHandlers():
    logger = setup_logging()

def validate_feature_columns(df, feature_columns, logger):
    """
    Validate that all specified feature columns exist in the dataset.

    Args:
        df (pd.DataFrame): The dataset.
        feature_columns (list): List of expected feature columns.
        logger (logging.Logger): Logger instance for logging.

    Raises:
        KeyError: If any feature column is missing.
    """
    missing_columns = [col for col in feature_columns if col not in df.columns]
    if missing_columns:
        logger.error(f"Missing feature columns: {missing_columns}")
        raise KeyError(f"Missing feature columns: {missing_columns}")

def normalize_feature_columns(df: pd.DataFrame, feat_cols: list, logger: logging.Logger) -> pd.DataFrame:
    """
    Normalize the feature columns using MinMaxScaler, including 'close'.

    Args:
        df (pd.DataFrame): DataFrame containing the data to normalize.
        feat_cols (list): List of feature column names to normalize.
        logger (logging.Logger): Logger instance for logging messages.

    Returns:
        pd.DataFrame: DataFrame with normalized feature columns.

    Raises:
        Exception: If scaler loading or fitting fails.
    """
    validate_feature_columns(df, feat_cols, logger)
    available_features = [col for col in feat_cols if col in df.columns]

    if len(available_features) != len(feat_cols):
        logger.warning(f"Missing columns in dataset: {set(feat_cols) - set(available_features)}")

    scaler_path = os.path.join('scalers', 'minmax_scaler.joblib')
    os.makedirs('scalers', exist_ok=True)

    if os.path.exists(scaler_path):
        logger.info(f"Loading scaler from {scaler_path}")
        scaler = joblib.load(scaler_path)
        if hasattr(scaler, 'feature_names_in_') and not np.array_equal(scaler.feature_names_in_, available_features):
            logger.warning("Scaler feature names do not match features to normalize. Re-fitting the scaler.")
            scaler = MinMaxScaler()
            scaler.fit(df[available_features])
            joblib.dump(scaler, scaler_path)
            logger.info(f"Scaler re-fitted and saved to {scaler_path}")
    else:
        logger.info(f"No scaler found at {scaler_path}. Fitting a new scaler.")
        scaler = MinMaxScaler()
        scaler.fit(df[available_features])
        joblib.dump(scaler, scaler_path)
        logger.info(f"Scaler fitted and saved to {scaler_path}")
    
    df[available_features] = scaler.transform(df[available_features])
    return df

def export_and_save_preprocessed_data(config: Dict, logger: logging.Logger) -> None:
    """
    Preprocess the raw data and split it into multiple PT files, implementing overlapping windows.

    Args:
        config (Dict): Configuration dictionary containing data paths and parameters.
        logger (logging.Logger): Logger instance for logging messages.

    Raises:
        FileNotFoundError: If the raw data file does not exist.
        KeyError: If required keys are missing in PT files.
        ValueError: If the preprocessed dataframe is empty after cleaning.
    """
    raw_data_path = config['data']['raw_data_path']
    pt_data_dir = config['data'].get('pt_data_dir', 'data/processed/')
    pt_data_prefix = config.get('data', {}).get('pt_data_prefix', 'processed_data_part')
    os.makedirs(pt_data_dir, exist_ok=True)

    # Check if already preprocessed
    existing_pt_files = [fname for fname in os.listdir(pt_data_dir) if fname.startswith(pt_data_prefix)]
    if existing_pt_files:
        logger.info(f"Processed data files already exist in '{pt_data_dir}'. Skipping preprocessing.")
        return

    try:
        logger.info(f"Reading raw data from: {raw_data_path}")
        if not os.path.exists(raw_data_path):
            logger.error(f"Raw data file not found at: {raw_data_path}")
            raise FileNotFoundError(f"Raw data file not found at: {raw_data_path}")
        
        df = pd.read_csv(raw_data_path)
        logger.info(f"Initial data shape: {df.shape}")

        # Ensure 'close' is in feature columns
        feat_cols = config['data']['feature_columns']
        validate_feature_columns(df, feat_cols, logger)

        # Replace infinite values with NaN
        df.replace([np.inf, -np.inf], np.nan, inplace=True)

        # Drop rows with NaN or infinite values
        initial_length = len(df)
        df.dropna(inplace=True)
        cleaned_length = len(df)
        logger.info(f"Removed {initial_length - cleaned_length} rows containing NaN or infinite values.")

        # Check if dataframe is empty after cleaning
        if df.empty:
            logger.error("Preprocessed dataframe is empty after removing NaNs and infinite values.")
            raise ValueError("No data left after preprocessing.")

        # Normalize features if required
        normalize_features = config['data'].get('normalize_features', True)
        if normalize_features:
            logger.info("Normalizing feature columns, including 'close'.")
            df = normalize_feature_columns(df, feat_cols, logger)

        # Save unnormalized 'close' prices separately
        unnormalized_close_prices = df['close'].values  # Assuming 'close' is in feat_cols
        unnormalized_close_path = os.path.join(pt_data_dir, 'unnormalized_close_prices.npy')
        np.save(unnormalized_close_path, unnormalized_close_prices)
        logger.info(f"Unnormalized 'close' prices saved to {unnormalized_close_path}")

        # Implement overlapping windows using window_stride
        window_stride = config['data'].get('window_stride', 1)  # Stride of 1 for full overlap
        n_steps = config['data']['n_steps']
        num_splits = config['data'].get('num_splits', 10)

        window_data = []
        window_close_prices = []
        for i in range(0, len(df) - n_steps + 1, window_stride):
            window = df[feat_cols].values[i : i + n_steps]
            window_data.append(window)
            window_close_prices.append(unnormalized_close_prices[i + n_steps - 1])  # Save last close price
        
        # Split data into parts and save as PT files
        logger.info(f"Splitting data into {num_splits} parts with overlapping windows.")
        data_splits = np.array_split(window_data, num_splits)
        close_price_splits = np.array_split(window_close_prices, num_splits)

        for idx, (split_data, split_close_prices) in enumerate(zip(data_splits, close_price_splits)):
            # Verify that each split has enough data
            if len(split_data) == 0:
                logger.warning(f"Split {idx} has no data. Skipping this split.")
                continue

            data_tensor = torch.tensor(split_data, dtype=torch.float32)
            close_price_tensor = torch.tensor(split_close_prices, dtype=torch.float32)
            pt_file_path = os.path.join(pt_data_dir, f"{pt_data_prefix}_{idx}.pt")
            torch.save({'data': data_tensor, 'close_prices': close_price_tensor}, pt_file_path)
            logger.info(f"Saved split {idx} to {pt_file_path} with {len(split_data)} windows.")

    except Exception as e:
        logger.error(f"Failed to preprocess data: {e}", exc_info=True)
        raise  this is the old version that was saving just fine : # File: data_processing/preprocess.py

import os
import logging
import pandas as pd
import numpy as np
import joblib
import torch
from sklearn.preprocessing import MinMaxScaler
from typing import Dict

def setup_logging(log_dir="logs", log_file="preprocess.log"):
    """
    Configure logging for the preprocessing script.

    Args:
        log_dir (str): Directory to store log files.
        log_file (str): Log file name.
    
    Returns:
        logging.Logger: Configured logger instance.
    """
    os.makedirs(log_dir, exist_ok=True)
    logger = logging.getLogger("Preprocessing")
    logger.setLevel(logging.INFO)

    # Create console handler
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)

    # Create file handler
    fh = logging.FileHandler(os.path.join(log_dir, log_file))
    fh.setLevel(logging.DEBUG)

    # Create formatter and add to handlers
    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s")
    ch.setFormatter(formatter)
    fh.setFormatter(formatter)

    # Add handlers to the logger if none exist
    if not logger.hasHandlers():
        logger.addHandler(ch)
        logger.addHandler(fh)

    return logger

# Initialize logger
logger = logging.getLogger("Preprocessing")
if not logger.hasHandlers():
    logger = setup_logging()


def validate_feature_columns(df, feature_columns, logger):
    """
    Validate that all specified feature columns exist in the dataset.

    Args:
        df (pd.DataFrame): The dataset.
        feature_columns (list): List of expected feature columns.
        logger (logging.Logger): Logger instance for logging.

    Raises:
        KeyError: If any feature column is missing.
    """
    missing_columns = [col for col in feature_columns if col not in df.columns]
    if missing_columns:
        logger.error(f"Missing feature columns: {missing_columns}")
        raise KeyError(f"Missing feature columns: {missing_columns}")


def normalize_feature_columns(df: pd.DataFrame, feat_cols: list, logger: logging.Logger) -> pd.DataFrame:
    """
    Normalize the feature columns using MinMaxScaler, including 'close'.

    Args:
        df (pd.DataFrame): DataFrame containing the data to normalize.
        feat_cols (list): List of feature column names to normalize.
        logger (logging.Logger): Logger instance for logging messages.

    Returns:
        pd.DataFrame: DataFrame with normalized feature columns.

    Raises:
        Exception: If scaler loading or fitting fails.
    """
    validate_feature_columns(df, feat_cols, logger)
    available_features = [col for col in feat_cols if col in df.columns]

    if len(available_features) != len(feat_cols):
        logger.warning(f"Missing columns in dataset: {set(feat_cols) - set(available_features)}")

    scaler_path = os.path.join('scalers', 'minmax_scaler.joblib')
    os.makedirs('scalers', exist_ok=True)

    if os.path.exists(scaler_path):
        logger.info(f"Loading scaler from {scaler_path}")
        scaler = joblib.load(scaler_path)
        if hasattr(scaler, 'feature_names_in_') and not np.array_equal(scaler.feature_names_in_, available_features):
            logger.warning("Scaler feature names do not match features to normalize. Re-fitting the scaler.")
            scaler = MinMaxScaler()
            scaler.fit(df[available_features])
            joblib.dump(scaler, scaler_path)
            logger.info(f"Scaler re-fitted and saved to {scaler_path}")
    else:
        logger.info(f"No scaler found at {scaler_path}. Fitting a new scaler.")
        scaler = MinMaxScaler()
        scaler.fit(df[available_features])
        joblib.dump(scaler, scaler_path)
        logger.info(f"Scaler fitted and saved to {scaler_path}")
    
    df[available_features] = scaler.transform(df[available_features])
    return df


def export_and_save_preprocessed_data(config: Dict, logger: logging.Logger) -> None:
    """
    Preprocess the raw data and split it into multiple PT files.

    Args:
        config (Dict): Configuration dictionary containing data paths and parameters.
        logger (logging.Logger): Logger instance for logging messages.

    Raises:
        FileNotFoundError: If the raw data file does not exist.
        KeyError: If required keys are missing in PT files.
        ValueError: If the preprocessed dataframe is empty after cleaning.
    """
    raw_data_path = config['data']['raw_data_path']
    pt_data_dir = config['data'].get('pt_data_dir', 'data/processed/')
    pt_data_prefix = config.get('data', {}).get('pt_data_prefix', 'processed_data_part')
    os.makedirs(pt_data_dir, exist_ok=True)

    # Check if already preprocessed
    existing_pt_files = [fname for fname in os.listdir(pt_data_dir) if fname.startswith(pt_data_prefix)]
    if existing_pt_files:
        logger.info(f"Processed data files already exist in '{pt_data_dir}'. Skipping preprocessing.")
        return

    try:
        logger.info(f"Reading raw data from: {raw_data_path}")
        if not os.path.exists(raw_data_path):
            logger.error(f"Raw data file not found at: {raw_data_path}")
            raise FileNotFoundError(f"Raw data file not found at: {raw_data_path}")
        
        df = pd.read_csv(raw_data_path)
        logger.info(f"Initial data shape: {df.shape}")

        # Ensure 'close' is in feature columns
        feat_cols = config['data']['feature_columns']
        validate_feature_columns(df, feat_cols, logger)

        # Replace infinite values with NaN
        df.replace([np.inf, -np.inf], np.nan, inplace=True)

        # Drop rows with NaN or infinite values
        initial_length = len(df)
        df.dropna(inplace=True)
        cleaned_length = len(df)
        logger.info(f"Removed {initial_length - cleaned_length} rows containing NaN or infinite values.")

        # Check if dataframe is empty after cleaning
        if df.empty:
            logger.error("Preprocessed dataframe is empty after removing NaNs and infinite values.")
            raise ValueError("No data left after preprocessing.")

        # Normalize features if required
        normalize_features = config['data'].get('normalize_features', True)
        if normalize_features:
            logger.info("Normalizing feature columns, including 'close'.")
            df = normalize_feature_columns(df, feat_cols, logger)

        # Save unnormalized 'close' prices separately
        unnormalized_close_prices = df['close'].values  # Assuming 'close' is in feat_cols
        unnormalized_close_path = os.path.join(pt_data_dir, 'unnormalized_close_prices.npy')
        np.save(unnormalized_close_path, unnormalized_close_prices)
        logger.info(f"Unnormalized 'close' prices saved to {unnormalized_close_path}")

        # Split data into parts and save as PT files
        num_splits = config['data'].get('num_splits', 10)
        
        # Determine minimum required steps per split to avoid issues
        n_steps = config['data'].get('n_steps', 20)
        min_required_steps = n_steps + 1  # To ensure at least one sample per split
        total_data_length = len(df)
        max_possible_splits = total_data_length // min_required_steps

        if max_possible_splits < num_splits:
            logger.warning(f"Requested num_splits={num_splits} is too high for the data size. Reducing to {max_possible_splits}.")
            num_splits = max_possible_splits

        if num_splits == 0:
            logger.error("Not enough data to create even one split with the required number of steps.")
            raise ValueError("Insufficient data for preprocessing.")

        data_splits = np.array_split(df[feat_cols].values, num_splits)
        close_price_splits = np.array_split(unnormalized_close_prices, num_splits)

        logger.info(f"Splitting data into {num_splits} parts.")

        for idx, (split_data, split_close_prices) in enumerate(zip(data_splits, close_price_splits)):
            # Verify that each split has enough data
            if len(split_data) < min_required_steps:
                logger.warning(f"Split {idx} has only {len(split_data)} steps, which is less than the minimum required {min_required_steps}. Skipping this split.")
                continue

            data_tensor = torch.tensor(split_data, dtype=torch.float32)
            pt_file_path = os.path.join(pt_data_dir, f"{pt_data_prefix}_{idx}.pt")
            torch.save({'data': data_tensor, 'close_prices': split_close_prices}, pt_file_path)
            logger.info(f"Saved split {idx} to {pt_file_path} with {len(split_data)} steps.")

    except Exception as e:
        logger.error(f"Failed to preprocess data: {e}", exc_info=True)
        raise make me a new version that does not bug on saving the pt files... and the unscalled close file