UPI System Design

UPI System Design - National & International Payments

Comprehensive system design for Unified Payment Interface supporting domestic and cross-border transactions

Overview

UPI (Unified Payment Interface) is India’s real-time payment system that enables instant money transfers between bank accounts through mobile devices. This design covers both national UPI operations and international payment integrations.

Key Statistics (2024):

• ~172 billion transactions in 2024 (NPCI); record 16.73 billion in Dec 2024 (~540M/day)
• Hundreds of millions of users active monthly
• Real-time processing (a few seconds end-to-end)
• High success rate for domestic transactions (exact rate varies by month/bank)

---

High-Level Architecture

National UPI Architecture

[Mobile App] → [PSP] → [NPCI Switch] → [Remitter Bank] ──debit──→ [NPCI Switch] → [Beneficiary Bank] → [Beneficiary Account]
     ↓           ↓           ↓              ↓                                        ↓
[UPI PIN    → [Device  → [VPA resolve,  → [Verify PIN                         → [Real-time Credit]
 encrypted     binding]    routing,        + balance]
 on device]                settlement]

International Payment Architecture

[Indian UPI] → [NPCI International (NIPL)] → [Partner Fast-Payment System] → [Foreign Bank]
     ↓              ↓                              ↓                          ↓
[INR Account] → [Currency Exchange] → [Direct system-to-system link,     → [Foreign Currency]
                                       e.g. UPI–PayNow (not SWIFT)]

---

🇮🇳 National UPI System Components

1. NPCI (National Payments Corporation of India)

class NPCISwitch:
    def __init__(self):
        self.participating_banks = self.load_bank_registry()
        self.transaction_router = TransactionRouter()
        self.fraud_detection = FraudDetectionEngine()
        self.settlement_engine = SettlementEngine()
    
    def process_transaction(self, upi_request):
        # 1. Validate request format
        if not self.validate_upi_format(upi_request):
            return self.create_error_response("INVALID_FORMAT")
        
        # 2. Resolve VPAs to banks (payee VPA resolved via payee PSP)
        remitter_bank = self.route_transaction(upi_request.payer_vpa)
        destination_bank = self.route_transaction(upi_request.payee_vpa)
        
        # 3. Fraud check
        fraud_score = self.fraud_detection.analyze(upi_request)
        if fraud_score > 0.8:
            return self.create_error_response("SUSPECTED_FRAUD")
        
        # 4. Debit first: remitter (issuer) bank decrypts/validates the UPI PIN,
        #    checks balance and debits the payer
        debit = remitter_bank.debit_account(upi_request)  # carries encrypted PIN block
        if debit.status != "SUCCESS":
            return debit
        
        # 5. Then credit the beneficiary (reversal of debit if credit fails)
        response = destination_bank.credit_account(upi_request)
        if response.status != "SUCCESS":
            remitter_bank.reverse_debit(upi_request)
        
        # 6. Interbank settlement (deferred net settlement, in cycles)
        if response.status == "SUCCESS":
            self.settlement_engine.queue_settlement(upi_request)
        
        return response

2. Payment Service Providers (PSPs)

class PaymentServiceProvider:
    def __init__(self, bank_code):
        self.bank_code = bank_code
        self.npci_client = NPCIClient()
        self.customer_db = CustomerDatabase()
        self.transaction_log = TransactionLogger()
    
    def initiate_payment(self, payment_request):
        # 1. Verify device binding / app session (PSP never sees the PIN or balance:
        #    the UPI PIN is encrypted on the device by NPCI's Common Library and
        #    validated by the payer's issuing (remitter) bank)
        user = self.verify_device_binding(payment_request.user_id, payment_request.device_id)
        if not user:
            return {"status": "FAILED", "error": "DEVICE_NOT_BOUND"}
        
        # 2. Create UPI transaction
        upi_transaction = {
            "payer_vpa": payment_request.payer_vpa,
            "payee_vpa": payment_request.payee_vpa,
            "amount": payment_request.amount,
            "reference_id": self.generate_reference_id(),
            "timestamp": datetime.now(),
            "merchant_id": payment_request.merchant_id,
            "encrypted_pin_block": payment_request.encrypted_pin  # opaque to PSP
        }
        
        # 3. Send to NPCI
        response = self.npci_client.send_transaction(upi_transaction)
        
        # 4. Log transaction
        self.transaction_log.log(upi_transaction, response)
        
        return response

3. UPI Transaction Flow

Step 1: User initiates payment in mobile app
Step 2: User enters UPI PIN; NPCI Common Library encrypts it on the device
Step 3: App sends request to PSP (Payment Service Provider), which checks device binding
Step 4: PSP forwards request to NPCI Switch (resolves payee VPA)
Step 5: NPCI sends debit request to remitter (payer's) bank
Step 6: Remitter bank validates UPI PIN and balance, debits payer
Step 7: NPCI sends credit request; beneficiary bank credits payee account
Step 8: Confirmation sent back through the chain; real-time notification to both parties

---

🇮🇳 IMPS (Immediate Payment Service)

Foundation technology underlying UPI for real-time fund transfers

Key Features:

• Real-time Processing: Sub-second transaction completion
• 24/7 Availability: Works on holidays and weekends
• Interbank Transfers: Seamless bank-to-bank transactions
• Multiple Channels: Mobile, Internet, ATM access
• Robust Security: Multi-factor authentication

IMPS Technical Implementation

class IMPSProcessor:
    def __init__(self):
        self.core_banking_interface = CoreBankingInterface()
        self.npci_gateway = NPCIGateway()
        self.encryption_service = EncryptionService()
    
    def process_imps_transaction(self, transaction):
        # 1. Validate transaction limits
        if not self.validate_transaction_limits(transaction):
            return {"status": "FAILED", "error": "LIMIT_EXCEEDED"}
        
        # 2. Debit payer account
        debit_response = self.core_banking_interface.debit_account(
            account_number=transaction.payer_account,
            amount=transaction.amount,
            reference=transaction.reference_id
        )
        
        if debit_response.status != "SUCCESS":
            return {"status": "FAILED", "error": "DEBIT_FAILED"}
        
        # 3. Send to NPCI for routing
        encrypted_payload = self.encryption_service.encrypt(transaction)
        npci_response = self.npci_gateway.forward_transaction(encrypted_payload)
        
        # 4. Handle response
        if npci_response.status == "SUCCESS":
            return {"status": "SUCCESS", "transaction_id": transaction.reference_id}
        else:
            # Reverse the debit
            self.core_banking_interface.credit_account(
                account_number=transaction.payer_account,
                amount=transaction.amount,
                reference=f"REV_{transaction.reference_id}"
            )
            return {"status": "FAILED", "error": npci_response.error}

IMPS vs UPI Comparison

Feature	IMPS	UPI
Identifier	Account + IFSC	VPA (@bank)
Authentication	OTP + PIN	UPI PIN
Transaction Limit	₹5 Lakh/txn	₹1 Lakh/txn (higher for some categories)
Channels	Mobile, Web, ATM	Mobile Apps, USSD *99#, UPI 123PAY
Launch Year	2010	2016
Adoption	Bank-specific	Universal

---

🌍 International Payment Integration

Cross-Border UPI Architecture

# Note: UPI's cross-border links (e.g. UPI–PayNow, live Feb 2023) are direct links
# between fast-payment systems, run via NPCI International (NIPL) - not SWIFT.
# The SWIFT MT103 / correspondent-bank flow below is the traditional (slower) rail.
class CrossBorderPaymentGateway:
    def __init__(self):
        self.currency_exchange = CurrencyExchangeService()
        self.correspondent_banks = CorrespondentBankRegistry()
        self.compliance_engine = ComplianceEngine()
        self.swift_gateway = SWIFTGateway()
    
    def process_international_payment(self, payment_request):
        # 1. Compliance checks (AML, KYC, sanctions)
        compliance_result = self.compliance_engine.validate(payment_request)
        if not compliance_result.approved:
            return {"status": "BLOCKED", "reason": compliance_result.reason}
        
        # 2. Currency conversion
        exchange_rate = self.currency_exchange.get_rate(
            from_currency="INR",
            to_currency=payment_request.destination_currency
        )
        
        converted_amount = payment_request.amount_inr * exchange_rate
        
        # 3. Route to correspondent bank
        correspondent_bank = self.correspondent_banks.find_optimal_route(
            destination_country=payment_request.destination_country,
            currency=payment_request.destination_currency
        )
        
        # 4. Create SWIFT message
        swift_message = self.create_swift_mt103(
            sender_details=payment_request.sender,
            beneficiary_details=payment_request.beneficiary,
            amount=converted_amount,
            currency=payment_request.destination_currency
        )
        
        # 5. Send through SWIFT network
        swift_response = self.swift_gateway.send_message(swift_message)
        
        return {
            "status": "SENT",
            "swift_reference": swift_response.reference,
            "exchange_rate": exchange_rate,
            "converted_amount": converted_amount,
            "estimated_arrival": swift_response.estimated_arrival
        }

International Payment Corridors

1. UPI-Singapore Integration

# Singapore PayNow integration
class UPIPayNowBridge:
    def process_cross_border_payment(self, upi_request):
        # Convert UPI format to PayNow format
        paynow_request = {
            "sender_proxy": upi_request.payer_vpa,
            "recipient_proxy": upi_request.payee_mobile,  # Singapore mobile
            "amount_sgd": self.convert_inr_to_sgd(upi_request.amount),
            "reference": upi_request.reference_id
        }
        
        # Route through Singapore's FAST network
        return self.singapore_fast_network.process(paynow_request)

2. UPI-UAE Integration

# UAE AANI integration (AANI = UAE's instant payment platform; Arabic for "instant", not an acronym)
class UPIAANIBridge:
    def process_uae_payment(self, upi_request):
        # Convert to UAE banking format
        aani_request = {
            "sender_vpa": upi_request.payer_vpa,
            "recipient_aani": upi_request.payee_aani,  # UAE identifier
            "amount_aed": self.convert_inr_to_aed(upi_request.amount),
            "purpose_code": "P2P_TRANSFER"
        }
        
        # Route through AANI (operated by Al Etihad Payments, a CBUAE subsidiary)
        return self.aani_platform.process(aani_request)

International Compliance Framework

class InternationalComplianceEngine:
    def __init__(self):
        self.sanctions_list = SanctionsDatabase()
        self.aml_rules = AMLRuleEngine()
        self.kyc_service = KYCService()
        self.reporting_service = RegulatoryReportingService()
    
    def validate_international_transaction(self, transaction):
        validation_results = {
            "sanctions_check": False,
            "aml_check": False,
            "kyc_check": False,
            "regulatory_approval": False
        }
        
        # 1. Sanctions screening
        if self.sanctions_list.is_sanctioned(transaction.beneficiary):
            return self.create_blocked_response("SANCTIONS_HIT")
        validation_results["sanctions_check"] = True
        
        # 2. AML checks
        aml_score = self.aml_rules.calculate_risk_score(transaction)
        if aml_score > 75:  # High risk threshold
            return self.create_review_response("HIGH_AML_RISK")
        validation_results["aml_check"] = True
        
        # 3. KYC validation
        kyc_status = self.kyc_service.validate_customer(transaction.sender_id)
        if kyc_status != "VERIFIED":
            return self.create_blocked_response("KYC_INCOMPLETE")
        validation_results["kyc_check"] = True
        
        # 4. Regulatory limits
        if transaction.amount > self.get_regulatory_limit(transaction.destination_country):
            return self.create_blocked_response("REGULATORY_LIMIT_EXCEEDED")
        validation_results["regulatory_approval"] = True
        
        # 5. Report to regulators if required
        if transaction.amount > 50000:  # Report large transactions
            self.reporting_service.report_transaction(transaction)
        
        return {"status": "APPROVED", "validations": validation_results}

---

💱 Currency Exchange & Settlement

Real-time Exchange Rate Management

class CurrencyExchangeService:
    def __init__(self):
        self.rate_providers = [
            ExchangeRateProvider("Reuters"),
            ExchangeRateProvider("Bloomberg"),
            ExchangeRateProvider("RBI")
        ]
        self.rate_cache = RedisCache()
        self.hedging_engine = CurrencyHedgingEngine()
    
    def get_real_time_rate(self, from_currency, to_currency):
        cache_key = f"{from_currency}_{to_currency}"
        
        # Check cache first (30 second TTL)
        cached_rate = self.rate_cache.get(cache_key)
        if cached_rate:
            return cached_rate
        
        # Fetch from multiple providers
        rates = []
        for provider in self.rate_providers:
            try:
                rate = provider.get_rate(from_currency, to_currency)
                rates.append(rate)
            except Exception as e:
                self.logger.error(f"Rate fetch failed from {provider.name}: {e}")
        
        if not rates:
            raise Exception("No exchange rate available")
        
        # Use median rate for stability
        final_rate = statistics.median(rates)
        
        # Cache the rate
        self.rate_cache.set(cache_key, final_rate, ttl=30)
        
        return final_rate
    
    def apply_fx_margin(self, base_rate, currency_pair, transaction_amount):
        """Apply spread based on currency pair and transaction size"""
        margin_config = {
            "INR_USD": {"base": 0.25, "volume_discount": 0.05},
            "INR_EUR": {"base": 0.30, "volume_discount": 0.05},
            "INR_SGD": {"base": 0.20, "volume_discount": 0.03},
            "INR_AED": {"base": 0.15, "volume_discount": 0.02}
        }
        
        config = margin_config.get(currency_pair, {"base": 0.35, "volume_discount": 0})
        
        # Volume-based pricing
        if transaction_amount > 100000:  # Large transaction
            margin = config["base"] - config["volume_discount"]
        else:
            margin = config["base"]
        
        return base_rate * (1 + margin / 100)

Settlement Architecture

class InternationalSettlementEngine:
    def __init__(self):
        self.nostro_accounts = NostroAccountManager()
        self.correspondent_banks = CorrespondentBankNetwork()
        self.settlement_queue = SettlementQueue()
        self.reconciliation_engine = ReconciliationEngine()
    
    def process_settlement(self, international_transaction):
        # 1. Determine settlement route
        settlement_route = self.determine_optimal_route(
            destination_country=international_transaction.destination_country,
            currency=international_transaction.currency,
            amount=international_transaction.amount
        )
        
        # 2. Check nostro account balance
        nostro_balance = self.nostro_accounts.get_balance(
            currency=international_transaction.currency,
            correspondent_bank=settlement_route.correspondent_bank
        )
        
        if nostro_balance < international_transaction.amount:
            # Trigger nostro funding
            self.nostro_accounts.fund_account(
                currency=international_transaction.currency,
                amount=international_transaction.amount * 1.1  # Buffer
            )
        
        # 3. Execute settlement
        settlement_instruction = {
            "amount": international_transaction.amount,
            "currency": international_transaction.currency,
            "beneficiary_bank": international_transaction.beneficiary_bank,
            "beneficiary_account": international_transaction.beneficiary_account,
            "reference": international_transaction.reference,
            "value_date": date.today()
        }
        
        settlement_result = settlement_route.correspondent_bank.execute_settlement(
            settlement_instruction
        )
        
        # 4. Update nostro position
        self.nostro_accounts.update_balance(
            currency=international_transaction.currency,
            correspondent_bank=settlement_route.correspondent_bank,
            amount=-international_transaction.amount
        )
        
        return settlement_result

---

Security Framework

Multi-layered Security Architecture

class UPISecurityFramework:
    def __init__(self):
        self.encryption_service = EncryptionService()
        self.tokenization_service = TokenizationService()
        self.fraud_detection = FraudDetectionEngine()
        self.device_fingerprinting = DeviceFingerprintingService()
    
    def secure_transaction_flow(self, transaction_request):
        security_layers = []
        
        # Layer 1: Device Authentication
        device_auth = self.device_fingerprinting.authenticate(
            device_id=transaction_request.device_id,
            user_id=transaction_request.user_id
        )
        security_layers.append(("DEVICE_AUTH", device_auth.trusted))
        
        # Layer 2: Biometric Validation (if available)
        if transaction_request.biometric_data:
            biometric_result = self.validate_biometric(
                transaction_request.biometric_data,
                transaction_request.user_id
            )
            security_layers.append(("BIOMETRIC", biometric_result.valid))
        
        # Layer 3: UPI PIN Validation (PIN encrypted on device via NPCI Common
        # Library; actually validated by the payer's issuing bank, not the app/PSP)
        pin_result = self.validate_upi_pin(
            transaction_request.encrypted_pin,
            transaction_request.user_id
        )
        security_layers.append(("UPI_PIN", pin_result.valid))
        
        # Layer 4: Transaction Encryption
        encrypted_payload = self.encryption_service.encrypt_aes256(
            data=transaction_request.transaction_data,
            key=self.get_session_key(transaction_request.session_id)
        )
        security_layers.append(("ENCRYPTION", True))
        
        # Layer 5: Fraud Scoring
        fraud_score = self.fraud_detection.calculate_risk_score(
            transaction_request
        )
        security_layers.append(("FRAUD_SCORE", fraud_score < 0.7))
        
        # Final security decision
        security_passed = all(layer[1] for layer in security_layers)
        
        return {
            "security_passed": security_passed,
            "security_layers": security_layers,
            "encrypted_payload": encrypted_payload if security_passed else None,
            "fraud_score": fraud_score
        }

Tokenization for International Payments

class PaymentTokenizationService:
    def __init__(self):
        self.token_vault = SecureTokenVault()
        self.hsm = HardwareSecurityModule()
    
    def tokenize_sensitive_data(self, payment_data):
        """Tokenize sensitive payment information for international transfers"""
        
        # Tokenize account numbers
        tokenized_sender_account = self.token_vault.tokenize(
            data=payment_data.sender_account_number,
            data_type="ACCOUNT_NUMBER"
        )
        
        tokenized_beneficiary_account = self.token_vault.tokenize(
            data=payment_data.beneficiary_account_number,
            data_type="ACCOUNT_NUMBER"
        )
        
        # Create international payment token
        international_payment_token = {
            "token_id": self.generate_unique_token_id(),
            "sender_token": tokenized_sender_account,
            "beneficiary_token": tokenized_beneficiary_account,
            "amount_token": self.tokenize_amount(payment_data.amount),
            "currency": payment_data.currency,
            "created_at": datetime.now(),
            "expires_at": datetime.now() + timedelta(hours=24)
        }
        
        # Store in HSM
        token_reference = self.hsm.store_token(international_payment_token)
        
        return {
            "payment_token": token_reference,
            "token_id": international_payment_token["token_id"],
            "expires_at": international_payment_token["expires_at"]
        }

---

Monitoring & Analytics

Real-time Transaction Monitoring

class UPIMonitoringDashboard:
    def __init__(self):
        self.metrics_collector = MetricsCollector()
        self.alert_manager = AlertManager()
        self.analytics_engine = AnalyticsEngine()
    
    def collect_real_time_metrics(self):
        metrics = {
            # Transaction Metrics
            "total_transactions_per_second": self.get_current_tps(),
            "success_rate": self.calculate_success_rate(),
            "average_response_time": self.get_avg_response_time(),
            
            # International Payment Metrics
            "cross_border_volume": self.get_cross_border_volume(),
            "currency_wise_distribution": self.get_currency_distribution(),
            "corridor_wise_success_rate": self.get_corridor_success_rates(),
            
            # Security Metrics
            "fraud_detection_rate": self.get_fraud_detection_rate(),
            "security_incidents": self.get_security_incidents(),
            "blocked_transactions": self.get_blocked_transactions(),
            
            # Infrastructure Metrics
            "system_availability": self.get_system_availability(),
            "database_performance": self.get_db_performance(),
            "api_gateway_health": self.get_api_gateway_health()
        }
        
        # Check for alerts
        self.check_and_trigger_alerts(metrics)
        
        return metrics
    
    def generate_business_intelligence(self):
        """Generate insights for business stakeholders"""
        
        analytics = {
            "daily_trends": self.analytics_engine.analyze_daily_trends(),
            "user_behavior": self.analytics_engine.analyze_user_patterns(),
            "merchant_insights": self.analytics_engine.analyze_merchant_data(),
            "geographic_distribution": self.analytics_engine.analyze_geographic_patterns(),
            "international_corridors": self.analytics_engine.analyze_corridor_performance(),
            "revenue_optimization": self.analytics_engine.suggest_revenue_optimization()
        }
        
        return analytics

Performance Metrics & SLAs

Metric	National UPI	International	SLA Target
Transaction Success Rate	99.95%	98.5%	> 99%
Response Time	< 2 seconds	< 30 seconds	< 5 seconds
Daily Volume	400M+ txns	1M+ txns	No limit
Peak TPS	100,000+	5,000+	Scale on demand
Fraud Rate	< 0.01%	< 0.05%	< 0.1%
System Uptime	99.99%	99.95%	> 99.9%

Except daily volume (NPCI: ~500M+/day by late 2024), these figures are illustrative design targets, not official NPCI statistics.

---

Scalability & Performance

Auto-scaling Architecture

class UPIAutoScalingManager:
    def __init__(self):
        self.kubernetes_client = KubernetesClient()
        self.metrics_monitor = MetricsMonitor()
        self.load_predictor = LoadPredictionEngine()
    
    def manage_auto_scaling(self):
        # Current system metrics
        current_metrics = self.metrics_monitor.get_current_metrics()
        
        # Predict next 15 minutes load
        predicted_load = self.load_predictor.predict_load(
            horizon_minutes=15,
            current_metrics=current_metrics
        )
        
        # Scaling decisions
        scaling_decisions = []
        
        # Scale API Gateway
        if predicted_load.api_requests_per_second > 50000:
            scaling_decisions.append({
                "service": "api-gateway",
                "action": "scale_up",
                "target_replicas": 20
            })
        
        # Scale Transaction Processors
        if predicted_load.transaction_volume > 80000:
            scaling_decisions.append({
                "service": "transaction-processor",
                "action": "scale_up",
                "target_replicas": 50
            })
        
        # Scale International Payment Gateway
        if predicted_load.international_volume > 5000:
            scaling_decisions.append({
                "service": "international-gateway",
                "action": "scale_up",
                "target_replicas": 10
            })
        
        # Execute scaling
        for decision in scaling_decisions:
            self.kubernetes_client.scale_deployment(
                deployment_name=decision["service"],
                replicas=decision["target_replicas"]
            )
        
        return scaling_decisions

This comprehensive UPI system design now covers both national and international payment systems with detailed implementation examples, security frameworks, compliance requirements, and scalability considerations. The design addresses the complexities of cross-border payments while maintaining the simplicity and speed of domestic UPI transactions.