from fastapi import FastAPI, HTTPException, Query from pydantic import BaseModel from typing import List, Optional, Dict, Any from datetime import datetime, date import re from difflib import SequenceMatcher import uvicorn app = FastAPI( title="Transaction Reconciliation API", description="Reconcile bank and credit card transactions using fuzzy matching", version="1.0.0" ) # Pydantic Models class Transaction(BaseModel): id: str date: str amount: float description: str type: str reference_number: Optional[str] = None class ReconciliationInput(BaseModel): bank_transactions: List[Transaction] credit_card_transactions: List[Transaction] class MatchedTransaction(BaseModel): bank_id: str credit_card_id: str match_score: float match_reason: str description: str amount: float class UnmatchedTransaction(BaseModel): id: str date: str amount: float description: str type: str reference_number: Optional[str] = None class ReconciliationOutput(BaseModel): matched_transactions: List[MatchedTransaction] unmatched_bank_transactions: List[UnmatchedTransaction] unmatched_credit_card_transactions: List[UnmatchedTransaction] class ReconciliationService: def __init__(self, description_threshold: float = 0.7, amount_tolerance: float = 0.01, max_date_diff_days: int = 7): self.description_threshold = description_threshold self.amount_tolerance = amount_tolerance self.max_date_diff_days = max_date_diff_days def fuzzy_match_description(self, desc1: str, desc2: str) -> float: """Calculate fuzzy match score between two descriptions""" # Clean descriptions for better matching clean_desc1 = self._clean_description(desc1.lower()) clean_desc2 = self._clean_description(desc2.lower()) # Use SequenceMatcher for fuzzy matching similarity = SequenceMatcher(None, clean_desc1, clean_desc2).ratio() # Additional check for common transaction patterns if self._check_common_patterns(clean_desc1, clean_desc2): similarity = max(similarity, 0.8) return similarity def _clean_description(self, description: str) -> str: """Clean description for better matching""" # Remove special characters and extra spaces cleaned = re.sub(r'[^\w\s]', ' ', description) cleaned = re.sub(r'\s+', ' ', cleaned).strip() return cleaned def _check_common_patterns(self, desc1: str, desc2: str) -> bool: """Check for common transaction patterns""" patterns = [ (r'uber', r'uber'), (r'amazon|amzn', r'amazon|amzn'), (r'invoice\s*#?\s*(\d+)', r'invoice\s*#?\s*(\d+)'), (r'payment.*invoice', r'payment.*invoice'), (r'trip\s*id\s*(\d+)', r'trip\s*id\s*(\d+)') ] for pattern1, pattern2 in patterns: if re.search(pattern1, desc1) and re.search(pattern2, desc2): return True return False def calculate_date_difference(self, date1: str, date2: str) -> int: """Calculate difference in days between two dates""" try: d1 = datetime.strptime(date1, "%Y-%m-%d").date() d2 = datetime.strptime(date2, "%Y-%m-%d").date() return abs((d1 - d2).days) except ValueError: return float('inf') def amounts_match(self, amount1: float, amount2: float) -> bool: """Check if amounts are close enough to match""" return abs(abs(amount1) - abs(amount2)) <= self.amount_tolerance def types_match(self, bank_type: str, cc_type: str) -> bool: """Check if transaction types match according to business logic""" type_mappings = { ('debit', 'payment'), ('credit', 'receipt'), ('withdrawal', 'payment'), ('deposit', 'receipt') } return (bank_type.lower(), cc_type.lower()) in type_mappings or bank_type.lower() == cc_type.lower() def calculate_match_score(self, bank_txn: Transaction, cc_txn: Transaction) -> tuple[float, str]: """Calculate overall match score and reason""" scores = [] reasons = [] # Amount matching (weight: 0.4) if self.amounts_match(bank_txn.amount, cc_txn.amount): scores.append(0.4) reasons.append("amounts match") else: amount_diff = abs(abs(bank_txn.amount) - abs(cc_txn.amount)) amount_score = max(0, 0.4 * (1 - amount_diff / max(abs(bank_txn.amount), abs(cc_txn.amount)))) scores.append(amount_score) if amount_score > 0.2: reasons.append("amounts close") # Reference number matching (weight: 0.3) if (bank_txn.reference_number and cc_txn.reference_number and bank_txn.reference_number == cc_txn.reference_number): scores.append(0.3) reasons.append("reference numbers match") else: scores.append(0) # Description matching (weight: 0.2) desc_score = self.fuzzy_match_description(bank_txn.description, cc_txn.description) scores.append(0.2 * desc_score) if desc_score >= self.description_threshold: reasons.append("descriptions match") # Date matching (weight: 0.1) date_diff = self.calculate_date_difference(bank_txn.date, cc_txn.date) if date_diff <= self.max_date_diff_days: date_score = 0.1 * (1 - date_diff / self.max_date_diff_days) scores.append(date_score) if date_diff <= 1: reasons.append("dates match") else: reasons.append("dates close") else: scores.append(0) total_score = sum(scores) reason = ", ".join(reasons) if reasons else "partial match" return total_score, reason def reconcile(self, input_data: ReconciliationInput) -> ReconciliationOutput: """Main reconciliation logic""" matched_transactions = [] unmatched_bank = list(input_data.bank_transactions) unmatched_cc = list(input_data.credit_card_transactions) # Find matches for bank_txn in input_data.bank_transactions: best_match = None best_score = 0 best_reason = "" for cc_txn in input_data.credit_card_transactions: # Check if types match first if not self.types_match(bank_txn.type, cc_txn.type): continue score, reason = self.calculate_match_score(bank_txn, cc_txn) # Minimum threshold for considering a match if score >= 0.6 and score > best_score: best_match = cc_txn best_score = score best_reason = reason if best_match: matched_transactions.append(MatchedTransaction( bank_id=bank_txn.id, credit_card_id=best_match.id, match_score=round(best_score, 2), match_reason=best_reason, description=bank_txn.description, amount=bank_txn.amount )) # Remove matched transactions from unmatched lists if bank_txn in unmatched_bank: unmatched_bank.remove(bank_txn) if best_match in unmatched_cc: unmatched_cc.remove(best_match) # Convert remaining unmatched transactions unmatched_bank_list = [ UnmatchedTransaction( id=txn.id, date=txn.date, amount=txn.amount, description=txn.description, type=txn.type, reference_number=txn.reference_number ) for txn in unmatched_bank ] unmatched_cc_list = [ UnmatchedTransaction( id=txn.id, date=txn.date, amount=txn.amount, description=txn.description, type=txn.type, reference_number=txn.reference_number ) for txn in unmatched_cc ] return ReconciliationOutput( matched_transactions=matched_transactions, unmatched_bank_transactions=unmatched_bank_list, unmatched_credit_card_transactions=unmatched_cc_list ) # Initialize service reconciliation_service = ReconciliationService() @app.get("/") async def root(): """Health check endpoint""" return { "message": "Transaction Reconciliation API is running", "status": "healthy", "version": "1.0.0" } @app.post("/reconcile", response_model=ReconciliationOutput) async def reconcile_transactions(input_data: ReconciliationInput): """ Reconcile bank and credit card transactions This endpoint matches transactions based on: - Amount similarity (within tolerance) - Date proximity (within 7 days) - Description fuzzy matching (70% threshold) - Transaction type compatibility - Reference number exact matching """ try: result = reconciliation_service.reconcile(input_data) return result except Exception as e: raise HTTPException(status_code=500, detail=f"Reconciliation failed: {str(e)}") @app.get("/health") async def health_check(): """Health check for deployment""" return {"status": "ok", "service": "Transaction Reconciliation API"} @app.post("/reconcile/custom", response_model=ReconciliationOutput) async def reconcile_with_custom_params( input_data: ReconciliationInput, description_threshold: float = Query(0.7, ge=0, le=1, description="Fuzzy match threshold for descriptions"), amount_tolerance: float = Query(0.01, ge=0, description="Maximum allowed difference in amounts"), max_date_diff_days: int = Query(7, ge=0, description="Maximum allowed date difference in days") ): """ Reconcile transactions with custom matching parameters """ try: custom_service = ReconciliationService( description_threshold=description_threshold, amount_tolerance=amount_tolerance, max_date_diff_days=max_date_diff_days ) result = custom_service.reconcile(input_data) return result except Exception as e: raise HTTPException(status_code=500, detail=f"Reconciliation failed: {str(e)}") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)