学習と評価で利用(15:評価ユーティリティセット)

学習

特徴の正規化・不均衡対策・早期停止・ラベル整形・評価集計を一括で支える、実運用向けの検証ユーティリティ群

今回は、学習・評価フェーズで使う補助ユーティリティのセットです。履歴特徴の正規化、クラス重みの算出、早期終了(過学習抑制)、ラベルの正規化、そして検証時の nDCG@3Top-1 的中率の集計を行います。

def normalize_hist_race(hist_race: torch.Tensor, eps=1e-6):

    hist_race = torch.nan_to_num(hist_race, nan=0.0)

    if torch.max(hist_race) > 0:
        max_val = torch.max(hist_race)
        min_val = torch.min(hist_race)
        if max_val != min_val:
            hist_race = (hist_race - min_val) / (max_val - min_val + eps)
        else:
            hist_race = torch.ones_like(hist_race)
    hist_race = torch.clamp(hist_race, 0.0, 1.0)
    return hist_race

def compute_class_weights(y_ranks):
    counts = Counter(y_ranks)
    total = sum(counts.values())
    weights = {cls: total / count for cls, count in counts.items()}
    return weights
    
class EarlyStopping:
    def __init__(self, patience=3, mode="max"):
        self.patience = patience
        self.counter = 0
        self.best_score = None
        self.early_stop = False
        self.mode = mode  # "max"(大きいほど良い) or "min"

    def __call__(self, current_score):
        if self.best_score is None:
            self.best_score = current_score
            return False

        if (self.mode == "max" and current_score > self.best_score) or \
           (self.mode == "min" and current_score < self.best_score):
            self.best_score = current_score
            self.counter = 0
        else:
            self.counter += 1

        if self.counter >= self.patience:
            self.early_stop = True
            return True

        return False
    
def normalize_targets_for_ranking(y: torch.Tensor):
    """
    y: [S] … 0/1(one-hot, float想定) か 0..K-1 / 1..K の順位(小さいほど良い)
    戻り: (正規化y, 種別)
    """
    # one-hot は「浮動小数の0/1」のときのみ認める
    if y.dtype.is_floating_point:
        if (y.ge(0) & y.le(1)).all() and y.sum() > 0:
            return y, "onehot"

    # それ以外は整数順位扱い → 1始まりに正規化
    y = y.to(torch.long)
    minv = y.min()
    if minv < 1:
        y = y - minv + 1   # 0..K-1 → 1..K
    return y, "rank"

def run_validation(model, val_dataloader, device, class_weights, autocast_ctx):
    model.eval()
    sum_ndcg3, sum_top1_acc, n_races = 0.0, 0.0, 0

    with torch.no_grad():
        for X_pack, y_pack, hist_pack, cat_pack, ids_pack, mask_pack in val_dataloader:
            X_pack = X_pack.to(device, non_blocking=True)
            y_pack = y_pack.to(device, non_blocking=True)
            mask_pack = mask_pack.to(device, non_blocking=True)
            if hist_pack is not None:
                hist_pack = hist_pack.to(device, non_blocking=True)
            cat_pack = {k: v.to(device, non_blocking=True) for k, v in cat_pack.items()}

            with autocast_ctx:
                preds = model(X_pack, cat_pack, mask=mask_pack)  # [B,S]

            valid_mask = ~mask_pack
            # hist が None の時のためのイテレータ
            hist_iter = (hist_pack if hist_pack is not None else [None]*preds.size(0))

            for r_valid, p_r, y_r, h_r in zip(valid_mask, preds, y_pack, hist_iter):
                # レース内の有効頭のみ
                if r_valid.sum() < 2:
                    continue
                p_r = p_r[r_valid]
                y_r = y_r[r_valid]
                y_r, _ = normalize_targets_for_ranking(y_r)

                # NDCG@3(= 1 - loss)
                lndcg3 = hard_ndcg3_loss_torch(p_r, y_r)
                ndcg3 = float(1.0 - lndcg3)

                # Top-1 的中率(参考)
                winners = (y_r == y_r.min()).nonzero(as_tuple=False).squeeze(-1)
                top1_acc = float((torch.argmax(p_r).unsqueeze(0) == winners).any().float())

                sum_ndcg3 += ndcg3
                sum_top1_acc += top1_acc
                n_races += 1

    return (sum_ndcg3 / n_races) if n_races > 0 else 0.0  # ←「大きいほど良い」

1) normalize_hist_race(hist_race, eps=1e-6):履歴スカラーのミニマックス正規化

  • 何をするか
    1. NaN→0 に置換(torch.nan_to_num)。
    2. 値域がある場合は (x−min)/(max−min+eps)[0,1] に正規化、一定値なら 1 に置換。
    3. 最後に clamp(0,1) で範囲外を切り落とし。
  • ねらい
    モデル入力へ渡す補助スカラー(例:履歴長や強度)をスケール不変にし、学習の安定性を高める。
  • 統計学的視点
    ミニマックス変換は単調変換で順序情報を保ちつつ、区間尺度に写像します。特徴量間の単位差を除去し、勾配の安定正則化効果(極端値の影響縮小)を狙えます。

2) compute_class_weights(y_ranks):逆頻度によるクラス重み

  • 何をするか
    クラス出現数 count_c に対し total / count_c を重みとする辞書を返します(希少クラスほど重く)。
  • ねらい
    不均衡データでの損失寄与を平準化し、頻度の少ない順位/クラスも学習可能にする。
  • 統計学的視点
    事後尤度最大化に重みを掛けるのは、コスト感度学習重要度サンプリングの一種と解釈できます(クラス事前分布の補正)。

3) EarlyStopping(patience=3, mode="max"):早期終了(過学習抑制)

  • 何をするか
    検証スコアが 連続 patience 改善しなければ 停止mode は「大きいほど良い/小さいほど良い」を切替え。
  • ねらい
    過学習の兆候(検証指標の悪化/停滞)を検知して計算資源の節約汎化性能の保持
  • 統計学的視点
    時系列の汎化誤差推定に対する逐次的な停止規則。過学習を早期の打ち切りで回避する、実務上の正則化手段です。

4) normalize_targets_for_ranking(y):ラベルの正規化(one-hot or 順位)

  • 何をするか
    • 浮動小数の0/1で合計>0なら one-hot としてそのまま返す。
    • それ以外は整数順位とみなし、最小値を1に揃えて 1..K のランクに正規化。
  • ねらい
    損失側(Top-1/ランキング系)が期待するラベル仕様に整え、実装分岐をシンプルにする。
  • 統計学的視点
    測定尺度を**名義(one-hot)序数(順位)**に正しくマッピングし、適切な尤度(クロスエントロピー/ランキング損失)へ橋渡しします。

5) run_validation(...):検証ループ(nDCG@3 と Top-1)

  • 処理の流れ
    1. model.eval()no_grad() で評価モード。
    2. バッチごとに パディング無視マスクを適用し、レース単位で有効頭だけ抽出。
    3. ラベルを normalize_targets_for_ranking で整形。
    4. nDCG@3hard_ndcg3_loss_torch1 − loss として算出。
    5. Top-1 的中率は「最小ランク群(勝者)のいずれか=argmax(pred) か」を判定。
    6. 全レース平均を返す(大きいほど良い指標)。
  • ねらい
    実運用に近い評価指標(上位重視の nDCG@K と勝者的中)で、汎化性能を定量化。
  • 統計学的視点
    nDCG は位置割引付き利得の正規化—情報検索での期待効用最大化に基づく指標。Top-1 は 0-1 損失に対応する単純指標で、モデルの決定境界の鋭さを補助的に評価します。
    なお autocast_ctx混合精度での数値安定と高速化に寄与します(分散推定に影響しにくい評価パスで使用)。

コメント

タイトルとURLをコピーしました