← All recipes

Recipe · ~25 min

UBO graph traversal + risk export

Walk N-hop ownership trees, screen every node, export to Graphviz.

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 not covered) + 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.",
    )

Step 6 — Read the OFAC 50%-Rule / BIS Affiliates determination

Every completed graph carries graph_json.ownership_exposure: per-regime derived-blocking determinations computed with correct aggregation — summed across distinct listed owners, deduped per owner, and NO percentage multiplication through tiers (an entity 50%+ owned by blocked persons is itself a blocked person per OFAC FAQ 401, so its full downstream stake counts). Regimes are never cross-aggregated, and thresholds differ: OFAC and BIS use "50% or more"; EU/UK/UN use "more than 50%".

exposure = graph["graph_json"]["ownership_exposure"]
for regime, r in exposure["regimes"].items():
    print(f"{regime}: {r['determination']}  aggregate={r['aggregate_listed_ownership_pct']}%")
for path in exposure["regimes"]["ofac_50pct"]["evidence_paths"]:
    chain = " -> ".join(f"{h['name']} ({h.get('direct_stake_pct', '')}%)" for h in path)
    print("PATH:", chain)

determination values: directly_listed | blocked | potentially_blocked_unknown_stakes (listed owners with undisclosed stakes — the aggregate shown is a FLOOR) | minority_listed_ownership | no_listed_ownership.

Coverage note. Ownership data comes from company registries (OpenCorporates + national registers). The United States has no functioning domestic beneficial-ownership registry (FinCEN's March 2025 interim rule exempted US companies from CTA BOI reporting), so US ownership chains may be incomplete. Absence of a listed owner in this result is not a clearance determination. The API returns this disclosure verbatim in ownership_exposure.coverage_disclosure — render it wherever you surface UBO results.

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