Deep Learning Foundation Course

# -*- coding: utf-8 -*-
"""lecture07_homework.ipynb
Automatically generated by Colab.
Original file is located at:
    https://colab.research.google.com/drive/1z1UgheoLYhrxhlewI2siinDwSYEhAJz8

Lecture 7 Homework: IMDb Sentiment Analysis
BiGRU + Attention + seed ensemble
ルール厳守：外部学習データ・事前学習Embeddingは使わない
"""

from google.colab import drive
drive.mount('/content/drive')
work_dir = 'drive/MyDrive/Colab Notebooks/DLBasics2026_colab'
!pip install portalocker
import os
import copy
import random
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
from torch.nn.utils.rnn import pad_sequence, pack_padded_sequence, pad_packed_sequence
from torch.nn.utils import clip_grad_norm_
from sklearn.metrics import f1_score
from sklearn.model_selection import train_test_split
from typing import List
seed = 1234
torch.manual_seed(seed)
np.random.seed(seed)
random.seed(seed)

x_train = np.load(work_dir + '/Lecture07/data/x_train.npy', allow_pickle=True)
t_train = np.load(work_dir + '/Lecture07/data/t_train.npy', allow_pickle=True)
x_train, x_valid, t_train, t_valid = train_test_split(
    x_train, t_train, test_size=0.2, random_state=seed
)
x_test = np.load(work_dir + '/Lecture07/data/x_test.npy', allow_pickle=True)
# Hyper Parameters
秘密！
TRAIN_FINAL_ON_ALL = True
# Model
秘密２

def fix_seed(seed: int = 1234):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.benchmark = True

def max_token_id(*arrays):
    m = 0
    for arr in arrays:
        for seq in arr:
            if len(seq) > 0:
                m = max(m, int(np.max(seq)))
    return m

word_num = max_token_id(x_train, x_valid, x_test) + 1
print(f"Vocabulary size: {word_num}")
def text_transform_v2(text, max_length=MAX_LEN):
    text = list(text)
    text = text[:max_length - 1] + [EOS_IDX]
    return text, len(text)
def collate_batch_v2(batch):
    label_list, text_list, len_seq_list = [], [], []
    for sample in batch:
        if isinstance(sample, tuple):
            label, text = sample
            label_list.append(float(label))
        else:
            text = sample
        text, len_seq = text_transform_v2(text)
        text_list.append(torch.tensor(text, dtype=torch.long))
        len_seq_list.append(len_seq)
    labels = torch.tensor(label_list, dtype=torch.float32)
    texts = pad_sequence(text_list, padding_value=PAD_IDX, batch_first=True).long()
    lengths = torch.tensor(len_seq_list, dtype=torch.long)
    return labels, texts, lengths
def make_loader(xs, ts=None, shuffle=False, seed=1234):
    dataset = list(xs) if ts is None else [(t, x) for t, x in zip(ts, xs)]
    generator = torch.Generator()
    generator.manual_seed(seed)
    return DataLoader(
        dataset,
        batch_size=batch_size,
        shuffle=shuffle,
        collate_fn=collate_batch_v2,
        pin_memory=torch.cuda.is_available(),
        generator=generator if shuffle else None,
    )
def get_pos_weight(labels):
    labels = np.asarray(labels).astype(int)
    pos = max(1, int(labels.sum()))
    neg = max(1, int(len(labels) - labels.sum()))
    return torch.tensor([neg / pos], dtype=torch.float32)
def find_best_threshold(y_true, y_prob):
    thresholds = np.arange(0.25, 0.751, 0.005)
    scores = [
        f1_score(y_true, (y_prob >= th).astype(int), average="macro")
        for th in thresholds
    ]
    best_i = int(np.argmax(scores))
    return float(thresholds[best_i]), float(scores[best_i])
def train_one_epoch(model, loader, optimizer, criterion, device):
    model.train()
    losses = []
    for label, line, len_seq in loader:
        x = line.to(device, non_blocking=True)
        t = label.to(device, non_blocking=True)
        lengths = len_seq.to(device, non_blocking=True)
        optimizer.zero_grad(set_to_none=True)
        logits = model(x, None, lengths)
        loss = criterion(logits, t)
        loss.backward()
        clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        losses.append(float(loss.detach().cpu()))
    return float(np.mean(losses))
@torch.no_grad()
def predict_proba(model, loader, device):
    model.eval()
    probs = []
    for _, line, len_seq in loader:
        x = line.to(device, non_blocking=True)
        lengths = len_seq.to(device, non_blocking=True)
        logits = model(x, None, lengths)
        prob = torch.sigmoid(logits)
        probs.append(prob.detach().cpu().numpy())
    return np.concatenate(probs)
@torch.no_grad()
def evaluate(model, loader, criterion, device):
    model.eval()
    losses, y_true, y_prob = [], [], []
    for label, line, len_seq in loader:
        x = line.to(device, non_blocking=True)
        t = label.to(device, non_blocking=True)
        lengths = len_seq.to(device, non_blocking=True)
        logits = model(x, None, lengths)
        loss = criterion(logits, t)
        losses.append(float(loss.detach().cpu()))
        y_true.extend(label.cpu().numpy().astype(int).tolist())
        y_prob.extend(torch.sigmoid(logits).detach().cpu().numpy().tolist())
    y_true = np.asarray(y_true).astype(int)
    y_prob = np.asarray(y_prob)
    th, f1 = find_best_threshold(y_true, y_prob)
    return float(np.mean(losses)), f1, th
# Main Training
# Seed ensemble、threshold tuning、final train、submission保存を実施。
# 詳細なハイパーパラメータとモデル本体は公開範囲に合わせて一部非表示。