src/entity/api/transaction_api.py

from typing import Optional, Literal, ClassVar
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from src.entity.transaction import Transaction
import logging

# Set up logging
logger = logging.getLogger(__name__)

class TransactionApi(BaseModel):
    """
    TransactionApi is a class that represents a transaction.
    """
    transaction_number: str = Field(..., title="Transaction number", description="The ID of the transaction.")
    transaction_timestamp: int = Field(..., title="Timestamp", description="The timestamp of the transaction.")
    transaction_amount: float = Field(..., ge=0, title="Amount", description="The transaction amount.")
    transaction_category: Optional[str] = Field(None, title="Product category", description="The category of product of the transaction.")
    transaction_is_real_fraud: Optional[bool] = Field(None, title="Ground truth", description="True if the transaction really is a fraud (ground truth).")
    merchant_name: str = Field(..., title="Name", description="The name of the merchant.")
    merchant_latitude: float = Field(..., title="Latitude", description="The latitude of the merchant.")
    merchant_longitude: float = Field(..., title="Longitude", description="The longitude of the merchant.")
    customer_credit_card_number: int|str = Field(..., title="Customer's credit card", description="The credit card number of the customer.")
    customer_gender: Optional[Literal['M', 'F']] = Field(None, title="Customer's gender", description="The gender of the customer.")
    customer_first_name: Optional[str] = Field(None, title="Customer's first name", description="The first name of the customer.")
    customer_last_name: Optional[str] = Field(None, title="Customer's last name", description="The last name of the customer.")
    customer_date_of_birth: Optional[str] = Field(None, title="Customer's date of birth", description="The date of birth of the customer.")
    customer_job: Optional[str] = Field(None, title="Customer's job", description="The job of the customer.")
    customer_street: Optional[str] = Field(None, title="Customer's street", description="The street of the customer.")
    customer_city: Optional[str] = Field(None, title="Customer's city", description="The city of the customer.")
    customer_state: Optional[str] = Field(None, title="Customer's state", description="The state of the customer.")
    customer_postal_code: Optional[int] = Field(None, title="Customer's postal code", description="The postal code of the customer.")
    customer_latitude: float = Field(..., title="Customer's latitude", description="The latitude of the customer.")
    customer_longitude: float = Field(..., title="Customer's longitude", description="The longitude of the customer.")
    customer_city_population: Optional[int] = Field(None, ge=0, title="Customer's city population", description="The population of the city.")

    def is_valid(self) -> bool:
        """
        Check if the transaction is valid.
        """
        logger.debug("Validating transaction...")
        now = datetime.now().replace(microsecond=0)
        
        # The transaction amount should be greater than 0
        if self.transaction_amount < 0:
            logger.error("Transaction amount is negative.")
            return False
        
        # Validate the transaction timestamp
        try:
            # Unix timestamps may not be very precise, so we divide by 1000 to get seconds
            dt_object = datetime.fromtimestamp(self.transaction_timestamp // 1000)
            if dt_object > now:
                logger.error(f"Transaction timestamp is in the future ({dt_object} > {now}).")
                return False
        except ValueError:
            logger.error("Invalid transaction timestamp.")
            return False
        
        # Validate the customer date of birth
        # The customer should be 10 years old at least to possess a credit card
        # If the customer date of birth is not provided, we assume the customer is older than 10
        # If the customer date of birth is provided, we check if the customer is older than 10
        if self.customer_date_of_birth:
            try:
                dob = datetime.strptime(self.customer_date_of_birth, "%Y-%m-%d")
                age = (now - dob).days // 365
                if age < 10:
                    logger.error("Customer is younger than 10 years old.")
                    return False
            except ValueError:
                logger.error("Invalid customer date of birth format. Expected YYYY-MM-DD.")
                return False

        return True
    
    def to_transaction(self) -> Transaction:
        return Transaction(
            transaction_number=self.transaction_number,
            transaction_amount=self.transaction_amount,
            transaction_datetime=datetime.fromtimestamp(self.transaction_timestamp / 1000),
            transaction_category=self.transaction_category,
            merchant_name=self.merchant_name,
            merchant_address_latitude=self.merchant_latitude,
            merchant_address_longitude=self.merchant_longitude,
            customer_credit_card_number=self.customer_credit_card_number,
            customer_gender=self.customer_gender,
            customer_firstname=self.customer_first_name,
            customer_lastname=self.customer_last_name,
            customer_dob=self.customer_date_of_birth,
            customer_job=self.customer_job,
            customer_address_street=self.customer_street,
            customer_address_city=self.customer_city,
            customer_address_state=self.customer_state,
            customer_address_zip=self.customer_postal_code,
            customer_address_latitude=self.customer_latitude,
            customer_address_longitude=self.customer_longitude,
            customer_address_city_population=self.customer_city_population,
            is_real_fraud=self.transaction_is_real_fraud,
        )

    class Config:
        json_schema_extra = {
            "example": {
                "transaction_number": "123456789",
                "transaction_timestamp": 1633036800,
                "transaction_amount": 100.0,
                "transaction_category": "Electronics",
                "merchant_name": "Best Buy",
                "merchant_latitude": 37.7749,
                "merchant_longitude": -122.4194,
                "customer_credit_card_number": "4111111111111111",
                "customer_gender": "M",
                "customer_first_name": "John",
                "customer_last_name": "Doe",
                "customer_date_of_birth": "1980-01-01",
                "customer_job": "Engineer",
                "customer_street": "123 Main St",
                "customer_city": "San Francisco",
                "customer_state": "CA",
                "customer_postal_code": 94105,
                "customer_latitude": 37.7749,
                "customer_longitude": -122.4194,
                "customer_city_population": 870000,
            }
        }

class TransactionProcessingOutput(BaseModel):
    """
    TransactionProcessingOutput is a class that represents the output of the transaction processing endpoint.
    It contains the fraud detection result and the fraud score.
    This class is used as the response model for the /transaction/process endpoint.
    """
    is_fraud: int = Field(description="The prediction result. 1 if the transaction is detected as fraudulent, 0 otherwise.")
    fraud_score: float = Field(description="Probability of transaction being fraudulent.")

    model_config: ClassVar[ConfigDict] = ConfigDict(
        json_schema_extra = {
            "example": {
                "is_fraud": 1,
                "fraud_score": 0.85
            }
        })