馬の履歴（時系列）と同一レース内の相互関係（集合）を同時に学習できる、埋め込み付きハイブリッド予測モデル
class SetTransformer(nn.Module):
def __init__(self, input_dim, hidden_dim=16, num_heads=2, num_blocks=1, dropout_p=0.3):
super().__init__()
self.encoder_linear = nn.Linear(input_dim, hidden_dim)
self.encoder_relu = nn.ReLU()
self.encoder_ln = nn.LayerNorm(hidden_dim)
self.encoder_dropout = nn.Dropout(p=dropout_p)
self.blocks = nn.ModuleList([
nn.TransformerEncoderLayer(
d_model=hidden_dim,
nhead=num_heads,
batch_first=True,
dropout=dropout_p
)
for _ in range(num_blocks)
])
self.output = nn.Sequential(
nn.Linear(hidden_dim, 32),
nn.ReLU(),
nn.Dropout(p=dropout_p),
nn.Linear(32, 1)
)
def forward(self, x, key_padding_mask=None):
squeeze_back = False
if x.dim() == 2:
x = x.unsqueeze(0)
squeeze_back = True
if key_padding_mask is not None and key_padding_mask.dim() == 1:
key_padding_mask = key_padding_mask.unsqueeze(0)
x = self.encoder_linear(x)
x = self.encoder_relu(x)
x = self.encoder_dropout(x)
x = self.encoder_ln(x)
for block in self.blocks:
x = block(x, src_key_padding_mask=key_padding_mask)
out = self.output(x).squeeze(-1)
if squeeze_back:
out = out.squeeze(0)
return out
class HybridRaceModelWithEmbedding(nn.Module):
def __init__(self, input_dim, sequence_length, lstm_hidden,
embedding_input_dims=None, embedding_output_dims=None,
set_hidden_dim=32, set_heads=2, set_blocks=1, set_dropout=0.3):
super().__init__()
self.sequence_length = sequence_length
self.embedding_layers = nn.ModuleDict()
self.embedding_dims = 0
if embedding_input_dims and embedding_output_dims:
for col in embedding_input_dims:
self.embedding_layers[col] = nn.Embedding(
num_embeddings=int(embedding_input_dims[col]),
embedding_dim=int(embedding_output_dims[col])
)
self.embedding_dims += int(embedding_output_dims[col])
self.lstm = nn.LSTM(
input_size=input_dim + self.embedding_dims,
hidden_size=lstm_hidden,
batch_first=True
)
self.dropout = nn.Dropout(p=0.3)
self.set_transformer = SetTransformer(
input_dim=lstm_hidden,
hidden_dim=set_hidden_dim,
num_heads=set_heads,
num_blocks=set_blocks,
dropout_p=set_dropout
)
def forward(self, x_num, x_cat_dict, mask=None):
squeeze_b = False
if x_num.dim() == 3:
x_num = x_num.unsqueeze(0)
squeeze_b = True
B, S, T, D = x_num.shape
device = x_num.device
emb_list = []
for col, x_cat in x_cat_dict.items():
if x_cat.dim() == 2:
x_cat = x_cat.unsqueeze(0)
x_cat = x_cat.to(device)
emb = self.embedding_layers[col](x_cat)
emb_list.append(emb)
if emb_list:
x = torch.cat([x_num] + emb_list, dim=-1)
else:
x = x_num
x = x.view(B * S, T, -1)
_, (h_n, _) = self.lstm(x)
h_out = self.dropout(h_n.squeeze(0))
h_out = h_out.view(B, S, -1)
if mask is None:
key_padding_mask = None
else:
key_padding_mask = mask
if key_padding_mask.dim() == 1 and not squeeze_b:
key_padding_mask = key_padding_mask.unsqueeze(0)
out = self.set_transformer(h_out, key_padding_mask=key_padding_mask)
if squeeze_b:
out = out.squeeze(0)
return out
class StackingEnsembleRaceModel(nn.Module):
def __init__(self, model_self_attn, model_set_transformer, hidden_dim=32, combine="head"):
super().__init__()
self.model_self_attn = model_self_attn
self.model_set_transformer = model_set_transformer
self.combine = combine
if combine == "head":
self.stacking_head = nn.Sequential(
nn.Linear(2, hidden_dim),
nn.ReLU(),
nn.Dropout(p=0.1),
nn.Linear(hidden_dim, 1)
)
@staticmethod
def _ensure_2d(t: torch.Tensor) -> torch.Tensor:
"""[S] -> [1,S] に揃える（[B,S]はそのまま）"""
return t if t.dim() == 2 else t.unsqueeze(0)
def forward(self, x_num, cat_features, mask=None):
out_self_attn = self.model_self_attn(x_num, cat_features, mask=mask)
out_set_trans = self.model_set_transformer(x_num, cat_features, mask=mask)
o1 = self._ensure_2d(out_self_attn)
o2 = self._ensure_2d(out_set_trans)
if self.combine == "mean":
out = (o1 + o2) / 2.0
else:
stacked = torch.stack([o1, o2], dim=-1)
out = self.stacking_head(stacked).squeeze(-1)
if out_self_attn.dim() == 1 and out_set_trans.dim() == 1:
out = out.squeeze(0)
return out
class HorseDataset(torch.utils.data.Dataset):
def __init__(self, X_num, X_cat, race_ids, horse_ids, y=None, history_lengths=None):
self.X_num = X_num
self.X_cat = X_cat
self.race_ids = race_ids
self.horse_ids = horse_ids
self.y = y
self.hist = history_lengths
def __len__(self):
return len(self.X_num)
def __getitem__(self, idx):
item = {
"X_num": torch.tensor(self.X_num[idx], dtype=torch.float32),
"race_id": self.race_ids[idx],
"horse_id": self.horse_ids[idx],
}
if self.hist is not None:
item["hist"] = torch.tensor(self.hist[idx], dtype=torch.float32)
if self.y is not None:
item["y"] = torch.tensor(self.y[idx], dtype=torch.long)
for col in self.X_cat:
item[col] = torch.tensor(self.X_cat[col][idx], dtype=torch.long)
return item
このコードは、時系列×集合（レース内の複数馬）の二層構造を扱うハイブリッドモデル群と、学習用データセットを定義しています。
流れとしては、時系列（各馬の履歴 T）を LSTM で圧縮 → レース内集合（S）を Transformer でモデリングし、最終的に 着順スコア を出力します。加えて、埋め込み特徴（カテゴリID）を LSTM 入力へ連結します。
1) SetTransformer：レース内の「集合」構造を表現
- 目的：同一レース内（集合 S）の馬同士の関係性を自己注意で表現し、各馬のスコアを出力。
- Encoder 前段（線形→ReLU→Dropout→LayerNorm）
self.encoder_linear = nn.Linear(input_dim, hidden_dim)
self.encoder_relu = nn.ReLU()
self.encoder_ln = nn.LayerNorm(hidden_dim)
self.encoder_dropout = nn.Dropout(p=dropout_p)- 入力次元を
hidden_dimに射影し、活性化・正則化で安定化。 - Transformer エンコーダ層（複数ブロック）
nn.TransformerEncoderLayer(d_model=hidden_dim, nhead=num_heads, batch_first=True)batch_first=True：テンソル形状を [B, S, H] として扱う前提。src_key_padding_mask=key_padding_mask：パディング位置を注意計算から除外。
- 出力ヘッド：
hidden_dim → 32 → 1の MLP で 各要素（馬）ごとのスコアを出力。 - 柔軟な入出力形状
- 入力
xが 2 次元の場合（[S, H]）、内部で バッチ次元を疑似的に付与（unsqueeze(0)）して処理 → 終了時に元形状へ戻す。 - これにより [S, H] / [B, S, H] の両方に対応。
- 入力
2) HybridRaceModelWithEmbedding：時系列×カテゴリ埋め込み×集合の統合モデル
- 目的：
- カテゴリ列（騎手ID 等）を Embedding に変換し、
- 数値列＋Embedding を 時系列 LSTM へ投入、
- LSTM の最終隠れ状態（各馬の時系列要約）を SetTransformer で集合モデリング。
- Embedding 層の構築
for col in embedding_input_dims: nn.Embedding(num_embeddings, embedding_dim)列ごとにテーブルを持ち、ID → 連続ベクトルへ変換。出力埋め込み次元の総和を LSTM 入力に足し込む。 - LSTM による時系列圧縮
- 期待形状は [B, S, T, D]：B=バッチ、S=レース内馬数、T=履歴長、D=数値特徴次元。
- 各カテゴリ列
x_cat_dict[col]は [B, S, T] の整数ID配列 →Embeddingで [B, S, T, E_col] に。 - 数値＋全列の埋め込みを 最終次元で結合 →
[B, S, T, D+ΣE]。 view(B*S, T, -1)で 馬ごとの系列へ並べ直し、LSTMで 最終隠れ状態 h_n を取得（各馬の履歴要約ベクトル）。h_out.view(B, S, -1)に戻して、レース内集合表現へ接続。
- SetTransformer で集合学習
LSTM の出力（[B, S, H]）をSetTransformerに投入。maskがあれば パディングを無視して注意計算。 - 可変形状の吸収
x_num.dim()==3の場合（[S,T,D]）は バッチ次元を補う（unsqueeze(0)）。maskについても次元を揃えてから渡すため、単発推論とバッチ推論に両対応。
3) StackingEnsembleRaceModel：2 つのモデルをアンサンブル
- 目的：異なるアーキテクチャの出力を 平均または小ヘッドで融合して精度を上げる。
- 入力
model_self_attn：別実装の自己注意系モデル（例：時系列自己注意など）。model_set_transformer：上記HybridRaceModelWithEmbeddingなど。
- 結合戦略
if combine == "mean": out = (o1 + o2) / 2.0 else: stacked = torch.stack([o1, o2], dim=-1) # [B,S,2] out = stacking_head(stacked).squeeze(-1) # [B,S,1]→[B,S]- 平均：単純・堅牢。
- head：
Linear → ReLU → Dropout → Linearの 学習可能結合で非線形融合。
- 形状ユーティリティ
_ensure_2d：[S]→[1,S]に整形してから演算 → バッチ有無に頑健。
4) HorseDataset：学習データ供給用 Dataset
- 役割：
DataLoaderに渡すための アイテム組み立て。 __getitem__の出力"X_num"：数値時系列テンソル（float32）。"y"：教師ラベル（存在時、long）。"hist"：履歴長などの補助（任意）。"race_id","horse_id"：メタ情報。- 各カテゴリ列（
X_cat[col]）はlongで返却（Embedding 層の入力要件）。
- ポイント：列ごとのテンソル化をここで統一しておくことで、バッチ化後もモデルの期待 dtype/shape に揃う。
モデル全体の意図（要約）
- 埋め込みでカテゴリIDを密ベクトルへ。
- LSTMで各馬の履歴（T ステップ）を要約。
- SetTransformerで同一レース内の相互作用（S 要素）を表現。
- （任意で）アンサンブルにより頑健性・精度を向上。
- Datasetにより入出力の dtype/shape を統制し、学習を安定化。
この設計により、時系列依存とレース内相対関係の両方を捉え、現実の勝敗要因に近い表現学習を狙っています。
コメント