diff options
| author | sillylaird <sillyfanboy@gmail.com> | 2026-09-03 00:33:59 +0000 |
|---|---|---|
| committer | sillylaird <sillyfanboy@gmail.com> | 2026-09-03 00:33:59 +0000 |
| commit | 898b52edcb47bcb3e9d6106e74ca73e74ea01e70 (patch) | |
| tree | 85c6ee5ad58b860144551184d4cf86b560c62b91 /.agents/skills/kaggle-learner/references/knowledge/time-series/birdclef-plus-2025.md | |
| download | www-898b52edcb47bcb3e9d6106e74ca73e74ea01e70.tar.gz www-898b52edcb47bcb3e9d6106e74ca73e74ea01e70.zip | |
Diffstat (limited to '')
| -rw-r--r-- | .agents/skills/kaggle-learner/references/knowledge/time-series/birdclef-plus-2025.md | 2312 |
1 files changed, 2312 insertions, 0 deletions
diff --git a/.agents/skills/kaggle-learner/references/knowledge/time-series/birdclef-plus-2025.md b/.agents/skills/kaggle-learner/references/knowledge/time-series/birdclef-plus-2025.md new file mode 100644 index 0000000..b25b5b1 --- /dev/null +++ b/.agents/skills/kaggle-learner/references/knowledge/time-series/birdclef-plus-2025.md @@ -0,0 +1,2312 @@ +# BirdCLEF\+ 2025 +> Last updated: 2026-01-23 +> Source count: 1 +--- + +### Mel-Spectrogram 特征提取(BirdCLEF+ 2025) + +基于前排方案,统一的 mel-spectrogram 提取流程: + +```python +import torch +import torchaudio +import torch.nn as nn +import numpy as np + +class MelSpectrogramExtractor: + """统一的 Mel-Spectrogram 提取器""" + + def __init__( + self, + sample_rate: int = 32000, + n_mels: int = 128, + n_fft: int = 2048, + hop_length: int = 512, + fmin: float = 0.0, + fmax: float = 16000.0, + power: float = 2.0, + normalize: bool = True, + ): + self.sample_rate = sample_rate + self.n_mels = n_mels + self.n_fft = n_fft + self.hop_length = hop_length + self.fmin = fmin + self.fmax = fmax + + # 使用 torchaudio 的 MelSpectrogram + self.mel_transform = torchaudio.transforms.MelSpectrogram( + sample_rate=sample_rate, + n_fft=n_fft, + hop_length=hop_length, + n_mels=n_mels, + f_min=fmin, + f_max=fmax, + power=power, + normalized=normalize, + ) + + def extract(self, waveform: torch.Tensor) -> torch.Tensor: + """ + 提取 mel-spectrogram + + Args: + waveform: (num_samples,) 或 (batch, num_samples) + + Returns: + mel_spec: (n_mels, time) 或 (batch, n_mels, time) + """ + if waveform.dim() == 1: + waveform = waveform.unsqueeze(0) + + mel_spec = self.mel_transform(waveform) + + # 转换为对数尺度 + mel_spec = torch.log(mel_spec + 1e-9) + + return mel_spec + + def extract_fixed_length( + self, waveform: torch.Tensor, target_length: int + ) -> torch.Tensor: + """ + 提取固定长度的 mel-spectrogram(用于 5 秒音频) + + Args: + waveform: (num_samples,) + target_length: 目标时间维度 + + Returns: + mel_spec: (n_mels, target_length) + """ + mel_spec = self.extract(waveform).squeeze(0) + + # 调整到固定长度 + if mel_spec.shape[1] < target_length: + # 填充 + pad_length = target_length - mel_spec.shape[1] + mel_spec = nn.functional.pad(mel_spec, (0, pad_length)) + else: + # 裁剪(从中心) + start = (mel_spec.shape[1] - target_length) // 2 + mel_spec = mel_spec[:, start:start + target_length] + + return mel_spec + + +# 常用配置(前排方案) +CONFIGS = { + "config_128": { # tf_efficientnet 系列 + "n_mels": 128, + "n_fft": 2048, + "hop_length": 512, + "fmin": 0.0, + "fmax": 16000.0, + }, + "config_96": { # 轻量级模型 + "n_mels": 96, + "n_fft": 2048, + "hop_length": 512, + "fmin": 0.0, + "fmax": 16000.0, + }, + "config_256": { # 高分辨率 + "n_mels": 256, + "n_fft": 4096, + "hop_length": 1024, + "fmin": 0.0, + "fmax": 16000.0, + }, +} + +# 使用示例 +extractor = MelSpectrogramExtractor(**CONFIGS["config_128"]) +waveform, sr = torchaudio.load("audio.wav") +mel_spec = extractor.extract_fixed_length(waveform.squeeze(0), target_length=313) # 5秒 -> 313帧 +``` + +### 伪标签生成(BirdCLEF+ 2025) + +```python +import torch +import torch.nn as nn +import numpy as np +from pathlib import Path + +class PseudoLabelGenerator: + """伪标签生成器 - 基于前排方案""" + + def __init__( + self, + model: nn.Module, + threshold: float = 0.4, + use_segmentwise: bool = True, + power_transform: float = 1.0, + ): + """ + Args: + model: 训练好的模型 + threshold: 置信度阈值(前排方案使用 0.3-0.5) + use_segmentwise: 是否使用 segmentwise_logit(更细粒度) + power_transform: 幂次变换参数(1st Place 使用) + """ + self.model = model + self.model.eval() + self.threshold = threshold + self.use_segmentwise = use_segmentwise + self.power_transform = power_transform + + @torch.no_grad() + def generate_pseudo_labels( + self, + audio_path: str, + segment_duration: int = 5, + overlap: float = 0.5, + ) -> list[dict]: + """ + 生成伪标签 + + Returns: + List of {"start": float, "end": float, "labels": np.ndarray} + """ + # 加载音频 + waveform, sr = torchaudio.load(audio_path) + + # 分段处理 + samples_per_segment = int(segment_duration * sr) + hop_length = int(samples_per_segment * (1 - overlap)) + + pseudo_labels = [] + + for start_idx in range(0, len(waveform) - samples_per_segment, hop_length): + end_idx = start_idx + samples_per_segment + segment = waveform[:, start_idx:end_idx] + + # 提取特征 + mel_spec = self.extract_mel(segment) + + # 模型预测 + if self.use_segmentwise: + # segmentwise_logit: 更细粒度的预测 + logits = self.model(mel_spec, return_segmentwise=True) + # 时间维度平均 + logits = logits.mean(dim=1) # (batch, num_classes) + else: + logits = self.model(mel_spec) + + # Sigmoid 激活 + probs = torch.sigmoid(logits).squeeze(0).cpu().numpy() + + # 幂次变换(1st Place 创新) + if self.power_transform != 1.0: + probs = np.power(probs, self.power_transform) + + # 高低阈值筛选(10th Place 方法) + mask = self._apply_threshold(probs) + + if mask.sum() > 0: + pseudo_labels.append({ + "start": start_idx / sr, + "end": end_idx / sr, + "labels": probs, + "mask": mask, + }) + + return pseudo_labels + + def _apply_threshold(self, probs: np.ndarray) -> np.ndarray: + """应用高低阈值筛选""" + # 高阈值:正样本 + high_threshold = 0.7 + # 低阈值:负样本 + low_threshold = 0.3 + + mask = np.zeros_like(probs, dtype=bool) + mask[probs >= high_threshold] = True # 高置信度正样本 + mask[probs <= low_threshold] = True # 低置信度负样本 + + return mask + + def extract_mel(self, waveform: torch.Tensor) -> torch.Tensor: + """提取 mel-spectrogram(简化版本)""" + # 实际使用中应该与训练时的提取器一致 + pass + + +# 使用示例(前排方案风格) +generator = PseudoLabelGenerator( + model=model, + threshold=0.4, + use_segmentwise=True, # 6th Place 关键 + power_transform=1.5, # 1st Place 幂次变换 +) + +pseudo_labels = generator.generate_pseudo_labels("train_soundscape_01.wav") +``` + +### MixUp 数据增强(BirdCLEF+ 2025) + +```python +import torch +import torch.nn as nn +import numpy as np + +class AudioMixUp: + """音频 MixUp 增强 - 前排方案风格""" + + def __init__( + self, + alpha: float = 0.5, + mixup_type: str = "hard", # "hard" 或 "soft" + probability: float = 0.5, + ): + """ + Args: + alpha: Beta 分布参数 + mixup_type: + - "soft": 标准混合标签(MixUp) + - "hard": 硬混合标签(8th Place 创新) + probability: 应用 MixUp 的概率 + """ + self.alpha = alpha + self.mixup_type = mixup_type + self.probability = probability + + def __call__( + self, + batch: dict, + ) -> dict: + """ + 应用 MixUp + + Args: + batch: {"mel": (B, C, H, W), "labels": (B, num_classes)} + + Returns: + Mixed batch + """ + if torch.rand(1).item() > self.probability: + return batch + + mel = batch["mel"] + labels = batch["labels"] + + batch_size = mel.size(0) + + # 生成混合权重 + lam = np.random.beta(self.alpha, self.alpha) + + # 随机排列 + index = torch.randperm(batch_size) + + # 混合特征 + mixed_mel = lam * mel + (1 - lam) * mel[index] + + # 混合标签 + if self.mixup_type == "soft": + # 标准 MixUp: 软标签混合 + mixed_labels = lam * labels + (1 - lam) * labels[index] + elif self.mixup_type == "hard": + # 硬 MixUp (8th Place): 混合标签的最大值 + mixed_labels = torch.maximum(labels, labels[index]) + else: + raise ValueError(f"Unknown mixup_type: {self.mixup_type}") + + return { + "mel": mixed_mel, + "labels": mixed_labels, + "lam": lam, # 可能用于损失调整 + } + + +# Sumix 增强(13th Place 使用) +class Sumix: + """Sumix: 原始信号上的 MixUp""" + + def __init__(self, alpha: float = 0.5, probability: float = 1.0): + self.alpha = alpha + self.probability = probability + + def __call__( + self, + waveform: torch.Tensor, + labels: torch.Tensor, + ) -> tuple[torch.Tensor, torch.Tensor]: + """ + 在原始波形上应用 Sumix + + Args: + waveform: (batch, num_samples) + labels: (batch, num_classes) + + Returns: + Mixed waveform and labels + """ + if torch.rand(1).item() > self.probability: + return waveform, labels + + batch_size = waveform.size(0) + lam = np.random.beta(self.alpha, self.alpha) + index = torch.randperm(batch_size) + + # 混合波形 + mixed_waveform = lam * waveform + (1 - lam) * waveform[index] + + # 混合标签(最大值) + mixed_labels = torch.maximum(labels, labels[index]) + + return mixed_waveform, mixed_labels + + +# 使用示例 +mixup = AudioMixUp(alpha=0.5, mixup_type="hard", probability=0.5) +sumix = Sumix(alpha=0.5, probability=1.0) + +# 训练循环中 +for batch in dataloader: + # Sumix 在原始波形 + waveform, labels = sumix(batch["waveform"], batch["labels"]) + + # 提取 mel-spectrogram + mel = extract_mel(waveform) + + # MixUp 在 mel-spectrogram + batch = mixup({"mel": mel, "labels": labels}) +``` + +### Soft AUC Loss(BirdCLEF+ 2025 - 4th Place) + +支持软标签的 AUC 损失函数: + +```python +import torch +import torch.nn as nn +import torch.nn.functional as F + +class SoftAUCLoss(nn.Module): + """ + Soft AUC Loss - 4th Place 创新 + + 支持 soft labels,适用于知识蒸馏和半监督学习 + """ + + def __init__(self, reduction: str = "mean"): + super().__init__() + self.reduction = reduction + + def forward( + self, + predictions: torch.Tensor, + targets: torch.Tensor, + ) -> torch.Tensor: + """ + Args: + predictions: (batch, num_classes) - 原始 logits + targets: (batch, num_classes) - 软标签 [0, 1] + + Returns: + AUC loss + """ + # Sigmoid 激活 + probs = torch.sigmoid(predictions) + + # 计算 AUC loss + # 对每个类别独立计算 + num_classes = predictions.size(1) + losses = [] + + for c in range(num_classes): + # 当前类别的预测和目标 + prob_c = probs[:, c] + target_c = targets[:, c] + + # 按目标值排序(软标签) + sorted_indices = torch.argsort(target_c, descending=True) + + # 计算正负样本的得分差异 + # 对于软标签,我们需要加权处理 + positive_scores = prob_c[sorted_indices[:len(sorted_indices)//2]] + negative_scores = prob_c[sorted_indices[len(sorted_indices)//2:]] + + # AUC 近似:正样本得分应该高于负样本 + # 使用 sigmoid 差异 + diff = positive_scores.unsqueeze(1) - negative_scores.unsqueeze(0) + loss_c = torch.sigmoid(-diff).mean() + + losses.append(loss_c) + + losses = torch.stack(losses) + + if self.reduction == "mean": + return losses.mean() + elif self.reduction == "sum": + return losses.sum() + else: + return losses + + +# 改进的 AUC Loss(更稳定) +class ImprovedAUCLoss(nn.Module): + """ + 改进的 AUC Loss - 更稳定且支持软标签 + """ + + def __init__(self, margin: float = 1.0): + super().__init__() + self.margin = margin + + def forward( + self, + predictions: torch.Tensor, + targets: torch.Tensor, + ) -> torch.Tensor: + """ + Args: + predictions: (batch, num_classes) + targets: (batch, num_classes) - 软标签 + """ + probs = torch.sigmoid(predictions) + num_classes = predictions.size(1) + + losses = [] + for c in range(num_classes): + prob_c = probs[:, c] + target_c = targets[:, c] + + # 计算成对损失 + # 对于每个样本对 (i, j): + # 如果 target_i > target_j,则希望 prob_i > prob_j + n = prob_c.size(0) + if n < 2: + continue + + # 创建样本对矩阵 + target_diff = target_c.unsqueeze(1) - target_c.unsqueeze(0) + prob_diff = prob_c.unsqueeze(1) - prob_c.unsqueeze(0) + + # 只考虑 target_i > target_j 的对 + mask = target_diff > 0 + + if mask.sum() > 0: + # Hinge loss: max(0, margin - (prob_i - prob_j)) + loss_c = F.relu(self.margin - prob_diff)[mask].mean() + losses.append(loss_c) + + if len(losses) == 0: + return torch.tensor(0.0, device=predictions.device) + + return torch.stack(losses).mean() + + +# 使用示例 +criterion = SoftAUCLoss(reduction="mean") + +# 训练循环 +for batch in dataloader: + predictions = model(batch["mel"]) + + # 支持软标签 + loss = criterion(predictions, batch["labels"]) + + loss.backward() + optimizer.step() +``` + +### 滑动窗口推理(BirdCLEF+ 2025 - 1st Place) + +```python +import torch +import torch.nn as nn +from scipy.ndimage import gaussian_filter1d + +class SlidingWindowInference: + """ + 滑动窗口推理 - 1st Place 创新 + + 使用帧预测的平均值,而不是仅使用中心窗口的最大值 + """ + + def __init__( + self, + model: nn.Module, + window_size: int = 5, # 秒 + hop_size: int = 5, # 秒(步长) + sample_rate: int = 32000, + smoothing_sigma: float = 1.0, + ): + self.model = model + self.model.eval() + self.window_size = window_size + self.hop_size = hop_size + self.sample_rate = sample_rate + self.smoothing_sigma = smoothing_sigma + + @torch.no_grad() + def predict( + self, + audio_path: str, + ) -> dict[str, float]: + """ + 对整个音频进行预测,返回 5 秒窗口的预测 + + Returns: + Dict of {row_id: {species_id: probability}} + """ + # 加载音频 + waveform, sr = torchaudio.load(audio_path) + + # 计算窗口参数 + samples_per_window = int(self.window_size * sr) + samples_per_hop = int(self.hop_size * sr) + + # 存储所有帧预测 + all_frame_predictions = [] + + # 滑动窗口 + window_id = 0 + for start_idx in range(0, len(waveform) - samples_per_window, samples_per_hop): + end_idx = start_idx + samples_per_window + window = waveform[:, start_idx:end_idx] + + # 提取特征 + mel_spec = self.extract_mel(window) + + # 模型预测 + frame_output = self.model(mel_spec) + + # 如果是 SED 模型,可能有 clipwise 和 segmentwise 输出 + if isinstance(frame_output, dict): + frame_pred = frame_output["clipwise_output"] + else: + frame_pred = frame_output + + all_frame_predictions.append(frame_pred.cpu().numpy()) + + window_id += 1 + + # 转换为 numpy array + all_frame_predictions = np.array(all_frame_predictions) # (num_windows, num_classes) + + # 1st Place 创新: 相邻窗口帧预测平均 + # 这是一种 1D 滑动窗口分割,类似于大图像的 2D 滑动窗口分割 + smoothed_predictions = self._smooth_predictions(all_frame_predictions) + + # 生成最终预测 + predictions = {} + for window_id in range(len(smoothed_predictions)): + row_id = f"soundscape_{window_id}_{self.window_size}" + predictions[row_id] = { + f"species_{i}": float(prob) + for i, prob in enumerate(smoothed_predictions[window_id]) + } + + return predictions + + def _smooth_predictions( + self, + predictions: np.ndarray, + ) -> np.ndarray: + """ + 平滑预测 - 使用高斯滤波和时间平均 + + Args: + predictions: (num_windows, num_classes) + + Returns: + Smoothed predictions + """ + # 1. 时间维度高斯平滑 + if self.smoothing_sigma > 0: + smoothed = gaussian_filter1d( + predictions, + sigma=self.smoothing_sigma, + axis=0, + mode="nearest", + ) + else: + smoothed = predictions + + # 2. 相邻窗口平均(1st Place 创新) + # 使用相邻 3 个窗口的平均 + kernel_size = 3 + if len(smoothed) >= kernel_size: + # Padding + padded = np.pad( + smoothed, + ((kernel_size // 2, kernel_size // 2), (0, 0)), + mode="edge", + ) + + # 一维卷积平均 + kernel = np.ones(kernel_size) / kernel_size + averaged = np.zeros_like(smoothed) + + for c in range(smoothed.shape[1]): + averaged[:, c] = np.convolve( + padded[:, c], + kernel, + mode="valid", + ) + + return averaged + else: + return smoothed + + def extract_mel(self, waveform: torch.Tensor) -> torch.Tensor: + """提取 mel-spectrogram""" + # 实际使用中应该与训练时的提取器一致 + pass + + +# 使用示例 +inference = SlidingWindowInference( + model=model, + window_size=5, + hop_size=5, + smoothing_sigma=1.0, +) + +predictions = inference.predict("test_soundscape_01.wav") + +# 后处理(可选) +# - Delta shift: 调整低置信度类别的概率 +# - Min-max 缩放 +# - 频率范围调整 +``` + +### SED 模型架构(BirdCLEF+ 2025 标准) + +前排方案广泛使用的 SED (Sound Event Detection) 模型架构: + +```python +import torch +import torch.nn as nn +import timm + +class SEDModel(nn.Module): + """ + Sound Event Detection 模型 + + 参考 BirdCLEF 2023 2nd Place 和 BirdCLEF+ 2025 前排方案 + """ + + def __init__( + self, + backbone: str = "tf_efficientnetv2_s.in21k", + num_classes: int = 206, + in_channels: int = 1, + pretrained: bool = True, + ): + super().__init__() + + self.num_classes = num_classes + + # 使用 timm 的 EfficientNet 作为 backbone + self.backbone = timm.create_model( + backbone, + pretrained=pretrained, + in_chans=in_channels, + num_classes=0, # 移除分类头 + ) + + # 获取 backbone 输出特征维度 + self.features_dim = self.backbone.num_features + + # 自定义注意力块 (6th Place AttBlockV2) + self.att_block = AttBlockV2( + self.features_dim, + num_classes, + activation="sigmoid", + ) + + def forward(self, x, return_segmentwise=False): + """ + Args: + x: (batch, in_channels, n_mels, time) + return_segmentwise: 是否返回 segmentwise_logit + + Returns: + 如果 return_segmentwise=False: + clipwise_output: (batch, num_classes) + 如果 return_segmentwise=True: + dict with: + clipwise_output: (batch, num_classes) + segmentwise_output: (batch, num_classes, time_frames) + """ + # Backbone 特征提取 + features = self.backbone(x) # (batch, features_dim, time_frames) + + # 全局池化 + pooled_features = features.mean(dim=[2]) # (batch, features_dim) + + # 片级预测 + clipwise_output = self.att_block(pooled_features) # (batch, num_classes) + + if not return_segmentwise: + return clipwise_output + + # 帧级预测(用于伪标签生成) + segmentwise_output = self.att_block(features) # (batch, num_classes, time_frames) + + return { + "clipwise_output": clipwise_output, + "segmentwise_output": segmentwise_output, + } + + +class AttBlockV2(nn.Module): + """ + 自定义注意力块 - 6th Place 创新 + + 使用 softmax 和 tanh 进行归一化,结合非线性变换 + """ + + def __init__( + self, + in_features: int, + out_features: int, + activation: str = "sigmoid", + ): + super().__init__() + + self.activation = activation + self.att = nn.Conv1d(in_features, out_features, kernel_size=1) + self.cla = nn.Conv1d(in_features, out_features, kernel_size=1) + + # 初始化权重(6th Place 关键) + self.apply(self._init_weights) + + def _init_weights(self, m): + if isinstance(m, nn.Conv1d): + nn.init.kaiming_normal_(m.weight, mode="fan_out", nonlinearity="relu") + if m.bias is not None: + nn.init.constant_(m.bias, 0) + + def forward(self, x): + """ + Args: + x: (batch, in_features, time_frames) 或 (batch, in_features) + + Returns: + output: (batch, out_features) 或 (batch, out_features, time_frames) + """ + if x.dim() == 2: + # 全局池化特征 + x = x.unsqueeze(-1) # (batch, in_features, 1) + + # 注意力权重 + att = self.att(x) + att = torch.softmax(att, dim=1) # 时间维度归一化 + + # 分类特征 + cla = self.cla(x) + + # 加权求和 + output = torch.clamp(torch.clamp((cla * att).sum(dim=-1), min=1e-7, max=1-1e-7), min=1e-7) + + # 激活函数 + if self.activation == "sigmoid": + output = torch.sigmoid(output) + elif self.activation == "none": + pass + else: + raise ValueError(f"Unknown activation: {self.activation}") + + return output.squeeze(-1) if output.size(-1) == 1 else output + + +# 常用 backbone 配置(前排方案) +BACKBONES = { + "tf_efficientnetv2_s.in21k": { + "features_dim": 1280, + "description": "2nd Place 使用,平衡性能和速度", + }, + "tf_efficientnetv2_b3.in21k": { + "features_dim": 1536, + "description": "6th Place 使用,更强性能", + }, + "tf_efficientnetv2_m.in21k": { + "features_dim": 2048, + "description": "14th Place 使用,更高精度", + }, + "eca_nfnet_l0": { + "features_dim": 2304, + "description": "2nd Place 使用,增加多样性", + }, +} + +# 使用示例 +model = SEDModel( + backbone="tf_efficientnetv2_s.in21k", + num_classes=206, + in_channels=1, + pretrained=True, +) + +# 训练时:片级预测 +clipwise_output = model(mel_spec) +loss = criterion(clipwise_output, labels) + +# 伪标签生成时:帧级预测 +output = model(mel_spec, return_segmentwise=True) +segmentwise_logits = output["segmentwise_output"] # (batch, 206, time_frames) +segmentwise_probs = torch.sigmoid(segmentwise_logits) +# 时间维度平均得到更稳定的伪标签 +avg_segmentwise_probs = segmentwise_probs.mean(dim=-1) # (batch, 206) +``` + +### 前排方案详细技术分析 + +#### 2nd Place - Xeno-Canto 预训练详细流程 + +**作者**: Volodymyr Vialactea +**核心创新**: 使用外部数据预训练 + 5秒音频片段训练 + +**完整流程:** + +```python +import torch +import torchaudio +import pandas as pd +from pathlib import Path + +class XenoCantoPretraining: + """ + 2nd Place 方案:Xeno-Canto 预训练流程 + + 关键点: + 1. 下载额外的 Xeno-Canto 数据 + 2. 数据清洗和预处理 + 3. 预训练 + 4. 在主数据集上微调 + """ + + def __init__( + self, + species_list: list, + target_sample_rate: int = 32000, + segment_duration: int = 5, + ): + self.species_list = species_list + self.target_sample_rate = target_sample_rate + self.segment_duration = segment_duration + + def download_xeno_canto_data(self, output_dir: str = "data/xeno_canto"): + """ + 步骤 1: 从 Xeno-Canto 下载数据 + + 注意事项: + - 过滤掉当年比赛中的物种(避免数据泄漏) + - 只下载高质量录音(评分 ≥ 3.0) + - 限制每个物种的下载量(避免数据不平衡) + """ + # 使用 xeno-canto-api 或手动下载 + # 这里提供框架代码 + + xc_species = [s for s in self.species_list if self._should_download(s)] + + for species in xc_species: + # 调用 Xeno-Canto API + # 下载音频文件 + # 保存到 output_dir/species_name/ + pass + + def _should_download(self, species: str) -> bool: + """检查物种是否应该下载(避免数据泄漏)""" + # 过滤比赛数据集中的物种 + competition_species = set(self._get_competition_species()) + return species not in competition_species + + def preprocess_xeno_canto(self, audio_dir: str): + """ + 步骤 2: 数据清洗和预处理 + + 2nd Place 的关键步骤: + 1. 去除人声(如果可能) + 2. 统一采样率到 32kHz + 3. 音频归一化 + 4. 质量检查(SNR、时长等) + """ + audio_files = list(Path(audio_dir).rglob("*.mp3")) + + cleaned_data = [] + + for audio_file in audio_files: + # 加载音频 + waveform, sr = torchaudio.load(audio_file) + + # 重采样到 32kHz + if sr != self.target_sample_rate: + resampler = torchaudio.transforms.Resample(sr, self.target_sample_rate) + waveform = resampler(waveform) + + # 质量检查 + if self._check_quality(waveform): + # 提取 5 秒片段 + segments = self._extract_segments(waveform) + + for segment in segments: + cleaned_data.append({ + "file_path": str(audio_file), + "species": audio_file.parent.name, + "waveform": segment, + }) + + return cleaned_data + + def _check_quality(self, waveform: torch.Tensor) -> bool: + """质量检查""" + # 检查 1: 时长至少 5 秒 + if waveform.shape[1] < self.target_sample_rate * self.segment_duration: + return False + + # 检查 2: SNR(信噪比) + # snr = self._calculate_snr(waveform) + # if snr < 10: # 最低 10dB + # return False + + # 检查 3: 削波检测 + if torch.abs(waveform).max() > 0.99: + return False + + return True + + def _extract_segments(self, waveform: torch.Tensor) -> list: + """ + 提取 5 秒音频片段 + + 2nd Place 使用了多种采样策略: + 1. 随机采样 + 2. 基于能量的采样(RMS) + 3. 重叠采样 + """ + segment_samples = self.segment_duration * self.target_sample_rate + + if waveform.shape[1] <= segment_samples: + # 填充到 5 秒 + padding = segment_samples - waveform.shape[1] + waveform = torch.nn.functional.pad(waveform, (0, padding)) + return [waveform] + + # 方法 1: 随机采样 + # 2nd Place 尝试了多种方法,最终发现随机采样效果最好 + + # 方法 2: 基于能量的采样(RMS) + # 计算每个 5 秒窗口的 RMS 能量 + # 选择能量最高的窗口 + + # 方法 3: 重叠采样 + # 滑动窗口,hop_size = 2.5 秒 + + # 这里实现随机采样 + max_start = waveform.shape[1] - segment_samples + start_idx = torch.randint(0, max_start, (1,)).item() + + segment = waveform[:, start_idx:start_idx + segment_samples] + return [segment] + + def pretrain(self, xc_data, model, save_path: str = "checkpoints/pretrained.pth"): + """ + 步骤 3: 预训练 + + 2nd Place 的预训练策略: + - 使用 Xeno-Canto 数据训练 + - BCE Loss + - SpecAugment 增强 + - 50-100 epochs + """ + # 创建 dataloader + train_loader = self._create_dataloader(xc_data) + + # 优化器 + optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3) + + # 学习率调度器 + scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( + optimizer, T_max=50, eta_min=1e-6 + ) + + # 损失函数 + criterion = nn.BCEWithLogitsLoss() + + # 训练循环 + model.train() + for epoch in range(50): # 50 epochs + for batch in train_loader: + mel_spec = self._extract_mel(batch["waveform"]) + labels = batch["labels"] + + # 前向传播 + logits = model(mel_spec) + loss = criterion(logits, labels) + + # 反向传播 + optimizer.zero_grad() + loss.backward() + optimizer.step() + + scheduler.step() + + print(f"Epoch {epoch+1}/50, Loss: {loss.item():.4f}") + + # 保存预训练模型 + torch.save(model.state_dict(), save_path) + print(f"Pretrained model saved to {save_path}") + + def finetune(self, model, train_data, val_data, pretrained_path: str): + """ + 步骤 4: 微调 + + 2nd Place 的微调策略: + - 加载预训练权重 + - 使用更小的学习率 + - 选择最佳 checkpoint(不是最后一个) + - 关键:AUC 从 0.83-0.84 跳升至 0.86-0.87 + """ + # 加载预训练权重 + model.load_state_dict(torch.load(pretrained_path)) + + # 优化器(更小的学习率) + optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4) + + # 学习率调度器 + scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( + optimizer, T_max=30, eta_min=1e-7 + ) + + # 损失函数 + criterion = nn.BCEWithLogitsLoss() + + best_val_score = 0 + best_epoch = 0 + + # 微调循环 + model.train() + for epoch in range(30): # 30 epochs + # 训练 + for batch in train_data: + mel_spec = self._extract_mel(batch["waveform"]) + labels = batch["labels"] + + logits = model(mel_spec) + loss = criterion(logits, labels) + + optimizer.zero_grad() + loss.backward() + optimizer.step() + + # 验证 + val_score = self._validate(model, val_data) + + print(f"Epoch {epoch+1}/30, Val AUC: {val_score:.4f}") + + # 保存最佳模型 + if val_score > best_val_score: + best_val_score = val_score + best_epoch = epoch + torch.save(model.state_dict(), f"checkpoints/best_finetuned_epoch{epoch}.pth") + + scheduler.step() + + print(f"Best epoch: {best_epoch}, Best Val AUC: {best_val_score:.4f}") + + def _extract_mel(self, waveform: torch.Tensor) -> torch.Tensor: + """提取 mel-spectrogram(应该与训练时一致)""" + # 实现 mel-spectrogram 提取 + pass + + def _create_dataloader(self, data): + """创建 dataloader""" + pass + + def _validate(self, model, val_data): + """验证""" + pass + + def _get_competition_species(self) -> list: + """获取竞赛数据集中的物种(避免数据泄漏)""" + pass + + def _calculate_snr(self, waveform: torch.Tensor) -> float: + """计算 SNR""" + pass + + +# 2nd Place 关键技术总结 +""" +关键发现(来自 2nd Place writeup): + +1. **预训练效果显著**: + - 无预训练:AUC 0.83-0.84 + - 有预训练:AUC 0.86-0.87 + - 提升:+0.02-0.03 AUC + +2. **Checkpoint 选择很重要**: + - 不是最后一个 epoch 最好 + - 需要验证集选择最佳 checkpoint + - 通常在 epoch 10-20 之间 + +3. **采样策略**: + - 随机采样效果最好 + - 基于能量的采样没有明显优势 + - 5 秒片段是最佳长度 + +4. **数据增强**: + - SpecAugment 必须保留 + - RandomFiltering 有效 + - 即使关闭略微提高 CV,但保留确保 LB 稳定性 +""" +``` + +#### 5th Place - Self-Distillation 详细实现 + +**作者**: Noir +**核心创新**: 三阶段自蒸馏 + Silero VAD 数据清洗 + +**完整流程:** + +```python +import torch +import torch.nn as nn +import numpy as np + +class SelfDistillationTrainer: + """ + 5th Place 方案:Self-Distillation 三阶段训练 + + 核心思想: + 1. 使用 Silero VAD 去除人声 + 2. 三阶段自蒸馏训练 + 3. 迭代丰富次要标签 + """ + + def __init__( + self, + model: nn.Module, + num_classes: int = 206, + ): + self.model = model + self.num_classes = num_classes + + def stage1_initial_training(self, train_loader, val_loader, epochs=30): + """ + 阶段 1: 初始训练 + + 使用清洗后的训练音频(train_audio)进行初始训练 + """ + print("=== Stage 1: Initial Training ===") + + optimizer = torch.optim.AdamW(self.model.parameters(), lr=1e-3) + scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( + optimizer, T_max=epochs, eta_min=1e-6 + ) + criterion = nn.BCEWithLogitsLoss() + + best_val_loss = float('inf') + + for epoch in range(epochs): + self.model.train() + train_loss = 0 + + for batch in train_loader: + mel_spec = batch['mel_spec'] + labels = batch['labels'] + + # 前向传播 + logits = self.model(mel_spec) + loss = criterion(logits, labels) + + # 反向传播 + optimizer.zero_grad() + loss.backward() + optimizer.step() + + train_loss += loss.item() + + # 验证 + val_loss = self._validate(self.model, val_loader, criterion) + + # 学习率更新 + scheduler.step() + + print(f"Epoch {epoch+1}/{epochs}, Train Loss: {train_loss/len(train_loader):.4f}, " + f"Val Loss: {val_loss:.4f}") + + # 保存最佳模型 + if val_loss < best_val_loss: + best_val_loss = val_loss + torch.save(self.model.state_dict(), "checkpoints/stage1_best.pth") + + print(f"Stage 1 complete. Best Val Loss: {best_val_loss:.4f}") + + # 加载最佳模型用于下一阶段 + self.model.load_state_dict(torch.load("checkpoints/stage1_best.pth")) + + def stage2_self_distillation_train_audio( + self, + train_loader, + epochs=20, + temperature=3.0, + alpha=0.7, + ): + """ + 阶段 2: 使用 train_audio 的自蒸馏 + + 使用 stage 1 模型的预测作为软标签进行蒸馏 + """ + print("=== Stage 2: Self-Distillation on train_audio ===") + + # stage 1 模型作为教师 + teacher_model = type(self.model)( + backbone=self.model.backbone, + num_classes=self.num_classes, + ) + teacher_model.load_state_dict(torch.load("checkpoints/stage1_best.pth")) + teacher_model.eval() + + # 学生模型(可以重置权重或继续训练) + # 5th Place 选择继续训练 + + optimizer = torch.optim.AdamW(self.model.parameters(), lr=5e-4) # 更小的学习率 + scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( + optimizer, T_max=epochs, eta_min=1e-7 + ) + + # 蒸馏损失 + distillation_criterion = nn.KLDivLoss(reduction="batchmean") + bce_criterion = nn.BCEWithLogitsLoss() + + best_val_loss = float('inf') + + for epoch in range(epochs): + self.model.train() + train_loss = 0 + + for batch in train_loader: + mel_spec = batch['mel_spec'] + hard_labels = batch['labels'] + + with torch.no_grad(): + # 教师模型预测(软标签) + teacher_logits = teacher_model(mel_spec) + teacher_probs = torch.sigmoid(teacher_logits / temperature) + + # 学生模型预测 + student_logits = self.model(mel_spec) + student_log_probs = torch.log_softmax(student_logits / temperature, dim=-1) + + # 蒸馏损失 + distill_loss = distillation_criterion(student_log_probs, teacher_probs) + + # 硬标签损失 + bce_loss = bce_criterion(student_logits, hard_labels) + + # 组合损失 + loss = alpha * (temperature ** 2) * distill_loss + (1 - alpha) * bce_loss + + # 反向传播 + optimizer.zero_grad() + loss.backward() + optimizer.step() + + train_loss += loss.item() + + # 验证 + val_loss = self._validate(self.model, train_loader, bce_criterion) # 用训练集验证 + + scheduler.step() + + print(f"Epoch {epoch+1}/{epochs}, Train Loss: {train_loss/len(train_loader):.4f}, " + f"Val Loss: {val_loss:.4f}") + + if val_loss < best_val_loss: + best_val_loss = val_loss + torch.save(self.model.state_dict(), "checkpoints/stage2_best.pth") + + print(f"Stage 2 complete. Best Val Loss: {best_val_loss:.4f}") + + self.model.load_state_dict(torch.load("checkpoints/stage2_best.pth")) + + def stage3_self_distillation_soundscape( + self, + train_audio_loader, + soundscape_files, + epochs=20, + temperature=3.0, + alpha=0.5, # 更重视伪标签 + ): + """ + 阶段 3: 结合 train_audio 和 train_soundscapes 的自蒸馏 + + 关键创新:丰富次要标签 + - 许多音频包含未标注的鸟叫声 + - 通过自蒸馏发现这些次要标签 + """ + print("=== Stage 3: Self-Distillation with soundscape ===") + + # stage 2 模型作为教师 + teacher_model = type(self.model)( + backbone=self.model.backbone, + num_classes=self.num_classes, + ) + teacher_model.load_state_dict(torch.load("checkpoints/stage2_best.pth")) + teacher_model.eval() + + optimizer = torch.optim.AdamW(self.model.parameters(), lr=3e-4) + scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( + optimizer, T_max=epochs, eta_min=1e-7 + ) + + distillation_criterion = nn.KLDivLoss(reduction="batchmean") + bce_criterion = nn.BCEWithLogitsLoss() + + # 生成 soundscape 的伪标签 + soundscape_pseudo_labels = self._generate_pseudo_labels( + teacher_model, soundscape_files + ) + + # 合并 train_audio 和 soundscape 数据 + # 50% train_audio + 50% soundscape + + best_val_loss = float('inf') + + for epoch in range(epochs): + self.model.train() + train_loss = 0 + + # 训练 train_audio(带硬标签) + for batch in train_audio_loader: + if np.random.rand() > 0.5: + continue # 50% 概率使用 train_audio + + mel_spec = batch['mel_spec'] + hard_labels = batch['labels'] + + with torch.no_grad(): + teacher_logits = teacher_model(mel_spec) + teacher_probs = torch.sigmoid(teacher_logits / temperature) + + student_logits = self.model(mel_spec) + student_log_probs = torch.log_softmax(student_logits / temperature, dim=-1) + + distill_loss = distillation_criterion(student_log_probs, teacher_probs) + bce_loss = bce_criterion(student_logits, hard_labels) + loss = alpha * (temperature ** 2) * distill_loss + (1 - alpha) * bce_loss + + optimizer.zero_grad() + loss.backward() + optimizer.step() + + train_loss += loss.item() + + # 训练 soundscape(伪标签) + for batch in soundscape_pseudo_labels: + if np.random.rand() <= 0.5: + continue # 50% 概率使用 soundscape + + mel_spec = batch['mel_spec'] + pseudo_labels = batch['labels'] # 软标签 + + with torch.no_grad(): + teacher_logits = teacher_model(mel_spec) + teacher_probs = torch.sigmoid(teacher_logits / temperature) + + student_logits = self.model(mel_spec) + student_log_probs = torch.log_softmax(student_logits / temperature, dim=-1) + + # 只使用蒸馏损失(没有硬标签) + distill_loss = distillation_criterion(student_log_probs, teacher_probs) + + optimizer.zero_grad() + distill_loss.backward() + optimizer.step() + + train_loss += distill_loss.item() + + scheduler.step() + + print(f"Epoch {epoch+1}/{epochs}, Train Loss: {train_loss:.4f}") + + # 保存检查点 + if epoch % 5 == 0: + torch.save(self.model.state_dict(), f"checkpoints/stage3_epoch{epoch}.pth") + + print("Stage 3 complete") + + def _generate_pseudo_labels( + self, + model: nn.Module, + audio_files: list, + ) -> list: + """ + 生成 soundscape 的伪标签 + + 关键:丰富次要标签 + - 使用帧级预测(segmentwise) + - 时间维度平均 + """ + model.eval() + pseudo_labels = [] + + with torch.no_grad(): + for audio_file in audio_files: + # 加载音频 + waveform, sr = torchaudio.load(audio_file) + + # 分段处理(5秒窗口) + segments = self._split_audio(waveform, sr) + + for segment in segments: + mel_spec = self._extract_mel(segment) + + # 获取帧级预测 + output = model(mel_spec, return_segmentwise=True) + segmentwise_logits = output["segmentwise_output"] # (1, 206, time) + segmentwise_probs = torch.sigmoid(segmentwise_logits) + + # 时间维度平均(关键:丰富次要标签) + avg_probs = segmentwise_probs.mean(dim=-1).squeeze(0) # (206,) + + pseudo_labels.append({ + "mel_spec": mel_spec, + "labels": avg_probs, + }) + + return pseudo_labels + + def _split_audio(self, waveform: torch.Tensor, sr: int) -> list: + """分段处理音频""" + segment_samples = 5 * sr + segments = [] + + for i in range(0, waveform.shape[1], segment_samples): + segment = waveform[:, i:i+segment_samples] + if segment.shape[1] == segment_samples: + segments.append(segment) + else: + # 填充 + padding = segment_samples - segment.shape[1] + segment = torch.nn.functional.pad(segment, (0, padding)) + segments.append(segment) + + return segments + + def _extract_mel(self, waveform: torch.Tensor) -> torch.Tensor: + """提取 mel-spectrogram""" + pass + + def _validate(self, model, val_loader, criterion): + """验证""" + model.eval() + total_loss = 0 + + with torch.no_grad(): + for batch in val_loader: + mel_spec = batch['mel_spec'] + labels = batch['labels'] + + logits = model(mel_spec) + loss = criterion(logits, labels) + total_loss += loss.item() + + return total_loss / len(val_loader) + + +class SileroVADDataCleaner: + """ + Silero VAD 数据清洗 + + 5th Place 使用 Silero VAD 检测并去除人声片段 + """ + + def __init__(self): + # 加载 Silero VAD 模型 + self.model, utils = torch.hub.load( + repo_or_dir='snakers4/silero-vad', + model='silero_vad', + force_reload=False, + onnx=False + ) + self.model.eval() + + def clean_audio(self, audio_path: str, output_path: str): + """ + 去除包含人声的音频片段 + + Returns: + 清洗后的音频(去除人声部分) + """ + waveform, sr = torchaudio.load(audio_path) + + # 转换为单声道 + if waveform.shape[0] > 1: + waveform = waveform.mean(dim=0, keepdim=True) + + # 重采样到 16kHz(Silero VAD 要求) + if sr != 16000: + resampler = torchaudio.transforms.Resample(sr, 16000) + waveform = resampler(waveform) + sr = 16000 + + # VAD 检测 + speech_chunks = self._detect_speech(waveform, sr) + + # 如果检测到人声,去除这些片段 + if speech_chunks: + cleaned_waveform = self._remove_speech_chunks(waveform, speech_chunks) + else: + cleaned_waveform = waveform + + # 保存清洗后的音频 + torchaudio.save(output_path, cleaned_waveform, sr) + + return cleaned_waveform + + def _detect_speech(self, waveform: torch.Tensor, sr: int) -> list: + """ + 检测人声片段 + + Returns: + List of (start_ms, end_ms) tuples + """ + # 获取语音概率 + speech_probs = [] + window_size = 512 # 32ms at 16kHz + + for i in range(0, waveform.shape[1], window_size): + chunk = waveform[:, i:i+window_size] + if chunk.shape[1] < window_size: + continue + + with torch.no_grad(): + speech_prob = self.model(chunk, sr).item() + speech_probs.append(speech_prob) + + # 阈值检测(人声概率 > 0.5) + speech_chunks = [] + in_speech = False + start_idx = 0 + + for i, prob in enumerate(speech_probs): + if prob > 0.5 and not in_speech: + in_speech = True + start_idx = i * window_size + elif prob <= 0.5 and in_speech: + in_speech = False + end_idx = i * window_size + speech_chunks.append((start_idx, end_idx)) + + # 转换为毫秒 + speech_chunks_ms = [(s * 1000 / sr, e * 1000 / sr) for s, e in speech_chunks] + + return speech_chunks_ms + + def _remove_speech_chunks( + self, + waveform: torch.Tensor, + speech_chunks: list, + ) -> torch.Tensor: + """去除人声片段""" + sr = 16000 + + # 将时间转换为样本索引 + speech_ranges = [(int(s * sr / 1000), int(e * sr / 1000)) for s, e in speech_chunks] + + # 创建掩码(True 表示保留) + mask = torch.ones(waveform.shape[1], dtype=torch.bool) + + for start, end in speech_ranges: + mask[start:end] = False + + # 应用掩码 + cleaned_waveform = waveform[:, mask] + + return cleaned_waveform + + +# 5th Place 关键技术总结 +""" +关键发现(来自 5th Place writeup): + +1. **Silero VAD 有效**: + - 去除人声减少误检 + - 清洗后数据质量提升 + +2. **三阶段自蒸馏**: + - Stage 1: 基础训练 + - Stage 2: train_audio 自蒸馏 + - Stage 3: 加入 soundscape 伪标签 + - 每个阶段都带来提升 + +3. **丰富次要标签**: + - 许多音频包含未标注的鸟叫声 + - 使用帧级预测和时间平均 + - 迭代训练发现更多标签 + +4. **数据平衡重要**: + - 样本 <20 的类别复制到 20 + - 样本 <30 的类别手动筛选 + - 使用前 30/60 秒数据 +""" +``` + +#### 1st Place - Multi-Iterative Noisy Student 详细流程 + +**作者**: Nikita Babych +**核心创新**: 多迭代 Noisy Student + MixUp + 幂次变换 + +**完整流程:** + +```python +import torch +import torch.nn as nn +import numpy as np + +class MultiIterativeNoisyStudent: + """ + 1st Place 方案:多迭代 Noisy Student 自训练 + + 核心创新: + 1. 多迭代自训练,每次使用 MixUp + 2. 伪标签幂次变换减少噪声 + 3. 滑动窗口推理,帧预测平均 + """ + + def __init__( + self, + model: nn.Module, + num_classes: int = 206, + num_iterations: int = 3, + ): + self.model = model + self.num_classes = num_classes + self.num_iterations = num_iterations + + # 1st Place 关键参数 + self.mixup_alpha = 0.5 + self.power_transform = 1.5 # 幂次变换参数(减少伪标签噪声) + + def train_iteration( + self, + train_audio_loader, + train_soundscape_files, + iteration: int, + epochs=30, + ): + """ + 执行一次 Noisy Student 迭代 + + Args: + iteration: 当前迭代编号(0, 1, 2, ...) + """ + print(f"=== Noisy Student Iteration {iteration + 1} ===") + + # 准备数据 + # 50% train_audio + 50% 伪标签 soundscape + if iteration == 0: + # 第一次迭代:只使用 train_audio + train_loader = train_audio_loader + else: + # 后续迭代:混合 train_audio 和伪标签 + train_loader = self._prepare_mixed_data( + train_audio_loader, + train_soundscape_files, + iteration, + ) + + # 优化器 + optimizer = torch.optim.AdamW(self.model.parameters(), lr=1e-3) + scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( + optimizer, T_max=epochs, eta_min=1e-6 + ) + criterion = nn.BCEWithLogitsLoss() + + best_val_loss = float('inf') + + for epoch in range(epochs): + self.model.train() + train_loss = 0 + + for batch in train_loader: + mel_spec = batch['mel_spec'] + labels = batch['labels'] + + # MixUp 数据增强(1st Place 关键) + if np.random.rand() < 0.5: # 50% 概率应用 MixUp + mel_spec, labels = self._apply_mixup(mel_spec, labels) + + # 前向传播 + logits = self.model(mel_spec) + loss = criterion(logits, labels) + + # 反向传播 + optimizer.zero_grad() + loss.backward() + optimizer.step() + + train_loss += loss.item() + + # 验证(使用训练集的一个子集) + val_loss = self._quick_validate(train_audio_loader, criterion) + + scheduler.step() + + print(f"Iteration {iteration+1}, Epoch {epoch+1}/{epochs}, " + f"Train Loss: {train_loss/len(train_loader):.4f}, " + f"Val Loss: {val_loss:.4f}") + + if val_loss < best_val_loss: + best_val_loss = val_loss + torch.save(self.model.state_dict(), + f"checkpoints/noisy_student_iter{iteration}_best.pth") + + print(f"Iteration {iteration+1} complete. Best Val Loss: {best_val_loss:.4f}") + + def _prepare_mixed_data( + self, + train_audio_loader, + soundscape_files, + iteration: int, + ): + """ + 准备混合数据:train_audio + 伪标签 soundscape + + 关键:幂次变换减少伪标签噪声(1st Place 创新) + """ + # 生成伪标签 + pseudo_labels = self._generate_pseudo_labels_power_transform( + soundscape_files, + self.power_transform, + ) + + # 创建混合 dataloader + mixed_data = [] + + # 添加 train_audio + for batch in train_audio_loader: + mixed_data.append(batch) + + # 添加伪标签 soundscape + for item in pseudo_labels: + mixed_data.append(item) + + # 打乱顺序 + np.random.shuffle(mixed_data) + + return mixed_data + + def _generate_pseudo_labels_power_transform( + self, + audio_files: list, + power: float = 1.5, + ) -> list: + """ + 生成伪标签并应用幂次变换 + + 1st Place 关键创新:幂次变换减少噪声 + + 原理: + - 直接对概率进行温度缩放会提高噪声的概率 + - 通过幂次变换,防止噪声的放大,并保留重要的标签信号 + """ + self.model.eval() + pseudo_labels = [] + + with torch.no_grad(): + for audio_file in audio_files: + waveform, sr = torchaudio.load(audio_file) + + # 分段处理(5秒窗口) + segments = self._split_audio(waveform, sr) + + for segment in segments: + mel_spec = self._extract_mel(segment) + + # 获取预测 + logits = self.model(mel_spec) + probs = torch.sigmoid(logits).squeeze(0).cpu().numpy() # (206,) + + # 幂次变换(1st Place 创新) + # power > 1: 压缩低概率,扩展高概率 + # power < 1: 扩展低概率,压缩高概率 + probs_transformed = np.power(probs, power) + + pseudo_labels.append({ + "mel_spec": mel_spec, + "labels": torch.tensor(probs_transformed, dtype=torch.float32), + }) + + return pseudo_labels + + def _apply_mixup( + self, + mel_spec: torch.Tensor, + labels: torch.Tensor, + ) -> tuple: + """ + MixUp 数据增强 + + 1st Place 关键:使用固定混合权重 0.5 + """ + batch_size = mel_spec.size(0) + + # 生成混合权重 + lam = np.random.beta(self.mixup_alpha, self.mixup_alpha) + # 1st Place 发现固定权重 0.5 效果更好 + # lam = 0.5 + + # 随机排列 + index = torch.randperm(batch_size) + + # 混合特征 + mixed_mel = lam * mel_spec + (1 - lam) * mel_spec[index] + + # 混合标签(取最大值) + mixed_labels = torch.maximum(labels, labels[index]) + + return mixed_mel, mixed_labels + + def _split_audio(self, waveform: torch.Tensor, sr: int) -> list: + """分段处理音频""" + segment_samples = 5 * sr + segments = [] + + for i in range(0, waveform.shape[1], segment_samples): + segment = waveform[:, i:i+segment_samples] + if segment.shape[1] == segment_samples: + segments.append(segment) + else: + padding = segment_samples - segment.shape[1] + segment = torch.nn.functional.pad(segment, (0, padding)) + segments.append(segment) + + return segments + + def _extract_mel(self, waveform: torch.Tensor) -> torch.Tensor: + """提取 mel-spectrogram""" + pass + + def _quick_validate(self, val_loader, criterion): + """快速验证""" + self.model.eval() + total_loss = 0 + count = 0 + + with torch.no_grad(): + for i, batch in enumerate(val_loader): + if i >= 10: # 只验证前 10 个 batch + break + + mel_spec = batch['mel_spec'] + labels = batch['labels'] + + logits = self.model(mel_spec) + loss = criterion(logits, labels) + total_loss += loss.item() + count += 1 + + return total_loss / max(count, 1) + + +# 1st Place 关键技术总结 +""" +关键发现(来自 1st Place writeup): + +1. **多迭代 Noisy Student 有效**: + - 每次迭代都带来提升 + - 3 次迭代是最优的 + - 更多迭代可能导致噪声累积 + +2. **幂次变换是关键**: + - 直接使用伪标签:性能提升有限 + - 幂次变换(power=1.5):显著提升 + - 防止噪声放大,保留信号 + +3. **MixUp 策略**: + - 固定权重 0.5 比随机权重更稳定 + - 迫使模型学习更鲁棒的特征 + - 减少过拟合 + +4. **滑动窗口推理**: + - 使用帧预测的平均值 + - 避免丢弃有价值的预测数据 + - 类似图像的 2D 滑动窗口分割 +""" + + +# 1st Place 完整训练流程示例 +def train_noisy_student_full_pipeline(): + """ + 完整的 Noisy Student 训练流程 + """ + # 初始化 + model = SEDModel(num_classes=206) + trainer = MultiIterativeNoisyStudent(model, num_iterations=3) + + # 准备数据 + train_audio_loader = ... # 训练音频 loader + soundscape_files = ... # soundscape 文件列表 + + # 迭代 0: 只使用 train_audio + print("=== Iteration 0: Training on train_audio only ===") + trainer.train_iteration(train_audio_loader, soundscape_files, iteration=0, epochs=30) + + # 迭代 1: 加入伪标签 soundscape + print("=== Iteration 1: Adding pseudo-labeled soundscape ===") + trainer.train_iteration(train_audio_loader, soundscape_files, iteration=1, epochs=30) + + # 迭代 2: 使用新的伪标签 + print("=== Iteration 2: Refreshing pseudo labels ===") + trainer.train_iteration(train_audio_loader, soundscape_files, iteration=2, epochs=30) + + # 最终集成:使用不同迭代的模型 + model_iter0 = SEDModel(num_classes=206) + model_iter0.load_state_dict(torch.load("checkpoints/noisy_student_iter0_best.pth")) + + model_iter1 = SEDModel(num_classes=206) + model_iter1.load_state_dict(torch.load("checkpoints/noisy_student_iter1_best.pth")) + + model_iter2 = SEDModel(num_classes=206) + model_iter2.load_state_dict(torch.load("checkpoints/noisy_student_iter2_best.pth")) + + # 集成预测 + def ensemble_predict(mel_spec): + pred0 = torch.sigmoid(model_iter0(mel_spec)) + pred1 = torch.sigmoid(model_iter1(mel_spec)) + pred2 = torch.sigmoid(model_iter2(mel_spec)) + + # 简单平均 + ensemble_pred = (pred0 + pred1 + pred2) / 3 + return ensemble_pred + + return ensemble_predict +``` + +#### 4th Place - Soft AUC Loss 详细分析 + +**作者**: dylan.liu +**核心创新**: 支持软标签的 AUC 损失函数 + +**问题背景:** +- 标准 AUC 损失函数不支持软标签(适用于知识蒸馏和半监督学习) +- 4th Place 通过自定义 soft AUC loss 解决这个问题 +- 效果:从 11 名跃升至 4 名(LB 从 0.850 → 0.901) + +```python +import torch +import torch.nn as nn +import torch.nn.functional as F + +class SoftAUCLoss_v4(nn.Module): + """ + 4th Place Soft AUC Loss 实现 + + 参考:4th Place writeup + 效果:LB 从 0.850 → 0.901(显著提升) + + 核心思想: + 1. 支持 soft labels(适用于知识蒸馏和半监督学习) + 2. 通过正负样本对的排序关系优化 AUC + 3. 减少 overfitting + """ + + def __init__( + self, + margin: float = 1.0, + reduction: str = "mean", + ): + super().__init__() + self.margin = margin + self.reduction = reduction + + def forward( + self, + predictions: torch.Tensor, + targets: torch.Tensor, + ) -> torch.Tensor: + """ + Args: + predictions: (batch, num_classes) - 原始 logits + targets: (batch, num_classes) - 软标签 [0, 1] + + Returns: + AUC loss + """ + probs = torch.sigmoid(predictions) + num_classes = predictions.size(1) + losses = [] + + for c in range(num_classes): + prob_c = probs[:, c] # (batch,) + target_c = targets[:, c] # (batch,) + + # 计算所有样本对的差异 + # 对于软标签,我们需要加权处理 + + # 创建样本对矩阵 + # target_diff > 0 表示 target_i > target_j + target_diff = target_c.unsqueeze(1) - target_c.unsqueeze(0) # (batch, batch) + prob_diff = prob_c.unsqueeze(1) - prob_c.unsqueeze(0) # (batch, batch) + + # 只考虑 target_i > target_j 的对 + # 即正样本(高 target)应该有更高的预测概率 + mask = target_diff > 0 + + if mask.sum() > 0: + # Hinge loss: max(0, margin - (prob_i - prob_j)) + # 理想情况:prob_i > prob_j(正样本预测高于负样本) + # margin - (prob_i - prob_j) 应该 <= 0 + # 如果 > 0,说明违反了排序关系 + + loss_c = F.relu(self.margin - prob_diff)[mask].mean() + + # 4th Place 发现加权版本更有效 + # 使用 target_diff 作为权重 + # weight = target_diff[mask] + # weighted_loss = F.relu(self.margin - prob_diff)[mask] * weight + # loss_c = weighted_loss.sum() / weight.sum() + + losses.append(loss_c) + + if len(losses) == 0: + return torch.tensor(0.0, device=predictions.device, requires_grad=True) + + losses = torch.stack(losses) + + if self.reduction == "mean": + return losses.mean() + elif self.reduction == "sum": + return losses.sum() + else: + return losses + + +class SoftAUCLoss_Advanced(nn.Module): + """ + 改进的 Soft AUC Loss + + 结合 4th Place 的发现和其他优化: + 1. 温度缩放 + 2. 自适应 margin + 3. 类别加权 + """ + + def __init__( + self, + margin: float = 1.0, + temperature: float = 1.0, + use_class_weighting: bool = True, + ): + super().__init__() + self.margin = margin + self.temperature = temperature + self.use_class_weighting = use_class_weighting + + def forward( + self, + predictions: torch.Tensor, + targets: torch.Tensor, + ) -> torch.Tensor: + """ + Args: + predictions: (batch, num_classes) + targets: (batch, num_classes) - 软标签 + """ + # 温度缩放 + probs = torch.sigmoid(predictions / self.temperature) + num_classes = predictions.size(1) + + losses = [] + + for c in range(num_classes): + prob_c = probs[:, c] + target_c = targets[:, c] + + # 样本对矩阵 + target_diff = target_c.unsqueeze(1) - target_c.unsqueeze(0) + prob_diff = prob_c.unsqueeze(1) - prob_c.unsqueeze(0) + + # mask: target_i > target_j + mask = target_diff > 0 + + if mask.sum() > 0: + # Hinge loss + base_loss = F.relu(self.margin - prob_diff)[mask] + + # 可选:使用 target_diff 作为权重 + # 这给予高 target 差异的样本对更高权重 + weights = target_diff[mask] + weighted_loss = base_loss * weights + + loss_c = weighted_loss.sum() / weights.sum() + + # 可选:类别权重(处理长尾分布) + if self.use_class_weighting: + # 稀有类别更高权重 + class_weight = self._get_class_weight(c, num_classes) + loss_c = loss_c * class_weight + + losses.append(loss_c) + + if len(losses) == 0: + return torch.tensor(0.0, device=predictions.device, requires_grad=True) + + return torch.stack(losses).mean() + + def _get_class_weight(self, class_idx: int, num_classes: int) -> float: + """ + 计算类别权重(处理长尾分布) + + 简单版本:可以基于样本频率 + """ + # 这里使用简单策略:可以替换为实际的类别频率 + #稀有类获得更高权重 + return 1.0 # 可以自定义 + + +# 4th Place 关键发现总结 +""" +关键发现(来自 4th Place writeup): + +1. **Soft AUC Loss 显著提升性能**: + - LB 从 0.850 → 0.901 + - 排名从 11 名 → 4 名 + - +0.05 AUC 提升是巨大的 + +2. **为什么 Soft AUC Loss 有效**: + - 标准 AUC loss 只支持硬标签(0 或 1) + - Soft AUC Loss 支持软标签(0 到 1 之间) + - 适用于知识蒸馏和半监督学习 + - 减少 overfitting + +3. **实现细节**: + - 使用样本对的排序关系 + - Hinge loss: max(0, margin - (prob_i - prob_j)) + - 只考虑 target_i > target_j 的对 + - margin 通常设为 1.0 + +4. **适用场景**: + - 半监督学习(伪标签) + - 知识蒸馏(软标签) + - 长尾分布(稀有类别) + - 标签噪声(软标签更鲁棒) + +5. **与其他损失函数对比**: + - BCE Loss: 简单但易过拟合 + - Focal Loss: 处理类别不平衡,但不优化 AUC + - Soft AUC Loss: 直接优化 AUC,支持软标签 +""" + + +# 使用示例 +def train_with_soft_auc_loss(): + """使用 Soft AUC Loss 训练""" + + model = SEDModel(num_classes=206) + + # 标准训练:BCE Loss + criterion_bce = nn.BCEWithLogitsLoss() + + # 半监督训练:Soft AUC Loss + criterion_soft_auc = SoftAUCLoss_v4(margin=1.0) + + # 优化器 + optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3) + + # 训练循环 + for epoch in range(30): + model.train() + + for batch in train_loader: + mel_spec = batch['mel_spec'] + labels = batch['labels'] # 可能是软标签 + + # 选择损失函数 + if batch.get('is_pseudo', False): # 伪标签数据 + # 使用 Soft AUC Loss + loss = criterion_soft_auc(model(mel_spec), labels) + else: # 真实标签 + # 可以使用 BCE Loss 或 Soft AUC Loss + loss = criterion_bce(model(mel_spec), labels) + + optimizer.zero_grad() + loss.backward() + optimizer.step() + + print(f"Epoch {epoch+1}/30, Loss: {loss.item():.4f}") +``` + +--- + +## Best Practices + +### 时间序列分类竞赛策略 + +| 策略 | 何时使用 | 说明 | +|------|---------|------| +| **CWT over STFT** | 非平稳信号 | CWT提供更好的时间-频率局部化 | +| **Entmax over Softmax** | 标签稀疏时 | Entmax产生更稀疏的输出 | +| **非负线性回归集成** | 多模型集成时 | 即使过拟合也能保持相关性 | +| **2-Stage Training** | 标签质量不均时 | Stage1全数据,Stage2高质量样本 | +| **Group K-Fold** | 有重复样本时 | 确保同一patient/EEG不分散 | +| **仅用高质量样本** | 评估时 | 使用votes≥10的样本建立验证集 | + +### 时频分析方法对比 + +| 方法 | 优点 | 缺点 | 适用场景 | +|------|------|------|---------| +| **STFT** | 简单,易实现 | 固定窗口,时频分辨率权衡 | 平稳信号 | +| **CWT** | 多分辨率分析,捕捉局部特征 | 需要选择小波函数 | 非平稳信号,EEG | +| **Superlet** | 最高时频分辨率 | 计算成本高 | 复杂脑波模式 | + +### 频率配置经验 + +| 配置 | 范围 | 说明 | +|------|------|------| +| **标准CWT** | 0.5-20 Hz | Kaggle提供的spectrogram默认范围 | +| **扩展CWT** | 0.5-40 Hz | 更好的结果 (suguuuuu) | +| **带通滤波** | 0.5-40 Hz | 高频噪声增加'other'投票 | + +### 数据增强策略 + +**时间序列 (1D):** +- 随机时间偏移 (±5秒) +- 随机带通滤波 (不同频率范围) +- 通道翻转 (水平/垂直) +- 幅值缩放 + +**Scalogram/Spectrogram (2D):** +- XYMasking (随机遮挡) +- Mixup +- 时间方向翻转 + +### Backbone选择 + +**时间序列 (1D):** +- 1D CNN + GRU +- Transformer (Time-series Transformer) +- LSTM/GRU + +**Scalogram (2D):** +- SwinV2: swinv2_tiny_window16 (最佳: CV 0.2229) +- MaxVIT: maxvit_base_tf_512 +- ConvNeXt: convnextv2_atto + +### 标签处理技巧 + +| 技巧 | 效果 | +|------|------| +| 标签平滑 (加0.02 offset) | 使低投票数标签获得更强正则化 | +| 仅用votes≥10评估 | CV/LB相关性接近1:1 | +| 投票数归一化 | 多专家投票转换为分布 | + +### 常见误区 + +| 误区 | 正确做法 | +|------|---------| +| STFT不够好就放弃时频分析 | 尝试CWT或Superlet | +| Softmax输出不够稀疏 | 使用Entmax | +| 集成权重手动调参 | 使用非负线性回归 | +| 用全部样本验证 | 仅用高质量样本 (votes≥10) | +| 忽略Group K-Fold | 防止同一patient的数据泄露 | + |
