Kaushik's Notes

FinTech System Design - Embedding Use Cases

9 min read

AI-powered solutions for modern payment systems and financial services

Overview

FinTech companies like JusPay leverage embedding technologies and AI to enhance payment processing, fraud detection, customer support, and compliance. These systems handle millions of transactions daily with high accuracy and low latency.

Key Areas:

  • Customer Support: RAG-based intelligent assistance
  • Anomaly Detection: Real-time transaction monitoring
  • Fraud Prevention: ML-powered risk assessment
  • Compliance: AML/KYC automation
  • Personalization: Recommendation systems

Customer Support using RAG Agent

Retrieval-Augmented Generation for intelligent customer service automation

System Architecture

User Query → [Intent Classification] → [RAG Pipeline] → [Response Generation]
     ↓              ↓                      ↓                    ↓
[NLP Processing] → [Vector Search] → [Context Retrieval] → [LLM Response]
                        ↓
                [Knowledge Base] ← [Document Embeddings]

Components

1. Document Processing Pipeline

# Document ingestion and embedding
class DocumentProcessor:
    def __init__(self):
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.vector_db = PineconeVectorDB()
    
    def process_documents(self, documents):
        chunks = []
        for doc in documents:
            # Chunk documents (~256 tokens with overlap): all-MiniLM-L6-v2 has a max
            # sequence length of 256 word pieces - longer chunks are silently truncated
            text_chunks = self.chunk_text(doc.content, chunk_size=256)
            
            for chunk in text_chunks:
                embedding = self.embedding_model.encode(chunk)
                metadata = {
                    'source': doc.source,
                    'category': doc.category,
                    'last_updated': doc.timestamp
                }
                chunks.append({
                    'text': chunk,
                    'embedding': embedding,
                    'metadata': metadata
                })
        
        # Store in vector database
        self.vector_db.upsert(chunks)

2. Query Processing & Retrieval

class RAGAgent:
    def __init__(self):
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.llm = OpenAI(model="gpt-4")
        self.vector_db = PineconeVectorDB()
    
    def answer_query(self, user_query):
        # 1. Generate query embedding
        query_embedding = self.embedding_model.encode(user_query)
        
        # 2. Retrieve relevant contexts
        contexts = self.vector_db.similarity_search(
            query_embedding, 
            top_k=5,
            filter={'category': 'payment_support'}
        )
        
        # 3. Generate response with context
        prompt = f"""
        Context: {' '.join([ctx['text'] for ctx in contexts])}
        
        User Question: {user_query}
        
        Provide a helpful, accurate response based on the context.
        If the information isn't available, say so clearly.
        """
        
        response = self.llm.generate(prompt)
        return {
            'answer': response,
            'sources': [ctx['metadata']['source'] for ctx in contexts],
            'confidence': self.calculate_confidence(contexts, user_query)
        }

Knowledge Base Structure

Example Knowledge Base (illustrative, payment-gateway style):
├── Payment Processing
│   ├── Transaction flows
│   ├── Payment methods (UPI, Cards, Wallets)
│   ├── Error codes and troubleshooting
│   └── Integration guides
├── Compliance & Security
│   ├── PCI DSS requirements
│   ├── RBI guidelines
│   ├── KYC procedures
│   └── Data protection policies
├── APIs & SDKs
│   ├── REST API documentation
│   ├── Mobile SDK guides
│   ├── Webhook implementations
│   └── Testing procedures
└── Business Logic
    ├── Merchant onboarding
    ├── Settlement processes
    ├── Refund policies
    └── Dispute resolution

Performance Metrics (illustrative targets, not published company figures)

  • Response Time: < 2 seconds
  • Accuracy: 95%+ for common queries
  • Coverage: 80%+ query resolution without human intervention
  • Customer Satisfaction: 4.5/5 rating

Anomaly Detection

Real-time transaction monitoring using ML embeddings and statistical analysis

System Architecture

[Transaction Stream] → [Feature Engineering] → [Embedding Generation]
         ↓                      ↓                       ↓
[Real-time Processing] → [Anomaly Detection] → [Alert System]
         ↓                      ↓                       ↓
[Kafka Pipeline] → [ML Models] → [Risk Scoring] → [Action Engine]

Feature Engineering

class TransactionFeatureExtractor:
    def __init__(self):
        self.scaler = StandardScaler()
        self.encoder = LabelEncoder()
    
    def extract_features(self, transaction):
        features = {}
        
        # Temporal features
        features['hour_of_day'] = transaction.timestamp.hour
        features['day_of_week'] = transaction.timestamp.weekday()
        features['is_weekend'] = transaction.timestamp.weekday() >= 5
        
        # Amount-based features
        features['amount'] = transaction.amount
        features['amount_log'] = np.log1p(transaction.amount)
        features['amount_zscore'] = self.calculate_amount_zscore(
            transaction.merchant_id, 
            transaction.amount
        )
        
        # User behavior features
        user_history = self.get_user_history(transaction.user_id, days=30)
        features['avg_transaction_amount'] = user_history.amount.mean()
        features['transaction_frequency'] = len(user_history)
        features['days_since_last_transaction'] = self.days_since_last(
            transaction.user_id
        )
        
        # Merchant features
        features['merchant_category'] = transaction.merchant_category
        features['merchant_risk_score'] = self.get_merchant_risk_score(
            transaction.merchant_id
        )
        
        # Geographic features
        features['location_country'] = transaction.country
        features['location_unusual'] = self.is_unusual_location(
            transaction.user_id, 
            transaction.country
        )
        
        return features

Embedding-based Anomaly Detection

class TransactionEmbeddingModel:
    def __init__(self):
        self.autoencoder, self.encoder = self.build_autoencoder()
        # contamination = expected anomaly fraction; real fraud rates are well under 1%
        # (0.1 would flag 10% of transactions)
        self.isolation_forest = IsolationForest(contamination=0.01)
        
    def build_autoencoder(self):
        """Autoencoder for transaction embeddings"""
        input_dim = 50  # Number of features
        encoding_dim = 16
        
        input_layer = Input(shape=(input_dim,))
        encoded = Dense(32, activation='relu')(input_layer)
        encoded = Dense(encoding_dim, activation='relu')(encoded)
        
        decoded = Dense(32, activation='relu')(encoded)
        decoded = Dense(input_dim, activation='sigmoid')(decoded)
        
        autoencoder = Model(input_layer, decoded)
        encoder = Model(input_layer, encoded)
        
        autoencoder.compile(optimizer='adam', loss='mse')
        return autoencoder, encoder
    
    def detect_anomalies(self, transaction_features):
        # 1. Generate embeddings
        embeddings = self.encoder.predict(transaction_features)
        
        # 2. Reconstruction error
        reconstructed = self.autoencoder.predict(transaction_features)
        reconstruction_error = np.mean(np.square(
            transaction_features - reconstructed
        ), axis=1)
        
        # 3. Isolation Forest: decision_function is centred near 0, LOWER = more
        #    anomalous; its scale differs from MSE, so normalise both to [0, 1]
        isolation_score = self.isolation_forest.decision_function(embeddings)
        if_anomaly = self.normalize(-isolation_score)        # higher = more anomalous
        recon_anomaly = self.normalize(reconstruction_error)
        
        # 4. Combined anomaly score
        anomaly_score = (
            0.6 * recon_anomaly + 
            0.4 * if_anomaly
        )
        
        return {
            'anomaly_score': anomaly_score,
            'is_anomaly': anomaly_score > self.threshold,
            'embedding': embeddings,
            'reconstruction_error': reconstruction_error
        }

Real-time Processing Pipeline

# Kafka consumer for real-time transaction processing
class AnomalyDetectionProcessor:
    def __init__(self):
        self.kafka_consumer = KafkaConsumer('transactions')
        self.feature_extractor = TransactionFeatureExtractor()
        self.anomaly_detector = TransactionEmbeddingModel()
        self.alert_system = AlertSystem()
    
    def process_stream(self):
        for message in self.kafka_consumer:
            transaction = json.loads(message.value)
            
            # Extract features
            features = self.feature_extractor.extract_features(transaction)
            
            # Detect anomalies
            result = self.anomaly_detector.detect_anomalies([features])
            
            if result['is_anomaly']:
                alert = {
                    'transaction_id': transaction['id'],
                    'anomaly_score': result['anomaly_score'],
                    'timestamp': datetime.now(),
                    'features': features,
                    'action_required': self.determine_action(
                        result['anomaly_score']
                    )
                }
                
                self.alert_system.send_alert(alert)
                
                # Take immediate action if high risk
                if result['anomaly_score'] > 0.9:
                    self.block_transaction(transaction['id'])

Anomaly Types Detected

Anomaly TypeDetection MethodAction
Unusual AmountZ-score analysisReview transaction
Velocity FraudFrequency patternsRate limiting
GeographicLocation embeddingAdditional verification
Merchant RiskRisk scoringEnhanced monitoring
Time-basedTemporal patternsTime-based rules
Account TakeoverBehavior changeAccount security check

Avoiding Frauds

Multi-layered fraud prevention using ML, rule engines, and real-time analysis

System Architecture

[Transaction] → [Pre-processing] → [Rule Engine] → [ML Models] → [Decision Engine]
      ↓              ↓                ↓              ↓              ↓
[Enrichment] → [Feature Eng] → [Hard Rules] → [Risk Scoring] → [Final Decision]
      ↓              ↓                ↓              ↓              ↓
[External APIs] → [Embeddings] → [Block/Allow] → [Probability] → [Approve/Decline/Review]

Rule Engine Implementation

class FraudRuleEngine:
    def __init__(self):
        self.rules = self.load_rules()
        self.blacklists = self.load_blacklists()
    
    def evaluate_rules(self, transaction):
        results = []
        
        # Velocity rules
        if self.check_velocity_rules(transaction):
            results.append({
                'rule': 'VELOCITY_EXCEEDED',
                'severity': 'HIGH',
                'action': 'BLOCK'
            })
        
        # Amount rules
        if transaction.amount > 100000:  # > 1 Lakh
            results.append({
                'rule': 'HIGH_AMOUNT',
                'severity': 'MEDIUM',
                'action': 'REVIEW'
            })
        
        # Geographic rules
        if self.is_high_risk_country(transaction.country):
            results.append({
                'rule': 'HIGH_RISK_COUNTRY',
                'severity': 'HIGH',
                'action': 'BLOCK'
            })
        
        # Blacklist checks
        if self.check_blacklists(transaction):
            results.append({
                'rule': 'BLACKLISTED',
                'severity': 'CRITICAL',
                'action': 'BLOCK'
            })
        
        return results
    
    def check_velocity_rules(self, transaction):
        """Check if user exceeds transaction velocity limits"""
        user_id = transaction.user_id
        
        # Last 1 hour
        recent_count = self.get_transaction_count(user_id, hours=1)
        if recent_count > 10:
            return True
        
        # Last 24 hours
        daily_count = self.get_transaction_count(user_id, hours=24)
        if daily_count > 50:
            return True
        
        # Amount velocity
        daily_amount = self.get_transaction_amount(user_id, hours=24)
        if daily_amount > 500000:  # > 5 Lakh per day
            return True
        
        return False

ML-based Fraud Detection

class FraudDetectionML:
    def __init__(self):
        self.models = {
            'xgboost': self.load_xgboost_model(),
            'neural_net': self.load_neural_network(),
            'isolation_forest': self.load_isolation_forest()
        }
        self.feature_importance = self.load_feature_importance()
    
    def predict_fraud(self, transaction_features):
        predictions = {}
        
        # XGBoost prediction
        xgb_prob = self.models['xgboost'].predict_proba(
            [transaction_features]
        )[0][1]
        
        # Neural network prediction
        nn_prob = self.models['neural_net'].predict(
            [transaction_features]
        )[0][0]
        
        # Isolation forest anomaly score (decision_function: lower = more anomalous,
        # centred near 0) -> normalise to [0, 1] with higher = more anomalous
        anomaly_score = self.models['isolation_forest'].decision_function(
            [transaction_features]
        )[0]
        if_anomaly = self.normalize(-anomaly_score)
        
        # Ensemble prediction
        ensemble_prob = (
            0.5 * xgb_prob + 
            0.3 * nn_prob + 
            0.2 * if_anomaly
        )
        
        return {
            'fraud_probability': ensemble_prob,
            'model_predictions': {
                'xgboost': xgb_prob,
                'neural_network': nn_prob,
                'isolation_forest': anomaly_score
            },
            'risk_level': self.calculate_risk_level(ensemble_prob),
            'feature_contributions': self.explain_prediction(
                transaction_features
            )
        }
    
    def calculate_risk_level(self, probability):
        if probability < 0.1:
            return 'LOW'
        elif probability < 0.5:
            return 'MEDIUM'
        elif probability < 0.8:
            return 'HIGH'
        else:
            return 'CRITICAL'

Decision Engine

class FraudDecisionEngine:
    def __init__(self):
        self.rule_engine = FraudRuleEngine()
        self.ml_engine = FraudDetectionML()
        self.risk_policies = self.load_risk_policies()
    
    def make_decision(self, transaction):
        # 1. Rule-based evaluation
        rule_results = self.rule_engine.evaluate_rules(transaction)
        
        # Check for hard blocks
        for rule in rule_results:
            if rule['action'] == 'BLOCK' and rule['severity'] == 'CRITICAL':
                return {
                    'decision': 'DECLINE',
                    'reason': f"Blocked by rule: {rule['rule']}",
                    'confidence': 1.0
                }
        
        # 2. Extract features for ML
        features = self.extract_ml_features(transaction)
        
        # 3. ML prediction
        ml_result = self.ml_engine.predict_fraud(features)
        
        # 4. Combine rule and ML results
        final_decision = self.combine_results(rule_results, ml_result)
        
        return final_decision
    
    def combine_results(self, rule_results, ml_result):
        fraud_prob = ml_result['fraud_probability']
        
        # Adjust probability based on rules
        for rule in rule_results:
            if rule['severity'] == 'HIGH':
                fraud_prob += 0.2
            elif rule['severity'] == 'MEDIUM':
                fraud_prob += 0.1
        
        fraud_prob = min(fraud_prob, 1.0)
        
        # Make final decision
        if fraud_prob > 0.8:
            decision = 'DECLINE'
        elif fraud_prob > 0.5:
            decision = 'REVIEW'
        else:
            decision = 'APPROVE'
        
        return {
            'decision': decision,
            'fraud_probability': fraud_prob,
            'rule_violations': len(rule_results),
            'ml_confidence': ml_result['risk_level'],
            'explanation': self.generate_explanation(rule_results, ml_result)
        }

Feature Engineering for Fraud Detection

class FraudFeatureEngineer:
    def extract_features(self, transaction):
        features = {}
        
        # Transaction features
        features['amount'] = transaction.amount
        features['amount_log'] = np.log1p(transaction.amount)
        features['currency'] = transaction.currency
        features['payment_method'] = transaction.payment_method
        
        # Temporal features
        features['hour'] = transaction.timestamp.hour
        features['day_of_week'] = transaction.timestamp.weekday()
        features['is_weekend'] = transaction.timestamp.weekday() >= 5
        features['is_night'] = transaction.timestamp.hour < 6
        
        # User behavior features
        user_stats = self.get_user_statistics(transaction.user_id)
        features['user_avg_amount'] = user_stats['avg_amount']
        features['user_transaction_count'] = user_stats['transaction_count']
        features['days_since_registration'] = user_stats['days_since_registration']
        features['amount_deviation'] = abs(
            transaction.amount - user_stats['avg_amount']
        ) / user_stats['std_amount']
        
        # Merchant features
        merchant_stats = self.get_merchant_statistics(transaction.merchant_id)
        features['merchant_risk_score'] = merchant_stats['risk_score']
        features['merchant_category'] = merchant_stats['category']
        features['merchant_avg_amount'] = merchant_stats['avg_amount']
        
        # Geographic features
        features['country'] = transaction.country
        features['country_risk_score'] = self.get_country_risk_score(
            transaction.country
        )
        features['is_cross_border'] = (
            transaction.user_country != transaction.merchant_country
        )
        
        # Device features (if available)
        if transaction.device_info:
            features['device_type'] = transaction.device_info.type
            features['browser'] = transaction.device_info.browser
            features['is_mobile'] = transaction.device_info.is_mobile
        
        # Network features
        features['ip_risk_score'] = self.get_ip_risk_score(transaction.ip_address)
        features['is_vpn'] = self.is_vpn_ip(transaction.ip_address)
        features['is_tor'] = self.is_tor_ip(transaction.ip_address)
        
        return features

JusPay-Specific System Design

Architecture below is an illustrative payment-orchestrator design, not JusPay’s documented internals.

Payment Processing Architecture

[Merchant] → [JusPay Gateway] → [Payment Processor] → [Bank/Wallet]
     ↓              ↓                    ↓                ↓
[Checkout] → [Risk Assessment] → [Routing Engine] → [Settlement]
     ↓              ↓                    ↓                ↓
[User] → [Fraud Detection] → [Optimal Route] → [Reconciliation]

Key Components

1. Payment Orchestration

  • Smart Routing: Route to best payment processor
  • Fallback Handling: Retry with alternative routes
  • Cost Optimization: Choose lowest-cost successful route

2. Risk Management

  • Real-time Scoring: ML-based fraud detection
  • Adaptive Rules: Dynamic rule updates
  • Merchant Risk Profiles: Customized risk policies

3. Data Processing

  • Stream Processing: Kafka for real-time events
  • Batch Processing: Apache Spark for analytics
  • Data Lake: Store transaction and user behavior data

Scalability Requirements (illustrative assumptions, not JusPay-published figures)

  • TPS: 10,000+ transactions per second
  • Latency: < 200ms for payment processing
  • Availability: 99.99% uptime
  • Data: Process 1TB+ daily transaction data

Interview Questions for SDE Intern

System Design Questions

  1. Design a real-time fraud detection system for JusPay
  2. Build a payment retry mechanism with smart routing
  3. Create a customer support chatbot using RAG
  4. Design a transaction reconciliation system
  5. Build an anomaly detection pipeline for transactions

Technical Questions

  1. How would you handle high-frequency trading scenarios?
  2. Explain PCI DSS compliance in payment systems
  3. How to ensure idempotency in payment APIs?
  4. Design rate limiting for payment APIs
  5. How to handle eventual consistency in distributed payments?

ML/AI Questions

  1. How to detect concept drift in fraud models?
  2. Explain feature importance in fraud detection
  3. How to handle imbalanced datasets in fraud detection?
  4. Design A/B testing for ML models in production
  5. How to ensure model explainability for regulatory compliance?

This comprehensive guide covers the essential aspects of FinTech system design with a focus on AI/ML applications, specifically tailored for companies like JusPay and suitable for SDE Intern interviews.

Graph View

Backlinks