Open Banking: PSD2, API Standards, and Ecosystem

# Example: UK Open Banking Account Information API

# Get Accounts
GET /accounts
Authorization: Bearer {access_token}
x-fapi-interaction-id: {uuid}

# Response
{
  "Data": {
    "Account": [
      {
        "AccountId": "ACC-123456",
        "Status": "Enabled",
        "StatusReason": "Account is active",
        "AccountType": "Personal",
        "AccountSubType": "CurrentAccount",
        "Nickname": "Main Account",
        "OpeningDate": "2020-01-15",
        "MaturityDate": null,
        "Account": [
          {
            "SchemeName": "UK.OBIE.IBAN",
            "Identification": "GB29 NWBK 6016 1331 9268 19",
            "Name": "Mr. John Doe"
          }
        ],
        "Servicer": {
          "SchemeName": "UK.OBIE.BICFI",
          "Identification": "NWBKGB22"
        }
      }
    ]
  },
  "Links": {
    "Self": "https://api.bank.com/open-banking/v3.1/accounts"
  },
  "Meta": {
    "TotalPages": 1
  }
}

# Example: Berlin Group XS2A Interface

import requests
from datetime import datetime

class OpenBankingClient:
    """
    Client for interacting with Open Banking APIs
    """
    
    def __init__(self, base_url: str, client_id: str, client_secret: str):
        self.base_url = base_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None
    
    def authenticate(self, certificate_path: str, key_path: str):
        """
        Authenticate using OAuth 2.0 with certificate
        """
        token_url = f"{self.base_url}/oauth/token"
        
        # Use certificate for mTLS
        response = requests.post(
            token_url,
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "scope": "accounts payments"
            },
            cert=(certificate_path, key_path)
        )
        
        self.access_token = response.json()["access_token"]
        return self.access_token
    
    def get_accounts(self, consent_id: str):
        """
        Get list of accounts for the user
        """
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "x-request-id": str(uuid.uuid4()),
            "PSU-ID": "customer-123",
            "Consent-ID": consent_id
        }
        
        response = requests.get(
            f"{self.base_url}/v1/accounts",
            headers=headers
        )
        
        return response.json()
    
    def get_account_transactions(
        self, 
        account_id: str, 
        date_from: str, 
        date_to: str
    ):
        """
        Get transactions for a specific account
        """
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "x-request-id": str(uuid.uuid4()),
            "PSU-ID": "customer-123"
        }
        
        params = {
            "dateFrom": date_from,
            "dateTo": date_to,
            "bookingStatus": "both"
        }
        
        response = requests.get(
            f"{self.base_url}/v1/accounts/{account_id}/transactions",
            headers=headers,
            params=params
        )
        
        return response.json()
    
    def initiate_payment(
        self,
        payment_type: str,
        debtor_iban: str,
        creditor_iban: str,
        amount: float,
        currency: str,
        remittance_info: str
    ):
        """
        Initiate a payment
        """
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "x-request-id": str(uuid.uuid4()),
            "PSU-ID": "customer-123",
            "Content-Type": "application/json"
        }
        
        payment_request = {
            "InstructedAmount": {
                "Amount": str(amount),
                "Currency": currency
            },
            "DebtorAccount": {
                "SchemeName": "UK.OBIE.IBAN",
                "Identification": debtor_iban
            },
            "CreditorAccount": {
                "SchemeName": "UK.OBIE.IBAN",
                "Identification": creditor_iban
            },
            "RemittanceInformation": {
                "Unstructured": remittance_info
            }
        }
        
        response = requests.post(
            f"{self.base_url}/v1/payments/{payment_type}",
            headers=headers,
            json=payment_request
        )
        
        return response.json()

import uuid
from datetime import datetime, timedelta

class OpenBankingConsent:
    """
    Handling Open Banking consent flow
    """
    
    def create_consent(
        self,
        user_id: str,
        bank_id: str,
        access_types: list,
        expiration_days: int = 90
    ):
        """
        Create consent for account access
        """
        
        consent_data = {
            "ConsentId": f"CONS-{uuid.uuid4().hex[:12]}",
            "CreationDateTime": datetime.utcnow().isoformat(),
            "Status": "received",
            "StatusUpdateDateTime": datetime.utcnow().isoformat(),
            "Permissions": access_types,
            # Permissions options:
            # - ReadAccountsBasic
            # - ReadAccountsDetail
            # - ReadBalances
            # - ReadBeneficiariesDetail
            # - ReadDirectDebits
            # - ReadTransactionsCredits
            # - ReadTransactionsDebits
            # - ReadTransactionsDetail
            # - ReadStandingOrdersDetail
            "ExpirationDateTime": (
                datetime.utcnow() + timedelta(days=expiration_days)
            ).isoformat(),
            "TransactionFromDateTime": (
                datetime.utcnow() - timedelta(days=90)
            ).isoformat(),
            "TransactionToDateTime": datetime.utcnow().isoformat()
        }
        
        # In production: store consent in database
        return consent_data
    
    def initiate_consent_flow(
        self,
        bank_id: str,
        redirect_url: str,
        consent_data: dict
    ):
        """
        Initiate OAuth consent flow with the bank
        """
        
        # Get bank's authorization endpoint
        auth_endpoint = self.get_bank_auth_endpoint(bank_id)
        
        # Generate state for CSRF protection
        state = uuid.uuid4().hex
        
        # Build authorization URL
        auth_url = (
            f"{auth_endpoint}?"
            f"response_type=code&"
            f"client_id={self.client_id}&"
            f"redirect_uri={redirect_url}&"
            f"scope=accounts+payments&"
            f"state={state}&"
            f"consent_id={consent_data['ConsentId']}"
        )
        
        return {
            "authorization_url": auth_url,
            "state": state,
            "consent_id": consent_data["ConsentId"]
        }
    
    def handle_callback(self, code: str, state: str):
        """
        Handle callback from bank after user consent
        """
        
        # Exchange code for access token
        token_response = self.exchange_code_for_token(code)
        
        return {
            "access_token": token_response["access_token"],
            "refresh_token": token_response.get("refresh_token"),
            "expires_in": token_response["expires_in"],
            "token_type": token_response["token_type"]
        }

class StrongCustomerAuthentication:
    """
    Implement SCA for Open Banking
    """
    
    def __init__(self):
        self.sca_methods = {
            "sca_purchase": "Two-factor authentication",
            "sca_login": "Login authentication",
            "sca_payment": "Payment authentication"
        }
    
    def check_sca_required(self, transaction: dict) -> bool:
        """
        Determine if SCA is required for this transaction
        """
        
        # Exemptions check
        exemptions = [
            # Low-value payments under €30 (cumulative limit applies)
            transaction.get("amount", 0) < 30,
            
            # Trusted beneficiaries
            transaction.get("is_trusted_beneficiary", False),
            
            # Recurring payments (same amount, same merchant)
            transaction.get("is_recurring", False),
            
            # Corporate payments with SCA policy
            transaction.get("is_corporate", False)
        ]
        
        return not any(exemptions)
    
    def initiate_sca(
        self,
        user_id: str,
        sca_method: str,
        authentication_method_id: str
    ):
        """
        Initiate SCA challenge
        """
        
        sca_challenge = {
            "sca_challenge_data": {
                "sca_method": sca_method,
                "sca_method_id": authentication_method_id,
                "challenge_data": {
                    "challenge_information": "Please authenticate using your mobile app"
                }
            },
            "_links": {
                "sca_method": {
                    "href": f"/verification/v1/channels/{authentication_method_id}"
                }
            }
        }
        
        return sca_challenge
    
    def verify_sca_result(self, sca_result: dict) -> bool:
        """
        Verify SCA authentication result
        """
        
        return (
            sca_result.get("sca_status") == "authenticated" and
            sca_result.get("authentication_method_id") == self.get_expected_method()
        )

class OpenBankingSecurity:
    """
    Security best practices for Open Banking APIs
    """
    
    @staticmethod
    def validate_jwt(token: str, public_key: str):
        """
        Validate JWT token from TPP
        """
        import jwt
        
        try:
            # Decode and validate token
            payload = jwt.decode(
                token,
                public_key,
                algorithms=["RS256"],
                audience="open-banking-api",
                issuer="https://auth.example.com"
            )
            return payload
        except jwt.InvalidTokenError as e:
            raise ValueError(f"Invalid token: {e}")
    
    @staticmethod
    def verify_signature(payload: bytes, signature: str, public_key: str):
        """
        Verify request signature
        """
        import cryptography
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.asymmetric import padding
        
        # Verify signature using public key
        public_key_obj = cryptography.serialization.load_pem_public_key(
            public_key.encode()
        )
        
        try:
            public_key_obj.verify(
                signature.encode(),
                payload,
                padding.PKCS1v15(),
                hashes.SHA256()
            )
            return True
        except Exception:
            return False
    
    @staticmethod
    def check_rate_limit(client_id: str, endpoint: str) -> bool:
        """
        Implement rate limiting per TPP
        """
        # Rate limit: 100 requests per minute per client
        # Implementation would use Redis or similar
        pass

# Anti-pattern: Assuming all banks have the same API
def bad_bank_integration():
    """
    Anti-pattern: One-size-fits-all integration
    """
    # Each bank may have different:
    # - API versions
    # - Field names
    # - Error formats
    # - Rate limits
    # - Authentication methods
    
    # Result: Fragile integrations, frequent failures
    
    return "Use a single integration approach for all banks"

# Good pattern: Adaptive integration
def good_bank_integration():
    """
    Good pattern: Handle bank-specific differences
    """
    
    # Use bank's API profile
    bank_profile = get_bank_profile(bank_id)
    
    # Adapt to bank's specific API
    adapter = BankAPIAdapter.get_adapter(
        bank_profile.api_version,
        bank_profile.authentication_type
    )
    
    return adapter.get_accounts(access_token)

# Anti-pattern: Not tracking consent expiration
def bad_consent_tracking():
    """
    Anti-pattern: Assuming consent is permanent
    """
    # User revokes consent
    # Bank updates account access
    # Application keeps trying with expired token
    
    return "All API calls will fail"

# Good pattern: Proactive consent management
def good_consent_tracking():
    """
    Track and refresh consent proactively
    """
    
    class ConsentManager:
        def __init__(self):
            self.consents = {}  # Would use database
        
        def check_consent_valid(self, consent_id: str) -> bool:
            consent = self.consents.get(consent_id)
            
            if not consent:
                return False
            
            # Check if consent is expired
            if datetime.fromisoformat(
                consent["expiration_date"]
            ) < datetime.utcnow():
                return False
            
            # Check if account access is still valid
            if not self.check_account_access(consent_id):
                return False
            
            return True
        
        def refresh_consent_if_needed(self, consent_id: str):
            """Proactively refresh consent before expiration"""
            
            consent = self.consents.get(consent_id)
            
            # Refresh if expiring within 7 days
            days_until_expiry = (
                datetime.fromisoformat(consent["expiration_date"]) - 
                datetime.utcnow()
            ).days
            
            if days_until_expiry < 7:
                # Initiate consent refresh flow
                return self.initiate_consent_refresh(consent_id)

# Anti-pattern: Generic error handling
def bad_error_handling():
    """
    Anti-pattern: Catch-all error handling
    """
    try:
        result = api.get_accounts(token)
        return result
    except Exception as e:
        # What went wrong? Network? Auth? Bank downtime?
        return None  # Silent failure!

# Good pattern: Specific error handling
def good_error_handling():
    """
    Specific error handling for different failure modes
    """
    
    from enum import Enum
    
    class APIErrorType(Enum):
        AUTHENTICATION = "authentication"
        AUTHORIZATION = "authorization"
        RATE_LIMIT = "rate_limit"
        TEMPORARY_UNAVAILABLE = "temporary"
        RESOURCE_NOT_FOUND = "not_found"
        INVALID_REQUEST = "invalid"
    
    def handle_api_error(error: Exception):
        if isinstance(error, TokenExpiredError):
            # Refresh token and retry
            new_token = refresh_access_token()
            return api.get_accounts(new_token)
        
        elif isinstance(error, InsufficientScopeError):
            # Request additional permissions
            return initiate_consent_reauthorization()
        
        elif isinstance(error, RateLimitError):
            # Implement backoff and retry
            import time
            time.sleep(60)  # Wait before retry
            return api.get_accounts(token)
        
        elif isinstance(error, ServiceUnavailableError):
            # Bank's API is down - notify user
            return {"status": "unavailable", "bank": bank_id}
        
        else:
            # Log and handle generic error
            log_error(error)
            raise

# Instead of integrating with each bank individually
# Consider using an aggregation platform

class UnifiedBankingAPI:
    """
    Abstraction over multiple Open Banking providers
    """
    
    def __init__(self):
        self.providers = {
            "uk": PlaidProvider(),      # For UK
            "eu": TinkProvider(),       # For Europe  
            "us": MXProvider()          # For US
        }
    
    def get_accounts(self, country_code: str, credentials: dict):
        provider = self.providers.get(country_code)
        return provider.get_accounts(credentials)

class DataMapper:
    """
    Map bank-specific data to standardized format
    """
    
    TRANSACTION_MAPPING = {
        # UK Bank -> Standard
        "transactionDate": "booking_date",
        "transactionInformation": "description",
        "amount": "amount",
        "creditDebitIndicator": "direction",  # CRDT/DBIT -> credit/debit
        "merchantDetails": "merchant",
    }
    
    def standardize_transaction(self, raw_transaction: dict) -> dict:
        standardized = {}
        
        for standard_field, bank_fields in self.TRANSACTION_MAPPING.items():
            # Try each bank-specific field
            for bank_field in bank_fields:
                if bank_field in raw_transaction:
                    value = raw_transaction[bank_field]
                    
                    # Apply transformations
                    if standard_field == "direction":
                        value = "credit" if value == "CRDT" else "debit"
                    
                    standardized[standard_field] = value
                    break
        
        return standardized

class OfflineStrategy:
    """
    Handle scenarios where bank APIs are unavailable
    """
    
    def __init__(self):
        self.cache = RedisCache()  # Cache for recent data
        self.queue = MessageQueue()  # Queue for failed requests
    
    def get_account_with_fallback(self, account_id: str):
        """
        Try API first, fall back to cached data
        """
        
        try:
            # Try live API
            data = api.get_account(account_id)
            
            # Update cache
            self.cache.set(f"account:{account_id}", data, ttl=3600)
            
            return {"data": data, "source": "live"}
        
        except ServiceUnavailableError:
            # Fall back to cache
            cached = self.cache.get(f"account:{account_id}")
            
            if cached:
                return {
                    "data": cached, 
                    "source": "cache",
                    "stale": True
                }
            
            raise NoDataAvailableError()