使用TimeSeriesDataSet/TemporalFusionTransformer遇NoneType错误求助

问题概述 #

针对单一金融产品的日内金融数据训练TemporalFusionTransformer模型，目标变量为已归一化的百分比差异形式。已确认数据集无空值，尝试过添加/移除归一化器，但加载DataLoader时触发TypeError，提示default_collate: batch must contain tensors, numpy arrays, numbers, dicts or lists; found <class 'NoneType'>。调试发现样本中存在(tensor([-2.1231]), None)的结构，怀疑与target_scale相关。

目标变量统计描述 #

count    1228.000000
mean        0.604246
std         3.276353
min       -30.971363
25%        -0.736436
50%         0.000000
75%         1.986204
max        29.972645

核心错误片段 #

error at index 15 with item ({'x_cat': tensor([], size=(31, 0), dtype=torch.int64), 'x_cont': tensor([[-0.5910,  0.5905, ...]]), ..., 'target_scale': array([0.60424554, 3.27732913])}, (tensor([-2.1231]), None)): default_collate: batch must contain tensors, numpy arrays, numbers, dicts or lists; found <class 'NoneType'>

完整代码 #

import pandas as pd
import numpy as np
from pytorch_forecasting import TimeSeriesDataSet, TemporalFusionTransformer, QuantileLoss
from pytorch_forecasting.data import GroupNormalizer
from torch.utils.data import DataLoader
from torch.optim import Adam
from torch.optim.lr_scheduler import ExponentialLR

data = pd.read_csv(f'{path_parent}/data/csv/stats_days_5.csv', parse_dates=['Date'])
data.reset_index(inplace=True, drop=True)
data = data.iloc[:-1]
data['time_index'] = np.arange(len(data))
data['group_id'] = 1

d_variable = 'spf_daily_prf_adj_sum_next_5_fixed'

ind_variables = [f'close_max_{num}', f'close_min_{num}', f'close_score_ave_{num}'
                 , f'count_ml_rev_{num}', f'mid_momentum_score_ave_{num}', f'late_momentum_score_ave_{num}'
                 ,f'mid_reversal_count_{num}', f'late_reversal_count_{num}']

# Train-test split
split_idx = int(data.shape[0] * 0.8)
training_cutoff = data.loc[split_idx, 'time_index']

max_prediction_length = 1  # how many steps to predict
max_encoder_length = 30   # how timesteps of history to use
training_cutoff = data["spf_daily_prf_adj_sum_next_5_fixed"].shape[0] - max_prediction_length

training = TimeSeriesDataSet (
    data[lambda x: x.time_index <= training_cutoff]
    ,time_idx='time_index'
    ,group_ids = ['group_id']
    ,min_encoder_length=max_encoder_length
    ,max_encoder_length=max_encoder_length
    ,min_prediction_length=1
    ,max_prediction_length=max_prediction_length
    ,target = d_variable
    ,static_categoricals = []
    ,time_varying_unknown_reals = ind_variables
    ,target_normalizer=GroupNormalizer(groups=['group_id'])
)

validation = TimeSeriesDataSet.from_dataset(training, data, predict=True, stop_randomization=True)

print("after transformation")

print(data.isnull().sum())
print(data.dtypes)  # Check data types
print(data.isin([np.nan, np.inf, -np.inf]).any())

print("got the line 61")

def debug_collate_fn(batch):
    for idx, item in enumerate(batch):
        if item is None:
            print(f"Found None at index {idx}")
        else:
            try:
                torch.utils.data.dataloader.default_collate([item])
            except TypeError as e:
                print(f"Error at index {idx} with item {item}: {e}")
    return torch.utils.data.dataloader.default_collate(batch)

def custom_collate(batch):
    batch = [item for item in batch if item is not None and item[1] is not None]
    return default_collate(batch)

train_dataloader = DataLoader(training, batch_size=16, shuffle=True, collate_fn=debug_collate_fn)

val_dataloader = DataLoader(validation, batch_size=32, shuffle=False, collate_fn=debug_collate_fn)

# model specs
model = TemporalFusionTransformer.from_dataset(
    training,
    learning_rate=0.03,
    hidden_size=16,
    attention_head_size=1,
    dropout=0.1,
    hidden_continuous_size=8,
    output_size=7,
    loss=QuantileLoss()
)

# training
def train_model(model, train_dataloader, val_dataloader, epochs=30):
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    scheduler = ExponentialLR(optimizer, gamma=0.9)
    history = {'train_loss': [], 'val_loss': []}
    
    for epoch in range(epochs):
        model.train()
        total_train_loss = 0
        for batch in train_dataloader:
            optimizer.zero_grad()
            output = model(batch)
            loss = output['loss']
            loss.backward()
            optimizer.step()
            total_train_loss += loss.item()
        train_loss = total_train_loss / len(train_dataloader)
        
        model.eval()
        total_val_loss = 0
        with torch.no_grad():
            for batch in val_dataloader:
                output = model(batch)
                loss = output['loss']
                total_val_loss += loss.item()
        val_loss = total_val_loss / len(val_dataloader)
        
        scheduler.step()
        history['train_loss'].append(train_loss)
        history['val_loss'].append(val_loss)
        print(f"Epoch {epoch}: Train Loss {train_loss:.4f}, Val Loss {val_loss:.4f}")
    
    return model, history

for i, batch in enumerate(train_dataloader):
    print(f"Batch {i}: {batch}")
    if any(item is None for item in batch):
        print("Found None in batch")

完整错误栈 #

Traceback (most recent call last):
  File "/mnt/c/candlestick/src/transfomer.py", line 142, in <module>
    for i, batch in enumerate(train_dataloader):
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/dataloader.py", line 630, in __next__
    data = self._next_data()
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/dataloader.py", line 673, in _next_data
    data = self._dataset_fetcher.fetch(index)  # may raise StopIteration
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/_utils/fetch.py", line 55, in fetch
    return self.collate_fn(data)
  File "/mnt/c/candlestick/src/transfomer.py", line 82, in debug_collate_fn
    return torch.utils.data.dataloader.default_collate(batch)
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/_utils/collate.py", line 317, in default_collate
    return collate(batch, collate_fn_map=default_collate_fn_map)
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/_utils/collate.py", line 174, in collate
    return [collate(samples, collate_fn_map=collate_fn_map) for samples in transposed]  # Backwards compatibility.
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/_utils/collate.py", line 174, in <listcomp>
    return [collate(samples, collate_fn_map=collate_fn_map) for samples in transposed]  # Backwards compatibility.
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/_utils/collate.py", line 174, in collate
    return [collate(samples, collate_fn_map=collate_fn_map) for samples in transposed]  # Backwards compatibility.
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/_utils/collate.py", line 174, in <listcomp>
    return [collate(samples, collate_fn_map=collate_fn_map) for samples in transposed]  # Backwards compatibility.
  File "/home/rgn/anaconda3/envs/rapids-23.08/lib/python3.10/site-packages/torch/utils/data/_utils/collate.py", line 192, in collate
    raise TypeError(default_collate_err_msg_format.format(elem_type))
TypeError: default_collate: batch must contain tensors, numpy arrays, numbers, dicts or lists; found <class 'NoneType'>

问题排查与解决方案 #

问题根源 #

1. training_cutoff逻辑冲突：代码中先基于训练集80%分割设置training_cutoff，随后又用目标变量长度覆盖该值，导致训练集截断逻辑混乱，部分样本无法生成有效目标序列对。
2. 验证集模式错误：使用predict=True创建验证集时，会生成无目标值的样本（用于预测场景），导致样本中出现(tensor, None)的结构，触发collate错误。
3. 单一Group的归一化器适配问题：使用GroupNormalizer但只有一个group_id，可能导致内部处理异常，生成的target_scale对应项为None。

具体修复步骤 #

1. 修正training_cutoff逻辑
   保留基于时间索引的分割逻辑，避免覆盖：

   # 移除错误的覆盖行
   # training_cutoff = data["spf_daily_prf_adj_sum_next_5_fixed"].shape[0] - max_prediction_length
   # 确保训练集有足够的历史数据和预测步长
   training_cutoff = training_cutoff - max_prediction_length
   

2. 调整验证集创建方式
   验证集需要包含目标值，重新定义验证集：

   validation = TimeSeriesDataSet(
       data[lambda x: x.time_index > training_cutoff],
       time_idx='time_index',
       group_ids=['group_id'],
       min_encoder_length=max_encoder_length,
       max_encoder_length=max_encoder_length,
       min_prediction_length=1,
       max_prediction_length=max_prediction_length,
       target=d_variable,
       static_categoricals=[],
       time_varying_unknown_reals=ind_variables,
       target_normalizer=training.target_normalizer  # 复用训练集的归一化器
   )
   

3. 修复collate函数
   使用自定义collate函数过滤掉包含None的样本：

   from torch.utils.data._utils.collate import default_collate
   
   def custom_collate(batch):
       # 过滤掉无效样本
       filtered_batch = []
       for item in batch:
           if item is not None:
               inputs, targets = item
               if targets is not None and not any(t is None for t in targets):
                   filtered_batch.append(item)
       return default_collate(filtered_batch) if filtered_batch else None
   

   替换DataLoader的collate_fn：

   train_dataloader = DataLoader(training, batch_size=16, shuffle=True, collate_fn=custom_collate)
   val_dataloader = DataLoader(validation, batch_size=32, shuffle=False, collate_fn=custom_collate)
   

4. 替换归一化器（可选）
   单一group场景下，改用SingleSeriesNormalizer更适配：

   from pytorch_forecasting.data import SingleSeriesNormalizer
   
   training = TimeSeriesDataSet (
       # ... 其他参数不变
       target_normalizer=SingleSeriesNormalizer()
   )
   

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