Data Streaming: Real-Time Fraud Detection Pipeline

Overview

Fraud detection in a KSA bank is not one pipeline — it is a set of overlapping pipelines operating on different event streams at different latency budgets. Card authorisations must be screened in under 80 ms (Mada network SLA). IPS credit transfers can tolerate 1–2 seconds before SAMA expects a status response. SADAD bill payments sit in the middle. Each stream has different feature requirements, different false-positive costs, and different consequences of being wrong.

What they share is the same structural problem: the signal for fraud lives in the pattern across events, not in any single event. A £200 card transaction is not suspicious; five £200 transactions in eight minutes across three cities is. The pipeline exists to materialise those patterns in real time so a scoring model can evaluate them before the payment instruction leaves the bank.

Architecture decisions

Two decisions shape the whole pipeline: which stream engine to use, and where to put the model. Neither has a universal answer.

Feature store design

Training-serving skew is the most common reason a fraud model that works in backtest fails in production. The root cause is always the same: the features computed offline for training are computed by a different code path than the features computed online for scoring. The fix is a unified feature store where the same feature definitions drive both paths.

Feast provides this: feature definitions in a Python registry, an offline store backed by a data warehouse (BigQuery, Redshift, Parquet on S3), and an online store (Redis) that is populated by a materialisation job. The Flink pipeline reads from the online store at scoring time; the training pipeline reads from the offline store at retraining time.

from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, Int64

card_entity = Entity(name="card_id", join_keys=["card_id"])

card_velocity_source = FileSource(
    path="s3://acmebank-features/card-velocity/**/*.parquet",
    timestamp_field="event_timestamp",
)

card_velocity_fv = FeatureView(
    name="card_velocity",
    entities=[card_entity],
    ttl=timedelta(hours=24),
    schema=[
        Field(name="txn_count_1min",  dtype=Int64),
        Field(name="txn_count_5min",  dtype=Int64),
        Field(name="txn_count_30min", dtype=Int64),
        Field(name="amt_sum_5min",    dtype=Float32),
        Field(name="unique_merch_1h", dtype=Int64),
        Field(name="distinct_geo_1h", dtype=Int64),
    ],
    source=card_velocity_source,
)

# materialise to Redis every 15 minutes
# feast materialize-incremental $(date -u +%Y-%m-%dT%H:%M:%S)

Velocity checks

Velocity features — transaction count and amount sum over sliding time windows — are the most reliable fraud signal for card-present and online card fraud. They are also the operationally hardest feature to get right because they require per-key stateful aggregation over time windows that vary from one minute to 24 hours.

Flink’s KeyedProcessFunction with explicit state timers is the right primitive. Do not use SlidingWindows from the DataStream Window API for low-latency paths — the default trigger fires only when the watermark advances past the window end, introducing latency proportional to the allowed lateness setting. Instead, maintain a ring buffer in state and compute aggregates inline on each event.

public class VelocityAggregator
    extends KeyedProcessFunction<String, CardAuth, EnrichedAuth> {

  // ring buffer: (timestamp, amount) pairs per card
  private ListState<TxnEntry> txnBuffer;

  private static final long W1  = 60_000L;   // 1 min
  private static final long W5  = 300_000L;  // 5 min
  private static final long W30 = 1_800_000L; // 30 min

  @Override
  public void processElement(CardAuth auth,
                              Context ctx,
                              Collector<EnrichedAuth> out) throws Exception {

    long now = auth.getEventTime();

    // evict entries outside the 30-min window
    List<TxnEntry> fresh = StreamSupport.stream(txnBuffer.get().spliterator(), false)
        .filter(e -> e.ts >= now - W30)
        .collect(Collectors.toList());

    fresh.add(new TxnEntry(now, auth.getAmount()));
    txnBuffer.update(fresh);

    VelocityFeatures v = computeVelocity(fresh, now);

    out.collect(new EnrichedAuth(auth, v));

    // register cleanup timer so state doesn't leak beyond 30 min
    ctx.timerService().registerEventTimeTimer(now + W30 + 1);
  }

  private VelocityFeatures computeVelocity(List<TxnEntry> buf, long now) {
    return new VelocityFeatures(
        buf.stream().filter(e -> e.ts >= now - W1 ).count(),
        buf.stream().filter(e -> e.ts >= now - W5 ).count(),
        buf.stream().filter(e -> e.ts >= now - W30).count(),
        buf.stream().filter(e -> e.ts >= now - W5 )
                    .mapToDouble(e -> e.amount).sum()
    );
  }
}

Model scoring on the stream

The trade-off between embedded scoring and remote model-serving comes down to latency and operational complexity. For a card authorisation pipeline where the budget is 80 ms end-to-end, a round-trip to an external model server (even a sidecar) adds 5–15 ms of network overhead plus serialisation on every event. Over millions of daily transactions that cost compounds into measurable P99 latency.

The better approach for latency-sensitive paths: export the model to ONNX and load it as an in-process operator inside Flink. The ONNX Runtime Java bindings support CPU inference at sub-millisecond latency for gradient-boosted models of the size typical in card fraud. Model updates require a Flink job restart, but with Flink’s savepoint mechanism this can be achieved with zero event loss.

public class FraudScoringOperator
    extends RichMapFunction<EnrichedAuth, ScoredAuth> {

  private transient OrtSession session;
  private transient OrtEnvironment env;

  private final String modelPath; // loaded from job config

  @Override
  public void open(Configuration cfg) throws Exception {
    env = OrtEnvironment.getEnvironment();
    OrtSession.SessionOptions opts = new OrtSession.SessionOptions();
    opts.setIntraOpNumThreads(2); // don't starve Flink task threads
    session = env.createSession(modelPath, opts);
  }

  @Override
  public ScoredAuth map(EnrichedAuth auth) throws Exception {
    float[] features = auth.toFeatureVector(); // 42 features
    OnnxTensor input = OnnxTensor.createTensor(
        env, new float[][] { features });

    try (OrtSession.Result result = session.run(
            Collections.singletonMap("features", input))) {
      float[][] proba = (float[][]) result.get(0).getValue();
      float pFraud = proba[0][1];
      return new ScoredAuth(auth, pFraud, classify(pFraud));
    }
  }

  private Decision classify(float p) {
    if (p > 0.90f) return Decision.BLOCK;
    if (p > 0.65f) return Decision.FLAG;
    return Decision.APPROVE;
  }
}

Feedback loops

A fraud model that cannot learn is a model that decays. Transaction patterns shift over months; card-not-present fraud especially evolves faster than a quarterly retraining cycle can track. The feedback loop turns resolved disputes back into training labels.

The label lag is the critical operational parameter. A dispute raised today might not be resolved for 45–90 days (chargeback cycle). The retraining pipeline must join labels to the original transaction features retrospectively, using the event_timestamp from the scoring-time feature snapshot — not the current online feature values, which will have changed.

Building the pipeline

1. Define Kafka topic layout

   One topic per event type (card-auth, ips-transfer, dispute-event, fraud-decisions). Over-partition upfront: 48 partitions on card-auth handles 3x projected peak throughput with room to scale Flink parallelism. Co-partition card-auth and the customer master KTable by card_id with the same partition count — stream-table joins require it.

2. Bootstrap the online feature store

   Run a full Feast materialisation from the offline store to Redis before the Flink job starts. An empty Redis means every scoring event gets zero-valued features until the ring buffer warms up — the first 30 minutes after deployment will have systematically lower fraud scores. Schedule a materialisation immediately before each deployment, not only nightly.

3. Deploy Flink with RocksDB state backend

   Configure incremental checkpoints to an S3-compatible store (Ceph in on-prem, AWS S3 on cloud). Checkpoint interval of 60 seconds balances recovery time against overhead. Set state.backend.incremental: true; full checkpoints for large velocity state can take minutes and block processing. Pin RocksDB block cache at 512 MB per TaskManager slot — undersizing causes excessive I/O and kills P99 latency.

4. Integrate ONNX model

   Store the ONNX artefact in a model registry (MLflow, or an S3 path with versioning). Load at job startup via open() on the scoring operator. Wire the model path as a Flink job parameter so it can be changed at savepoint-restore time without code changes. Include the feature vector schema version in the ONNX metadata — a schema mismatch at restore time should fail fast, not silently pass wrong features.

5. Wire the decision sink

   The fraud-decisions topic is consumed by the authorisation gateway to approve, flag (step-up authentication), or block the transaction. Use an idempotent Kafka sink keyed by txn_id to prevent duplicate decisions under Flink task failure and recovery. The downstream authorisation gateway must deduplicate on txn_id too — Flink’s at-least-once recovery can re-emit the last checkpoint’s output.

6. Activate the feedback pipeline

   Deploy the label writer job separately from the scorer — it runs at lower parallelism and does not need to meet the same latency SLA. Verify that the dispute-event topic includes the original event_timestamp and txn_id as mandatory fields; without them the retrospective feature join is impossible.

7. Validate with shadow mode

   Run the scoring pipeline in shadow mode for two weeks before switching the authorisation gateway to act on its decisions. Shadow mode: score all events, emit to fraud-decisions-shadow, compare with the existing rule-engine output, measure precision and recall against historical confirmed fraud. Ship only when shadow precision matches or exceeds the baseline at the same recall.

Exactly-once in fraud

In most stream processing, “at least once” is acceptable because downstream systems are idempotent. In a fraud pipeline, a duplicated BLOCK decision can create two chargeback records against the same transaction, visible to the customer and reportable to SAMA. Exactly-once delivery from Flink to Kafka is achievable at a throughput cost of roughly 5–8%.

execution:
  checkpointing:
    interval: 60000   # 60 seconds
    mode: EXACTLY_ONCE
    min-pause-between-checkpoints: 30000
    timeout: 120000
    max-concurrent-checkpoints: 1
  state-backend: rocksdb
  state-backend-incremental: true
  savepoint-dir: s3://acmebank-flink/savepoints/fraud-scorer

kafka-sink:
  semantic: EXACTLY_ONCE
  transaction-timeout: 120000 # must exceed checkpoint interval

parallelism: 12   # = card-auth partition count / 4

Operational concerns

A fraud pipeline that works in testing and fails at 02:00 on a Friday is worse than no pipeline at all — it blocks card authorisations while the incident response team catches up. Three operational investments pay for themselves within months:

• Latency percentile monitoring per operator. P99 latency at the scoring operator, not just end-to-end job latency. When the ONNX model grows (more features, deeper trees) the scoring operator’s P99 spikes first. You want to see it before the authorisation gateway SLA is affected.
• Consumer lag alerting on the decision sink topic. If the authorisation gateway stops consuming fraud-decisions, lag grows and eventually Flink back-pressures. This shows up as increasing card authorisation latency before it shows up as an error. Alert at 10 seconds of lag; page at 30 seconds.
• Weekly shadow-mode comparison. Continuously run the current production model and a shadow challenger. When the challenger’s precision at a given recall threshold exceeds production by >2%, trigger a promotion review. Without this you have no visibility into model drift until fraud losses increase.

Common pitfalls

When not to stream

Not every fraud detection requirement belongs in a streaming pipeline. Three cases where batch or rule-engine approaches are clearly better:

• Account-level pattern analysis. Detecting a slow-burn account takeover that spans 90 days does not need sub-second latency. A nightly SQL job over the data warehouse is simpler, cheaper, and produces the same alert 24 hours after the pattern completes — fast enough for account fraud where the loss accumulates over days, not milliseconds.
• AML transaction monitoring. SAMA’s AML monitoring requirements for CTR/STR filings operate on daily aggregates and multi-day patterns. Running AML logic in Flink adds operational complexity without reducing the regulatory response time. Use a purpose-built AML platform (Actimize, NICE Actimize, Quantexa) that is pre-certified for SAMA reporting.
• Startup pipelines with under 1 M transactions/day. Below that volume, a Kafka consumer plus PostgreSQL with window aggregation functions in SQL delivers the same velocity features at a fraction of the operational complexity. Reserve Flink for the scale where SQL aggregation latency becomes the bottleneck.