AML Monitoring Systems: How to Build It

# STEP 2. THE DATA MODEL. THE PART YOU CANNOT RETROFIT.
# Monitoring asks questions about BEHAVIOUR OVER TIME. A transactions table
# alone forces every scenario into a full scan, and the system becomes too
# slow to run daily -- at which point someone quietly reduces the frequency.

AGGREGATES = {
    # Maintain these incrementally, per customer, per window.
    "windows": ["1d", "7d", "30d", "90d", "365d"],
    "per_window": [
        "credit_count", "credit_value", "debit_count", "debit_value",
        "cash_credit_value",          # CTR aggregation runs off this
        "distinct_counterparties",
        "distinct_geographies",
        "max_single_value",
        "round_amount_count",         # structuring signal
    ],
    # Store the BASELINE the customer declared, and the OBSERVED baseline.
    # A scenario compares against both; they diverge, and the divergence is
    # itself the finding.
    "baselines": ["declared_at_onboarding", "observed_trailing_180d"],
}

# STEP 3. NAMED SCENARIOS, NOT A SCORE.
# You must be able to tell an examiner WHY an alert fired. A gradient-boosted
# score cannot answer that, and an unexplainable alert is worse than none.
# Start deterministic. Add a model LATER, to RANK the queue, never to raise
# the alert -- ranking makes no decision, so it needs no defence.

SCENARIOS = {
    "structuring": "cash credits clustering just under Rs 10 lakh across a "
                   "rolling calendar month",
    "rapid_movement": "credits followed by near-total debits within 24-48h, "
                      "with little balance retained",
    "profile_departure": "value or volume materially above the declared AND "
                         "observed baseline",
    "unexpected_geography": "counterparties outside the declared markets",
    "dormant_reactivation": "a long-dormant account resuming at volume",
    "round_amounts": "repeated exact round figures, which real trade rarely "
                     "produces",
}

# STEP 4. TUNE BEFORE LAUNCH, AGAINST HISTORY AND AGAINST YOUR STAFFING.
def tune(scenario, history, analysts_per_day, minutes_per_alert):
    capacity = analysts_per_day * 60 * 7.5 / minutes_per_alert
    out = []
    for t in scenario["threshold_range"]:
        alerts = history.replay(scenario, threshold=t)
        out.append({
            "threshold": t,
            "alerts_per_day": alerts.per_day,
            # The number that decides it. A threshold generating more alerts
            # than you can clear is not a sensitive threshold -- it is an
            # unreviewed backlog with a compliance failure inside it.
            "within_capacity": alerts.per_day <= capacity,
            "known_cases_caught": alerts.recall_against(history.confirmed_cases),
        })
    return out   # record this table. The RATIONALE is the deliverable.

# WHAT TO CHECK
# [ ] aggregates are maintained incrementally, not computed on read
# [ ] both baselines are stored -- what they declared and what they do
# [ ] every scenario has a NAME and a sentence a non-engineer understands
# [ ] no model raises an alert. Ranking only
# [ ] tuning is replayed against real history, not reasoned about
# [ ] the chosen threshold is within the volume you can actually clear
# [ ] the tuning table is SAVED. "Why this threshold" is the most common
#     examination question and the most common thing nobody kept

from datetime import date, timedelta

# STEP 5. FOUR TIMESTAMPS. A SYSTEM THAT COLLAPSES THEM CANNOT ANSWER.
CASE_TIMESTAMPS = (
    "alert_raised_at",      # machine. The scenario fired
    "case_opened_at",       # an analyst picked it up
    "suspicion_formed_on",  # HUMAN JUDGEMENT. The 7-working-day clock
    "report_filed_at",      # the filing
)

def form_suspicion(case, analyst, reason, store):
    # Deliberate, attributed, reasoned -- and not editable afterwards.
    assert case["suspicion_formed_on"] is None, "already formed; do not backdate"
    assert reason and len(reason.split()) >= 10, "a reason, not a checkbox"
    store.set_immutable(case["id"], "suspicion_formed_on", today_ist(),
                        by=analyst, reason=reason)
    # Without this field an examiner reads suspicion as forming at the ALERT,
    # and every filing you have ever made looks late.
    return str_due(today_ist())

def str_due(formed_on, holidays):
    # 7 WORKING days. Not 7 calendar days.
    d, left = formed_on, 7
    while left:
        d += timedelta(days=1)
        if d.weekday() < 5 and d not in holidays:
            left -= 1
    return d

# STEP 6. TWO CLOCKS, TWO MECHANISMS. DO NOT BUILD THEM AS ONE.
def ctr_batch(month, ledger):
    # A QUERY, not a judgement. Cash above Rs 10 lakh in a calendar month,
    # aggregated across RELATED transactions. Due the 15th of the next month.
    rows = ledger.cash_credits_by_customer(month)          # related, not single
    return [r for r in rows if r.total > 1_000_000]

# STEP 7. FEED THE OUTCOME BACK. THE LOOP NOBODY CLOSES.
def close_case(case, outcome, rating, stats):
    assert outcome in ("cleared", "filed")
    # A cleared case and a filed case should move the rating differently.
    rating.apply_outcome(case["customer_id"], outcome)
    # And both change what you know about the scenario that raised it.
    stats.record(case["scenario"], outcome)
    return stats.precision(case["scenario"], window="180d")

# WHAT TO CHECK
# [ ] suspicion_formed_on exists, is set by a named analyst with a reason,
#     and cannot be edited after the fact
# [ ] the STR clock counts WORKING days against the actual holiday calendar
# [ ] CTR aggregates RELATED cash transactions across the month, and runs
#     automatically. An analyst noticing it is not a control
# [ ] case outcomes feed both the customer rating and scenario precision
# [ ] scenario precision is reported on a rolling window, so drift is
#     visible before an examiner finds it
# [ ] every timestamp in IST. A UTC filing date near a month boundary is
#     a deadline argument you will lose