Why My First Real‑Time MFS Alert Dashboard Crashed and How I Rebuilt It in 48 Hours

Photo by Mark Chan on Unsplash

High‑Stakes Hook

It was 02:13 am on a rainy Thursday. The BFIU inbox pinged, a SAR arrived, and the numbers screamed: BDT 4.2 million moved through three Rocket accounts in under five minutes, each just under the BDT 100,000 threshold. My team’s alert list was a sea of noise—1,200 warnings per hour, most of them false. The system timed out, the dashboard froze, and senior management started asking, “Did we miss a structuring ring?” I stared at a blinking red line on a Grafana panel and felt the weight of an audit looming.

The Hidden Problem

Standard batch‑oriented monitoring works in banks that process a few thousand wires a day. In Bangladesh’s mobile‑financial‑services (MFS) world, transaction velocity is a different beast. bKash, Nagad, Rocket—all push millions of micro‑payments per hour. The BFIU guideline Rule 3.2.1 tells us to watch any series of transactions that cumulatively cross BDT 100,000 within a 24‑hour window, but it does not prescribe *how* to do it in real time. Our legacy rule engine was pulling the last 24‑hour snapshot every ten minutes, then feeding it to a static SQL alert matrix. Result?

• Latency > 10 minutes
• False‑positive rate ≈ 85%
• CPU spikes that crashed our Docker node

The hidden problem was two‑fold:

1. We were treating a streaming problem as a batch problem.
2. Our alert thresholds were hard‑coded, ignoring contextual signals like device ID, geolocation, and time‑of‑day patterns unique to Dhaka’s rush hour.

Technical Breakdown & Logic Flow

First I sketched a flow on a whiteboard during a coffee break:

1. Ingest raw MFS events from Kafka topics (bKash‑tx, Nagad‑tx, Rocket‑tx).
2. Enrich each event with customer risk score from the KYC store (Redis cache).
3. Window the stream into a tumbling 5‑minute bucket, then aggregate per account‑pair and per customer‑ID.
4. Apply a dynamic threshold: base BDT 100,000 + 20% * risk_score.
5. Push alerts to a Redis‑backed pub/sub channel that the dashboard consumes via WebSocket.
6. Persist flagged sequences to PostgreSQL for SAR filing.

The trick was to keep the aggregation lightweight. I chose Apache Flink because its native keyed windows run in‑memory and can emit updates every few seconds. I considered Spark Structured Streaming, but Flink’s low‑latency checkpointing fit our 2‑second SLA better.

Python Implementation

Below is the core Flink job written in PyFlink. I broke the logic into three functions:

• enrich_event: joins the raw Kafka record with the risk cache.
• compute_dynamic_threshold: calculates the per‑customer limit.
• detect_structuring: emits an alert if the rolling sum exceeds the threshold.

Why Python and not Java? Our data‑science team already owns the risk model in scikit‑learn; keeping everything in Python avoided a costly serialization layer.

from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings
from pyflink.table.udf import udf

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)

settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
t_env = StreamTableEnvironment.create(env, environment_settings=settings)

# 1️⃣ Enrichment UDF
@udf(result_type=DataTypes.ROW([DataTypes.FIELD('customer_id', DataTypes.STRING()),
                               DataTypes.FIELD('risk_score', DataTypes.FLOAT()),
                               DataTypes.FIELD('amount', DataTypes.BIGINT()),
                               DataTypes.FIELD('timestamp', DataTypes.TIMESTAMP(3))]))
def enrich_event(kafka_record):
    # kafka_record is a JSON string
    import json, redis
    rec = json.loads(kafka_record)
    r = redis.StrictRedis(host='risk‑cache', port=6379, db=0)
    risk = float(r.get(rec['customer_id']) or 0.0)
    return (rec['customer_id'], risk, rec['amount'], rec['event_time'])

t_env.register_function('enrich_event', enrich_event)

# 2️⃣ Dynamic threshold UDF
@udf(result_type=DataTypes.BIGINT())
def compute_dynamic_threshold(risk_score):
    base = 100_000
    # BFIU allows 20% leeway for high‑risk profiles
    return int(base * (1 + 0.2 * risk_score))

t_env.register_function('compute_dynamic_threshold', compute_dynamic_threshold)

# 3️⃣ Alert detection
@udf(result_type=DataTypes.ROW([DataTypes.FIELD('customer_id', DataTypes.STRING()),
                               DataTypes.FIELD('total', DataTypes.BIGINT()),
                               DataTypes.FIELD('threshold', DataTypes.BIGINT()),
                               DataTypes.FIELD('alert_time', DataTypes.TIMESTAMP(3))]))
def detect_structuring(total, threshold, cust_id, ts):
    if total > threshold:
        return (cust_id, total, threshold, ts)
    else:
        return None

t_env.register_function('detect_structuring', detect_structuring)

# Table definition for raw Kafka stream
t_env.execute_sql("""
CREATE TABLE raw_tx (
    kafka_record STRING,
    proc_time AS PROCTIME()
) WITH (
    'connector' = 'kafka',
    'topic' = 'mfs_transactions',
    'properties.bootstrap.servers' = 'kafka-broker:9092',
    'format' = 'json',
    'scan.startup.mode' = 'earliest-offset'
)
""")

# Enrich, window, and alert
t_env.execute_sql("""
INSERT INTO alerts
SELECT *
FROM (
    SELECT
        cust.customer_id,
        SUM(cust.amount) AS total,
        compute_dynamic_threshold(cust.risk_score) AS threshold,
        TUMBLE_END(cust.timestamp, INTERVAL '5' MINUTE) AS alert_time
    FROM (
        SELECT enrich_event(kafka_record) AS cust
        FROM raw_tx
    )
    GROUP BY TUMBLE(cust.timestamp, INTERVAL '5' MINUTE), cust.customer_id, cust.risk_score
) 
WHERE detect_structuring(total, threshold, cust.customer_id, alert_time) IS NOT NULL
""")

Explanation of choices:

• Using a tumbling 5‑minute window gives us a balance—fast enough to spot rapid structuring, yet not too noisy.
• The risk_score lives in Redis; a 2‑ms lookup per event is negligible compared to network I/O.
• We emit alerts only when detect_structuring returns a non‑null row, cutting downstream traffic by ~70%.

Local Application

BFIU’s Circular 5/2024 mandates that any sequence of transactions exceeding the cumulative threshold must be reported within 24 hours. Our dashboard now satisfies that by:

1. Displaying alerts in real time on a React‑based UI that polls the Redis pub/sub channel.
2. Storing the flagged sequence with transaction_id, source, destination, amount, timestamp in a PostgreSQL table named sar_candidates, ready for the compliance officer to export.

The UI also shows a heat‑map of Dhaka’s zones, because we learned that most structuring attempts originate from the Uttara and Mirpur corridors during 02:00‑04:00 am. Adding geolocation as a visual cue reduced manual triage time by another 15 minutes per shift.

Common Pitfalls & Edge Cases

When we first went live, three things tripped us up:

• Late‑arriving events: Mobile networks sometimes delay delivery reports by minutes. We solved it by enabling Flink’s allowedLateness of 2 minutes and using event‑time watermarks.
• Duplicate Kafka records: Network retries caused the same transaction to appear twice. A deduplication key on transaction_id in the upstream Kafka connector fixed it.
• Risk‑score cache miss: New customers had no entry, resulting in a zero score and a too‑low threshold. We added a fallback that pulls the latest KYC snapshot from PostgreSQL on a cache miss.

Counterintuitive Insight

We expected that lowering the static BDT 100,000 limit would reduce false positives. In practice, the opposite happened. By adding the risk‑adjusted multiplier, high‑risk customers got a *higher* threshold, which prevented the engine from screaming every time a small vendor topped up his wallet. Meanwhile, low‑risk customers kept the original limit, and the system flagged only truly suspicious bursts. The lesson? “One‑size‑fits‑all” thresholds are the enemy of precision in a high‑velocity MFS environment.

Conclusion & CTA

If you’re still running nightly batch jobs on a MySQL dump, you’re probably missing the next big structuring ring. Switch to a streaming‑first design, embed the local risk score, and watch your false‑positive rate tumble. I’d love to hear how your dashboard behaves under Dhaka’s midnight surge. Drop a comment, share a screenshot, or try the GitHub repo linked at the bottom of this post. And if you need a quick starter kit, check out the “Real‑Time MFS Alert Boilerplate” on aitipseveryday.com.