Real-Time AML, KYC, and Payment Analytics with Streaming SQL

Real-time anti-money laundering (AML), know-your-customer (KYC), and payment analytics with streaming SQL means the transaction patterns compliance and payment operations teams watch, structuring below reporting thresholds, velocity spikes, sanctioned-jurisdiction wires, overdue reviews, gateway failure rates, stuck settlements, are maintained continuously as materialized views and routed to case management or alerting within seconds of the triggering transaction. This guide builds that monitoring layer on RisingWave, a PostgreSQL-compatible streaming database, covering four AML typologies, event-driven KYC, payment method and gateway analytics, payment rail health and settlement aging, and a general financial alerting pattern with dynamic thresholds and deduplication.

Why Compliance and Payment Monitoring Are Streaming Problems

Traditional AML monitoring runs nightly. A transaction at 9 AM is reviewed, at best, when the overnight job completes, and FinCEN guidance, the EU's AMLD6, and central bank rules in many jurisdictions now reference prompt detection as the standard. KYC has been organized around periodic reviews, every one to three years, while regulators cite "inadequate ongoing monitoring" as the specific deficiency in most enforcement actions, with fines routinely above $100 million. Payment operations teams learn about a degraded gateway or a stalled settlement queue from customer complaints because the batch dashboard refreshes hourly.

Every one of these patterns is a windowed aggregation, a join against reference data, or a time-bounded sequence over a transaction stream. Flink can express them, but a structuring detector in Flink is a keyed process function with state descriptors and timers, and changing the window from 24 to 48 hours means rebuilding, redeploying, and migrating state. In SQL it is changing one INTERVAL. AML rules change often as typologies evolve, so iteration speed is a material compliance advantage.

RisingWave maintains each view below incrementally, persists state to S3, and serves results over the PostgreSQL wire protocol. It is open source under Apache 2.0 and written in Rust. NOW() in a WHERE or FILTER clause acts as a temporal filter, which is how rolling 24-hour, 7-day, 30-day, and 90-day windows coexist without separate pipelines.

Sources: Transactions, Payments, and Reference Data

One banking transaction stream feeds AML and KYC. Payment gateway and rail events feed payment analytics. Customer profiles, risk tiers, high-risk jurisdictions, merchants, and alert thresholds are reference tables, several of them fed by CDC from core banking.

CREATE SOURCE transactions (
    transaction_id        VARCHAR,
    account_id            VARCHAR,
    customer_id           VARCHAR,
    transaction_type      VARCHAR,    -- 'cash_deposit', 'cash_withdrawal', 'wire_in', 'wire_out', 'transfer', 'debit', 'credit'
    amount                NUMERIC,
    currency              VARCHAR,
    counterparty_account  VARCHAR,
    counterparty_bank     VARCHAR,
    counterparty_country  VARCHAR,
    channel               VARCHAR,
    balance_after         NUMERIC,
    event_time            TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'transactions',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE TABLE customer_profiles (
    customer_id        VARCHAR PRIMARY KEY,
    risk_tier          VARCHAR,       -- 'LOW', 'MEDIUM', 'HIGH', 'PROHIBITED'
    is_pep             BOOLEAN,       -- politically exposed person
    is_sanctioned      BOOLEAN,
    nationality        VARCHAR(2),
    residence_country  VARCHAR(2),
    onboarding_date    DATE,
    last_kyc_review    DATE
) WITH (
    connector = 'postgres-cdc',
    hostname = 'core-banking.internal', port = '5432',
    username = 'rw_reader', password = 'secret',
    database.name = 'customers', schema.name = 'public', table.name = 'customer_profiles'
);

CREATE TABLE high_risk_jurisdictions (
    country_code  VARCHAR PRIMARY KEY,
    risk_level    VARCHAR,            -- 'HIGH', 'SANCTIONED', 'MONITORED'
    fatf_status   VARCHAR,            -- 'BLACKLIST', 'GREYLIST', 'COMPLIANT'
    updated_at    TIMESTAMPTZ
);

AML Typology 1: Structuring

Structuring, or smurfing, is breaking cash activity into amounts below the Currency Transaction Report threshold ($10,000 in the US) to avoid mandatory reporting. No single transaction looks wrong; the rolling 24-hour total does.

CREATE MATERIALIZED VIEW cash_aggregates_24h AS
SELECT
    customer_id,
    account_id,
    SUM(amount)                                                    AS total_cash_24h,
    SUM(amount) FILTER (WHERE transaction_type = 'cash_deposit')   AS cash_deposits_24h,
    COUNT(*)                                                       AS cash_txn_count_24h,
    MAX(amount)                                                    AS largest_single_cash_txn
FROM transactions
WHERE event_time > NOW() - INTERVAL '24 HOURS'
  AND transaction_type IN ('cash_deposit', 'cash_withdrawal')
GROUP BY customer_id, account_id;

CREATE MATERIALIZED VIEW aml_alert_structuring AS
SELECT customer_id, account_id, total_cash_24h, cash_txn_count_24h, largest_single_cash_txn,
       'STRUCTURING' AS typology, 'HIGH' AS alert_severity,
       FORMAT('%s in cash over 24h across %s transactions, none above the CTR threshold (max %s)',
              total_cash_24h::TEXT, cash_txn_count_24h::TEXT, largest_single_cash_txn::TEXT) AS narrative
FROM cash_aggregates_24h
WHERE total_cash_24h >= 9000
  AND largest_single_cash_txn < 10000
  AND cash_txn_count_24h >= 2;

The narrative column matters. Investigators need the reason in plain language inside the case record, and generating it in the view means every alert carries it.

AML Typology 2: Velocity Spikes Against a Baseline

Laundering often shows up as activity far above a customer's own norm. A seven-day window compared with a 30-day baseline normalized to weekly rates catches it without a fixed threshold that would be wrong for both a retail customer and a wholesaler:

CREATE MATERIALIZED VIEW transaction_velocity_7d AS
SELECT customer_id,
       COUNT(*)                                                     AS txn_count_7d,
       SUM(amount)                                                  AS total_volume_7d,
       COUNT(DISTINCT counterparty_account)                         AS unique_counterparties_7d,
       COUNT(DISTINCT counterparty_country)                         AS unique_countries_7d,
       SUM(amount) FILTER (WHERE transaction_type = 'wire_out')     AS wire_out_7d
FROM transactions
WHERE event_time > NOW() - INTERVAL '7 DAYS'
GROUP BY customer_id;

CREATE MATERIALIZED VIEW transaction_baseline_30d AS
SELECT customer_id,
       COUNT(*) / 4.0      AS avg_weekly_txn_count,
       SUM(amount) / 4.0   AS avg_weekly_volume
FROM transactions
WHERE event_time > NOW() - INTERVAL '30 DAYS'
GROUP BY customer_id;

CREATE MATERIALIZED VIEW aml_alert_velocity_spike AS
SELECT v.customer_id, v.txn_count_7d, v.total_volume_7d, b.avg_weekly_volume,
       v.total_volume_7d / NULLIF(b.avg_weekly_volume, 0) AS volume_ratio,
       'VELOCITY_SPIKE' AS typology, 'MEDIUM' AS alert_severity,
       FORMAT('Weekly volume %s is %.1fx the 30-day weekly average %s',
              v.total_volume_7d::TEXT, v.total_volume_7d / NULLIF(b.avg_weekly_volume, 0), b.avg_weekly_volume::TEXT) AS narrative
FROM transaction_velocity_7d v
JOIN transaction_baseline_30d b ON v.customer_id = b.customer_id
WHERE b.avg_weekly_volume > 0
  AND v.total_volume_7d > b.avg_weekly_volume * 3;

AML Typology 3: High-Risk and Sanctioned Jurisdictions

Wires to or from FATF blacklist or greylist countries need immediate attention. The jurisdiction table is maintained by compliance and joined on the counterparty country:

CREATE MATERIALIZED VIEW hri_wire_activity_7d AS
SELECT t.customer_id, t.counterparty_country, j.risk_level, j.fatf_status,
       COUNT(*) AS wire_count_7d, SUM(t.amount) AS wire_total_7d, MAX(t.event_time) AS last_wire_time
FROM transactions t
JOIN high_risk_jurisdictions j ON t.counterparty_country = j.country_code
WHERE t.event_time > NOW() - INTERVAL '7 DAYS'
  AND t.transaction_type IN ('wire_out', 'wire_in')
GROUP BY t.customer_id, t.counterparty_country, j.risk_level, j.fatf_status;

CREATE MATERIALIZED VIEW aml_alert_sanctions AS
SELECT customer_id, counterparty_country, wire_total_7d, wire_count_7d, fatf_status,
       'SANCTIONS_EXPOSURE' AS typology,
       CASE WHEN fatf_status = 'BLACKLIST' THEN 'CRITICAL' WHEN fatf_status = 'GREYLIST' THEN 'HIGH' ELSE 'MEDIUM' END AS alert_severity,
       FORMAT('%s wires totaling %s with %s (%s) in the last 7 days',
              wire_count_7d::TEXT, wire_total_7d::TEXT, counterparty_country, fatf_status) AS narrative
FROM hri_wire_activity_7d
WHERE fatf_status IN ('BLACKLIST', 'GREYLIST');

An hourly tumbling window over the same join gives compliance a view of total institutional exposure per high-risk country, which is a reporting requirement in its own right.

AML Typology 4: Round Trips and Rapid Withdrawals

Layering moves money through accounts to obscure its origin. Two sequence patterns catch common forms with a time-bounded self-join: funds leaving and returning to the same account within 24 hours, and a large deposit followed by withdrawal of most of it within two hours, the money mule signature.

CREATE MATERIALIZED VIEW aml_alert_round_trip AS
SELECT o.customer_id, o.account_id,
       o.transaction_id AS outflow_txn_id, i.transaction_id AS inflow_txn_id,
       o.amount AS outflow_amount, i.amount AS inflow_amount,
       o.counterparty_account AS via_account, o.event_time AS outflow_time, i.event_time AS inflow_time,
       'ROUND_TRIP' AS typology, 'HIGH' AS alert_severity,
       FORMAT('Outflow of %s to %s followed by near-equal inflow of %s within 24h',
              o.amount::TEXT, o.counterparty_account, i.amount::TEXT) AS narrative
FROM transactions o
JOIN transactions i
  ON o.account_id = i.account_id
 AND o.transaction_type IN ('wire_out', 'transfer')
 AND i.transaction_type IN ('wire_in', 'transfer')
 AND o.amount > 5000
 AND i.amount BETWEEN o.amount * 0.85 AND o.amount * 1.15
 AND i.event_time > o.event_time
 AND i.event_time < o.event_time + INTERVAL '24 HOURS';

CREATE MATERIALIZED VIEW aml_alert_rapid_withdrawal AS
SELECT c.account_id, c.customer_id,
       c.transaction_id AS deposit_txn_id, d.transaction_id AS withdrawal_txn_id,
       c.amount AS deposit_amount, d.amount AS withdrawal_amount,
       c.event_time AS deposit_time, d.event_time AS withdrawal_time,
       'RAPID_WITHDRAWAL' AS typology, 'HIGH' AS alert_severity,
       FORMAT('Deposit of %s followed by withdrawal of %s within 2 hours', c.amount::TEXT, d.amount::TEXT) AS narrative
FROM transactions c
JOIN transactions d
  ON c.account_id = d.account_id
 AND c.transaction_type IN ('credit', 'wire_in')
 AND d.transaction_type IN ('debit', 'wire_out', 'transfer')
 AND c.amount > 5000
 AND d.amount >= c.amount * 0.9
 AND d.event_time > c.event_time
 AND d.event_time < c.event_time + INTERVAL '2 HOURS';

Prioritizing Alerts by Customer Risk and Routing to Case Management

A structuring alert on a politically exposed person is more urgent than the same pattern on a standard retail customer. Union the typologies, join the profile, and escalate severity before the sink:

CREATE MATERIALIZED VIEW aml_alerts_enriched AS
SELECT a.customer_id, a.typology, a.alert_severity, a.narrative, NOW() AS detected_at,
       p.risk_tier, p.is_pep, p.residence_country,
       CASE
           WHEN p.is_pep AND a.alert_severity <> 'CRITICAL'          THEN 'CRITICAL'
           WHEN p.risk_tier = 'HIGH' AND a.alert_severity = 'MEDIUM' THEN 'HIGH'
           ELSE a.alert_severity
       END AS effective_severity
FROM (
    SELECT customer_id, typology, alert_severity, narrative FROM aml_alert_structuring
    UNION ALL SELECT customer_id, typology, alert_severity, narrative FROM aml_alert_velocity_spike
    UNION ALL SELECT customer_id, typology, alert_severity, narrative FROM aml_alert_sanctions
    UNION ALL SELECT customer_id, typology, alert_severity, narrative FROM aml_alert_round_trip
    UNION ALL SELECT customer_id, typology, alert_severity, narrative FROM aml_alert_rapid_withdrawal
) a
LEFT JOIN customer_profiles p ON a.customer_id = p.customer_id;

CREATE SINK aml_alerts_sink AS
SELECT * FROM aml_alerts_enriched WHERE effective_severity IN ('HIGH', 'CRITICAL')
WITH (connector = 'kafka', topic = 'aml-alerts', properties.bootstrap.server = 'kafka:9092')
FORMAT PLAIN ENCODE JSON;

Case management systems such as Actimize or Quantexa, or an in-house tool, consume the topic and open investigations. Critical alerts arrive seconds after the transaction instead of the next morning.

Event-Driven KYC: Ongoing Monitoring Instead of Periodic Review

KYC becomes continuous when the customer's behavioral profile is a live view joined against their onboarding data. The signals that should trigger re-verification are cash intensity, cross-border wire breadth, volume far above the assigned tier, and an overdue review for a high-risk customer:

CREATE MATERIALIZED VIEW customer_behavioral_profile AS
SELECT customer_id,
       COUNT(*)                                                     AS tx_count_90d,
       SUM(amount)                                                  AS total_volume_90d,
       AVG(amount)                                                  AS avg_tx_amount,
       COUNT(DISTINCT counterparty_bank)                            AS distinct_banks,
       COUNT(DISTINCT counterparty_country)                         AS distinct_countries,
       COUNT(*)    FILTER (WHERE transaction_type IN ('wire_in', 'wire_out'))  AS wire_count,
       SUM(amount) FILTER (WHERE transaction_type = 'cash_deposit')             AS cash_deposit_volume,
       MAX(event_time)                                              AS last_activity
FROM transactions
WHERE event_time >= NOW() - INTERVAL '90 DAYS'
GROUP BY customer_id;

CREATE MATERIALIZED VIEW kyc_risk_signals AS
SELECT bp.customer_id, cp.risk_tier, cp.last_kyc_review, cp.is_pep,
       (bp.cash_deposit_volume > 10000)::INT                                                  AS cash_intensity_flag,
       (bp.wire_count > 20 AND bp.distinct_countries > 5)::INT                                AS cross_border_flag,
       (bp.total_volume_90d > 500000 AND cp.risk_tier = 'LOW')::INT                           AS volume_anomaly_flag,
       (cp.last_kyc_review < CURRENT_DATE - INTERVAL '365 DAYS' AND cp.risk_tier = 'HIGH')::INT AS overdue_review_flag,
       bp.tx_count_90d, bp.total_volume_90d, bp.distinct_countries, bp.last_activity
FROM customer_behavioral_profile bp
JOIN customer_profiles cp ON bp.customer_id = cp.customer_id;

CREATE SINK kyc_review_triggers AS
SELECT customer_id, risk_tier, last_activity, NOW() AS trigger_time,
       cash_intensity_flag + cross_border_flag + volume_anomaly_flag + overdue_review_flag AS total_flags
FROM kyc_risk_signals
WHERE cash_intensity_flag + cross_border_flag + volume_anomaly_flag + overdue_review_flag >= 2
   OR overdue_review_flag = 1
WITH (connector = 'kafka', properties.bootstrap.server = 'kafka:9092', topic = 'kyc-review-queue')
FORMAT PLAIN ENCODE JSON;

Because customer_profiles is a CDC table, a sanctions list update that flips is_sanctioned or a tier change made by an analyst propagates into every dependent view within seconds, which is the event-driven behavior regulators now expect.

Payment Analytics for Merchants: Methods, Gateways, and Exposure

E-commerce platforms accept cards, wallets such as Alipay and PayPal, buy now pay later, cash on delivery, and installments, each with different fees, failure rates, and chargeback patterns. Four operational needs make this real time: a gateway degrading shows as a success-rate drop within one five-minute window, BNPL providers cap daily volume per merchant, cash-on-delivery counts drive same-day dispatch planning, and flash sale incentives can be adjusted based on which methods shoppers are actually using.

CREATE SOURCE payment_events (
    payment_id       VARCHAR,
    order_id         VARCHAR,
    merchant_id      VARCHAR,
    channel          VARCHAR,      -- 'web', 'mobile', 'pos', 'marketplace'
    payment_method   VARCHAR,      -- 'card', 'alipay', 'paypal', 'cod', 'bnpl', 'installment'
    gateway          VARCHAR,      -- 'stripe', 'adyen', 'alipay_gateway'
    amount           NUMERIC,
    currency         VARCHAR,
    outcome          VARCHAR,      -- 'success', 'failure', 'pending', 'refunded'
    failure_code     VARCHAR,
    occurred_at      TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'ecommerce.payments',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;

CREATE MATERIALIZED VIEW mv_payment_hourly AS
SELECT merchant_id, payment_method, channel, window_start AS hour_start,
       COUNT(*)                                                      AS transaction_count,
       SUM(amount) FILTER (WHERE outcome = 'success')                AS successful_volume,
       COUNT(*)    FILTER (WHERE outcome = 'failure')                AS failure_count,
       COUNT(*)    FILTER (WHERE outcome = 'refunded')               AS refund_count,
       ROUND(COUNT(*) FILTER (WHERE outcome = 'success')::NUMERIC / NULLIF(COUNT(*), 0) * 100, 2) AS success_rate_pct,
       ROUND(AVG(amount) FILTER (WHERE outcome = 'success'), 2)      AS avg_transaction_value
FROM TUMBLE(payment_events, occurred_at, INTERVAL '1 HOUR')
GROUP BY merchant_id, payment_method, channel, window_start;

CREATE MATERIALIZED VIEW mv_gateway_health_5min AS
SELECT gateway, payment_method, window_start,
       COUNT(*)                                            AS attempts,
       COUNT(*) FILTER (WHERE outcome = 'success')         AS successes,
       ROUND(COUNT(*) FILTER (WHERE outcome = 'success')::NUMERIC / NULLIF(COUNT(*), 0) * 100, 2) AS success_rate_pct,
       COUNT(DISTINCT failure_code)                        AS distinct_failure_codes
FROM TUMBLE(payment_events, occurred_at, INTERVAL '5 MINUTES')
GROUP BY gateway, payment_method, window_start;

CREATE MATERIALIZED VIEW mv_payment_mix AS
SELECT merchant_id, payment_method, hour_start, successful_volume,
       ROUND(successful_volume / NULLIF(SUM(successful_volume) OVER (PARTITION BY merchant_id, hour_start), 0) * 100, 2) AS method_share_pct
FROM mv_payment_hourly;

CREATE MATERIALIZED VIEW mv_bnpl_daily_exposure AS
SELECT merchant_id, DATE_TRUNC('day', hour_start) AS day, SUM(successful_volume) AS bnpl_volume, SUM(transaction_count) AS bnpl_count
FROM mv_payment_hourly
WHERE payment_method = 'bnpl'
GROUP BY merchant_id, DATE_TRUNC('day', hour_start);

The payment mix and BNPL exposure views read from the hourly view rather than the raw stream, so they cost almost nothing to maintain. Gateway health goes to a Kafka alert topic; a consumer pages when success_rate_pct for a gateway falls below the configured floor for two consecutive windows.

Payment Rail Operations: Throughput, Success Rates, and Settlement Aging

Payment processors see the same problem from the network side. Card networks, ACH, RTP, FedNow, and SWIFT each have their own authorization, clearing, and settlement lifecycle, and operations teams need throughput per rail and the aging of payments stuck between clearing and settlement.

CREATE SOURCE rail_events (
    event_id      VARCHAR,
    payment_id    VARCHAR,
    merchant_id   VARCHAR,
    payment_rail  VARCHAR,       -- 'visa', 'mastercard', 'ach', 'rtp', 'wire'
    event_type    VARCHAR,       -- 'authorized', 'cleared', 'settled', 'declined', 'reversed'
    amount        NUMERIC,
    region        VARCHAR,
    error_code    VARCHAR,
    event_time    TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'payments.events',
    properties.bootstrap.server = 'broker:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE MATERIALIZED VIEW payment_throughput_metrics AS
SELECT payment_rail, region, window_start, window_end,
       COUNT(*)                                            AS total_events,
       COUNT(*) FILTER (WHERE event_type = 'authorized')   AS authorized_count,
       COUNT(*) FILTER (WHERE event_type = 'declined')     AS declined_count,
       COUNT(*) FILTER (WHERE event_type = 'reversed')     AS reversed_count,
       SUM(amount) FILTER (WHERE event_type = 'settled')   AS settled_volume,
       ROUND(COUNT(*) FILTER (WHERE event_type = 'authorized') * 100.0
             / NULLIF(COUNT(*) FILTER (WHERE event_type IN ('authorized', 'declined')), 0), 2) AS auth_success_rate_pct
FROM HOP(rail_events, event_time, INTERVAL '1 MINUTE', INTERVAL '5 MINUTES')
GROUP BY payment_rail, region, window_start, window_end;

CREATE MATERIALIZED VIEW settlement_exceptions AS
SELECT payment_id, merchant_id, payment_rail, amount, cleared_at, settled_at,
       EXTRACT(EPOCH FROM (settled_at - cleared_at)) / 3600 AS settlement_hours,
       CASE WHEN settled_at IS NULL AND cleared_at < NOW() - INTERVAL '24 HOURS' THEN 'STUCK_IN_CLEARING'
            WHEN EXTRACT(EPOCH FROM (settled_at - cleared_at)) / 3600 > 48      THEN 'LATE_SETTLEMENT'
       END AS exception_type
FROM (
    SELECT payment_id, merchant_id, payment_rail, MAX(amount) AS amount,
           MIN(event_time) FILTER (WHERE event_type = 'cleared') AS cleared_at,
           MAX(event_time) FILTER (WHERE event_type = 'settled') AS settled_at
    FROM rail_events
    GROUP BY payment_id, merchant_id, payment_rail
) lifecycle
WHERE (settled_at IS NULL AND cleared_at < NOW() - INTERVAL '24 HOURS')
   OR EXTRACT(EPOCH FROM (settled_at - cleared_at)) / 3600 > 48;

A five-minute hopping window updated every minute is the right grain for an operations dashboard: trend visibility without noise. The settlement view is a per-payment lifecycle reconstruction, and a payment appears in it the moment it crosses 24 hours in clearing, not after batch reconciliation.

A General Financial Alerting Pattern

Every fintech has Kafka carrying transactions, ledger events, and balance changes. Kafka moves data; it does not watch it. The usual answer is a consumer application with threshold logic, and the moment the alert is stateful ("spending in the last hour above $10,000") the consumer is managing in-memory state, restarts, and deployments. Financial alerts need temporal state, cross-entity joins, rate-of-change detection, and sequence detection, which is exactly what windowed SQL provides.

The AML views above are one instance of the pattern. Customer-facing alerts follow the same shape:

CREATE TABLE alert_config (
    alert_type       VARCHAR PRIMARY KEY,
    enabled          BOOLEAN,
    threshold_value  NUMERIC,
    severity         VARCHAR
);

INSERT INTO alert_config VALUES
    ('LARGE_TRANSACTION', TRUE, 10000, 'HIGH'),
    ('LOW_BALANCE',       TRUE,   100, 'MEDIUM'),
    ('HOURLY_VELOCITY',   TRUE,  5000, 'HIGH');

CREATE MATERIALIZED VIEW alert_large_transaction AS
SELECT t.transaction_id, t.account_id, t.customer_id, t.amount, t.event_time,
       'LARGE_TRANSACTION' AS alert_type, c.severity,
       FORMAT('Transaction of %s exceeds threshold %s', t.amount::TEXT, c.threshold_value::TEXT) AS message
FROM transactions t
JOIN alert_config c ON c.alert_type = 'LARGE_TRANSACTION' AND c.enabled
WHERE t.transaction_type = 'debit' AND t.amount > c.threshold_value;

CREATE MATERIALIZED VIEW alert_low_balance AS
SELECT t.transaction_id, t.account_id, t.customer_id, t.balance_after, t.event_time,
       'LOW_BALANCE' AS alert_type, c.severity,
       FORMAT('Balance dropped to %s after a debit of %s', t.balance_after::TEXT, t.amount::TEXT) AS message
FROM transactions t
JOIN alert_config c ON c.alert_type = 'LOW_BALANCE' AND c.enabled
WHERE t.transaction_type IN ('debit', 'transfer_out')
  AND t.balance_after >= 0 AND t.balance_after < c.threshold_value;

CREATE MATERIALIZED VIEW hourly_spending AS
SELECT account_id, SUM(amount) AS total_spent_1h, COUNT(*) AS txn_count_1h
FROM transactions
WHERE transaction_type = 'debit' AND event_time > NOW() - INTERVAL '1 HOUR'
GROUP BY account_id;

CREATE MATERIALIZED VIEW baseline_hourly_rate_7d AS
SELECT account_id, SUM(amount) / (7 * 24) AS avg_hourly_spend
FROM transactions
WHERE transaction_type = 'debit' AND event_time > NOW() - INTERVAL '7 DAYS'
GROUP BY account_id;

CREATE MATERIALIZED VIEW alert_spending_spike AS
SELECT h.account_id, h.total_spent_1h, b.avg_hourly_spend,
       h.total_spent_1h / NULLIF(b.avg_hourly_spend, 0) AS spike_ratio,
       NOW() AS alert_time, 'SPENDING_SPIKE' AS alert_type, 'HIGH' AS severity,
       FORMAT('Spending rate is %.1fx the 7-day average', h.total_spent_1h / NULLIF(b.avg_hourly_spend, 0)) AS message
FROM hourly_spending h
JOIN baseline_hourly_rate_7d b ON h.account_id = b.account_id
WHERE b.avg_hourly_spend > 0 AND h.total_spent_1h > b.avg_hourly_spend * 5;

Thresholds live in alert_config, so changing one is an UPDATE, not a view rebuild. A rolling-window alert can fire repeatedly for the same condition, so add a cooldown before the sink:

CREATE MATERIALIZED VIEW alert_dedup_state AS
SELECT account_id, alert_type, MAX(alert_time) AS last_alerted_at
FROM alert_spending_spike
GROUP BY account_id, alert_type;

CREATE MATERIALIZED VIEW alert_spending_spike_deduped AS
SELECT s.*
FROM alert_spending_spike s
LEFT JOIN alert_dedup_state d ON s.account_id = d.account_id AND s.alert_type = d.alert_type
WHERE d.last_alerted_at IS NULL OR s.alert_time > d.last_alerted_at + INTERVAL '30 MINUTES';

CREATE SINK alerts_kafka_sink AS
SELECT transaction_id, account_id, customer_id, alert_type, severity, message, event_time FROM alert_large_transaction
UNION ALL
SELECT transaction_id, account_id, customer_id, alert_type, severity, message, event_time FROM alert_low_balance
UNION ALL
SELECT account_id, account_id, NULL, alert_type, severity, message, alert_time FROM alert_spending_spike_deduped
WITH (connector = 'kafka', topic = 'financial-alerts', properties.bootstrap.server = 'kafka:9092')
FORMAT PLAIN ENCODE JSON;

End to end, a transaction published to Kafka reaches the user's phone in roughly 100 to 300 milliseconds: 5 to 50 ms for RisingWave to consume it, 10 to 50 ms for the incremental view update, 5 to 20 ms for the sink, and the rest in the notification service. Alerts that must block a transaction before authorization do not go through the async path; the authorization engine queries the risk views synchronously instead.

Batch vs Streaming for Compliance and Payment Monitoring

Frequently Asked Questions

How do I avoid alerting twice on the same structuring pattern as the 24-hour window slides? Apply the cooldown pattern from the alerting section to the AML views, or key the case management system on customer and typology so a second alert within the investigation window attaches to the open case rather than opening a new one.

Can the AML and KYC views share the transaction source with fraud detection? Yes, and they should. One Kafka source feeds fraud rules, AML typologies, KYC profiles, and customer alerts as separate views. Reference tables such as risk tiers and jurisdictions are shared as well.

How are sanctions and high-risk jurisdiction lists kept current? Load them into tables that compliance owns, either by direct upsert or by CDC from the system of record. Every dependent view picks up a list change on the next event, with no restart.

What about typologies that need strict ordered matching across many steps? Time-bounded self-joins cover two-step sequences such as round trips and rapid withdrawals. RisingWave does not support MATCH_RECOGNIZE, so multi-step ordered patterns are usually expressed as chained joins or handled by a downstream rules engine fed by the sink.

How do I audit what a view saw when an alert fired? Sink the alert views, including the narrative and the aggregate values that triggered them, to Iceberg or PostgreSQL. Because Kafka retains the source log, you can also replay the exact window into a scratch view to reproduce the computation for an examiner.