#!/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()