test_model.py

#!/usr/bin/env python3
"""
Model Testing Script

Test a trained GNN model on sample data.
"""

import argparse
import logging
import sys
from pathlib import Path

# Add project root to path
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))

import torch
from backend.core.graph_builder import GraphBuilder
from backend.core.gnn_model import GNNWrapper
from backend.core.feature_extractor import GraphFeatureExtractor
from backend.core.scorer import FraudScorer

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def create_sample_claim():
    """Create a sample claim for testing."""
    return {
        "claim_id": "TEST-001",
        "policy_number": "POL-TEST-001",
        "claim_date": "2024-06-15T10:30:00",
        "claim_type": "collision",
        "amount": 8500.0,
        "estimated_damage": 7500.0,
        "status": "pending",
        "description": "Rear-end collision at intersection",
        "driver": {
            "driver_id": "DRV-TEST-001",
            "age": 35,
            "license_number": "LIC-TEST-001",
            "risk_score": 0.15,
            "claims_count": 1,
        },
        "vehicle": {
            "vehicle_id": "VEH-TEST-001",
            "vin": "VIN-TEST-001",
            "make": "Honda",
            "model": "Accord",
            "year": 2020,
            "category": "sedan",
        },
        "location": {
            "latitude": 40.7128,
            "longitude": -74.0060,
            "city": "New York",
            "state": "NY",
        },
        "repair_shop": {
            "shop_id": "SHOP-TEST-001",
            "name": "Test Auto Shop",
            "address": "123 Test St",
            "city": "New York",
            "state": "NY",
            "tier": "B",
        },
        "witnesses": [],
        "police_report_filed": True,
        "injury_reported": False,
        "towed": False,
    }


def test_model(
    checkpoint_path: str,
    claim_data: dict,
    device: str = "cpu",
):
    """Test a trained model on sample data."""
    logger.info(f"Loading model from {checkpoint_path}")

    # Create graph builder
    graph_builder = GraphBuilder(enable_implicit_links=True)

    # Add claim to graph
    logger.info("Adding claim to graph...")
    graph_builder.add_claim(claim_data)
    graph_builder.generate_implicit_links()

    # Load model
    logger.info("Loading GNN model...")
    device = torch.device(device)
    gnn_model = GNNWrapper(device=device)
    gnn_model.load_model(checkpoint_path)

    # Create scorer
    scorer = FraudScorer(gnn_model=gnn_model)

    # Score claim
    logger.info("Scoring claim...")
    score_result = scorer.score_claim(claim_data["claim_id"], graph_builder.graph)

    # Display results
    logger.info("\n" + "=" * 60)
    logger.info("Fraud Detection Results")
    logger.info("=" * 60)
    logger.info(f"Claim ID: {score_result.claim_id}")
    logger.info(f"Fraud Probability: {score_result.fraud_probability:.1%}")
    logger.info(f"Risk Level: {score_result.risk_level}")
    logger.info(f"Confidence: {score_result.confidence:.1%}")
    logger.info(f"\nPrimary Factors:")
    for factor in score_result.primary_factors:
        logger.info(f"  - {factor}")

    if hasattr(score_result, 'network_summary'):
        logger.info(f"\nNetwork Summary:")
        ns = score_result.network_summary
        logger.info(f"  Component Size: {ns.get('component_size', 'N/A')}")
        logger.info(f"  Neighbor Count: {ns.get('neighbor_count', 'N/A')}")
        logger.info(f"  Has Cycles: {ns.get('has_cycles', 'N/A')}")

    return score_result


def parse_arguments():
    """Parse command line arguments."""
    parser = argparse.ArgumentParser(
        description="Test trained Ghost Network model",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument(
        "--checkpoint",
        type=str,
        default="checkpoints/best_model.pt",
        help="Path to model checkpoint",
    )
    parser.add_argument(
        "--device",
        type=str,
        default="cpu",
        choices=["cpu", "cuda"],
        help="Device to run on",
    )
    parser.add_argument(
        "--claim_id",
        type=str,
        default="TEST-001",
        help="Claim ID for test",
    )
    return parser.parse_args()


def main():
    """Main testing function."""
    args = parse_arguments()

    logger.info("=" * 60)
    logger.info("Ghost Network Model Testing")
    logger.info("=" * 60)

    # Check if checkpoint exists
    checkpoint_path = Path(args.checkpoint)
    if not checkpoint_path.exists():
        logger.error(f"Checkpoint not found: {args.checkpoint}")
        logger.info("Please train a model first:")
        logger.info("  python run_training_pipeline.py")
        return 1

    # Create test claim
    claim_data = create_sample_claim()
    claim_data["claim_id"] = args.claim_id

    # Test model
    try:
        result = test_model(
            checkpoint_path=str(checkpoint_path),
            claim_data=claim_data,
            device=args.device,
        )
        logger.info("\n✓ Model testing completed successfully")
        return 0
    except Exception as e:
        logger.exception(f"Model testing failed: {e}")
        return 1


if __name__ == "__main__":
    sys.exit(main())