scripts/07_anomaly_detection.py

#!/usr/bin/env python
"""Stage 07 — mispricing / anomaly detection prototype.

CRITICAL DATA CONSTRAINT: this uses the labeled TRAIN split's held-out
validation rows, not the test split. test.csv has no price column — it's
what stage 05 predicts for submission — so there is no "actual price" to
compare against for test rows, and residual-based anomaly detection is
structurally impossible there. The validation split (same one stage 04
scores against) is the only labeled data the model wasn't directly
trained on, so it's the right choice for this evaluation.

HONESTY NOTE (do not drop this when reporting results): there are no real
fraud/mispricing labels anywhere for this dataset. This script injects
SYNTHETIC anomalies (src/anomaly/injection.py) as the only available
ground truth. Results measure recovery of injected perturbations, not
real-world fraud detection — state this explicitly in any writeup.

No GPU is required for the detectors themselves (scikit-learn, CPU-only).
A GPU is only used, if available, to get the trained model's predictions
on the validation split; it falls back to CPU automatically otherwise,
and this step is fast regardless since it reuses cached embeddings.

Usage:
    python scripts/07_anomaly_detection.py --config configs/base.yaml
"""

import argparse
import json
import sys
from pathlib import Path

import numpy as np
import torch
from torch.utils.data import DataLoader, random_split

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from src.anomaly.detectors import IsolationForestDetector, ZScoreDetector
from src.anomaly.evaluate import evaluate_detector
from src.anomaly.features import add_embedding_features, build_residual_features
from src.anomaly.injection import inject_price_anomalies
from src.data.dataset import EmbeddingDataset
from src.models.price_model import PriceModel
from src.utils.config import load_config
from src.utils.exceptions import CheckpointError, PricePredictorError
from src.utils.logging import get_logger
from src.utils.seed import set_seed

logger = get_logger(__name__)


def run(config_path: str, output_path: str, anomaly_fraction: float) -> dict:
    config = load_config(config_path)
    set_seed(config["seed"])

    embeddings_dir = Path(config["data"]["embeddings_dir"])
    price_path = embeddings_dir / "train_price.npy"
    if not price_path.exists():
        raise PricePredictorError(
            f"{price_path} not found — run scripts/02_extract_embeddings.py first. "
            "This script requires labeled prices, which only exist for the train split."
        )
    prices = np.load(price_path)

    dataset = EmbeddingDataset(
        text_embeddings_path=str(embeddings_dir / "train_text.npy"),
        image_embeddings_path=str(embeddings_dir / "train_image.npy"),
        prices=prices,
    )

    # Same split logic as scripts/03_train.py / 04_evaluate.py, so this
    # evaluates on rows the model wasn't directly optimized against.
    val_split = config["training"].get("val_split", 0.15)
    n_val = max(1, int(len(dataset) * val_split))
    n_train = len(dataset) - n_val
    generator = torch.Generator().manual_seed(config["seed"])
    _, val_ds = random_split(dataset, [n_train, n_val], generator=generator)
    loader = DataLoader(val_ds, batch_size=256, shuffle=False)

    model = PriceModel.from_config(config)
    checkpoint_path = Path(config["checkpoint_dir"]) / "best.pt"
    if not checkpoint_path.exists():
        raise CheckpointError(f"No checkpoint found at {checkpoint_path} — run scripts/03_train.py first")
    checkpoint = torch.load(checkpoint_path, map_location="cpu")
    model.load_state_dict(checkpoint["model_state_dict"])

    device = "cuda" if torch.cuda.is_available() else "cpu"
    model.to(device)
    model.eval()

    all_actual, all_predicted, all_fused = [], [], []
    with torch.no_grad():
        for text_emb, image_emb, price in loader:
            text_emb, image_emb = text_emb.to(device), image_emb.to(device)
            predicted = model(text_emb, image_emb)
            fused = model.forward_fused(text_emb, image_emb)
            all_actual.append(price.numpy())
            all_predicted.append(predicted.cpu().numpy())
            all_fused.append(fused.cpu().numpy())

    actual_price = np.concatenate(all_actual)
    predicted_price = np.concatenate(all_predicted)
    fused_embeddings = np.concatenate(all_fused)

    logger.info(
        "Collected %d validation rows (actual + predicted price, %d-dim fused embedding)",
        len(actual_price), fused_embeddings.shape[1],
    )

    perturbed_price, is_anomaly = inject_price_anomalies(
        actual_price, fraction=anomaly_fraction, seed=config["seed"]
    )
    logger.info(
        "Injected %d synthetic anomalies (%.1f%% of rows) — this is SYNTHETIC ground truth, "
        "not real fraud labels. Results below measure recovery of these injected "
        "perturbations only.",
        int(is_anomaly.sum()), 100 * anomaly_fraction,
    )

    base_features = build_residual_features(perturbed_price, predicted_price)
    rich_features = add_embedding_features(base_features, fused_embeddings)

    results = {}

    baseline = ZScoreDetector(residual_column=1)
    baseline_scores = baseline.fit_score(base_features)
    results["zscore_baseline"] = evaluate_detector(baseline_scores, is_anomaly)
    logger.info("Z-score baseline:    %s", results["zscore_baseline"])

    iso_forest = IsolationForestDetector(contamination=anomaly_fraction, seed=config["seed"])
    iso_scores = iso_forest.fit_score(rich_features)
    results["isolation_forest"] = evaluate_detector(iso_scores, is_anomaly)
    logger.info("Isolation Forest:    %s", results["isolation_forest"])

    results["_honesty_note"] = (
        "Ground truth is SYNTHETIC (randomly injected price perturbations), "
        "not real fraud/mispricing labels. These metrics measure recovery of "
        "injected anomalies, not real-world fraud detection capability."
    )

    output_file = Path(output_path)
    output_file.parent.mkdir(parents=True, exist_ok=True)
    with output_file.open("w") as f:
        json.dump(results, f, indent=2)
    logger.info("Wrote anomaly detection comparison to %s", output_file)

    return results


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Stage 07: mispricing/anomaly detection prototype (synthetic-label evaluation)"
    )
    parser.add_argument("--config", default="configs/base.yaml")
    parser.add_argument("--output", default="reports/anomaly_detection_comparison.json")
    parser.add_argument("--anomaly-fraction", type=float, default=0.05)
    args = parser.parse_args()

    try:
        run(args.config, args.output, args.anomaly_fraction)
    except PricePredictorError as e:
        logger.error("Anomaly detection failed: %s", e)
        sys.exit(1)
    except Exception as e:
        logger.exception("Unexpected error during anomaly detection: %s", e)
        sys.exit(1)


if __name__ == "__main__":
    main()