ml-failfast-validation

ML のフェイルファスト検証

長時間実行する ML 実験を開始する前に問題を検出するための、POC 検証パターンです。

自己進化型スキル：このスキルは使用するたびに改善されます。手順が間違っていた場合、パラメーターがずれていた場合、または回避策が必要だった場合は、先送りせずにこのファイルをすぐに修正してください。実際に再現可能な問題がある場合にのみ更新してください。

このスキルを使うタイミング

次の場合にこのスキルを使用してください。

• 数時間かかる新しい ML 実験を開始するとき
• 本格的なトレーニングの前にモデルアーキテクチャを検証するとき
• 勾配フローとデータパイプラインの整合性を確認するとき
• POC 検証チェックリストを実装するとき
• 予測の崩壊や勾配爆発の問題をデバッグするとき

---

1. なぜフェイルファストが必要か？

フェイルファストなし	フェイルファストあり
4 時間後にクラッシュを発見	30 秒で検出
分かりにくいエラーからデバッグ	明確なエラーメッセージ
GPU 時間を無駄にする	コミット前に検証
データの問題が見過ごされる	スキーマを明示的に確認

原則：コストのかかる計算を始める前に、問題になりうるすべてのことを検証してください。

---

2. POC 検証チェックリスト

最小限の POC（5 項目）

def run_poc_validation():
    """Fast validation before full experiment."""

    print("=" * 60)
    print("FAIL-FAST POC VALIDATION")
    print("=" * 60)

    # [1/5] Model instantiation
    print("\n[1/5] Model instantiation...")
    model = create_model(architecture, input_size=n_features)
    x = torch.randn(32, seq_len, n_features).to(device)
    out = model(x)
    assert out.shape == (32, 1), f"Output shape wrong: {out.shape}"
    print(f"   Input: (32, {seq_len}, {n_features}) -> Output: {out.shape}")
    print("   Status: PASS")

    # [2/5] Gradient flow
    print("\n[2/5] Gradient flow...")
    y = torch.randn(32, 1).to(device)
    loss = F.mse_loss(out, y)
    loss.backward()
    grad_norms = [p.grad.norm().item() for p in model.parameters() if p.grad is not None]
    assert len(grad_norms) > 0, "No gradients!"
    assert all(np.isfinite(g) for g in grad_norms), "NaN/Inf gradients!"
    print(f"   Max grad norm: {max(grad_norms):.4f}")
    print("   Status: PASS")

    # [3/5] NDJSON artifact validation
    print("\n[3/5] NDJSON artifact validation...")
    log_path = output_dir / "experiment.jsonl"
    with open(log_path, "a") as f:
        f.write(json.dumps({"phase": "poc_start", "timestamp": datetime.now().isoformat()}) + "\n")
    assert log_path.exists(), "Log file not created"
    print(f"   Log file: {log_path}")
    print("   Status: PASS")

    # [4/5] Epoch selector variation
    print("\n[4/5] Epoch selector variation...")
    epochs = []
    for seed in [1, 2, 3]:
        selector = create_selector()
        # Simulate different validation results
        for e in range(10, 201, 10):
            selector.record(epoch=e, sortino=np.random.randn() * 0.1, sparsity=np.random.rand())
        epochs.append(selector.select())
    print(f"   Selected epochs: {epochs}")
    assert len(set(epochs)) > 1 or all(e == epochs[0] for e in epochs), "Selector not varying"
    print("   Status: PASS")

    # [5/5] Mini training (10 epochs)
    print("\n[5/5] Mini training (10 epochs)...")
    model = create_model(architecture, input_size=n_features).to(device)
    optimizer = torch.optim.AdamW(model.parameters(), lr=0.0005)
    initial_loss = None
    for epoch in range(10):
        loss = train_one_epoch(model, train_loader, optimizer)
        if initial_loss is None:
            initial_loss = loss
    print(f"   Initial loss: {initial_loss:.4f}")
    print(f"   Final loss: {loss:.4f}")
    print("   Status: PASS")

    print("\n" + "=" * 60)
    print("POC RESULT: ALL 5 CHECKS PASSED")
    print("=" * 60)

拡張 POC（10 項目）

包括的な検証を行うには、以下を追加してください。

# [6/10] Data loading
print("\n[6/10] Data loading...")
df = fetch_data(symbol, threshold)
assert len(df) > min_required_bars, f"Insufficient data: {len(df)} bars"
print(f"   Loaded: {len(df):,} bars")
print("   Status: PASS")

# [7/10] Schema validation
print("\n[7/10] Schema validation...")
validate_schema(df, required_columns, "raw_data")
print("   Status: PASS")

# [8/10] Feature computation
print("\n[8/10] Feature computation...")
df = compute_features(df)
validate_schema(df, feature_columns, "features")
print(f"   Features: {len(feature_columns)}")
print("   Status: PASS")

# [9/10] Prediction sanity
print("\n[9/10] Prediction sanity...")
preds = model(X_test).detach().cpu().numpy()
pred_std = preds.std()
target_std = y_test.std()
pred_ratio = pred_std / target_std
assert pred_ratio > 0.005, f"Predictions collapsed: ratio={pred_ratio:.4f}"
print(f"   Pred std ratio: {pred_ratio:.2%}")
print("   Status: PASS")

# [10/10] Checkpoint save/load
print("\n[10/10] Checkpoint save/load...")
torch.save(model.state_dict(), checkpoint_path)
model2 = create_model(architecture, input_size=n_features)
model2.load_state_dict(torch.load(checkpoint_path))
print("   Status: PASS")

---

3. スキーマ検証パターン

問題

# BAD: Cryptic error 2 hours into experiment
KeyError: 'returns_vs'  # Which file? Which function? What columns exist?

解決策

def validate_schema(df, required: list[str], stage: str) -> None:
    """Fail-fast schema validation with actionable error messages."""
    # Handle both DataFrame columns and DatetimeIndex
    available = list(df.columns)
    if hasattr(df.index, 'name') and df.index.name:
        available.append(df.index.name)

    missing = [c for c in required if c not in available]
    if missing:
        raise ValueError(
            f"[{stage}] Missing columns: {missing}\n"
            f"Available: {sorted(available)}\n"
            f"DataFrame shape: {df.shape}"
        )
    print(f"  Schema validation PASSED ({stage}): {len(required)} columns", flush=True)


# Usage at pipeline boundaries
REQUIRED_RAW = ["open", "high", "low", "close", "volume"]
REQUIRED_FEATURES = ["returns_vs", "momentum_z", "atr_pct", "volume_z",
                     "rsi_14", "bb_pct_b", "vol_regime", "return_accel", "pv_divergence"]

df = fetch_data(symbol)
validate_schema(df, REQUIRED_RAW, "raw_data")

df = compute_features(df)
validate_schema(df, REQUIRED_FEATURES, "features")

---

4. 勾配の健全性チェック

基本的な勾配チェック

def check_gradient_health(model: nn.Module, sample_input: torch.Tensor) -> dict:
    """Verify gradients flow correctly through model."""
    model.train()
    out = model(sample_input)
    loss = out.sum()
    loss.backward()

    stats = {"total_params": 0, "params_with_grad": 0, "grad_norms": []}

    for name, param in model.named_parameters():
        stats["total_params"] += 1
        if param.grad is not None:
            stats["params_with_grad"] += 1
            norm = param.grad.norm().item()
            stats["grad_norms"].append(norm)

            # Check for issues
            if not np.isfinite(norm):
                raise ValueError(f"Non-finite gradient in {name}: {norm}")
            if norm > 100:
                print(f"  WARNING: Large gradient in {name}: {norm:.2f}")

    stats["max_grad"] = max(stats["grad_norms"]) if stats["grad_norms"] else 0
    stats["mean_grad"] = np.mean(stats["grad_norms"]) if stats["grad_norms"] else 0

    return stats

アーキテクチャ固有のチェック

def check_lstm_gradients(model: nn.Module) -> dict:
    """Check LSTM-specific gradient patterns."""
    stats = {}

    for name, param in model.named_parameters():
        if param.grad is None:
            continue

        # Check forget gate bias (should not be too negative)
        if "bias_hh" in name or "bias_ih" in name:
            # LSTM bias: [i, f, g, o] gates
            hidden_size = param.shape[0] // 4
            forget_bias = param.grad[hidden_size:2*hidden_size]
            stats["forget_bias_grad_mean"] = forget_bias.mean().item()

        # Check hidden-to-hidden weights
        if "weight_hh" in name:
            stats["hh_weight_grad_norm"] = param.grad.norm().item()

    return stats

---

5. 予測の妥当性チェック

崩壊の検出

def check_prediction_sanity(preds: np.ndarray, targets: np.ndarray) -> dict:
    """Detect prediction collapse or explosion."""
    stats = {
        "pred_mean": preds.mean(),
        "pred_std": preds.std(),
        "pred_min": preds.min(),
        "pred_max": preds.max(),
        "target_std": targets.std(),
    }

    # Relative threshold (not absolute!)
    stats["pred_std_ratio"] = stats["pred_std"] / stats["target_std"]

    # Collapse detection
    if stats["pred_std_ratio"] < 0.005:  # < 0.5% of target variance
        raise ValueError(
            f"Predictions collapsed!\n"
            f"  pred_std: {stats['pred_std']:.6f}\n"
            f"  target_std: {stats['target_std']:.6f}\n"
            f"  ratio: {stats['pred_std_ratio']:.4%}"
        )

    # Explosion detection
    if stats["pred_std_ratio"] > 100:  # > 100x target variance
        raise ValueError(
            f"Predictions exploded!\n"
            f"  pred_std: {stats['pred_std']:.2f}\n"
            f"  target_std: {stats['target_std']:.6f}\n"
            f"  ratio: {stats['pred_std_ratio']:.1f}x"
        )

    # Unique value check
    stats["unique_values"] = len(np.unique(np.round(preds, 6)))
    if stats["unique_values"] < 10:
        print(f"  WARNING: Only {stats['unique_values']} unique prediction values")

    return stats

相関チェック

def check_prediction_correlation(preds: np.ndarray, targets: np.ndarray) -> float:
    """Check if predictions have any correlation with targets."""
    corr = np.corrcoef(preds.flatten(), targets.flatten())[0, 1]

    if not np.isfinite(corr):
        print("  WARNING: Correlation is NaN (likely collapsed predictions)")
        return 0.0

    # Note: negative correlation may still be useful (short signal)
    print(f"  Prediction-target correlation: {corr:.4f}")
    return corr

---

6. NDJSON ロギングの検証

必須イベントタイプ

REQUIRED_EVENTS = {
    "experiment_start": ["architecture", "features", "config"],
    "fold_start": ["fold_id", "train_size", "val_size", "test_size"],
    "epoch_complete": ["epoch", "train_loss", "val_loss"],
    "fold_complete": ["fold_id", "test_sharpe", "test_sortino"],
    "experiment_complete": ["total_folds", "mean_sharpe", "elapsed_seconds"],
}

def validate_ndjson_schema(log_path: Path) -> None:
    """Validate NDJSON log has all required events and fields."""
    events = {}
    with open(log_path) as f:
        for line in f:
            event = json.loads(line)
            phase = event.get("phase", "unknown")
            if phase not in events:
                events[phase] = []
            events[phase].append(event)

    for phase, required_fields in REQUIRED_EVENTS.items():
        if phase not in events:
            raise ValueError(f"Missing event type: {phase}")

        sample = events[phase][0]
        missing = [f for f in required_fields if f not in sample]
        if missing:
            raise ValueError(f"Event '{phase}' missing fields: {missing}")

    print(f"  NDJSON schema valid: {len(events)} event types")

---

7. POC の所要時間ガイド

チェック	通常の所要時間	最大時間	超過した場合の対応
モデルのインスタンス化	< 1s	5s	デバイスを確認し、モデルサイズを縮小
勾配フロー	< 2s	10s	バッチサイズを確認
スキーマ検証	< 0.1s	1s	データ読み込みを確認
ミニトレーニング（10 エポック）	< 30s	2min	バッチを縮小し、データローダーを確認
POC 全体（10 項目）	< 2min	5min	何か問題があります

---

8. 失敗時の対応ガイド

失敗	考えられる原因	修正方法
形状の不一致	input_size または seq_len が誤っている	特徴量の数を確認
NaN 勾配	LR が高すぎる、初期化不良	LR を下げ、初期化を確認
勾配がゼロ	レイヤーが機能していない、パラメーター不足	モデルアーキテクチャを確認
予測の崩壊	正規化器の問題、損失関数の不良	sLSTM 正規化器を確認
予測の爆発	勾配爆発	勾配クリッピングを追加または強化
スキーマに列がない	データソースが誤っている	fetch 関数を確認
チェックポイントの読み込み失敗	state dict のキーが一致しない	モデルアーキテクチャの一致を確認

---

9. 統合例

def main():
    # Parse args, setup output dir...

    # PHASE 1: Fail-fast POC
    print("=" * 60)
    print("FAIL-FAST POC VALIDATION")
    print("=" * 60)

    try:
        run_poc_validation()
    except Exception as e:
        print(f"\n{'=' * 60}")
        print(f"POC FAILED: {type(e).__name__}")
        print(f"{'=' * 60}")
        print(f"Error: {e}")
        print("\nFix the issue before running full experiment.")
        sys.exit(1)

    # PHASE 2: Full experiment (only if POC passes)
    print("\n" + "=" * 60)
    print("STARTING FULL EXPERIMENT")
    print("=" * 60)

    run_full_experiment()

---

10. 避けるべきアンチパターン

しないこと：「時間を節約する」ために検証を省略する

# BAD: "I'll just run it and see"
run_full_experiment()  # 4 hours later: crash

しないこと：相対的な量に絶対しきい値を使う

# BAD: Absolute threshold
assert pred_std > 1e-4  # Meaningless for returns ~0.001

# GOOD: Relative threshold
assert pred_std / target_std > 0.005  # 0.5% of target variance

しないこと：すべての例外を黙って捕捉する

# BAD: Hides real issues
try:
    result = risky_operation()
except Exception:
    result = default_value  # What went wrong?

# GOOD: Catch specific exceptions
try:
    result = risky_operation()
except (ValueError, RuntimeError) as e:
    logger.error(f"Operation failed: {e}")
    raise

しないこと：flush せずに出力する

# BAD: Output buffered, can't see progress
print(f"Processing fold {i}...")

# GOOD: See output immediately
print(f"Processing fold {i}...", flush=True)

---

参考資料

• データパイプラインでのスキーマ検証
• PyTorch の勾配デバッグ
• NDJSON 仕様

---

トラブルシューティング

問題	原因	解決策
POC で勾配が NaN になる	学習率が高すぎる	LR を 10 分の 1 に下げ、重みの初期化を確認します
勾配がゼロになる	レイヤーが機能していない、またはパラメーターが不足している	モデルのアーキテクチャを確認し、requires_grad=True を検証します
予測が収束する	正規化処理の問題、または損失関数の問題	目的変数の正規化を確認し、損失関数を検証します
予測値が発散する	勾配爆発	勾配クリッピングを追加し、学習率を下げます
スキーマの列が不足している	データソースまたは変換処理が誤っている	fetch 関数が想定される列を返すことを確認します
チェックポイントの読み込みに失敗する	state dict のキーが一致しない	モデルのアーキテクチャが保存済みチェックポイントと一致することを確認します
POC がタイムアウトする（5 分超）	データの読み込み、またはモデルが大きすぎる	バッチサイズを小さくし、DataLoader の num_workers を確認します
ミニトレーニングが進まない	学習率が低すぎる、またはパラメーターが凍結されている	LR を上げ、オプティマイザーがすべてのパラメーターを更新することを確認します
NDJSON の検証に失敗する	必須のイベントタイプが不足している	すべてのフェーズで想定されるフィールドが出力されることを確認します
形状の不一致エラー	input_size または seq_len が誤っている	特徴量の数がモデルの入力次元と一致することを確認します

実行後の振り返り

このスキルの実行が完了したら、終了する前に次の点を確認してください。

1. コマンドは成功しましたか？ — 成功していない場合は、失敗の原因となった指示またはエラーテーブルを修正してください。
2. パラメーターや出力に変更はありましたか？ — 基盤となるツールのインターフェースに変更があった場合は、使用例とパラメーターテーブルを更新して合わせてください。
3. 回避策が必要でしたか？ — 異なるフラグや追加の手順など、臨時の対応が必要だった場合は、次回同じ回避策が必要にならないよう、この SKILL.md を更新してください。

問題が実際に発生し、再現可能な場合にのみ更新してください。推測に基づく更新はしないでください。