Audit Trail Analytics: Tracing P2P and O2C Document Flows
Every payment should trace back through an invoice, a goods receipt, and a purchase order. Here is how to reconstruct, validate, and audit those document chains with Python.
Every payment should trace back through an invoice, a goods receipt, and a purchase order. That chain — PO to GR to invoice to payment — is the backbone of procure-to-pay controls, and verifying its completeness is one of the first things an auditor checks in an AP review. In practice, tracing these chains across large datasets is tedious and error-prone when done manually.
VynFi generates complete P2P and O2C document chains with full referential integrity. Every purchase order links to its goods receipt, every goods receipt links to its invoice, and the explicit linkages are stored in a document_references.json file that makes chain reconstruction straightforward. This tutorial covers the complete audit trail analysis workflow from the VynFi document flow notebook.
**DataSynth 3.1.1 update:** Document→JE fraud propagation is now wired correctly — when `fraud.documentFraudRate > 0` and `propagate_to_lines` is on, every JE derived from a fraudulent source document gets `is_fraud_propagated = true` and `fraud_source_document_id = <doc-id>`. Previously the flag stayed `false` because the source-document reference was never set on the JE header. Auditors can now trace a fraudulent payment back through invoice → GR → PO with the propagation flag as the join key. Ready-to-use dataset: VynFi/vynfi-audit-p2p.
Generate P2P and O2C Data
To get the full document_flows directory, request both the o2c and p2p process models in your generation config. The archive will contain eight document type files plus a document_references.json that links them into end-to-end chains.
import osfrom vynfi import VynFiclient = VynFi(api_key=os.environ["VYNFI_API_KEY"])config = { "sector": "retail", "country": "US", "accountingFramework": "us_gaap", "rows": 1000, "companies": 5, "periods": 3, "periodLength": "monthly", "processModels": ["o2c", "p2p"], "exportFormat": "json", "fraudPacks": [], "fraudRate": 0.0,}job = client.jobs.generate_config(config=config)completed = client.jobs.wait(job.id)archive = client.jobs.download_archive(completed.id)# The document_flows/ directory contains:# P2P: purchase_orders, goods_receipts, vendor_invoices, payments# O2C: sales_orders, deliveries, customer_invoices, customer_receipts# Links: document_references.json, lineage_graph.jsonprint("Files in archive:")for f in archive.files(): print(f" {f}")Build the Forward-Link Index
The document_references.json file contains explicit source-to-target references between documents. Building a forward-link index from this file is the foundation of all chain reconstruction work. Once you have the index, tracing a chain forward from any document is a simple breadth-first search.
from collections import defaultdictimport pandas as pddf_refs = pd.DataFrame(archive.json("document_flows/document_references.json"))forward_links = defaultdict(list)backward_links = defaultdict(list)for _, ref in df_refs.iterrows(): src_type = ref.get("source_doc_type") or ref.get("source_type") or ref.get("from_type") src_id = ref.get("source_doc_id") or ref.get("source_id") or ref.get("from_id") tgt_type = ref.get("target_doc_type") or ref.get("target_type") or ref.get("to_type") tgt_id = ref.get("target_doc_id") or ref.get("target_id") or ref.get("to_id") if all([src_type, src_id, tgt_type, tgt_id]): forward_links[(src_type, src_id)].append((tgt_type, tgt_id)) backward_links[(tgt_type, tgt_id)].append((src_type, src_id))print(f"Forward links built: {len(forward_links):,} source documents")def trace_chain_forward(doc_type, doc_id, forward_links, max_depth=10): """BFS: trace a document chain forward through references.""" chain, visited, queue = [(doc_type, doc_id)], {(doc_type, doc_id)}, [(doc_type, doc_id)] while queue and len(chain) < max_depth: current = queue.pop(0) for target in forward_links.get(current, []): if target not in visited: visited.add(target) chain.append(target) queue.append(target) return chainP2P Chain Reconstruction
Starting from each purchase order, trace the chain forward through goods receipts, vendor invoices, and payments. The chain length distribution tells you immediately how far most orders progress through the cycle. A PO that reaches length 4 has completed the full P2P cycle. Shorter chains indicate breaks — open commitments, uninvoiced receipts, or unpaid invoices.
# Reconstruct P2P chains starting from each POp2p_chains = []for _, po in dfs["purchase_orders"].iterrows(): po_id = po[id_columns["purchase_orders"]] chain = trace_chain_forward("purchase_order", po_id, forward_links) p2p_chains.append(chain)# Chain length distributionchain_lengths = pd.Series([len(c) for c in p2p_chains])print("P2P chain length distribution:")print(chain_lengths.value_counts().sort_index())# Drop-off analysisp2p_stages = ["purchase_order", "goods_receipt", "vendor_invoice", "payment"]stage_labels = ["PO", "GR", "Invoice", "Payment"]stage_counts = defaultdict(int)for chain in p2p_chains: for dtype, _ in chain: stage_counts[dtype] += 1print("P2P Drop-off Analysis:")for i in range(len(p2p_stages) - 1): current = stage_counts.get(p2p_stages[i], 0) nxt = stage_counts.get(p2p_stages[i + 1], 0) rate = nxt / current * 100 if current > 0 else 0 print(f" {stage_labels[i]:>10s} -> {stage_labels[i+1]:<10s}: " f"{current:>5,} -> {nxt:>5,} ({rate:.1f}% flow-through)")Three-Way Matching
Three-way matching compares three documents that should agree: the purchase order (what was ordered), the goods receipt (what was received), and the vendor invoice (what is being charged). Discrepancies between these three documents are control failures — short shipments, price inflation, unauthorized substitutions, or duplicate billing.
def find_amount(record): for field in ["total_net_amount", "total_gross_amount", "total_amount", "amount", "total", "net_amount", "value"]: if field in record: try: return float(record[field]) except (ValueError, TypeError): continue return NoneTOLERANCE = 0.01 # 1% tolerance for floating-point differencesmatch_results = []for po_id, po_rec in po_lookup.items(): result = {"po_id": po_id, "po_amount": find_amount(po_rec), "has_gr": False, "gr_amount": None, "has_invoice": False, "invoice_amount": None, "flags": []} linked_grs = [(t, i) for t, i in forward_links.get(("purchase_order", po_id), []) if "goods_receipt" in str(t)] if linked_grs: result["has_gr"] = True gr_id = linked_grs[0][1] if gr_id in gr_lookup: result["gr_amount"] = find_amount(gr_lookup[gr_id]) linked_invs = [(t, i) for t, i in forward_links.get(("goods_receipt", gr_id), []) if "invoice" in str(t)] if linked_invs: result["has_invoice"] = True inv_id = linked_invs[0][1] if inv_id in vi_lookup: result["invoice_amount"] = find_amount(vi_lookup[inv_id]) else: result["flags"].append("NO_GR") # Check amount agreement if result["po_amount"] and result["gr_amount"]: if abs(result["po_amount"] - result["gr_amount"]) / max(result["po_amount"], 1) > TOLERANCE: result["flags"].append("AMOUNT_MISMATCH_PO_GR") match_results.append(result)df_match = pd.DataFrame(match_results)full_match = ((df_match["has_gr"]) & (df_match["has_invoice"])).sum()print(f"Full three-way match: {full_match}/{len(df_match)} ({full_match/len(df_match):.1%})")Add fraudPacks: ['vendor_kickback'] to your generation config to inject inflated invoices, shell company payments, and split POs designed to evade approval thresholds. The three-way matching logic will then flag real mismatches rather than just demonstrating the mechanics on clean data.
Gap Analysis
A gap is a document that should have a successor but does not. Gaps reveal process bottlenecks and control failures. For each gap type, calculating the age in days since the source document was created identifies stale items that require follow-up.
def analyze_gaps(source_type, source_lookup, expected_target_type, forward_links): """Find source documents with no forward link to the expected target type.""" gaps = [] for doc_id, doc_rec in source_lookup.items(): targets = forward_links.get((source_type, doc_id), []) has_target = any(expected_target_type in str(tgt_type) for tgt_type, _ in targets) if not has_target: doc_date = find_date_field(doc_rec) amount = find_amount(doc_rec) age_days = (pd.Timestamp.now() - doc_date).days if doc_date else None gaps.append({"doc_id": doc_id, "doc_date": doc_date, "amount": amount, "age_days": age_days}) return gapsgap_types = { "POs without GRs": ("purchase_order", po_lookup, "goods_receipt"), "GRs without Invoices": ("goods_receipt", gr_lookup, "invoice"), "Invoices without Payments": ("vendor_invoice", vi_lookup, "payment"), "SOs without Deliveries": ("sales_order", so_lookup, "delivery"), "Deliveries without Invoices": ("delivery", del_lookup, "invoice"),}print("Gap Analysis Summary")print("=" * 55)for gap_name, (src_type, src_lookup, tgt_type) in gap_types.items(): gaps = analyze_gaps(src_type, src_lookup, tgt_type, forward_links) ages = [g["age_days"] for g in gaps if g["age_days"] is not None] avg_age = sum(ages) / len(ages) if ages else 0 total_at_risk = sum(g["amount"] or 0 for g in gaps) print(f" {gap_name:<30s} {len(gaps):>4} gaps avg {avg_age:.0f} days ${total_at_risk:>12,.0f}")Lead Time Analysis
Visualizing how long each stage transition takes reveals where the process slows down. For P2P, the three transitions are PO-to-GR (procurement lead time), GR-to-invoice (vendor billing lag), and invoice-to-payment (AP processing time). Outliers in any of these distributions warrant investigation.
In a typical retail dataset, GR-to-invoice lag tends to be the most variable — some vendors invoice immediately, others wait weeks. Invoice-to-payment is tightly controlled by payment terms, so its distribution is narrower. PO-to-GR duration reflects supplier lead times and varies significantly by product category.
Audit Findings Summary
Combining the three-way match results, gap analysis, and lead time outliers produces an audit findings summary that maps directly to assertions. POs without GRs map to the completeness assertion for goods received. Amount mismatches map to the accuracy assertion for invoice processing. Long unpaid aging maps to the cutoff assertion for period-end liabilities.
- Three-way match rate: the percentage of POs with a complete GR-invoice chain is the primary completeness metric for AP controls
- Amount mismatch flags: any invoice that differs from its PO by more than your tolerance threshold is a potential cutoff or accuracy exception
- Gap aging by bucket (0-30, 31-60, 61-90, 90+ days): items in the 90+ bucket represent stale open items that likely require accrual or write-off
- Vendor-level gap rates: suppliers with consistently high gap rates may have systemic issues with invoicing accuracy or delivery confirmation
The archive also includes a relationships/cross_process_links.json file that connects P2P and O2C flows through shared inventory. This enables completeness testing across the full procure-to-sell cycle — verifying that purchased inventory eventually flows to revenue.
The full notebook is available at 04_document_flow_audit_trail.ipynb in the VynFi Python SDK repository. It includes master data joins, document timeline visualization, and lead time analysis across all process stages.