如何用PyTorch Lightning加载非连续CSV时序数据训练模型

核心结论 #

不需要用seq_len填充——文件间的时序断裂是真实的业务边界（比如不同股票），填充会引入无效的时序依赖，反而干扰模型学习。正确的做法是确保每个采样序列完全来自单个文件，通过以下方式实现：

步骤1：批量加载并拆分每个文件的数据 #

遍历文件夹内所有CSV，对每个文件单独做训练/验证拆分（避免跨文件拆分破坏时序连续性）：

def setup(self, stage=None):
    import glob
    # 替换为你的CSV文件夹路径
    file_paths = glob.glob('./stock_data/*.csv')
    
    # 存储每个文件的训练/验证数据
    self.train_X = []
    self.train_y = []
    self.val_X = []
    self.val_y = []
    
    for path in file_paths:
        df = pd.read_csv(
            path, 
            sep=',', 
            infer_datetime_format=True, 
            low_memory=False, 
            na_values=['nan','?'], 
            index_col='Time'
        )
        
        # 提取特征和标签
        X = df[['cols']].values  # 替换为你的特征列名
        y = df['label'].values   # 替换为你的标签列名
        
        # 单个文件内无shuffle拆分，保证时序连续
        X_train, X_val, y_train, y_val = train_test_split(
            X, y, test_size=0.25, shuffle=False
        )
        
        self.train_X.append(X_train)
        self.train_y.append(y_train)
        self.val_X.append(X_val)
        self.val_y.append(y_val)

步骤2：实现多文件兼容的Dataset #

有两种可选方案：

方案A：用ConcatDataset合并单个文件的Dataset

复用你已有的TimeseriesDataset，将每个文件的数据集合并：

def train_dataloader(self):
    train_datasets = []
    for X, y in zip(self.train_X, self.train_y):
        ds = TimeseriesDataset(X, y, seq_len=self.seq_len)
        train_datasets.append(ds)
    
    # 合并所有文件的数据集
    concat_dataset = ConcatDataset(train_datasets)
    
    return DataLoader(
        concat_dataset,
        batch_size=self.batch_size,
        shuffle=True,  # 这里可以shuffle，每个序列都是单个文件的有效数据
        num_workers=self.num_workers
    )

def val_dataloader(self):
    val_datasets = []
    for X, y in zip(self.val_X, self.val_y):
        ds = TimeseriesDataset(X, y, seq_len=self.seq_len)
        val_datasets.append(ds)
    
    concat_dataset = ConcatDataset(val_datasets)
    
    return DataLoader(
        concat_dataset,
        batch_size=self.batch_size,
        shuffle=False,
        num_workers=self.num_workers
    )

方案B：自定义多文件Dataset（更高效）

直接在Dataset内部管理多个文件的数据，避免多次创建Dataset实例：

class MultiFileTimeseriesDataset(Dataset):
    def __init__(self, X_list: list[np.ndarray], y_list: list[np.ndarray], seq_len: int = 1):
        self.seq_len = seq_len
        # 转换为Tensor并存储每个文件的可生成序列数
        self.X_tensors = [torch.tensor(x).float() for x in X_list]
        self.y_tensors = [torch.tensor(y).float() for y in y_list]
        
        # 计算累计长度，用于快速定位索引所属文件
        self.cumulative_lengths = []
        total = 0
        for X in self.X_tensors:
            seq_count = X.shape[0] - (seq_len - 1)
            self.cumulative_lengths.append(total)
            total += seq_count
        self.total_length = total

    def __len__(self):
        return self.total_length

    def __getitem__(self, index):
        # 找到索引对应的文件
        file_idx = 0
        while file_idx < len(self.cumulative_lengths)-1 and index >= self.cumulative_lengths[file_idx+1]:
            file_idx += 1
        
        # 计算文件内的局部索引
        local_idx = index - self.cumulative_lengths[file_idx]
        X = self.X_tensors[file_idx]
        y = self.y_tensors[file_idx]
        
        return (X[local_idx:local_idx+self.seq_len], y[local_idx+self.seq_len-1])

使用时直接传入所有文件的训练/验证数据：

def train_dataloader(self):
    dataset = MultiFileTimeseriesDataset(
        self.train_X, self.train_y, seq_len=self.seq_len
    )
    return DataLoader(
        dataset,
        batch_size=self.batch_size,
        shuffle=True,
        num_workers=self.num_workers
    )

关键注意事项 #

1. 禁止全局合并后拆分：必须每个文件内部单独拆分训练/验证集，否则会出现训练集包含某文件后半段、验证集包含该文件前半段的情况，破坏时序连续性。
2. 特征归一化策略：如果不同股票的特征分布差异大，建议按文件单独做归一化；如果是同类型资产的全局时序，可以用所有训练数据的统计量做归一化。
3. Shuffle权限：训练时可以开启shuffle=True，因为每个样本都是单个文件的有效序列，Shuffle不会破坏序列内部的时序逻辑。

内容的提问来源于stack exchange，提问作者dorien