Graph-Based Transactional-Flow-Checker

The problem

Verifying financial transactions is difficult because it is hard to see if a series of steps actually connects logically. It is easy to miss gaps or errors when looking at a long list of numbers.

What it does

It maps out financial steps as a connected map to see if each transaction flows correctly into the next. It checks if every piece of data can actually reach its destination.

Why it matters

It treats financial data as a connected path rather than a simple list, making it easier to spot broken links in a sequence.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 GraphBasedTransactionalFlowChecker.py
Running example usage...
Test Case 1 (Valid): True, Message: 
Test Case 2 (Cycle): False, Message: No root transaction found
Test Case 3 (Disconnected): True, Message:

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 97 lines, one file, standard library only.

# Graph-Based Transactional-Flow-Checker
import json
from collections import deque

def build_graph(transactions):
    adj_list = {}
    in_degree = {}
    for tx in transactions:
        node_id = tx['id']
        adj_list[node_id] = []
        in_degree[node_id] = 0
    for tx in transactions:
        node_id = tx['id']
        depends_on = tx.get('depends_on', [])
        for dep in depends_on:
            if dep not in adj_list:
                adj_list[dep] = []
                in_degree[dep] = 0
            adj_list[dep].append(node_id)
            in_degree[node_id] = in_degree.get(node_id, 0) + 1
    return adj_list, in_degree

def has_cycle(adj_list, in_degree):
    queue = deque([node for node in in_degree if in_degree[node] == 0])
    visited = set()
    temp_in_degree = in_degree.copy()
    while queue:
        node = queue.popleft()
        visited.add(node)
        for neighbor in adj_list.get(node, []):
            temp_in_degree[neighbor] -= 1
            if temp_in_degree[neighbor] == 0:
                queue.append(neighbor)
    return len(visited) != len(adj_list)

def is_reachable(adj_list, roots):
    visited = set()
    queue = deque(roots)
    while queue:
        node = queue.popleft()
        if node not in visited:
            visited.add(node)
            for neighbor in adj_list.get(node, []):
                queue.append(neighbor)
    return len(visited) == len(adj_list)

def check_integrity(transactions):
    adj_list, in_degree = build_graph(transactions)
    roots = [node for node in in_degree if in_degree[node] == 0]
    if not roots:
        return False, "No root transaction found"
    if has_cycle(adj_list, in_degree):
        return False, "Cycle detected in transaction dependencies"
    all_nodes = set(in_degree.keys())
    reachable = set()
    for root in roots:
        queue = deque([root])
        visited = set()
        while queue:
            node = queue.popleft()
            if node not in visited:
                visited.add(node)
                for neighbor in adj_list.get(node, []):
                    queue.append(neighbor)
        reachable.update(visited)
    if reachable != all_nodes:
        return False, "Not all transactions are reachable from roots"
    return True, ""

def example_usage():
    # Valid chain test
    valid, message = check_integrity([
        {'id': '1', 'depends_on': []},
        {'id': '2', 'depends_on': ['1']},
        {'id': '3', 'depends_on': ['2']},
    ])
    print(f"Test Case 1 (Valid): {valid}, Message: {message}")

    # Cycle test
    valid, message = check_integrity([
        {'id': '1', 'depends_on': ['2']},
        {'id': '2', 'depends_on': ['1']},
    ])
    print(f"Test Case 2 (Cycle): {valid}, Message: {message}")

    # Disconnected nodes test
    valid, message = check_integrity([
        {'id': '1', 'depends_on': []},
        {'id': '2', 'depends_on': []},
        {'id': '3', 'depends_on': ['1']},
        {'id': '4', 'depends_on': ['2']},
    ])
    print(f"Test Case 3 (Disconnected): {valid}, Message: {message}")

if __name__ == "__main__":
    print("Running example usage...")
    example_usage()