Recipe 04 — UBO graph traversal + risk export

25 minutes. Result: a Python script that walks a counterparty's ownership tree N hops, screens every node, and exports the result as JSON + a Graphviz DOT file for your reports.

What this recipe does

For risk-based KYC, you need to look past the immediate counterparty to the ultimate beneficial owner. OilFlow's /api/v1/ubo/screen traverses up to N hops, screens every entity in the chain against the 8 sanctions lists + PEP + cluster blocklist, and runs shell-company detection on each.

Step 1 — Kick off the traversal

from oilflow import Client
import json
import time

client = Client()

initial = client.ubo.screen(
    root_entity_name="Acme Holdings BVI",
    root_jurisdiction="BVI",
    max_hops=3,
)
graph_id = initial["graph_id"]
print(f"queued: {graph_id}")

Step 2 — Poll until ready

Traversal is async; small graphs return in seconds, large ones can take 60–90s.

def wait_for_graph(client, graph_id, max_wait_s=120):
    started = time.time()
    while time.time() - started < max_wait_s:
        graph = client.ubo.graph(graph_id)
        if graph["status"] == "ready":
            return graph
        if graph["status"] == "failed":
            raise RuntimeError("UBO traversal failed")
        time.sleep(3)
    raise TimeoutError("UBO traversal exceeded max_wait_s")

graph = wait_for_graph(client, graph_id)

Or subscribe to the ubo.graph_ready webhook event for production.

Step 3 — Inspect the result

print(f"aggregate_risk: {graph['aggregate_risk']}")
print(f"nodes: {len(graph['nodes'])}")
print(f"edges: {len(graph['edges'])}")

# Flag any sanctions / PEP hits in the chain
hits = [n for n in graph["nodes"] if n["sanctions_match"] or n["pep_match"]]
for n in hits:
    print(f"HIT  {n['name']}  jurisdiction={n['jurisdiction']}  pep={n['pep_match']}  sanctions={n['sanctions_match']}")

# Flag shell-company verdicts
shells = [n for n in graph["nodes"] if (n.get("shell_score") or 0) > 0.7]
for n in shells:
    print(f"SHELL {n['name']}  shell_score={n['shell_score']:.2f}")

Step 4 — Export to Graphviz for your report

def to_dot(graph):
    lines = ["digraph UBO {", '  rankdir="LR";']
    for n in graph["nodes"]:
        color = "red" if n["sanctions_match"] else "orange" if (n.get("shell_score") or 0) > 0.7 else "lightgray"
        label = n["name"].replace('"', '\\"')
        lines.append(f'  "{n["id"]}" [label="{label}", style=filled, fillcolor={color}];')
    for e in graph["edges"]:
        pct = e.get("ownership_pct")
        lines.append(f'  "{e["from"]}" -> "{e["to"]}" [label="{pct or ""}%"];')
    lines.append("}")
    return "\n".join(lines)

with open("ubo.dot", "w") as f:
    f.write(to_dot(graph))

# Render:
#   dot -Tpng ubo.dot -o ubo.png

For a richer JS visualization, the same graph["nodes"] + graph["edges"] plug straight into `vis.js Network`.

Step 5 — Persist + alert

If aggregate_risk is high or critical, fire your own escalation:

if graph["aggregate_risk"] in ("high", "critical"):
    notify_compliance_lead(
        subject=f"UBO chain blocker on {root}",
        body=f"Graph {graph_id}: {len(hits)} sanctions/PEP hits, {len(shells)} shells.",
    )

Common gotchas

• `max_hops` capped at 5. Most real ownership chains are 1–3 hops.
• Ambiguous root names. Pass root_jurisdiction to narrow ("Acme Holdings BVI" + "BVI" resolves correctly; "Acme" alone returns disambiguation candidates).
• Cost. UBO is heavier than KYC (each hop screens N entities). Rate-limited to 10 req/min on default plan.

Next steps

• Recipe 05: LC discrepancy validation
• Recipe 01: Pipe results into Slack