클론 코딩
TimeSeriesTransformer Modeling
ParkS2
2024. 10. 7. 14:47
# Define Time2Vec layer
class Time2Vec(nn.Module):
def __init__(self, input_dim):
super(Time2Vec, self).__init__()
self.linear = nn.Linear(input_dim, 1)
self.periodic = nn.Linear(input_dim, input_dim-1)
def forward(self, x):
linear_out = self.linear(x)
periodic_out = torch.sin(self.periodic(x))
return torch.cat([linear_out, periodic_out], dim=-1)
# Define Transformer Encoder Block
class TransformerBlock(nn.Module):
def __init__(self, input_dim, num_heads, dropout):
super(TransformerBlock, self).__init__()
self.attention = nn.MultiheadAttention(input_dim, num_heads, dropout=dropout)
self.norm1 = nn.LayerNorm(input_dim)
self.norm2 = nn.LayerNorm(input_dim)
self.ff = nn.Sequential(
nn.Linear(input_dim, 4 * input_dim),
nn.GELU(),
nn.Linear(4 * input_dim, input_dim),
nn.Dropout(dropout)
)
def forward(self, x):
attended, _ = self.attention(x, x, x)
x = self.norm1(attended + x)
feedforward = self.ff(x)
return self.norm2(feedforward + x)
# Define main model architecture
class TimeSeriesTransformer(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, num_heads, output_size, dropout):
super(TimeSeriesTransformer, self).__init__()
self.time2vec = Time2Vec(input_size)
self.embedding = nn.Linear(input_size, hidden_size)
self.position_encoding = self.generate_position_encoding(hidden_size, 10)
self.dropout = nn.Dropout(dropout)
self.transformer_blocks = nn.ModuleList([
TransformerBlock(hidden_size, num_heads, dropout)
for _ in range(num_layers)
])
self.output_layer = nn.Sequential(
nn.Linear(hidden_size, hidden_size // 2),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_size // 2, output_size)
)
def generate_position_encoding(self, hidden_size, max_len):
pe = torch.zeros(max_len, hidden_size)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(torch.arange(0, hidden_size, 2).float() * (-np.log(10000.0) / hidden_size))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
return pe.unsqueeze(0)
def forward(self, x):
b, s, f = x.shape
x = self.time2vec(x)
x = self.embedding(x)
x = x + self.position_encoding[:, :s, :].to(x.device)
x = self.dropout(x)
for transformer in self.transformer_blocks:
x = transformer(x)
x = x.mean(dim=1)
return self.output_layer(x)
# Define custom dataset class
class AgriculturePriceDataset(Dataset):
def __init__(self, dataframe, window_size=9, prediction_length=3, is_test=False):
self.data = dataframe
self.window_size = window_size
self.prediction_length = prediction_length
self.is_test = is_test
self.price_column = [col for col in self.data.columns if '평균가격(원)' in col and len(col.split('_')) == 1][0]
self.numeric_columns = self.data.select_dtypes(include=[np.number]).columns.tolist()
self.sequences = []
if not self.is_test:
for i in range(len(self.data) - self.window_size - self.prediction_length + 1):
x = self.data[self.numeric_columns].iloc[i:i+self.window_size].values
y = self.data[self.price_column].iloc[i+self.window_size:i+self.window_size+self.prediction_length].values
self.sequences.append((x, y))
else:
self.sequences = [self.data[self.numeric_columns].values]
def __len__(self):
return len(self.sequences)
def __getitem__(self, idx):
if not self.is_test:
x, y = self.sequences[idx]
return torch.FloatTensor(x), torch.FloatTensor(y)
else:
return torch.FloatTensor(self.sequences[idx])
# Training function with mixed precision training
def train_model(model, train_loader, criterion, optimizer, scheduler, scaler, device):
model.train()
total_loss = 0
for batch_x, batch_y in train_loader:
batch_x, batch_y = batch_x.to(device), batch_y.to(device)
optimizer.zero_grad()
with autocast():
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
scaler.scale(loss).backward()
scaler.step(optimizer)
scaler.update()
scheduler.step()
total_loss += loss.item()
return total_loss / len(train_loader)
# Evaluation function
def evaluate_model(model, test_loader, criterion, device):
model.eval()
total_loss = 0
with torch.no_grad():
for batch_x, batch_y in test_loader:
batch_x, batch_y = batch_x.to(device), batch_y.to(device)
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
total_loss += loss.item()
return total_loss / len(test_loader)
# Main training loop
def train_and_predict(품목_리스트):
품목별_predictions = {}
품목별_scalers = {}
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
pbar_outer = tqdm(품목_리스트, desc="품목 처리 중", position=0)
for 품목명 in pbar_outer:
pbar_outer.set_description(f"품목별 전처리 및 모델 학습 -> {품목명}")
# Data preprocessing
train_data, scaler = process_all_items_with_full_conditions("./train/train.csv",
"./train/meta/TRAIN_산지공판장_2018-2021.csv",
"./train/meta/TRAIN_전국도매_2018-2021.csv",
품목명)
품목별_scalers[품목명] = scaler
# Dataset and DataLoader setup
dataset = AgriculturePriceDataset(train_data)
train_data, val_data = train_test_split(dataset, test_size=0.2, random_state=42)
train_loader = DataLoader(train_data, CFG.batch_size, shuffle=True, num_workers=4)
val_loader = DataLoader(val_data, CFG.batch_size, shuffle=False, num_workers=4)
# Model setup
input_size = len(dataset.numeric_columns)
model = TimeSeriesTransformer(
input_size=input_size,
hidden_size=CFG.hidden_size,
num_layers=CFG.num_layers,
num_heads=CFG.num_heads,
output_size=CFG.output_size,
dropout=CFG.dropout
).to(device)
# Training setup
criterion = nn.HuberLoss()
optimizer = torch.optim.AdamW(model.parameters(), lr=CFG.learning_rate, weight_decay=CFG.weight_decay)
scheduler = CosineAnnealingWarmRestarts(optimizer, T_0=10, T_mult=2)
scaler = GradScaler()
# Training loop
best_val_loss = float('inf')
os.makedirs('models', exist_ok=True)
for epoch in range(CFG.epoch):
train_loss = train_model(model, train_loader, criterion, optimizer, scheduler, scaler, device)
val_loss = evaluate_model(model, val_loader, criterion, device)
if val_loss < best_val_loss:
best_val_loss = val_loss
torch.save(model.state_dict(), f'models/best_model_{품목명}.pth')
print(f'Epoch {epoch+1}/{CFG.epoch}, Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}')
# Inference
품목_predictions = []
pbar_inner = tqdm(range(25), desc="테스트 파일 추론 중", position=1, leave=False)
for i in pbar_inner:
test_file = f"./test/TEST_{i:02d}.csv"
산지공판장_file = f"./test/meta/TEST_산지공판장_{i:02d}.csv"
전국도매_file = f"./test/meta/TEST_전국도매_{i:02d}.csv"
test_data, _ = process_data(test_file, 산지공판장_file, 전국도매_file, 품목명, scaler=품목별_scalers[품목명])
test_dataset = AgriculturePriceDataset(test_data, is_test=True)
test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False)
model.eval()
predictions = []
with torch.no_grad():
for batch in test_loader:
batch = batch.to(device)
output = model(batch)
predictions.append(output.cpu().numpy())
predictions_array = np.concatenate(predictions)
# Inverse transform predictions
price_column_index = test_data.columns.get_loc(test_dataset.price_column)
predictions_reshaped = predictions_array.reshape(-1, 1)
price_scaler = MinMaxScaler()
price_scaler.min_ = 품목별_scalers[품목명].min_[price_column_index]
price_scaler.scale_ = 품목별_scalers[품목명].scale_[price_column_index]
predictions_original_scale = price_scaler.inverse_transform(predictions_reshaped)
if np.isnan(predictions_original_scale).any():
pbar_inner.set_postfix({"상태": "NaN"})
else:
pbar_inner.set_postfix({"상태": "정상"})
품목_predictions.extend(predictions_original_scale.flatten())
품목별_predictions[품목명] = 품목_predictions
pbar_outer.update(1)
return 품목별_predictions
# Run training and prediction
품목별_predictions = train_and_predict(품목_리스트)
# Prepare submission file
sample_submission = pd.read_csv('./sample_submission.csv')
for 품목명, predictions in 품목별_predictions.items():
sample_submission[품목명] = predictions
# Save results
sample_submission.to_csv('./optimized_submission.csv', index=False)
Time2Vec Layer의 개념
Time2Vec는 기본적으로 시간 데이터를 벡터로 변환하여 입력에 추가하는 레이어이다. 여기서 시간은 단순한 선형적인 변화뿐만 아니라, 주기적 패턴(예: 주기적 이벤트, 계절적 변화)을 포착하는 것이 중요하다. 따라서 Time2Vec는 선형 변환과 **주기적 변환(사인 함수)**을 모두 사용하여 시간을 벡터로 표현한다.
class Time2Vec(nn.Module):
def __init__(self, input_dim):
super(Time2Vec, self).__init__()
# 선형 변환을 위한 레이어: 시간 정보를 1차원으로 변환
self.linear = nn.Linear(input_dim, 1)
# 주기적 변환을 위한 레이어: 시간 정보를 input_dim - 1 크기의 벡터로 변환
self.periodic = nn.Linear(input_dim, input_dim - 1)
def forward(self, x):
# 선형 변환 결과 (일차적으로 시간 값을 그대로 모델에 전달)
linear_out = self.linear(x)
# 주기적 변환 결과 (사인 함수를 통해 시간의 주기성을 반영)
periodic_out = torch.sin(self.periodic(x))
# 선형 변환 결과와 주기적 변환 결과를 결합하여 최종 Time2Vec 출력 생성
return torch.cat([linear_out, periodic_out], dim=-1)Time2Vec Layer의 역할
- Time2Vec 레이어는 시간 정보를 더 풍부하게 표현하는 역할을 한다. 시계열 데이터의 시간 정보는 단순한 숫자(예: 시간, 일, 분 등)일 수 있지만, 이러한 시간 정보는 주기적 변화(예: 하루, 일주일, 계절 등)와 관련이 있을 수 있다. Time2Vec는 이러한 주기적 패턴을 모델이 학습할 수 있도록 한다.
- 선형 변환: 시간을 단순히 선형적으로 변환하는 부분이다. 이 부분은 시간의 일차적인 흐름을 표현한다. 시간의 흐름을 그대로 모델에 전달하는 역할을 한다.
- 주기적 변환: 사인 함수를 사용하여 주기성을 반영하는 부분이다. 예를 들어, 하루의 시간대나 계절적인 변화와 같은 주기적인 특성을 포착할 수 있다. 이 부분은 주기적인 패턴을 학습하는 데 유용하다.
forward 함수 -> 모델의 "순전파" 정의
TransformerBlock
TransformerBlock은 트랜스포머 아키텍처에서 중요한 요소 중 하나로, 기본적으로 입력 시퀀스의 관계를 학습하여 중요한 정보를 추출하는 역할을 한다. 이 블록은 멀티헤드 어텐션, 레이어 정규화, 그리고 피드포워드 네트워크로 구성되어 있다. 트랜스포머는 이러한 구조를 통해 입력 데이터의 패턴을 효과적으로 학습하고, 시퀀스 내의 장기적인 의존성을 파악할 수 있다.
class TransformerBlock(nn.Module):
def __init__(self, input_dim, num_heads, dropout):
super(TransformerBlock, self).__init__()
# 멀티헤드 어텐션 모듈
self.attention = nn.MultiheadAttention(input_dim, num_heads, dropout=dropout)
# 어텐션 이후의 첫 번째 레이어 정규화
self.norm1 = nn.LayerNorm(input_dim)
# 피드포워드 네트워크 이후의 두 번째 레이어 정규화
self.norm2 = nn.LayerNorm(input_dim)
# 피드포워드 네트워크 (두 개의 선형 레이어와 활성화 함수 GELU로 구성)
self.ff = nn.Sequential(
nn.Linear(input_dim, 4 * input_dim), # 첫 번째 선형 레이어
nn.GELU(), # 활성화 함수
nn.Linear(4 * input_dim, input_dim), # 두 번째 선형 레이어
nn.Dropout(dropout) # 드롭아웃으로 과적합 방지
)
def forward(self, x):
# 1. 어텐션 레이어
# Q(쿼리), K(키), V(밸류) 모두 같은 x를 사용하여 멀티헤드 어텐션 수행
attended, _ = self.attention(x, x, x)
# 2. 어텐션 결과와 입력을 더하고, 레이어 정규화 적용
x = self.norm1(attended + x)
# 3. 피드포워드 네트워크를 통과한 결과와 다시 입력을 더하고, 레이어 정규화 적용
feedforward = self.ff(x)
return self.norm2(feedforward + x)self.attention = nn.MultiheadAttention(input_dim, num_heads, dropout=dropout)
- 멀티헤드 어텐션은 트랜스포머 모델의 핵심 개념으로, 입력 시퀀스 내의 각 요소가 다른 요소와 어떻게 관계를 맺고 있는지 학습하는 메커니즘이다.
- 쿼리(Q), 키(K), 밸류(V) 세 가지 요소를 기반으로 어텐션 스코어를 계산하며, Q, K, V는 모두 동일한 입력 데이터(x)에서 생성된다.
- 멀티헤드 어텐션에서 **쿼리(Q)**는 "내가 다른 토큰들과 얼마나 관련이 있는가?"를 묻는 역할, **키(K)**는 "내가 그 질문과 얼마나 일치하는가?"에 대한 답변, **밸류(V)**는 그 토큰의 실제 정보를 나타낸다. 쿼리와 키를 사용해 어텐션 스코어를 계산하고, 이 스코어로 밸류를 가중합하여 최종 출력을 생성한다. 멀티헤드 어텐션은 이를 여러 번 병렬로 수행해 다양한 패턴을 학습할 수 있게 한다.
- 여러 개의 어텐션 헤드를 사용하여, 데이터 내의 다양한 관계(패턴)를 병렬로 학습할 수 있다.
- 각 헤드는 독립적으로 어텐션을 계산한 후, 이를 결합하여 최종 결과를 도출한다. 이는 다양한 각도에서 데이터 관계를 학습할 수 있도록 도와준다.
- dropout을 사용하여 모델의 과적합을 방지한다.
self.norm1 = nn.LayerNorm(input_dim)
self.norm2 = nn.LayerNorm(input_dim)
- 레이어 정규화는 각 층의 출력이 일정한 분포를 유지할 수 있도록 정규화하는 과정이다. 이는 학습을 더 안정적으로 만들어 주며, 네트워크가 더 잘 수렴하도록 돕는다.
- self.norm1은 멀티헤드 어텐션 블록에서 나온 결과와 원본 입력을 더한 값에 대해 적용된다. 이는 Residual Connection(잔차 연결) 후에 바로 적용된다.
- self.norm2는 피드포워드 네트워크를 통과한 결과에 적용된다.
self.ff = nn.Sequential(
nn.Linear(input_dim, 4 * input_dim), # 첫 번째 선형 레이어
nn.GELU(), # 활성화 함수
nn.Linear(4 * input_dim, input_dim), # 두 번째 선형 레이어
nn.Dropout(dropout) # 드롭아웃으로 과적합 방지
)- *피드포워드 네트워크(Feedforward Neural Network, FFN)**는 트랜스포머 모델에서 입력 데이터가 한 방향으로만 흐르며 레이어를 통과해 최종 출력을 계산하는 네트워크
- 피드포워드 네트워크는 두 개의 선형 레이어로 구성되어 있으며, 중간에 GELU라는 활성화 함수가 사용된다.
- 첫 번째 선형 레이어는 입력 차원(input_dim)을 4배로 확장한 후, GELU 활성화 함수를 거친다.
- 두 번째 선형 레이어는 다시 원래의 차원(input_dim)으로 줄어들게 된다.
- 이 과정은 입력 데이터에 대해 비선형 변환을 적용하여, 더 복잡한 패턴을 학습할 수 있도록 돕는다.
- Dropout은 네트워크가 과적합되지 않도록 일부 뉴런을 무작위로 학습하지 않도록 하는 정규화 기법이다.
TimeSeriesTransformer
TimeSeriesTransformer는 시계열 데이터를 처리하여 미래의 가격을 예측하는 모델로, Time2Vec 레이어, 트랜스포머 블록들, 그리고 **위치 인코딩(Position Encoding)**을 사용하여 모델이 시간의 흐름과 시퀀스 내 패턴을 학습하도록 설계되어 있다. 이 모델은 주로 금융 데이터나 농산물 가격 예측과 같은 시계열 데이터에 유용하다.
class TimeSeriesTransformer(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, num_heads, output_size, dropout):
super(TimeSeriesTransformer, self).__init__()
# Time2Vec 레이어
self.time2vec = Time2Vec(input_size)
# 입력 데이터를 임베딩(차원 축소)하는 선형 레이어
self.embedding = nn.Linear(input_size, hidden_size)
# 위치 인코딩을 생성하는 함수 호출
self.position_encoding = self.generate_position_encoding(hidden_size, 10)
# 드롭아웃 레이어 (과적합 방지)
self.dropout = nn.Dropout(dropout)
# 트랜스포머 블록 여러 개 쌓기 (ModuleList를 사용하여 리스트로 생성)
self.transformer_blocks = nn.ModuleList([
TransformerBlock(hidden_size, num_heads, dropout)
for _ in range(num_layers)
])
# 출력층: 예측을 위한 선형 레이어와 활성화 함수
self.output_layer = nn.Sequential(
nn.Linear(hidden_size, hidden_size // 2),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_size // 2, output_size)
)
def generate_position_encoding(self, hidden_size, max_len):
# 위치 인코딩 생성
pe = torch.zeros(max_len, hidden_size)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(torch.arange(0, hidden_size, 2).float() * (-np.log(10000.0) / hidden_size))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
return pe.unsqueeze(0)
def forward(self, x):
# 입력 데이터의 배치 크기(b), 시퀀스 길이(s), 특성 수(f)를 추출
b, s, f = x.shape
# 1. Time2Vec 레이어로 시간 정보를 추가한 벡터 생성
x = self.time2vec(x)
# 2. 입력 데이터를 임베딩 (차원 축소)
x = self.embedding(x)
# 3. 위치 인코딩을 더해줌으로써 시계열 데이터의 순서 정보를 추가
x = x + self.position_encoding[:, :s, :].to(x.device)
# 4. 드롭아웃을 적용하여 과적합 방지
x = self.dropout(x)
# 5. 트랜스포머 블록을 여러 번 거침
for transformer in self.transformer_blocks:
x = transformer(x)
# 6. 시퀀스의 평균값을 계산하여 하나의 벡터로 변환 (차원 축소)
x = x.mean(dim=1)
# 7. 최종 출력 레이어를 통해 예측값 도출
return self.output_layer(x)주요 구성 요소 설명
- Time2Vec 레이어
- 시계열 데이터에서 시간을 벡터로 변환하는 역할을 한다.
- 선형 변환과 주기적 변환(사인 함수)을 사용하여 시간 정보를 더 효과적으로 모델에 반영한다.
self.time2vec = Time2Vec(input_size)
2. Embedding Layer (임베딩 레이어)
연속적인 벡터로 변환
- 입력 데이터를 차원 축소하여 모델이 더 쉽게 처리할 수 있도록 한다.
- 입력의 크기인 input_size에서 모델의 내부 크기인 hidden_size로 변환하는 역할을 한다.
self.embedding = nn.Linear(input_size, hidden_size)
3.Position Encoding (위치 인코딩)
- 시계열 데이터에서 각 데이터 포인트가 시간적으로 어디에 위치해 있는지를 모델이 인식할 수 있도록 돕는 기법이다.
- generate_position_encoding 함수는 **사인(sin)**과 코사인(cos) 함수를 사용하여 위치 인코딩을 계산한다.
def generate_position_encoding(self, hidden_size, max_len):
pe = torch.zeros(max_len, hidden_size)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(torch.arange(0, hidden_size, 2).float() * (-np.log(10000.0) / hidden_size))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
return pe.unsqueeze(0)
여기서 position은 시퀀스 내의 각 요소의 위치이며, 이를 사인과 코사인으로 변환해 위치 정보를 벡터화한다. 이 정보를 모델에 더해줌으로써 모델이 시퀀스의 순서 정보를 학습할 수 있다.
- 이렇게 계산된 위치 인코딩은 입력 데이터 x에 더해진다:
div_term = torch.exp(torch.arange(0, hidden_size, 2).float() * (-np.log(10000.0) / hidden_size))*위치 인코딩(position encoding)**을 생성하는 과정에서, 각 위치에 고유한 값을 부여하기 위해 사용되는 수식이다.
x = x + self.position_encoding[:, :s, :].to(x.device)
4. Transformer Blocks (트랜스포머 블록)
- 여러 개의 TransformerBlock을 쌓아서, 시퀀스 내 패턴을 학습한다. 각 블록은 멀티헤드 어텐션과 피드포워드 네트워크로 구성되어 있으며, 입력 데이터를 반복적으로 처리한다.
- ModuleList를 사용하여 여러 개의 트랜스포머 블록을 정의한다.
self.transformer_blocks = nn.ModuleList([
TransformerBlock(hidden_size, num_heads, dropout)
for _ in range(num_layers)
])
forward 함수에서는 각 트랜스포머 블록을 반복적으로 적용하여 입력 데이터를 처리한다:
for transformer in self.transformer_blocks:
x = transformer(x)
5. Global Average Pooling (시퀀스의 평균 계산)
- 트랜스포머 블록을 거친 후, 각 시퀀스의 요소들에 대해 평균을 계산하여 하나의 벡터로 변환한다. 이는 시퀀스의 중요한 정보를 하나의 벡터에 모으는 역할을 한다.
x = x.mean(dim=1)
6. Output Layer (출력 레이어)
- 최종 예측을 위한 출력층으로, ReLU 활성화 함수와 드롭아웃을 거친 후, 최종적으로 output_size 크기의 예측 값을 도출한다.
self.output_layer = nn.Sequential(
nn.Linear(hidden_size, hidden_size // 2),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_size // 2, output_size)
)
AgriculturePriceDataset (데이터 분할)
AgriculturePriceDataset 클래스는 농산물 가격 예측을 위한 데이터를 생성하고, 훈련 데이터와 테스트 데이터를 다르게 처리하여 모델에 입력할 수 있도록 하는 사용자 정의 데이터셋 클래스이다. 주어진 시계열 데이터를 일정한 크기(윈도우 크기)로 나누어 모델이 학습할 수 있도록 준비하며, 예측하려는 목표 값(타겟 값)도 함께 제공한다.
class AgriculturePriceDataset(Dataset):
def __init__(self, dataframe, window_size=9, prediction_length=3, is_test=False):
# 데이터프레임과 윈도우 크기, 예측 길이, 테스트 모드 여부를 받아 초기화
self.data = dataframe
self.window_size = window_size
self.prediction_length = prediction_length
self.is_test = is_test
# 가격 정보가 포함된 컬럼 찾기
self.price_column = [col for col in self.data.columns if '평균가격(원)' in col and len(col.split('_')) == 1][0]
# 숫자형 데이터를 포함한 모든 열을 리스트로 저장
self.numeric_columns = self.data.select_dtypes(include=[np.number]).columns.tolist()
# 데이터셋 생성 (훈련용 또는 테스트용 데이터에 따라 다르게 처리)
self.sequences = []
if not self.is_test:
# 훈련 데이터일 경우
for i in range(len(self.data) - self.window_size - self.prediction_length + 1):
x = self.data[self.numeric_columns].iloc[i:i+self.window_size].values # 입력 데이터 (x)
y = self.data[self.price_column].iloc[i+self.window_size:i+self.window_size+self.prediction_length].values # 예측할 타겟 값 (y)
self.sequences.append((x, y))
else:
# 테스트 데이터일 경우 전체 데이터를 시퀀스로 처리
self.sequences = [self.data[self.numeric_columns].values]
def __len__(self):
# 데이터셋의 길이를 반환 (훈련 데이터의 경우 시퀀스의 개수, 테스트 데이터는 하나의 시퀀스)
return len(self.sequences)
def __getitem__(self, idx):
# 데이터셋에서 인덱스를 기준으로 샘플을 반환
if not self.is_test:
x, y = self.sequences[idx]
return torch.FloatTensor(x), torch.FloatTensor(y)
else:
return torch.FloatTensor(self.sequences[idx])
주요 구성 요소 설명
1. __init__ 메서드 (초기화)
def __init__(self, dataframe, window_size=9, prediction_length=3, is_test=False):
# 데이터프레임과 윈도우 크기, 예측 길이, 테스트 모드 여부를 받아 초기화
self.data = dataframe
self.window_size = window_size
self.prediction_length = prediction_length
self.is_test = is_test
- dataframe: 학습 또는 테스트에 사용할 데이터가 포함된 pandas 데이터프레임이다. 이 데이터프레임에는 다양한 특성과 농산물의 가격 데이터가 포함되어 있다.
- window_size: 입력 데이터의 시퀀스 길이를 나타낸다. 예를 들어, window_size=9이면 과거 9일치 데이터를 입력으로 사용하여 이후 며칠간의 가격을 예측한다.
- prediction_length: 예측할 타겟 값(미래의 가격)의 길이를 의미한다. 예를 들어, prediction_length=3이면 3일 후의 가격을 예측한다.
- is_test: 테스트 데이터인지 여부를 나타내는 플래그다. 테스트 데이터일 경우, 타겟 값이 없기 때문에 시퀀스만 생성한다.
self.price_column = [col for col in self.data.columns if '평균가격(원)' in col and len(col.split('_')) == 1][0]
self.numeric_columns = self.data.select_dtypes(include=[np.number]).columns.tolist()
- price_column: 데이터프레임에서 평균 가격 정보를 포함하는 컬럼을 자동으로 찾아 설정한다. 이 컬럼이 타겟 값이 된다.
- numeric_columns: 데이터프레임에서 숫자형 데이터를 포함하는 모든 열을 리스트로 저장한다. 이 열들이 모델의 입력으로 사용된다.
# 데이터셋 생성 (훈련용 또는 테스트용 데이터에 따라 다르게 처리)
self.sequences = []
if not self.is_test:
# 훈련 데이터일 경우
for i in range(len(self.data) - self.window_size - self.prediction_length + 1):
x = self.data[self.numeric_columns].iloc[i:i+self.window_size].values # 입력 데이터 (x)
y = self.data[self.price_column].iloc[i+self.window_size:i+self.window_size+self.prediction_length].values # 예측할 타겟 값 (y)
self.sequences.append((x, y))
- 입력 데이터(x): window_size 크기만큼 데이터를 슬라이딩 윈도우 방식으로 자른다. 각 시퀀스는 self.numeric_columns에 포함된 숫자형 데이터를 사용하여 입력으로 처리된다.
- 타겟 값(y): prediction_length 크기만큼 이후의 가격 데이터를 타겟 값으로 설정한다.
- 이 과정을 통해 훈련 데이터는 입력과 타겟 값이 쌍으로 저장된 시퀀스로 구성된다.
else:
# 테스트 데이터일 경우 전체 데이터를 시퀀스로 처리
self.sequences = [self.data[self.numeric_columns].values]
테스트 데이터일 경우 타겟 값은 필요하지 않기 때문에, 전체 데이터를 시퀀스로 저장한다. 여기서는 테스트 데이터를 예측하기 위한 입력만 준비한다.
def __len__(self):
# 데이터셋의 길이를 반환 (훈련 데이터의 경우 시퀀스의 개수, 테스트 데이터는 하나의 시퀀스)
return len(self.sequences)
데이터셋의 길이를 반환한다. 훈련 데이터일 경우, 생성된 시퀀스의 개수를 반환하며, 테스트 데이터는 하나의 시퀀스만 존재할 수 있다.
def __getitem__(self, idx):
if not self.is_test:
x, y = self.sequences[idx]
return torch.FloatTensor(x), torch.FloatTensor(y)
else:
return torch.FloatTensor(self.sequences[idx])
- 데이터셋에서 주어진 인덱스 idx에 해당하는 데이터를 반환한다.
- 훈련 데이터: 입력 시퀀스 x와 타겟 값 y를 반환한다.
- 테스트 데이터: 타겟 값 없이 입력 시퀀스 x만 반환한다.
- 반환되는 값은 torch.FloatTensor로 변환되어 PyTorch 모델에서 사용할 수 있는 형태로 제공된다.
요약
- **AgriculturePriceDataset**은 주어진 데이터에서 일정한 기간 동안의 입력과 그 이후의 타겟 값을 쌍으로 만들어 모델 학습에 필요한 데이터를 제공한다.
- 훈련 데이터와 테스트 데이터를 다르게 처리하며, 훈련 데이터의 경우 입력과 타겟을, 테스트 데이터의 경우 입력만을 반환한다.
- 이를 통해 시계열 데이터를 모델에 효율적으로 제공할 수 있으며, 특히 농산물 가격 예측과 같은 시계열 예측 문제에서 사용된다.
5. Training and Evaluation
train_model 함수는 모델을 훈련시키고, **혼합 정밀도 학습(Mixed Precision Training)**을 사용하여 메모리 사용량을 최적화하고 훈련 속도를 향상시킨다.
evaluate_model 함수는 검증 데이터셋을 사용하여 모델의 성능을 평가하는 역할을 한다.
train_model 함수
이 함수는 모델을 학습시키기 위해 배치 단위로 데이터를 받아, 모델의 예측 결과와 실제 값의 차이를 기반으로 손실을 계산한 후, 이를 바탕으로 모델의 가중치를 업데이트하는 과정이다.
# Training function with mixed precision training
def train_model(model, train_loader, criterion, optimizer, scheduler, scaler, device):
model.train() # 모델을 학습 모드로 설정
total_loss = 0 # 총 손실 초기화
# 미니배치 학습
for batch_x, batch_y in train_loader:
# 배치를 GPU 또는 CPU로 이동
batch_x, batch_y = batch_x.to(device), batch_y.to(device)
# 옵티마이저의 기울기 초기화
optimizer.zero_grad()
# 혼합 정밀도 학습을 위한 autocast 사용
with autocast():
# 모델을 통해 예측값 생성
outputs = model(batch_x)
# 예측값과 실제 값의 손실 계산
loss = criterion(outputs, batch_y)
# 손실을 역전파 (backpropagation) 하여 모델의 가중치 업데이트
scaler.scale(loss).backward() # 손실 값에 대한 스케일링 적용
scaler.step(optimizer) # 스케일링된 손실 값으로 옵티마이저를 사용해 가중치 업데이트
scaler.update() # 스케일러 업데이트
# 스케줄러를 사용해 학습률 조정
scheduler.step()
# 배치의 손실을 총 손실에 더함
total_loss += loss.item()
# 평균 손실 반환
return total_loss / len(train_loader)
설명
- model.train():
모델을 학습 모드로 전환한다. 이 설정은 드롭아웃과 배치 정규화 등의 레이어가 학습 모드에서 동작하도록 한다. - 데이터 로드 및 장치로 이동:
미니배치 단위로 데이터를 train_loader에서 받아, device(GPU 또는 CPU)로 이동시킨다.
batch_x, batch_y = batch_x.to(device), batch_y.to(device)
3.혼합 정밀도 학습 (autocast):
**autocast()**는 혼합 정밀도 학습을 가능하게 하는 컨텍스트 매니저이다. 이를 통해 일부 연산이 FP16(반 정밀도)로 처리되어 메모리 효율성과 학습 속도가 향상된다.
with autocast():
outputs = model(batch_x) # 모델 예측
loss = criterion(outputs, batch_y) # 손실 계산
4.손실 역전파 및 옵티마이저 업데이트:
scaler.scale(loss)는 손실 값을 스케일링하여, 낮은 정밀도의 연산에서 발생할 수 있는 수치 불안정을 방지한다. 이후 backward()를 통해 손실을 역전파하고, 옵티마이저로 가중치를 업데이트한다.
손실역전파 : 딥러닝 모델에서 손실 함수를 기준으로 가중치를 업데이트하기 위해 사용하는 알고리즘
scaler.scale(loss).backward() # 손실 역전파
scaler.step(optimizer) # 옵티마이저 업데이트
scaler.update() # 스케일러 업데이트
학습률 스케줄링:
스케줄러는 학습률을 동적으로 조정하는 역할을 한다. 여기서는 CosineAnnealingWarmRestarts 같은 스케줄러를 사용해 학습률을 조정하며, 매 미니배치마다 학습률을 업데이트한다.
스케줄러 : 딥러닝 모델을 학습할 때 **학습률(learning rate)**을 동적으로 조절하는 방법을 말한다. 학습률은 가중치 업데이트의 크기를 결정하는 중요한 하이퍼파라미터로, 학습이 진행됨에 따라 학습률을 변화시키면 더 좋은 성능을 얻을 수 있다. 스케줄러는 학습 과정 중에 학습률을 어떻게 변화시킬지를 제어하는 역할
scheduler.step()
손실 계산:
각 미니배치의 손실을 누적하여, 총 손실을 계산하고 평균 손실을 반환한다.
return total_loss / len(train_loader)
evaluate_model 함수
evaluate_model 함수는 학습된 모델이 검증 데이터셋에서 어떻게 성능을 발휘하는지 평가하는 함수다. 이 함수는 모델을 평가 모드로 전환하고, 검증 데이터셋에서 예측을 수행한 뒤 손실 값을 계산한다.
# Evaluation function
def evaluate_model(model, test_loader, criterion, device):
model.eval() # 모델을 평가 모드로 설정
total_loss = 0 # 총 손실 초기화
with torch.no_grad(): # 평가에서는 기울기 계산을 하지 않음
for batch_x, batch_y in test_loader:
# 배치를 GPU 또는 CPU로 이동
batch_x, batch_y = batch_x.to(device), batch_y.to(device)
# 모델을 통해 예측값 생성
outputs = model(batch_x)
# 예측값과 실제 값의 손실 계산
loss = criterion(outputs, batch_y)
# 배치의 손실을 총 손실에 더함
total_loss += loss.item()
# 평균 손실 반환
return total_loss / len(test_loader)
설명
- model.eval():
모델을 평가 모드로 전환한다. 이 모드는 드롭아웃과 배치 정규화 레이어가 평가 모드에서 동작하도록 하며, 학습 시에만 필요한 기울기 계산을 하지 않는다. - 기울기 비활성화 (torch.no_grad()):
평가 과정에서는 기울기 계산이 필요 없으므로 **torch.no_grad()**를 사용하여 메모리 사용량을 줄이고, 계산 속도를 높인다.
with torch.no_grad():
데이터 로드 및 장치로 이동:
검증 데이터(test_loader)에서 배치를 로드하여, device로 이동시킨다.
batch_x, batch_y = batch_x.to(device), batch_y.to(device)
모델 예측 및 손실 계산:
모델의 예측 값을 생성한 후, 실제 값과의 손실을 계산하여 누적한다.
outputs = model(batch_x) # 예측값 생성
loss = criterion(outputs, batch_y) # 손실 계산
평균 손실 계산:
각 미니배치의 손실을 모두 더한 후, 검증 데이터 전체에 대한 평균 손실을 반환한다.
return total_loss / len(test_loader)
train_and_predict 함수
train_and_predict 함수는 농산물 품목 리스트에 대해 데이터를 전처리한 후, 각 품목별로 모델을 학습시키고, 테스트 데이터를 기반으로 가격을 예측하는 전체적인 학습 및 예측 파이프라인을 정의한 함수이다. 각 품목에 대해 모델을 학습시키며, 최적의 모델을 저장하고, 이를 바탕으로 예측 값을 생성한다.
# Main training loop
def train_and_predict(품목_리스트):
품목별_predictions = {} # 각 품목에 대한 예측 결과 저장
품목별_scalers = {} # 각 품목별로 사용된 스케일러 저장
device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # GPU 사용 가능 시 CUDA 사용
pbar_outer = tqdm(품목_리스트, desc="품목 처리 중", position=0)
# 품목 리스트에 대해 반복
for 품목명 in pbar_outer:
pbar_outer.set_description(f"품목별 전처리 및 모델 학습 -> {품목명}")
# 1. 데이터 전처리 (품목별로 데이터 로드 및 스케일링 처리)
train_data, scaler = process_all_items_with_full_conditions(
"./train/train.csv",
"./train/meta/TRAIN_산지공판장_2018-2021.csv",
"./train/meta/TRAIN_전국도매_2018-2021.csv",
품목명
)
품목별_scalers[품목명] = scaler # 각 품목별로 스케일러 저장
# 2. Dataset 및 DataLoader 설정
dataset = AgriculturePriceDataset(train_data) # 전처리된 데이터를 Dataset으로 변환
train_data, val_data = train_test_split(dataset, test_size=0.2, random_state=42) # 학습/검증 데이터 분리
train_loader = DataLoader(train_data, CFG.batch_size, shuffle=True, num_workers=4) # 학습 데이터 로더
val_loader = DataLoader(val_data, CFG.batch_size, shuffle=False, num_workers=4) # 검증 데이터 로더
# 3. 모델 설정
input_size = len(dataset.numeric_columns) # 입력 데이터의 크기
model = TimeSeriesTransformer(
input_size=input_size,
hidden_size=CFG.hidden_size,
num_layers=CFG.num_layers,
num_heads=CFG.num_heads,
output_size=CFG.output_size,
dropout=CFG.dropout
).to(device) # 모델을 GPU 또는 CPU로 이동
# 4. 손실 함수, 옵티마이저, 스케줄러 설정
criterion = nn.HuberLoss() # 손실 함수: HuberLoss
optimizer = torch.optim.AdamW(model.parameters(), lr=CFG.learning_rate, weight_decay=CFG.weight_decay) # AdamW 옵티마이저
scheduler = CosineAnnealingWarmRestarts(optimizer, T_0=10, T_mult=2) # 학습률 스케줄러: CosineAnnealingWarmRestarts
scaler = GradScaler() # 혼합 정밀도 학습을 위한 GradScaler
# 5. 학습 루프
best_val_loss = float('inf') # 초기 최저 검증 손실 설정
os.makedirs('models', exist_ok=True) # 모델 저장을 위한 디렉토리 생성
for epoch in range(CFG.epoch): # 에포크 반복
train_loss = train_model(model, train_loader, criterion, optimizer, scheduler, scaler, device) # 학습
val_loss = evaluate_model(model, val_loader, criterion, device) # 검증
# 최적의 검증 손실을 기록하고, 해당 모델 저장
if val_loss < best_val_loss:
best_val_loss = val_loss
torch.save(model.state_dict(), f'models/best_model_{품목명}.pth') # 최적의 모델 저장
print(f'Epoch {epoch+1}/{CFG.epoch}, Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}')
# 6. 테스트 데이터에 대한 추론 (Inference)
품목_predictions = [] # 각 품목에 대한 예측 결과 저장
pbar_inner = tqdm(range(25), desc="테스트 파일 추론 중", position=1, leave=False)
for i in pbar_inner:
# 테스트 데이터 파일 로드
test_file = f"./test/TEST_{i:02d}.csv"
산지공판장_file = f"./test/meta/TEST_산지공판장_{i:02d}.csv"
전국도매_file = f"./test/meta/TEST_전국도매_{i:02d}.csv"
# 7. 테스트 데이터 전처리
test_data, _ = process_data(test_file, 산지공판장_file, 전국도매_file, 품목명, scaler=품목별_scalers[품목명])
test_dataset = AgriculturePriceDataset(test_data, is_test=True) # 테스트 데이터셋 생성
test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False) # 테스트 데이터 로더
model.eval() # 모델을 평가 모드로 전환
predictions = []
with torch.no_grad(): # 기울기 계산 비활성화 (추론 시에는 불필요)
for batch in test_loader:
batch = batch.to(device) # 배치를 GPU 또는 CPU로 이동
output = model(batch) # 모델로부터 예측값 생성
predictions.append(output.cpu().numpy()) # 예측 결과 저장
predictions_array = np.concatenate(predictions) # 예측 결과를 배열로 변환
# 8. 예측값을 원래 스케일로 역변환
price_column_index = test_data.columns.get_loc(test_dataset.price_column) # 가격 컬럼의 인덱스 찾기
predictions_reshaped = predictions_array.reshape(-1, 1)
price_scaler = MinMaxScaler() # 가격을 역변환할 스케일러
price_scaler.min_ = 품목별_scalers[품목명].min_[price_column_index] # 저장된 스케일러 값 복원
price_scaler.scale_ = 품목별_scalers[품목명].scale_[price_column_index]
predictions_original_scale = price_scaler.inverse_transform(predictions_reshaped) # 원래 스케일로 복원
if np.isnan(predictions_original_scale).any():
pbar_inner.set_postfix({"상태": "NaN"}) # NaN이 발생하면 상태 출력
else:
pbar_inner.set_postfix({"상태": "정상"}) # 정상일 경우 상태 출력
품목_predictions.extend(predictions_original_scale.flatten()) # 예측 결과 저장
품목별_predictions[품목명] = 품목_predictions # 품목별 예측 결과 저장
pbar_outer.update(1) # 진행 바 업데이트
return 품목별_predictions # 모든 품목에 대한 예측 결과 반환
전처리 및 데이터 로딩:
- process_all_items_with_full_conditions 함수를 사용해 훈련 데이터를 전처리하고, AgriculturePriceDataset을 통해 데이터를 PyTorch의 Dataset으로 변환한다.
- train_test_split을 사용해 훈련 데이터와 검증 데이터를 나눈 후, 이를 DataLoader로 감싸서 배치 단위로 학습에 사용할 수 있도록 준비한다.
train_data, scaler = process_all_items_with_full_conditions(...)
dataset = AgriculturePriceDataset(train_data)
train_data, val_data = train_test_split(dataset, test_size=0.2, random_state=42)
train_loader = DataLoader(train_data, ...)
val_loader = DataLoader(val_data, ...)
모델 학습 및 검증:
- TimeSeriesTransformer 모델을 설정한 후, 학습 과정에서 사용될 손실 함수(HuberLoss), 옵티마이저(AdamW), 스케줄러(CosineAnnealingWarmRestarts), 그리고 혼합 정밀도 학습을 위한 GradScaler를 정의한다.
- 매 에포크마다 train_model 함수를 사용해 학습을 진행하고, evaluate_model 함수를 사용해 검증 데이터를 평가한다. 검증 손실이 가장 낮을 때 모델을 저장한다.
model = TimeSeriesTransformer(...)
criterion = nn.HuberLoss()
optimizer = torch.optim.AdamW(model.parameters(), lr=CFG.learning_rate, weight_decay=CFG.weight_decay)
scheduler = CosineAnnealingWarmRestarts(optimizer, T_0=10, T_mult=2)
scaler = GradScaler()
for epoch in range(CFG.epoch):
train_loss = train_model(model, train_loader, ...)
val_loss = evaluate_model(model, val_loader, ...)
if val_loss < best_val_loss:
torch.save(model.state_dict(), f'models/best_model_{품목명}.pth')
테스트 데이터 추론 및 예측 값 저장:
- 테스트 데이터 파일을 로드한 후, process_data로 전처리하여 테스트 데이터를 생성한다.
- model.eval()을 호출하여 모델을 평가 모드로 전환한 후, 기울기 계산을 비활성화(torch.no_grad())하고, 테스트 데이터에 대한 예측을 수행한다.
- 예측된 값을 원래 스케일로 복원한 후, 최종 예측 값을 저장한다.
model.eval()
with torch.no_grad():
for batch in test_loader:
batch = batch.to(device)
output = model(batch)
predictions.append(output.cpu().numpy())
predictions_array = np.concatenate(predictions)
predictions_original_scale = price_scaler.inverse_transform(predictions_reshaped)
품목_predictions.extend(predictions_original_scale.flatten())
결과 반환:
- 모든 품목에 대한 예측 결과를 품목별_predictions에 저장하고 반환한다.
return 품목별_predictions
1. Time2Vec Layer
- Time2Vec 레이어는 시간 정보를 벡터화하는 역할을 한다. 이는 Linear 변환과 주기적(sin) 변환으로 구성되어 있으며, 시간을 특성으로 입력받아 이를 다양한 주기로 변환하여 모델에 추가적인 정보를 제공한다.
2. Transformer Encoder Block
- TransformerBlock은 기본적인 트랜스포머의 인코더 블록을 정의한다. 멀티헤드 어텐션, 레이어 정규화, 피드포워드 네트워크를 포함하고 있으며, 입력 데이터의 패턴을 학습하여 중요한 정보를 추출한다.
3. Main Model (TimeSeriesTransformer)
- TimeSeriesTransformer 모델은 입력 데이터를 받아 Time2Vec 레이어로 처리한 후, 여러 개의 트랜스포머 블록을 거쳐 최종적으로 가격을 예측한다. 위치 인코딩(position encoding)을 사용하여 시계열 데이터의 순서를 모델이 인식할 수 있도록 한다.
4. Dataset Class (AgriculturePriceDataset)
- AgriculturePriceDataset 클래스는 데이터셋을 생성하는데 사용된다. 주어진 기간 동안의 농산물 가격 데이터를 윈도우 크기로 나누어 모델에 학습 데이터를 제공하며, 훈련용과 테스트용 데이터셋을 다르게 처리한다.
5. Training and Evaluation
- train_model 함수는 모델을 훈련시키는 함수이며, 혼합 정밀도 학습(mixed precision training)을 통해 성능을 최적화한다.
- evaluate_model 함수는 검증 데이터셋을 사용하여 모델의 성능을 평가한다.
6. Main Training Loop
- train_and_predict 함수는 품목 리스트에 대해 데이터를 전처리하고, 각 품목에 대해 모델을 학습시킨 후 테스트 데이터를 추론하여 예측값을 생성한다. 각 품목별로 최적의 모델을 저장하며, 이를 바탕으로 예측값을 저장한다.
7. Prediction and Submission
- 최종적으로, 예측된 값들을 sample_submission 파일에 저장하고 이를 제출 파일로 변환한다.