+─────────────────────────────────────────────────────────────────────────────+
|               GLOBAL PAYMENT AUTHORIZATION LATENCY BUDGET (ms)              |
+─────────────────────────────────────────────────────────────────────────────+
|  Total Allowed Round-Trip Budget: ~1,500 ms                                 |
|                                                                             |
|  Merchant POS / Web Checkout:          │ 80ms                               |
|  Acquirer & Gateway Routing:           ││ 120ms                             |
|  Card Scheme Switch (Visa/Mastercard): │││ 220ms                            |
|  HSM Decryption & PIN Verification:    ││ 140ms                             |
|  =============================================================              |
|  >> INTERNAL FRAUD ENGINE SLA BUDGET:  │││││ MAX 50ms (Target: < 25ms)      |
|  =============================================================              |
|  Core Banking Ledger Write:            │││ 180ms                            |
|  Egress Switch & Auth Response:        ││ 160ms                             |
|  Merchant Terminal Render:             │ 70ms                               |
|                                                                             |
|  [WARNING] If Fraud Engine crosses 50ms, the card network triggers          |
|  a "Stand-In Processing" (STIP) timeout or soft decline, costing            |
|  merchants up to 4.2% in unnecessary checkout abandonment.                  |
+─────────────────────────────────────────────────────────────────────────────+

flowchart LR
    PaymentGateway["Payment Ingress Gateway<br/>(ISO 8583 / ISO 20022)"] -->|Sub-2ms Ingress| Tokenizer["Tokenization & Sanitization<br/>(PCI-DSS v4.0 Zero-PAN)"]
    
    subgraph IN_MEMORY_TIER ["In-Memory Buffer & Feature Layer (Sub-5ms)"]
        Tokenizer -->|XADD stream:payments| RedisBuffer[("Redis 7.2 In-Memory Stream<br/>(Append-Only FIFO Buffer)")]
        Tokenizer -->|Pipelined ZADD / INCRBY| VelocityCache[("Redis Sliding Window Cache<br/>(ZSET Time-Index + GEO)")]
    end
    
    subgraph WORKER_POOL ["Stateless Event-Driven Risk Workers"]
        RedisBuffer -->|XREADGROUP Consumer Group| WorkerPool["Distributed Scoring Workers<br/>(Go / Python ONNX Runtimes)"]
        VelocityCache -.->|Sub-1ms Feature Pull| WorkerPool
        WorkerPool -->|Embedded Inference| InferenceModel["XGBoost / LightGBM<br/>(< 12ms Scored Vector)"]
    end
    
    subgraph VERDICT_GATEWAY ["Decision Router (< 30ms Round-Trip)"]
        InferenceModel --> DecisionRouter{"Risk Score Matrix<br/>(0 - 1000)"}
        DecisionRouter -->|Score < 300| Approve["APPROVE (98.6%)<br/>Direct Settlement"]
        DecisionRouter -->|300 <= Score <= 750| Challenge["CHALLENGE (1.1%)<br/>EMV 3DS 2.2 Trigger"]
        DecisionRouter -->|Score > 750| Decline["DECLINE (0.3%)<br/>Hard Rejection"]
    end
    subgraph ASYNC_PERSISTENCE ["Dual-Path Asynchronous Persistence Tier"]
        DecisionRouter -.->|Micro-Batched Multi-Row| PostgresLedger[("PostgreSQL 16 Primary<br/>(ACID Settlement Ledger)")]
        DecisionRouter -.->|Kafka / Debezium Stream| ClickHouseAudit[("ClickHouse OLAP Cluster<br/>(7-Year Immutable Audit Trail)")]
    end

+─────────────────────────────────────────────────────────────────────────────+
|               DECOUPLED REAL-TIME FRAUD DETECTION PIPELINE                  |
+─────────────────────────────────────────────────────────────────────────────+
|                                                                             |
|  [POS / E-Comm Ingress] ──► [Stateless Tokenizer] ──► [Redis 7.2 Streams]   |
|                                                               │             |
|          ┌────────────────────────────────────────────────────┘             |
|          ▼ (Sub-3ms Atomic Feature Enrichment)                              |
|  [Redis Sliding Window Cache] ◄──► [Consumer Group: 32 Scoring Workers]     |
|  - Card Velocity (ZSET)            - Rule Gates: Hard Failures              |
|  - Geo Haversine (Redis GEO)       - Embedded ML: ONNX Inference (12ms)     |
|          │                                                    │             |
|          │                                                    ▼             |
|          │                                        [Decision Verdict Engine] |
|          │                                        - Score < 300: APPROVE    |
|          │                                        - 300-750: 3DS CHALLENGE  |
|          │                                        - Score > 750: DECLINE    |
|          │                                                    │             |
|          └──────────────────────────┬─────────────────────────┘             |
|                                     ▼                                       |
|                  [Dual-Path Asynchronous Persistence]                       |
|                  ├─► PostgreSQL 16 (ACID Double-Entry Ledger)               |
|                  └─► ClickHouse OLAP (7-Year Audit & Model Retraining)      |
|                                                                             |
+─────────────────────────────────────────────────────────────────────────────+

-- DANGEROUS: Kills database connection pools under 10,000+ TPS
SELECT COUNT(), SUM(amount) 
FROM transactions 
WHERE card_token = 'tok_98a72b1' 
  AND created_at >= NOW() - INTERVAL '10 minutes';

+─────────────────────────────────────────────────────────────────────────────+
|                MATHEMATICAL RISK FORMULATION: VELOCITY & GEO                |
+─────────────────────────────────────────────────────────────────────────────+
|                                                                             |
|  1. Sliding-Window Card Velocity:                                           |
|                                                                             |
|     V_count(c, Delta t) = Sum_{i in Tx(c)} I(t_now - t_i <= Delta t)       |
|                                                                             |
|     V_amount(c, Delta t) = Sum_{i in Tx(c)} A_i  I(t_now - t_i <= Delta t) |
|                                                                             |
|  2. Great-Circle Geospatial Anomaly ("Impossible Travel Speed"):            |
|                                                                             |
|     D_haversine = 2R  arcsin( sqrt( sin^2(Delta phi / 2) +                 |
|                   cos(phi_1)  cos(phi_2)  sin^2(Delta lambda / 2) ) )     |
|                                                                             |
|     v_geo = D_haversine / max(1, t_k - t_{k-1})                            |
|                                                                             |
|     If v_geo > 900 km/h (Commercial Aircraft Velocity Limit):                |
|     Anomaly Multiplier Gamma_geo = 3.5                                       |
|                                                                             |
|  3. Composite Calibrated Risk Probability:                                  |
|                                                                             |
|     P(Fraud | x) = 1 / ( 1 + exp( - ( beta_0 + Sum w_j  f_j(x)             |
|                    + beta_ML  Model_ONNX(x) ) ) )                          |
|                                                                             |
|     Risk Score S = round( P(Fraud | x)  1000 )    [Range: 0 to 1000]       |
|                                                                             |
+─────────────────────────────────────────────────────────────────────────────+

package main
import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"time"
	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
)
type AuthRequest struct {
	CardPAN        string  json:"card_pan"
	MerchantID     string  json:"merchant_id"
	MCC            string  json:"mcc"
	Amount         float64 json:"amount"
	Currency       string  json:"currency"
	Latitude       float64 json:"latitude"
	Longitude      float64 json:"longitude"
	TerminalID     string  json:"terminal_id"
	DeviceFingerprint string json:"device_fingerprint"
}
type IngressGateway struct {
	redisClient redis.Client
	hmacSecret  []byte
}
func NewIngressGateway(redisURL string, secret string) IngressGateway {
	opt, err := redis.ParseURL(redisURL)
	if err != nil {
		panic(err)
	}
	// Pool tuning for 50,000+ TPS
	opt.PoolSize = 256
	opt.MinIdleConns = 64
	opt.ReadTimeout = 15  time.Millisecond
	opt.WriteTimeout = 15  time.Millisecond
	return &IngressGateway{
		redisClient: redis.NewClient(opt),
		hmacSecret:  []byte(secret),
	}
}
// TokenizePAN generates an irreversible PCI-compliant token
func (g IngressGateway) TokenizePAN(pan string) string {
	mac := hmac.New(sha256.New, g.hmacSecret)
	mac.Write([]byte(pan))
	return "tok_" + hex.EncodeToString(mac.Sum(nil))[:24]
}
func (g IngressGateway) HandleAuthorize(w http.ResponseWriter, r http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 45time.Millisecond)
	defer cancel()
	var req AuthRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, {"error":"invalid_payload"}, http.StatusBadRequest)
		return
	}
	txID := uuid.New().String()
	cardToken := g.TokenizePAN(req.CardPAN)
	now := time.Now().UnixMilli()
	// Atomic Redis Pipeline: Push to Stream + Record Velocity
	pipe := g.redisClient.Pipeline()
	// 1. Append transaction to Redis Ingestion Stream (FIFO Log)
	pipe.XAdd(ctx, &redis.XAddArgs{
		Stream: "stream:transactions:inbound",
		MaxLen: 500000,
		Approx: true, // Amortized O(1) memory bounding
		Values: map[string]interface{}{
			"tx_id":              txID,
			"card_token":         cardToken,
			"merchant_id":        req.MerchantID,
			"mcc":                req.MCC,
			"amount":             req.Amount,
			"currency":           req.Currency,
			"latitude":           req.Latitude,
			"longitude":          req.Longitude,
			"timestamp":          now,
			"device_fingerprint": req.DeviceFingerprint,
		},
	})
	// 2. Add to Sliding Window Velocity Sorted Set (Key: vel:{card_token}:600s)
	velocityKey := fmt.Sprintf("vel:%s:600s", cardToken)
	pipe.ZAdd(ctx, velocityKey, redis.Z{
		Score:  float64(now),
		Member: fmt.Sprintf("%s:%.2f", txID, req.Amount),
	})
	pipe.Expire(ctx, velocityKey, 650time.Second)
	// 3. Update Last Known Location for Geospatial Velocity
	geoKey := fmt.Sprintf("geo:%s", cardToken)
	pipe.GeoAdd(ctx, geoKey, &redis.GeoLocation{
		Name:      txID,
		Longitude: req.Longitude,
		Latitude:  req.Latitude,
	})
	pipe.Expire(ctx, geoKey, 86400time.Second)
	// Execute Pipeline in single TCP round-trip
	_, err := pipe.Exec(ctx)
	if err != nil {
		http.Error(w, {"decision":"STAND_IN_DECLINE","reason":"ingestion_timeout"}, http.StatusServiceUnavailable)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	json.NewEncoder(w).Encode(map[string]interface{}{
		"status": "QUEUED_FOR_EVALUATION",
		"tx_id":  txID,
	})
}

"""
Real-Time Fraud Scoring Worker
Technology: Python 3.11, Redis 7.2 (redis-py), ONNX Runtime, NumPy
Throughput: ~3,200 transactions/sec per 4-core worker process
"""
import time
import json
import math
import numpy as np
import onnxruntime as ort
import redis
Redis Connection with Connection Pooling
r = redis.Redis(
    host='127.0.0.1',
    port=6379,
    db=0,
    decode_responses=True,
    socket_timeout=0.025,
    socket_connect_timeout=0.010,
    max_connections=64
)
STREAM_NAME = "stream:transactions:inbound"
GROUP_NAME = "fraud_scoring_group"
CONSUMER_ID = f"worker_{int(time.time()  1000)}"
Ensure Consumer Group exists

try:
    r.xgroup_create(STREAM_NAME, GROUP_NAME, id="0", mkstream=True)
except redis.exceptions.ResponseError:
    pass  # Already exists
Load Pre-Compiled LightGBM / XGBoost Model in ONNX format
session_options = ort.SessionOptions()
session_options.intra_op_num_threads = 2
session_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
model = ort.InferenceSession("models/fraud_xgboost_v4.onnx", session_options)
def compute_haversine_velocity(card_token, cur_lat, cur_lon, cur_ts):
    """Calculates km/h between current and immediately preceding transaction"""
    geo_key = f"geo:{card_token}"
    history_key = f"geo_meta:{card_token}"
    
    last_meta = r.get(history_key)
    if not last_meta:
        # Cache current as baseline
        r.setex(history_key, 86400, json.dumps({"lat": cur_lat, "lon": cur_lon, "ts": cur_ts}))
        return 0.0
    last = json.loads(last_meta)
    prev_lat, prev_lon, prev_ts = last["lat"], last["lon"], last["ts"]
    time_diff_hours = max((cur_ts - prev_ts) / 3600000.0, 0.0001)
    # Great-Circle Haversine Formula
    R = 6371.0 # Earth radius in km
    dlat = math.radians(cur_lat - prev_lat)
    dlon = math.radians(cur_lon - prev_lon)
    a = (math.sin(dlat / 2)2 + 
         math.cos(math.radians(prev_lat))  math.cos(math.radians(cur_lat))  math.sin(dlon / 2)2)
    c = 2  math.atan2(math.sqrt(a), math.sqrt(1 - a))
    distance_km = R  c
    # Update cache with current coordinates
    r.setex(history_key, 86400, json.dumps({"lat": cur_lat, "lon": cur_lon, "ts": cur_ts}))
    return distance_km / time_diff_hours
def extract_sliding_features(card_token, current_amount, now_ms):
    """Extracts 10-minute velocity count and aggregated sum from Redis Sorted Set"""
    vel_key = f"vel:{card_token}:600s"
    window_start = now_ms - 600000 # 10 minutes ago
    # Pipeline: Clean expired entries & fetch active window
    pipe = r.pipeline()
    pipe.zremrangebyscore(vel_key, "-inf", window_start)
    pipe.zrange(vel_key, 0, -1)
    results = pipe.execute()
    active_items = results[1] # List of "tx_id:amount"
    tx_count_10m = len(active_items)
    
    amount_sum_10m = 0.0
    for item in active_items:
        try:
            amount_sum_10m += float(item.split(":")[1])
        except (IndexError, ValueError):
            pass
    return float(tx_count_10m), float(amount_sum_10m)
def process_scoring_batch():
    """Reads and scores micro-batches of up to 100 transactions"""
    while True:
        try:
            # Read from Consumer Group (Blocks up to 50ms)
            entries = r.xreadgroup(
                GROUP_NAME, 
                CONSUMER_ID, 
                {STREAM_NAME: ">"}, 
                count=100, 
                block=50
            )
            if not entries:
                continue
            for stream, messages in entries:
                ack_ids = []
                for msg_id, payload in messages:
                    start_time = time.perf_counter()
                    
                    tx_id = payload["tx_id"]
                    card_token = payload["card_token"]
                    amount = float(payload["amount"])
                    cur_lat = float(payload["latitude"])
                    cur_lon = float(payload["longitude"])
                    ts = int(payload["timestamp"])
                    # 1. Feature Extraction in Memory (< 3ms)
                    vel_count, vel_sum = extract_sliding_features(card_token, amount, ts)
                    geo_speed_kmh = compute_haversine_velocity(card_token, cur_lat, cur_lon, ts)
                    # 2. Hard Rule Gate: Impossible Travel Velocity
                    if geo_speed_kmh > 900.0:
                        verdict = "DECLINE"
                        risk_score = 990
                    else:
                        # 3. ONNX Model Inference (< 10ms)
                        # Vector: [amount, vel_count_10m, vel_sum_10m, geo_speed_kmh, mcc_risk_weight]
                        feature_vector = np.array([[amount, vel_count, vel_sum, geo_speed_kmh, 1.0]], dtype=np.float32)
                        ort_inputs = {model.get_inputs()[0].name: feature_vector}
                        raw_prob = model.run(None, ort_inputs)[1][0][1] # Class 1 (Fraud) Probability
                        risk_score = int(raw_prob  1000)
                        if risk_score < 300:
                            verdict = "APPROVE"
                        elif risk_score <= 750:
                            verdict = "CHALLENGE_3DS"
                        else:
                            verdict = "DECLINE"
                    elapsed_ms = (time.perf_counter() - start_time)  1000
                    # 4. Write Verdict to Fast Decision Hash
                    r.hset(f"decision:{tx_id}", mapping={
                        "verdict": verdict,
                        "risk_score": risk_score,
                        "elapsed_ms": f"{elapsed_ms:.2f}",
                        "evaluated_at": int(time.time()  1000)
                    })
                    r.expire(f"decision:{tx_id}", 300) # 5-minute TTL
                    ack_ids.append(msg_id)
                # Batch Acknowledge in Redis Streams
                if ack_ids:
                    r.xack(STREAM_NAME, GROUP_NAME, ack_ids)
        except Exception as e:
            time.sleep(0.01) # Circuit protection on Redis disconnect

+─────────────────────────────────────────────────────────────────────────────+
|               DUAL-PATH ASYNCHRONOUS PERSISTENCE TOPOLOGY                   |
+─────────────────────────────────────────────────────────────────────────────+
|                                                                             |
|                      [Scored Transaction Envelope]                          |
|                                    │                                        |
|            ┌───────────────────────┴───────────────────────┐                 |
|            ▼ (Transactional Path)                          ▼ (Audit Path)   |
|  [PostgreSQL 16 Primary Ledger]            [ClickHouse OLAP Ingestion]      |
|  - Write-Path: Multi-Row Micro-Batch       - Ingestion: Vectorized Block    |
|  - Engine: PostgreSQL WAL + PgBouncer      - Engine: MergeTree Partitioned  |
|  - Table: settled_authorizations           - Table: fraud_audit_log         |
|  - Guarantee: Strict ACID Consistency      - Guarantee: Append-Only Immutable|
|  - Retention: 90 Days Hot Operational      - Retention: 7 Years Partitioned |
|                                                                             |
+─────────────────────────────────────────────────────────────────────────────+

-- Operational Ledger: PostgreSQL 16
CREATE TABLE settled_authorizations (
    tx_id UUID PRIMARY KEY,
    card_token VARCHAR(32) NOT NULL,
    merchant_id VARCHAR(64) NOT NULL,
    amount NUMERIC(12, 2) NOT NULL,
    currency VARCHAR(3) NOT NULL,
    risk_score SMALLINT NOT NULL CHECK (risk_score BETWEEN 0 AND 1000),
    verdict VARCHAR(16) NOT NULL CHECK (verdict IN ('APPROVE', 'CHALLENGE_3DS', 'DECLINE')),
    latency_ms NUMERIC(6, 2) NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_settled_card_token_created 
ON settled_authorizations (card_token, created_at DESC);

-- Regulatory Audit Engine: ClickHouse OLAP
CREATE TABLE default.fraud_audit_log (
    tx_id UUID,
    card_token LowCardinality(String),
    merchant_id LowCardinality(String),
    mcc LowCardinality(String),
    amount Float64,
    currency LowCardinality(String),
    latitude Float32,
    longitude Float32,
    velocity_count_10m UInt16,
    velocity_sum_10m Float64,
    geo_speed_kmh Float32,
    risk_score UInt16,
    verdict LowCardinality(String),
    decision_latency_ms Float32,
    model_version LowCardinality(String),
    created_date Date DEFAULT toDate(created_at),
    created_at DateTime64(3, 'UTC')
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_date)
ORDER BY (card_token, created_at, tx_id)
TTL created_date + INTERVAL 7 YEAR;

# Reclaim messages pending for more than 45,000ms from dead workers
XAUTOCLAIM stream:transactions:inbound fraud_scoring_group recovery_worker 45000 0-0 COUNT 50