Building an AML Transaction Monitoring Script in Python & SQL

What transaction monitoring actually does

At its core, transaction monitoring is a job that reads new transactions on a schedule, runs each one through a set of rules, and raises an alert when something looks off. The complicated products you see on vendor websites are mostly a nicer UI on top of that loop. If you can write SQL and a Python script, you can build the loop.

The three pieces you need: a source of transactions (usually a database), a set of rules (thresholds, patterns, watchlists), and somewhere to send alerts (Slack, email, a case-management table).

The rules you start with

Real AML programs cover dozens of typologies. For a first script, four rules cover most of what regulators expect to see and are easy to reason about:

• Large single transaction — anything over a fixed threshold (e.g. $10,000).
• Structuring — several transactions under the threshold that sum above it within a short window.
• High-risk geography — sends to or from countries on a sanctions or high-risk list.
• Velocity — a user's transaction count or volume spiking versus their own recent baseline.

SQL for the heavy lifting

Push as much work into SQL as you can — it's faster and easier to audit than equivalent pandas. Structuring is a good example: a window function does the rolling sum per user cleanly.

-- Flag structuring: 24h rolling volume per user
SELECT
  user_id,
  transaction_id,
  amount,
  created_at,
  SUM(amount) OVER (
    PARTITION BY user_id
    ORDER BY created_at
    RANGE BETWEEN INTERVAL '24 hours' PRECEDING AND CURRENT ROW
  ) AS rolling_24h_volume
FROM transactions
WHERE created_at >= NOW() - INTERVAL '48 hours'
  AND amount < 10000
QUALIFY rolling_24h_volume >= 10000;

Python for the loop and the alerts

Python's job is to run the SQL, apply anything awkward to express in SQL (like fuzzy name matching against a watchlist), and push alerts out.

import os, requests, pandas as pd
from sqlalchemy import create_engine

engine = create_engine(os.environ["DATABASE_URL"])
SLACK = os.environ["SLACK_WEBHOOK_URL"]

def run_rule(name, sql):
    df = pd.read_sql(sql, engine)
    for _, row in df.iterrows():
        alert(name, row.to_dict())

def alert(rule, payload):
    requests.post(SLACK, json={
        "text": f":rotating_light: *{rule}* triggered\n```{payload}```"
    })

if __name__ == "__main__":
    run_rule("large_transaction",
             "SELECT * FROM transactions WHERE amount >= 10000 "
             "AND created_at >= NOW() - INTERVAL '15 minutes'")
    run_rule("structuring", open("sql/structuring.sql").read())

Schedule this with cron, a workflow tool, or (my preference) a small scheduler service that also records every run — you'll want an audit trail the first time compliance asks whether a rule fired last Tuesday.

What to log, and why it matters

For each alert, write a row to an alerts table with the rule name, the transaction and user IDs, a snapshot of the fields the rule evaluated, and the timestamp. This gives you two things for free: a queue analysts can work through, and a dataset you can measure precision on later to tune thresholds down.

Where this connects to my work

The Reddit competitor and remittance monitor on my portfolio home has the same skeleton: pull new records on a schedule, apply keyword and semantic rules, alert Slack. Swap Reddit posts for payments and the keyword rules for AML typologies and you've built the first useful version of a transaction monitoring system.