Modernización de la Clase de Carga de Datos

A continuación, se presenta una versión refactorizada de la clase encargada de procesar conjuntos de datos personalizados para modelos de series temporales. Se han implementado mejoras en la tipificación, se ha optimizado el procesamiento de fechas utilizando los accesores nativos de Pandas (dt) en lugar de funciones lambda, y se han renombrado las variables para lograr una semántica más clara.

import os
import pandas as pd
import numpy as np
import torch
from torch.utils.data import Dataset
from sklearn.preprocessing import StandardScaler

class CustomTimeSeriesDataset(Dataset):
    def __init__(self, data_dir: str, split: str = 'train', window_config: list = None,
                 feature_mode: str = 'M', filename: str = 'custom_data.csv',
                 target_col: str = 'OT', apply_scaling: bool = True,
                 time_encoding_type: int = 1, freq: str = 'h', train_only: bool = False):
        
        self.data_dir = data_dir
        self.filename = filename
        self.target_col = target_col
        self.apply_scaling = apply_scaling
        self.time_encoding_type = time_encoding_type
        self.freq = freq
        self.feature_mode = feature_mode
        self.train_only = train_only
        
        # Configuración de ventanas temporales
        if window_config is None:
            self.input_window = 96
            self.overlap_window = 48
            self.forecast_horizon = 96
        else:
            self.input_window, self.overlap_window, self.forecast_horizon = window_config
            
        assert split in ['train', 'val', 'test'], "El parámetro split debe ser 'train', 'val' o 'test'"
        self.split_idx = {'train': 0, 'val': 1, 'test': 2}[split]
        
        self.scaler = StandardScaler()
        self._load_and_preprocess()

    def _load_and_preprocess(self):
        file_path = os.path.join(self.data_dir, self.filename)
        raw_df = pd.read_csv(file_path)
        
        # Filtrado de columnas según el modo de características
        all_cols = list(raw_df.columns)
        all_cols.remove('date')
        
        if self.feature_mode == 'S':
            if self.target_col in all_cols:
                all_cols.remove(self.target_col)
            feature_cols = [self.target_col]
        else:
            feature_cols = all_cols
            
        # Cálculo de límites para la partición de datos
        total_len = len(raw_df)
        train_limit = int(total_len * (1.0 if self.train_only else 0.7))
        test_limit = int(total_len * 0.2)
        
        borders_start = [0, train_limit - self.input_window, total_len - test_limit - self.input_window]
        borders_end = [train_limit, train_limit + (total_len - train_limit - test_limit), total_len]
        
        start_idx = borders_start[self.split_idx]
        end_idx = borders_end[self.split_idx]
        
        # Escalado de datos numéricos
        data_subset = raw_df[feature_cols]
        if self.apply_scaling:
            train_data = data_subset.iloc[borders_start[0]:borders_end[0]]
            self.scaler.fit(train_data.values)
            scaled_data = self.scaler.transform(data_subset.values)
        else:
            scaled_data = data_subset.values
            
        self.features_x = scaled_data[start_idx:end_idx]
        self.features_y = scaled_data[start_idx:end_idx]
        
        # Codificación de características temporales
        dates = pd.to_datetime(raw_df['date'].iloc[start_idx:end_idx])
        if self.time_encoding_type == 0:
            time_df = pd.DataFrame({
                'month': dates.dt.month,
                'day': dates.dt.day,
                'weekday': dates.dt.weekday,
                'hour': dates.dt.hour
            })
            self.time_marks = time_df.values
        else:
            self.time_marks = self._generate_cyclical_features(dates)

    def _generate_cyclical_features(self, dates):
        month_sin = np.sin(2 * np.pi * dates.dt.month / 12)
        month_cos = np.cos(2 * np.pi * dates.dt.month / 12)
        day_sin = np.sin(2 * np.pi * dates.dt.day / 31)
        day_cos = np.cos(2 * np.pi * dates.dt.day / 31)
        return np.column_stack([month_sin, month_cos, day_sin, day_cos])

    def __len__(self):
        return len(self.features_x) - self.input_window - self.forecast_horizon + 1

    def __getitem__(self, idx):
        in_start = idx
        in_end = in_start + self.input_window
        
        out_start = in_end - self.overlap_window
        out_end = out_start + self.overlap_window + self.forecast_horizon
        
        x_seq = self.features_x[in_start:in_end]
        y_seq = self.features_y[out_start:out_end]
        x_mark = self.time_marks[in_start:in_end]
        y_mark = self.time_marks[out_start:out_end]
        
        return x_seq, y_seq, x_mark, y_mark

    def inverse_transform(self, data):
        return self.scaler.inverse_transform(data)

1. Casos de Uso para Conjuntos de Datos Personalizados

Esta implementación está diseñada para procesar datos propietarios o de fuentes externas. El requisito fundamental es que el archivo CSV mantenga una estructura específica: la primera columna debe contener las marcas de tiempo (date) y la última columna debe corresponder a la variable objetivo (OT u otro identificador de destino). Es común utilizar esta estructura al trabajar con bases de datos como tráfico, clima o consumo eléctrico, donde el número de canales (variables de entrada) varía dinámicamente, requiriendo un ajuste manual del parámetro de dimensiones de entrada en el modelo.

2. Compatibilidad de los Parámetros de Inicialización

La firma del método __init__ mantiene similitudes con clases específicas de otros datasets (como los de la familia ETT) por razones de compatibilidad histórica con arquitecturas como Informer. Aunque los parámetros parezcan estáticos, en un entorno de producción real, los hiperparámetros efectivos son inyectados dinámicamente a través de funciones proveedoras de datos (data providers), lo que permite reutilizar el mismo código para múltiples experimentos sin modificar la clase base.

3. Modelado Univariado vs. Multivariado

• Univariado (Modo 'S'): El modelo asume una relación del tipo $y = f(x_1)$, donde $x_1$ representa exclusivamente el historial de la variable objetivo, y $y$ es su pryoección futura.
• Multivariado (Modo 'M' o 'MS'): La relación se expande a $y = f(x_1, x_2, ..., x_n)$. El sistema utiliza el historial de todas las variables exógenas disponibles, incluyendo la propia variable objetivo, para inferir los valores futuros de dicha variable objetivo.

4. Sustitución de Datasets Específicos (ETT)

Aunque esta clase genérica tiene la flexibilidad para leer archivos con diversas granularidades temporales, reemplazar clases especializadas (como Dataset_ETT_hour) requiere precaución:

• Frecuencia Temporal: Las clases especializadas asumen frecuencias predefinidas ('h' para horas, 't' para minutos). En la versión personalizada, el parámetro de frecuencia debe configurarse explícitamente para evitar errores en la extracción de características cíclicas.
• Estrategia de Partición: Los datasets de energía (ETT) suelen utilizar divisiones cronológicas estrictas debido a su alta estacionalidad. La clase genérica aplica divisiones porcentuales (70/10/20), lo cual podría alterar la distribución temporal si no se ajusta manualmente.
• Aumento de Datos: Las implementaciones específicas pueden incluir transformaciones propias del dominio que deben ser integradas manualmente en la clase genérica si se desea replicar el comportamiento.

5. Prevención de Fuga de Datos en el Escalado

El método StandardScaler se ajusta (fit) utilizando exclusivamente los datos correspondientes al conjunto de entrenamiento. Posteriormente, este mismo escalador se aplica (transform) a la totalidad del dataset, incluyendo las particiones de validación y prueba. Si se calcularan la media y la desviación estándar sobre el dataset completo, la información estadística del futuro se filtraría hacia la etapa de entrenamiento (Data Leakage), generando métricas de evaluación artificialmente infladas y poco realistas.

6. Importancia de las Características Temporales

En el pronóstico de series temporales, los patrones cíclicos (horas pico, estacionalidad mensual) son predictores críticos. La extracción de la columna date en una matriz separada (time_marks) permite al modelo contextualizar cada observación numérica dentro de su marco temporal correspondiente.

7. Estructura de las Matrices de Tiempo

Dependiendo de la configuración, las marcas de tiempo pueden representarse de dos formas:

Codificación Manual

Extracción directa de componentes discretos a partir del objeto datetime.

Codificación Cíclica Automática

Transformación de variables discretas a un espacio continuo utilizando funciones trigonométricas (seno y coseno) para preservar la naturaleza circular del tiempo (por ejemplo, la hora 23 está cerca de la hora 0).

8. Integración de Embeddings Temporales en la Red

Los tensores de tiempo (x_mark, y_mark) se retornan junto con los datos numéricos en el método __getitem__. Dentro del modelo de aprendizaje profundo, estas matrices se concatenan con las series originales a lo largo de la dimensión de características. De este modo, un valor de temperatura en un instante específico se enriquece con su contexto estacional.

# Ejemplo de concatenación en PyTorch
# seq_x shape: [batch, seq_len, num_features]
# seq_x_mark shape: [batch, seq_len, time_features]
combined_input = torch.cat([seq_x, seq_x_mark], dim=-1)

9. El Rol de la Ventana de Solapamiento (label_len)

En arquitecturas tradicionales (LSTM, CNN), la longitud de la etiqueta suele ser cero. Sin embargo, en modelos basados en arquitecturas Transformer (como Autoformer o Informer), se introduce un parámetro de solapamiento entre la secuencia de entrada del codificador y la secuencia de entrada del decodificador.

Si el objetivo es predecir 3 pasos futuros (forecast_horizon = 3) usando 10 pasos históricos (input_window = 10), una ventana de solapamiento de 5 (overlap_window = 5) generará un tensor de salida objetivo (seq_y) de longitud 8. Los primeros 5 valores actúan como contexto conocido (o "prompt") para inicializar el decodificador de manera auto-regresiva, facilitando una transición más suave hacia la predicción de los 3 pasos desconocidos.

10. Dimensionalidad de los Tensores en un Batch

Al utilizar un DataLoader de PyTorch, las muestras individuales se apilan en lotes (batches). Para una configuración donde el tamaño del lote es B, la ventana de entrada es 96, el solapamiento es 48 y el horizonte de predicción es 96, los tensores resultantes tendrán las siguientes formas geométricas:

Donde C representa el número total de variables exógenas y endógenas, y T es la dimensionalidad del vector de características temporales extraídas.