Getting started: flag a transaction and prove it (AML)#

In this tutorial we will build a tiny anti money laundering (AML) check from scratch. We will model Customer and Transaction entities, record a couple of customers and a few transfers, write one rule that flags any transfer at or above a reporting threshold of 10000, evaluate it, and then read the self explaining proof of the flag condition by condition.

By the end we will have a working graph that does not just say “this transaction is reportable”, it shows us exactly why, atom by atom, in plain text.

Along the way we will meet the four tools you reach for most often in factgraph: the schema classes (Entity, Identity, Field), the write surface (fg.entities.create, fg.fields.set), the rule DSL (vars, build_application_rule), and the evaluation surface (fg.eval.evaluate, row.explain().narrate()).

Prerequisites#

We need Python 3.10 or newer (the factgraph package declares requires-python = ">=3.10").

Install factgraph into a fresh virtual environment:

python -m venv .venv
source .venv/bin/activate
pip install factgraph

We are using factgraph v0.2.0. Everything in this tutorial runs on the native evaluation engine, which is built in and needs no extra dependencies.

Create a file called aml.py and work through the steps below. We will add to the same file as we go and run it after each step.

1. Model the customers and transactions#

First, we declare our schema as Python classes. We subclass Entity and annotate each member as either an Identity (part of the entity’s stable identity) or a Field (an ordinary attribute). Every entity must declare at least one Identity.

We also give each class a Meta.repr template and give the fields we care about a repr= template. These templates control how the proof reads later, and the two kinds of template accept different placeholders. A field or identity repr= template may use %CLS (the class name), %ENT (the entity), and %FLD (the field value). A Meta.repr template may use %CLS plus the entity’s identity field placeholders (here %customer_id and %txn_id); it cannot reference %ENT or %FLD. These templates do not change behaviour, only how the explanation narrates.

Put this at the top of aml.py:

from factgraph.sdk import FactGraph, Entity, Identity, Field
from factgraph.sdk.dsl import vars, build_application_rule


class Customer(Entity):
    class Meta:
        repr = "Customer %customer_id"

    customer_id: str = Identity()
    name: str = Field(repr="%ENT name %FLD")


class Transaction(Entity):
    class Meta:
        repr = "Transaction %txn_id"

    txn_id: str = Identity()
    sender: str = Field(repr="%ENT sent by %FLD")
    amount: int = Field(repr="%ENT amount %FLD")

Notice that the field cardinality is inferred from the annotation: str and int are scalar annotations, so name, sender, and amount are all single value fields.

Now build a graph from these classes and check that it loaded. Add:

fg = FactGraph.create(schema_classes=[Customer, Transaction])
print("capabilities:", fg.meta.capabilities())

Run it:

python aml.py

The output should look something like:

capabilities: {'value_kinds': frozenset({'scalar', 'entity_ref'}), 'scalar_tags': frozenset({'string', 'int', 'float64', 'bool', 'bytes', 'time', 'uuid'}), 'cardinalities': frozenset({'single', 'multi'})}

(The order of the items inside each frozenset may differ; sets are unordered.)

We now have an empty graph that knows about Customer and Transaction. The int tag and the single cardinality in the capabilities are the building blocks our schema just used.

2. Create the customers#

Next, we create two customers. fg.entities.create takes the entity class plus the identity values as keyword arguments, and returns a stable reference string (an e_ref) for that entity.

Add this below the fg = FactGraph.create(...) line:

alice_ref = fg.entities.create(Customer, customer_id="c-1")
bob_ref = fg.entities.create(Customer, customer_id="c-2")
print("alice_ref:", alice_ref)

Run the file again. The output should look something like:

alice_ref: idref_v1:Customer:...

Notice that the returned e_ref starts with idref_v1:Customer:. It is deterministic: creating the same customer identity again returns the same string. We will hold on to alice_ref and bob_ref to attach fields to these customers.

Let us confirm the customers are really there. Add:

print("alice exists:", fg.entities.exists(Customer, customer_id="c-1"))
print("charlie exists:", fg.entities.exists(Customer, customer_id="c-3"))

The output should look something like:

alice exists: True
charlie exists: False

fg.entities.exists reports True only for an entity whose identity is active in the graph. We never created c-3, so it is False.

3. Record the transactions and their fields#

Now we create three transactions and set their fields. We create each transaction the same way we created the customers, then write the sender and amount cells with fg.fields.set. fg.fields.set takes the field descriptor (for example Transaction.amount), the entity reference, and the value.

Add:

# A transfer below the threshold.
t1 = fg.entities.create(Transaction, txn_id="t-1")
fg.fields.set(Transaction.sender, t1, "c-1")
fg.fields.set(Transaction.amount, t1, 4200)

# A transfer exactly at the threshold.
t2 = fg.entities.create(Transaction, txn_id="t-2")
fg.fields.set(Transaction.sender, t2, "c-1")
fg.fields.set(Transaction.amount, t2, 10000)

# A transfer well above the threshold.
t3 = fg.entities.create(Transaction, txn_id="t-3")
fg.fields.set(Transaction.sender, t3, "c-2")
fg.fields.set(Transaction.amount, t3, 25000)

print("t-3 amount:", fg.fields.get(Transaction.amount, t3))

Run the file. The output should look something like:

t-3 amount: 25000

Notice that fg.fields.get reads the current value of a cell straight back. We have now written nine facts in total (three transactions, each with a sender and an amount), and they are all readable.

4. Write the reporting rule#

Now the interesting part. We write one rule: a transaction is reportable if it exists and its amount is at or above 10000.

We use vars to introduce two logic variables, one for the transaction and one for its amount, then build_application_rule to assemble the rule body. The when= list is the conjunction of conditions (every condition must hold). We refer to a transaction’s field with Transaction(t).amount, and we compare it to a literal with the ordinary Python operators.

Add:

with vars("t", "amt") as (t, amt):
    reportable = build_application_rule(
        id="reportable_transfer",
        when=[
            Transaction(t).amount == amt,
            amt >= 10000,
        ],
        ports={"transaction": t, "amount": amt},
        repr="transaction %transaction is reportable (amount %amount)",
    )

print("rule id:", reportable.id)

Run the file. The output should look something like:

rule id: reportable_transfer

A few things to notice. Transaction(t).amount == amt binds the amount of transaction t to the variable amt, and it also implies that the transaction exists. The amt >= 10000 line is the threshold condition. The ports= mapping names the variables we want to read back on each result, and repr= is the sentence the proof will use as its conclusion.

5. Evaluate the rule#

Now we run the rule against the graph. We pass the rule to fg.eval.evaluate, and because we want a concrete, fully bound conclusion we pass the same rule as the head=. We pin the engine to "native".

Add:

result = fg.eval.evaluate(reportable, head=reportable, engine="native")

print("reportable count:", result.count())
for row in result:
    print(" -", row.bindings["transaction"]["value"], "amount", row.bindings["amount"]["value"])

Run the file. The output should look something like:

reportable count: 2
 - idref_v1:Transaction:... amount 10000
 - idref_v1:Transaction:... amount 25000

Notice that exactly two rows came back: t-2 (exactly 10000) and t-3 (25000). The 4200 transfer t-1 is correctly absent, because 4200 is below the threshold. The >= comparison includes the boundary value 10000, just as a reporting threshold should.

Notice also how we read a binding. Each value in row.bindings is a small wrapper, not the bare value. An entity reference looks like {"kind": "entity_ref", "value": "idref_v1:Transaction:..."} and a scalar looks like {"kind": "literal", "tag": "int", "value": 25000}. The raw value lives under the "value" key, which is why we read row.bindings["amount"]["value"] rather than row.bindings["amount"].

We can also ask the result direct questions. Add:

print("any reportable?", result.exists())
print("first row:", result.first().bindings["transaction"]["value"])

The output should look something like:

any reportable? True
first row: idref_v1:Transaction:...

result.exists() tells us whether anything matched at all, and result.first() hands us the first row. These are the everyday shortcuts on an evaluation result.

6. Read the proof, condition by condition#

A count of matches is useful, but the whole point of factgraph is that every result carries its own proof. Each row can produce an Explanation, and Explanation.narrate() renders that proof as human readable lines.

Let us narrate the proof for the 25000 transfer. Add:

big_row = next(row for row in result if row.bindings["amount"]["value"] == 25000)
explanation = big_row.explain()

print()
for line in explanation.narrate():
    print(line)

Run the file one last time. The narrated proof should look something like:

Conclusion ── transaction Transaction t-3 is reportable (amount 25000)
  reportable_transfer [head] ── "transaction Transaction t-3 is reportable (amount 25000)"  [holds]
       ✓ Transaction t-3 amount 25000  [a0]  holds
       ✓ 25000 >= 10000  [a1]  holds

Read it top to bottom. The Conclusion line is our rule’s repr= sentence, filled in with this row’s values. Below it, each condition of the rule is checked off: the ✓ marks a condition that holds. The first ✓ confirms the transaction’s amount really is 25000 (rendered with the %ENT amount %FLD template), and the second ✓ confirms 25000 >= 10000, the threshold check itself.

This is the audit trail: not “trust me, it is reportable”, but a line for every fact and every comparison that led to the flag.

For contrast, run the same narration logic against a row that holds at the boundary, the 10000 transfer:

edge_row = next(row for row in result if row.bindings["amount"]["value"] == 10000)
print()
for line in edge_row.explain().narrate():
    print(line)

The proof should look something like:

Conclusion ── transaction Transaction t-2 is reportable (amount 10000)
  reportable_transfer [head] ── "transaction Transaction t-2 is reportable (amount 10000)"  [holds]
       ✓ Transaction t-2 amount 10000  [a0]  holds
       ✓ 10000 >= 10000  [a1]  holds

Notice that 10000 >= 10000 holds: the boundary is included, and the proof says so explicitly.

What we built#

We have built a working AML check that flags every transfer at or above a 10000 reporting threshold, and, for each flagged transfer, produces a condition by condition proof of why it was flagged. We modelled entities with Entity/Identity/Field, recorded them with fg.entities.create and fg.fields.set, expressed the policy as one rule with vars and build_application_rule, evaluated it with fg.eval.evaluate, and read the evidence with row.explain().narrate().

From here:

Backward Tutorials How-to guides Forward
comments powered by Disqus
factgraph - © CC BY 4.0