test/example_usage.py

"""Example usage of the fraud detection system."""

import sys
from pathlib import Path

# Add parent directory to path to allow importing src modules
sys.path.insert(0, str(Path(__file__).parent.parent))

import logging
import warnings
import os

# Suppress warnings for cleaner output
warnings.filterwarnings('ignore', category=FutureWarning)
warnings.filterwarnings('ignore', category=DeprecationWarning)
warnings.filterwarnings('ignore', category=UserWarning)
warnings.filterwarnings('ignore', message='.*LangChain.*')

from src.data.processor import FraudDataProcessor
from src.llm.groq_client import GroqClient
from src.rag.document_loader import DocumentLoader
from src.rag.vector_store import VectorStore
from src.services.fraud_analyzer import FraudAnalyzer
from src.config.config import settings

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def example_basic_llm():
    """Example: Basic LLM usage."""
    logger.info("=== Example: Basic LLM Usage ===")
    
    client = GroqClient()
    
    response = client.invoke(
        prompt="What are common indicators of credit card fraud?",
        system_message="You are an expert fraud detection analyst.",
    )
    
    print("\nResponse:")
    print(response)
    print("\n")


def example_rag_system():
    """Example: RAG system setup."""
    logger.info("=== Example: RAG System Setup ===")
    
    # Load documents
    document_loader = DocumentLoader(
        chunk_size=settings.chunk_size,
        chunk_overlap=settings.chunk_overlap,
    )
    
    pdf_documents = document_loader.load_pdfs_from_directory(settings.pdf_dir)
    
    if pdf_documents:
        # Create vector store
        vector_store = VectorStore()
        vector_store.add_documents(pdf_documents)
        
        # Search for relevant documents
        query = "What are the main types of payment fraud?"
        results = vector_store.similarity_search(query, k=3)
        
        print(f"\nFound {len(results)} relevant documents for query: {query}")
        for i, doc in enumerate(results, 1):
            print(f"\n--- Document {i} ---")
            print(doc.page_content[:200] + "...")
    else:
        logger.warning("No PDF documents found")
    
    print("\n")


def example_fraud_analysis():
    """Example: Fraud analysis."""
    logger.info("=== Example: Fraud Analysis ===")
    
    # Initialize components
    groq_client = GroqClient()
    
    # Setup RAG (optional)
    vector_store = None
    try:
        document_loader = DocumentLoader()
        pdf_documents = document_loader.load_pdfs_from_directory(settings.pdf_dir)
        if pdf_documents:
            vector_store = VectorStore()
            vector_store.add_documents(pdf_documents)
    except Exception as e:
        logger.warning(f"RAG setup failed: {e}")
    
    # Create analyzer
    analyzer = FraudAnalyzer(
        groq_client=groq_client,
        vector_store=vector_store,
    )
    
    # Analyze a transaction
    try:
        result = analyzer.analyze_transaction(
            transaction_id=0,
            use_rag=vector_store is not None,
        )
        
        print("\n=== Analysis Result ===")
        print(f"Transaction: {result['transaction'].get('merchant', 'N/A')}")
        print(f"\nAnalysis:\n{result['analysis']}")
    except Exception as e:
        logger.error(f"Analysis failed: {e}")
    
    print("\n")


def example_data_processing():
    """Example: Data processing."""
    logger.info("=== Example: Data Processing ===")
    
    processor = FraudDataProcessor()
    
    try:
        # Load data
        train_df = processor.load_train_data()
        print(f"\nLoaded {len(train_df)} training samples")
        
        # Get summary
        summary = processor.get_transaction_summary()
        print(f"\n=== Dataset Summary ===")
        print(f"Total transactions: {summary['total_transactions']}")
        print(f"Fraud count: {summary['fraud_count']}")
        print(f"Fraud percentage: {summary['fraud_percentage']:.2f}%")
        print(f"Average amount: ${summary['average_amount']:.2f}")
        
        # Format a transaction
        if len(train_df) > 0:
            transaction = train_df.iloc[0].to_dict()
            formatted = processor.format_transaction_for_llm(transaction)
            print(f"\n=== Formatted Transaction ===")
            print(formatted)
    except Exception as e:
        logger.error(f"Data processing failed: {e}")
    
    print("\n")


if __name__ == "__main__":
    print("Fraud Detection System - Example Usage\n")
    print("=" * 50)
    
    # Run examples
    try:
        example_basic_llm()
    except Exception as e:
        logger.error(f"Basic LLM example failed: {e}")
    
    try:
        example_data_processing()
    except Exception as e:
        logger.error(f"Data processing example failed: {e}")
    
    try:
        example_rag_system()
    except Exception as e:
        logger.error(f"RAG system example failed: {e}")
    
    try:
        example_fraud_analysis()
    except Exception as e:
        logger.error(f"Fraud analysis example failed: {e}")
    
    print("=" * 50)
    print("\nExamples completed!")