path: root/.agents/skills/kaggle-learner/references/knowledge/time-series/birdclef-plus-2025.md

# 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的数据泄露 |