Banking Fraud DAG: how it works¶

This page documents the akko_banking_fraud_demo Airflow DAG task by task, with real, verified examples. It is the reference pipeline that proves the AKKO differentiator: governed AI functions called directly from SQL on a sovereign lakehouse, under strict access control.

Source: airflow/dags/akko_banking_fraud_demo.py (mirrored into the chart by helm/scripts/sync-dags.sh and shipped as the akko-airflow-dags ConfigMap).

Task graph¶

seed_transactions --> replicate_to_iceberg --> train_fraud_model --> score_transactions --> refresh_superset

The DAG is idempotent: score_transactions only scores rows where fraud_score IS NULL, so re-running it scores the next unscored batch instead of rewriting existing values.

Task 1: seed_transactions¶

Generates a PaySim-style synthetic dataset (about 0.13% fraud rate, no personal data) into Postgres banking.transactions. The row IDs start at TRANSACTION_ID_BASE = 1000000 so the demo rows are isolable. Schema and generator live in helm/akko/files/demos/banking-fraud/seed/{schema.sql,generate_transactions.py}.

Verified: 10000 demo transactions created.

Task 2: replicate_to_iceberg¶

Replicates the Postgres source into the Iceberg lakehouse using Spark Connect, writing iceberg.banking.transactions.

Verified: 10000 rows in iceberg.banking.transactions (replication completes in about 30s once the Spark Connect driver is healthy).

Task 3: train_fraud_model¶

Trains a baseline fraud model and logs it to MLflow (tracking server, sovereign). This task proves the MLflow layer is wired end to end (experiment, run, model artifact).

Task 4: score_transactions (the governed AI step)¶

This is the heart of the demo. A single transactional Trino MERGE calls two governed AI functions per row and writes the results back into Iceberg.

The scored sample is bounded by SCORE_SAMPLE_LIMIT (default 60) so a live demo stays fast (each AI call is a CPU LLM inference). The MERGE:

MERGE INTO iceberg.banking.transactions t
USING (
    SELECT transaction_id,
           CAST(amount AS DOUBLE) AS amt,
           lower(COALESCE(TRY(json_extract_scalar(
               akko_ai_anomaly(
                   CAST(amount AS VARCHAR),
                   'banking transaction, avg ~2000 EUR, fraud is rare'
               ), '$.is_anomaly')), 'false')) AS is_anom,
           akko_ai_sentiment(COALESCE(description, merchant)) AS sentiment
    FROM iceberg.banking.transactions
    WHERE fraud_score IS NULL AND transaction_id >= 1000000
    LIMIT 60
) s
ON t.transaction_id = s.transaction_id
WHEN MATCHED THEN UPDATE SET
    akko_ai_anomaly = CASE WHEN s.is_anom = 'true' THEN 1.0 ELSE 0.0 END,
    fraud_score  = CASE
        WHEN s.is_anom = 'true'
        THEN round(least(0.99, 0.55 + (s.amt / 25000.0)), 3)
        ELSE round(least(0.45, s.amt / 50000.0), 3)
    END,
    akko_ai_sentiment = s.sentiment,
    scored_at    = CURRENT_TIMESTAMP

What the AI functions return (real output)¶

akko_ai_anomaly(value, context) returns a JSON verdict, not a bare number. Real responses:

akko_ai_anomaly('4909.74',  'banking transaction, avg ~2000 EUR, fraud is rare')
  -> {"is_anomaly": true, "reason": "The value of 4909.74 EUR is significantly higher
      than the average transaction amount ..."}

akko_ai_anomaly('250.00',   ...) -> {"is_anomaly": true,  "reason": "..."}

The DAG parses is_anomaly (TRY-guarded against a malformed reply) and derives a bounded, varied fraud_score that combines the AI flag with the transaction exposure (amount).

akko_ai_sentiment(text) returns a clean label. Real distribution over the scored sample:

NEUTRAL  59
NEGATIVE  1

Real result after a run¶

scored            = 60
distinct scores   = 54          (fraud_score is varied, not a constant)
min fraud_score   = 0.119
max fraud_score   = 0.784
akko_ai_anomaly   = 60 non null
akko_ai_sentiment = 60 non null

top risks (transaction_id | amount | fraud_score | anomaly | sentiment)
  1000440 | 5840.81 | 0.784 | 1.0 | NEUTRAL
  1000793 | 5746.65 | 0.780 | 1.0 | NEUTRAL
  1000503 | 5656.61 | 0.776 | 1.0 | NEUTRAL

Task 5: refresh_superset¶

Busts the Superset caches for the fraud dashboard so the freshly scored data is visible. The task tolerates a missing dashboard (it logs a warning and returns) so a provisioning hiccup never fails the pipeline.

Governance: why this is the differentiator¶

The AI functions are not open. The pipeline runs under the machine identity svc-airflow, which is granted a dedicated, least-privilege AI tier akko-service:

Function allowlist: only akko_ai_anomaly and akko_ai_sentiment (nothing else).
Two enforcement layers: the OPA policy akko.ai (precheck, package akko.ai) AND the ai-service inline RBAC both must allow the call. Every decision is audited (audit_type: AI_RBAC, ALLOW/DENY, user, role, function).
Quota: a batch identity carries no human daily cap (it is already bounded by data ABAC), so akko-service has an unlimited AI quota while human roles stay cost-capped.
Rate limiting: trusted service-token callers are exempt from the interactive 20/min guard; their calls are serialized (one in-flight LLM call), so they cannot flood the model.

A human analyst calling the same functions is subject to their own role tier and daily quota, and a viewer with no AI grant is denied. Nothing escapes the gate.

How to run it¶

# On the cluster (idempotent: scores the next unscored batch)
kubectl -n akko exec akko-scheduler-0 -c scheduler -- \
  airflow dags trigger akko_banking_fraud_demo -r my_run_id

# Watch the run
kubectl -n akko exec akko-scheduler-0 -c scheduler -- \
  airflow dags state akko_banking_fraud_demo my_run_id

Tune the scored sample without touching code via the env var FRAUD_DEMO_SCORE_SAMPLE_LIMIT (0 means no limit, which scores every unscored row).

Files in the repo¶

File	Role
`airflow/dags/akko_banking_fraud_demo.py`	The DAG (source of truth)
`helm/akko/files/airflow/dags/akko_banking_fraud_demo.py`	Chart mirror (via `sync-dags.sh`)
`helm/akko/files/demos/banking-fraud/seed/`	Seed schema + generator
`helm/examples/values-netcup.yaml`	`akko-service` AI tier + `svc-airflow` mapping
`helm/akko/charts/akko-opa/templates/configmap.yaml`	`akko.ai` precheck matrix