Financial Services Reference Architecture¶

# Immutable audit logging with blockchain verification
from azure.storage.blob import BlobServiceClient, ImmutabilityPolicy
from datetime import datetime, timedelta
import hashlib
import json

class SOXAuditLogger:
    def __init__(self, connection_string):
        self.blob_service = BlobServiceClient.from_connection_string(connection_string)
        self.container = self.blob_service.get_container_client("audit-logs")

    def log_data_change(self, user, action, table, record_id, before_value, after_value):
        """Log data changes with immutability and blockchain verification"""

        audit_record = {
            "timestamp": datetime.utcnow().isoformat(),
            "user": user,
            "action": action,
            "table": table,
            "record_id": record_id,
            "before": before_value,
            "after": after_value,
            "ip_address": self.get_client_ip(),
            "session_id": self.get_session_id()
        }

        # Calculate hash for blockchain
        record_json = json.dumps(audit_record, sort_keys=True)
        audit_record["hash"] = hashlib.sha256(record_json.encode()).hexdigest()
        audit_record["previous_hash"] = self.get_last_hash()

        # Store with immutability policy
        blob_name = f"{datetime.utcnow().strftime('%Y/%m/%d/%H')}/{audit_record['hash']}.json"
        blob_client = self.container.get_blob_client(blob_name)

        blob_client.upload_blob(
            json.dumps(audit_record),
            overwrite=False  # Prevent overwrites
        )

        # Set immutability policy (7 years for SOX)
        immutability_policy = ImmutabilityPolicy(
            expiry_time=datetime.utcnow() + timedelta(days=2555),  # 7 years
            policy_mode="Locked"
        )
        blob_client.set_immutability_policy(immutability_policy)

        # Set legal hold for ongoing investigations
        if self.is_under_investigation(record_id):
            blob_client.set_legal_hold(True)

        return audit_record["hash"]

    def verify_audit_chain(self, from_date, to_date):
        """Verify blockchain integrity of audit logs"""
        blobs = self.container.list_blobs()
        previous_hash = None

        for blob in blobs:
            audit_record = json.loads(
                self.container.download_blob(blob.name).readall()
            )

            if previous_hash and audit_record["previous_hash"] != previous_hash:
                raise ValueError(f"Audit chain broken at {blob.name}")

            # Verify hash
            temp_record = audit_record.copy()
            stored_hash = temp_record.pop("hash")
            temp_record.pop("previous_hash")
            calculated_hash = hashlib.sha256(
                json.dumps(temp_record, sort_keys=True).encode()
            ).hexdigest()

            if stored_hash != calculated_hash:
                raise ValueError(f"Hash mismatch in {blob.name}")

            previous_hash = stored_hash

        return True

# PCI-DSS compliant card tokenization
from azure.keyvault.keys import KeyClient
from azure.keyvault.keys.crypto import CryptographyClient, EncryptionAlgorithm
from azure.identity import DefaultAzureCredential
import secrets
import hashlib

class PCITokenizationService:
    def __init__(self, vault_url):
        credential = DefaultAzureCredential()
        self.key_client = KeyClient(vault_url, credential)
        self.encryption_key = self.key_client.get_key("card-encryption-key")
        self.crypto_client = CryptographyClient(self.encryption_key, credential)

    def tokenize_pan(self, pan):
        """Tokenize PAN with Format-Preserving Encryption"""

        # Validate PAN
        if not self.luhn_check(pan):
            raise ValueError("Invalid PAN")

        # Generate token
        token = self.generate_token()

        # Encrypt PAN
        encrypted_pan = self.crypto_client.encrypt(
            EncryptionAlgorithm.rsa_oaep_256,
            pan.encode()
        ).ciphertext

        # Store mapping in Key Vault (HSM-protected)
        self.key_client.get_key_vault_secret(
            f"pan-token-{token}",
            value=encrypted_pan.hex(),
            tags={
                "bin": pan[:6],  # Bank Identification Number
                "last4": pan[-4:],
                "token_date": datetime.utcnow().isoformat()
            }
        )

        return {
            "token": token,
            "bin": pan[:6],
            "last4": pan[-4:],
            "expiry_date": self.calculate_expiry()
        }

    def detokenize_pan(self, token):
        """Detokenize PAN (audit logged)"""

        # Log access
        self.log_pan_access(token)

        # Retrieve encrypted PAN
        secret = self.key_client.get_secret(f"pan-token-{token}")
        encrypted_pan = bytes.fromhex(secret.value)

        # Decrypt
        decrypted = self.crypto_client.decrypt(
            EncryptionAlgorithm.rsa_oaep_256,
            encrypted_pan
        ).plaintext

        return decrypted.decode()

    @staticmethod
    def luhn_check(pan):
        """Validate PAN using Luhn algorithm"""
        def digits_of(n):
            return [int(d) for d in str(n)]

        digits = digits_of(pan)
        odd_digits = digits[-1::-2]
        even_digits = digits[-2::-2]
        checksum = sum(odd_digits)
        for d in even_digits:
            checksum += sum(digits_of(d * 2))
        return checksum % 10 == 0

# GDPR compliance implementation
from azure.purview.catalog import PurviewCatalogClient
from azure.identity import DefaultAzureCredential
from datetime import datetime
import json

class GDPRComplianceManager:
    def __init__(self, purview_endpoint, storage_account):
        self.purview_client = PurviewCatalogClient(
            purview_endpoint,
            DefaultAzureCredential()
        )
        self.storage_account = storage_account

    def right_to_be_forgotten(self, customer_id):
        """Erase all customer PII across data estate"""

        # 1. Find all assets containing customer data
        assets = self.purview_client.discovery.query(
            keywords=customer_id,
            filter={
                "classifications": ["PII", "Customer Data"]
            }
        )

        deletion_log = {
            "customer_id": customer_id,
            "request_date": datetime.utcnow().isoformat(),
            "deleted_records": []
        }

        # 2. Delete from each identified location
        for asset in assets:
            if asset["entityType"] == "azure_datalake_gen2_path":
                self.delete_from_data_lake(asset["qualifiedName"], customer_id)
            elif asset["entityType"] == "azure_sql_table":
                self.delete_from_sql(asset["qualifiedName"], customer_id)
            elif asset["entityType"] == "azure_cosmosdb_collection":
                self.delete_from_cosmos(asset["qualifiedName"], customer_id)

            deletion_log["deleted_records"].append({
                "asset": asset["name"],
                "location": asset["qualifiedName"],
                "deleted_at": datetime.utcnow().isoformat()
            })

        # 3. Create immutable deletion certificate
        self.create_deletion_certificate(deletion_log)

        return deletion_log

    def data_portability(self, customer_id, export_format="json"):
        """Export all customer data in machine-readable format"""

        # Find all customer data
        customer_data = {}

        # Profile data
        customer_data["profile"] = self.export_customer_profile(customer_id)

        # Transaction history
        customer_data["transactions"] = self.export_transactions(customer_id)

        # Account information
        customer_data["accounts"] = self.export_accounts(customer_id)

        # Consents
        customer_data["consents"] = self.export_consents(customer_id)

        # Generate export file
        export_file = f"customer-data-{customer_id}-{datetime.utcnow().strftime('%Y%m%d')}.{export_format}"

        if export_format == "json":
            return json.dumps(customer_data, indent=2)
        elif export_format == "xml":
            return self.to_xml(customer_data)
        else:
            raise ValueError(f"Unsupported format: {export_format}")

    def consent_management(self, customer_id, consent_type, action):
        """Manage customer consent with audit trail"""

        consent_record = {
            "customer_id": customer_id,
            "consent_type": consent_type,
            "action": action,  # granted, revoked, updated
            "timestamp": datetime.utcnow().isoformat(),
            "ip_address": self.get_client_ip(),
            "user_agent": self.get_user_agent()
        }

        # Store consent with blockchain verification
        self.store_consent_record(consent_record)

        # Update data processing rules
        if action == "revoked":
            self.revoke_processing_rights(customer_id, consent_type)

        return consent_record

# Real-time fraud detection with Azure ML
from azureml.core import Workspace, Model
from azureml.core.webservice import Webservice
import pandas as pd
import numpy as np

class FraudDetectionService:
    def __init__(self, workspace_name, model_name):
        self.ws = Workspace.from_config()
        self.model = Model(self.ws, model_name)
        self.service = Webservice(self.ws, f"{model_name}-service")

    def score_transaction(self, transaction):
        """Score transaction for fraud in real-time (<100ms)"""

        # Feature engineering
        features = self.engineer_features(transaction)

        # Get model prediction
        fraud_probability = self.service.run(input_data=json.dumps({
            "data": [features]
        }))

        # Risk-based decision
        if fraud_probability > 0.95:
            decision = "BLOCK"
            action = "block_transaction"
        elif fraud_probability > 0.75:
            decision = "CHALLENGE"
            action = "request_3ds_authentication"
        elif fraud_probability > 0.50:
            decision = "REVIEW"
            action = "flag_for_manual_review"
        else:
            decision = "APPROVE"
            action = "approve_transaction"

        # Log decision for model monitoring
        self.log_decision({
            "transaction_id": transaction["id"],
            "fraud_score": fraud_probability,
            "decision": decision,
            "latency_ms": self.get_latency()
        })

        return {
            "fraud_score": fraud_probability,
            "decision": decision,
            "action": action,
            "reasons": self.explain_prediction(features, fraud_probability)
        }

    def engineer_features(self, transaction):
        """Real-time feature engineering"""

        # Transaction features
        features = {
            "amount": transaction["amount"],
            "merchant_category": transaction["merchant_category_code"],
            "country": transaction["country_code"],
            "channel": transaction["channel"],
        }

        # Velocity features (from cache)
        customer_id = transaction["customer_id"]
        card_id = transaction["card_id"]

        features["txn_count_24h"] = self.get_txn_count(customer_id, hours=24)
        features["txn_count_1h"] = self.get_txn_count(customer_id, hours=1)
        features["amount_sum_24h"] = self.get_amount_sum(customer_id, hours=24)

        # Card features
        features["card_age_days"] = self.get_card_age(card_id)
        features["international_txn_ratio"] = self.get_international_ratio(card_id)

        # Behavioral features
        features["hour_of_day"] = transaction["timestamp"].hour
        features["day_of_week"] = transaction["timestamp"].weekday()
        features["is_unusual_merchant"] = self.check_unusual_merchant(customer_id, transaction["merchant_id"])
        features["is_unusual_location"] = self.check_unusual_location(customer_id, transaction["location"])

        # Network features
        features["device_reputation"] = self.get_device_reputation(transaction["device_id"])
        features["ip_reputation"] = self.get_ip_reputation(transaction["ip_address"])

        return features

    def explain_prediction(self, features, fraud_score):
        """Generate human-readable fraud indicators"""

        reasons = []

        if features["txn_count_1h"] > 5:
            reasons.append("High velocity: Multiple transactions in short time")

        if features["amount"] > 1000 and features["amount_sum_24h"] / features["amount"] < 2:
            reasons.append("Large transaction: Unusual amount for customer")

        if features["is_unusual_location"]:
            reasons.append("Location anomaly: Transaction from new location")

        if features["device_reputation"] < 0.5:
            reasons.append("Device risk: Suspicious device characteristics")

        if features["international_txn_ratio"] < 0.1 and features["country"] != "US":
            reasons.append("International: Rare international transaction")

        return reasons[:3]  # Top 3 reasons

# Process market data at scale
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.window import Window

spark = SparkSession.builder \
    .appName("HFT-Analytics") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .config("spark.sql.shuffle.partitions", "400") \
    .getOrCreate()

# Read tick data (billions of records)
tick_data = (spark
    .read
    .format("delta")
    .load("/silver/market_data/ticks")
    .where(col("trade_date") == current_date())
)

# Calculate VWAP by symbol
vwap_window = Window.partitionBy("symbol").orderBy("timestamp") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)

vwap_data = tick_data.withColumn(
    "vwap",
    sum(col("price") * col("volume")).over(vwap_window) /
    sum(col("volume")).over(vwap_window)
)

# Calculate tick imbalance
imbalance_window = Window.partitionBy("symbol").orderBy("timestamp") \
    .rowsBetween(-100, 0)

imbalance_data = vwap_data.withColumn(
    "buy_volume_100",
    sum(when(col("side") == "BUY", col("volume")).otherwise(0)).over(imbalance_window)
).withColumn(
    "sell_volume_100",
    sum(when(col("side") == "SELL", col("volume")).otherwise(0)).over(imbalance_window)
).withColumn(
    "imbalance_ratio",
    col("buy_volume_100") / (col("buy_volume_100") + col("sell_volume_100"))
)

# Write to gold layer with Z-ordering
(imbalance_data
    .write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .partitionBy("trade_date", "exchange")
    .save("/gold/market_analytics/vwap_imbalance")
)

# Optimize table
spark.sql("""
    OPTIMIZE delta.`/gold/market_analytics/vwap_imbalance`
    ZORDER BY (symbol, timestamp)
""")

# Multi-region DR with automated failover
from azure.cosmos import CosmosClient, PartitionKey
from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.identity import DefaultAzureCredential

class DisasterRecoveryManager:
    def __init__(self, subscription_id, resource_group):
        self.credential = DefaultAzureCredential()
        self.cosmos_mgmt = CosmosDBManagementClient(self.credential, subscription_id)
        self.resource_group = resource_group

    def configure_multi_region(self, account_name):
        """Configure multi-region writes with automatic failover"""

        locations = [
            {"locationName": "East US", "failoverPriority": 0, "isZoneRedundant": True},
            {"locationName": "West US 2", "failoverPriority": 1, "isZoneRedundant": True},
            {"locationName": "UK South", "failoverPriority": 2, "isZoneRedundant": True}
        ]

        self.cosmos_mgmt.database_accounts.begin_create_or_update(
            self.resource_group,
            account_name,
            {
                "location": "East US",
                "locations": locations,
                "enableMultipleWriteLocations": True,
                "enableAutomaticFailover": True,
                "consistencyPolicy": {
                    "defaultConsistencyLevel": "BoundedStaleness",
                    "maxStalenessPrefix": 100,
                    "maxIntervalInSeconds": 5
                },
                "backupPolicy": {
                    "type": "Continuous",
                    "continuousModeProperties": {
                        "tier": "Continuous7Days"
                    }
                }
            }
        )

    def test_failover(self, account_name):
        """Test regional failover"""

        # Trigger manual failover
        self.cosmos_mgmt.database_accounts.begin_failover_priority_change(
            self.resource_group,
            account_name,
            {
                "failoverPolicies": [
                    {"locationName": "West US 2", "failoverPriority": 0},
                    {"locationName": "East US", "failoverPriority": 1},
                    {"locationName": "UK South", "failoverPriority": 2}
                ]
            }
        )

    def point_in_time_restore(self, account_name, restore_timestamp):
        """Restore account to specific point in time"""

        restore_account_name = f"{account_name}-restored"

        self.cosmos_mgmt.database_accounts.begin_create_or_update(
            self.resource_group,
            restore_account_name,
            {
                "location": "East US",
                "createMode": "Restore",
                "restoreParameters": {
                    "restoreMode": "PointInTime",
                    "restoreSource": f"/subscriptions/{subscription_id}/resourceGroups/{self.resource_group}/providers/Microsoft.DocumentDB/databaseAccounts/{account_name}",
                    "restoreTimestampInUtc": restore_timestamp
                }
            }
        )