NOWNESS · invention
✓ VALIDATED — its own code really ran here
Path-Dependency Influence Ranking
Invented and built autonomously on 2026-08-02 12:40
The problem
It is difficult to see which specific choices in a complex chain of events actually lead to a final result. We often struggle to distinguish between minor details and the key decisions that shape an outcome.
What it does
It maps out a series of choices and ranks them based on how much they influence the final result. It identifies the specific steps that carry the most weight.
Why it matters
It allows you to see exactly which parts of a process are the most important drivers of the final outcome.
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 path_dependency_influence.py [Generated DAG: 60 edges, 25 nodes, seed=42] ==================================================================================================== Path-Dependency Influence Ranking ==================================================================================================== Target : 'n24' Nodes : 25 Edges : 60 Sources: n13, n1, n3, n5, n0 # Score Edge Paths Depth Reach Div BN W ----------------------------------------------------------------------------------------------- 1 0.8826 n3→n8 [review] 2 2 1.1281 0.5774 1.0000 2.210 2 0.8059 n1→n10 [approve] 4 1 0.6800 0.4083 1.0000 0.940 3 0.8046 n1→n12 [split] 4 1 0.5975 0.4083 1.0000 1.110 4 0.7848 n1→n2 [approve] 4 3 0.1151 0.4083 0.7071 2.420 5 0.7047 n8→n18 [reject] 1 1 0.5600 1.0000 0.7071 2.370 6 0.6981 n12→n24 [merge] 1 0 1.0000 0.5000 0.5000 2.390 7 0.6747 n10→n24 [merge] 1 0 1.000
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 — 381 lines, one file, standard library only.
#!/usr/bin/env python3
"""
Path-Dependency Influence Ranking
Recursive graph-traversal path scoring that identifies which decision branches
contribute most to a final outcome node.
Algorithm (Recursive Graph-Traversal Path Scoring):
1. Set the target/outcome node.
2. For every node, recursively count how many forward paths reach the target
(path_count_to_target).
3. For every node, recursively propagate a "reachability signal" forward
toward the target, applying decay per hop.
4. Score each directed edge (src→dst) as a weighted composite of:
- Path multiplicity (35%): share of total paths passing through src
- Reachability (25%): how well dst connects to target
- Divergence penalty (15%): high fan-out nodes split influence
- Bottleneck factor (15%): edges into high in-degree nodes
- Edge weight (10%): user-defined importance
5. Rank edges by composite score descending.
Usage:
python path_dependency_influence.py
python path_dependency_influence.py --seed 42 --nodes 30 --edges 80
python path_dependency_influence.py --graph-file graph.json --target outcome
python path_dependency_influence.py --export ranked.json --top 10
"""
import argparse
import json
import math
import random
import sys
from collections import defaultdict, deque
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
# ---------------------------------------------------------------------------
# Data types
# ---------------------------------------------------------------------------
@dataclass
class Edge:
src: str
dst: str
weight: float = 1.0
label: str = ""
@dataclass
class RankedBranch:
src: str
dst: str
label: str
score: float
path_count: int
depth: int
reachability: float
normalized_path_frac: float
divergence_penalty: float
bottleneck_factor: float
weight: float
def to_dict(self) -> dict:
return {
"src": self.src,
"dst": self.dst,
"label": self.label,
"score": round(self.score, 5),
"path_count": self.path_count,
"depth": self.depth,
"reachability": round(self.reachability, 5),
"normalized_path_frac": round(self.normalized_path_frac, 5),
"divergence_penalty": round(self.divergence_penalty, 5),
"bottleneck_factor": round(self.bottleneck_factor, 5),
"edge_weight": self.weight,
}
class NodeInfo:
"""Lightweight per-node state."""
__slots__ = ("nid", "is_target", "reachability_score", "path_count_to_target")
def __init__(self, nid: str):
self.nid: str = nid
self.is_target: bool = False
self.reachability_score: float = 0.0
self.path_count_to_target: int = 0
# ---------------------------------------------------------------------------
# Core ranker
# ---------------------------------------------------------------------------
class PathDependencyInfluenceRanker:
def __init__(self, decay_factor: float = 0.85, max_depth: int = 200):
self.decay_factor = decay_factor
self.max_depth = max_depth
self.nodes: Dict[str, NodeInfo] = {}
self.incoming: Dict[str, List[Edge]] = defaultdict(list)
self.outgoing: Dict[str, List[Edge]] = defaultdict(list)
# Memoization tables (cleared per compute run)
self._path_memo: Dict[Tuple[str, int], int] = {}
self._reach_memo: Dict[Tuple[str, int], float] = {}
self._depth_memo: Dict[Tuple[str, str], int] = {}
# ---- construction -------------------------------------------------
def add_edge(self, src: str, dst: str, weight: float = 1.0, label: str = "") -> None:
e = Edge(src=src, dst=dst, weight=weight, label=label)
self.incoming[dst].append(e)
self.outgoing[src].append(e)
for nid in (src, dst):
if nid not in self.nodes:
self.nodes[nid] = NodeInfo(nid)
# ---- recursive metrics -------------------------------------------
def _path_count(self, node_id: str, depth: int, visited: set) -> int:
"""Return number of forward paths from node_id to the target."""
if depth > self.max_depth:
return 0
if node_id in visited:
return 0 # cycle guard
key = (node_id, depth)
if key in self._path_memo:
return self._path_memo[key]
node = self.nodes.get(node_id)
if node is None:
return 0
if node.is_target:
return 1
visited.add(node_id)
total = 0
for e in self.outgoing[node_id]:
total += self._path_count(e.dst, depth + 1, visited)
visited.discard(node_id)
self._path_memo[key] = total
return total
def _reachability(self, node_id: str, depth: int, visited: set) -> float:
"""Recursive reachability signal from node_id towards target."""
if depth > self.max_depth:
return 0.0
if node_id in visited:
return 0.0 # cycle guard
key = (node_id, depth)
if key in self._reach_memo:
return self._reach_memo[key]
node = self.nodes.get(node_id)
if node is None:
return 0.0
if node.is_target:
return 1.0
visited.add(node_id)
out_edges = self.outgoing[node_id]
if not out_edges:
visited.discard(node_id)
self._reach_memo[key] = 0.0
return 0.0
total = 0.0
for e in out_edges:
child = self._reachability(e.dst, depth + 1, visited)
total += e.weight * child
avg = total * (self.decay_factor ** depth) / len(out_edges)
visited.discard(node_id)
self._reach_memo[key] = avg
return avg
def _bfs_depth(self, src: str, target: str) -> int:
"""Shortest-hop distance from src to target. -1 if unreachable."""
cache_key = (src, target)
if cache_key in self._depth_memo:
return self._depth_memo[cache_key]
if src == target:
return 0
q = deque([(src, 0)])
seen = {src}
while q:
cur, d = q.popleft()
for e in self.outgoing.get(cur, []):
if e.dst == target:
self._depth_memo[cache_key] = d + 1
return d + 1
if e.dst not in seen:
seen.add(e.dst)
q.append((e.dst, d + 1))
self._depth_memo[cache_key] = -1
return -1
# ---- compute ------------------------------------------------------
def compute(self, target_node: str) -> List[RankedBranch]:
if target_node not in self.nodes:
raise ValueError(f"Target node '{target_node}' not in the graph.")
# Reset memos for fresh run
self._path_memo.clear()
self._reach_memo.clear()
self._depth_memo.clear()
# Tag target
self.nodes[target_node].is_target = True
# ---- Phase 1: compute per-node metrics recursively ------------
for nid in self.nodes:
node = self.nodes[nid]
node.path_count_to_target = self._path_count(nid, 0, set())
node.reachability_score = self._reachability(nid, 0, set())
# ---- Phase 2: normalisers ------------------------------------
max_paths = max((n.path_count_to_target for n in self.nodes.values()), default=1)
max_reach = max((n.reachability_score for n in self.nodes.values()), default=1e-9)
if max_reach < 1e-9:
max_reach = 1.0 # prevent division by zero on degenerate graphs
# ---- Phase 3: score every directed edge -----------------------
scored: List[Tuple[float, RankedBranch]] = []
for src, edges in self.outgoing.items():
src_node = self.nodes[src]
src_fan_out = len(edges)
for e in edges:
dst_node = self.nodes[e.dst]
dst_in_deg = len(self.incoming[e.dst])
# Component 1: path fraction through source
path_frac = src_node.path_count_to_target / max_paths
# Component 2: reachability ratio of destination
reach_ratio = dst_node.reachability_score / max_reach
# Component 3: divergence penalty (sqrt so edges aren't crushed)
div_penalty = 1.0 / math.sqrt(max(src_fan_out, 1))
# Component 4: bottleneck factor
bottleneck = 1.0 / math.sqrt(max(dst_in_deg, 1))
# Composite score — weights tuned for path-dependency ranking
score = (
0.35 * path_frac
+ 0.25 * reach_ratio
+ 0.15 * div_penalty
+ 0.15 * bottleneck
+ 0.10 * e.weight
)
# Shortest-hop depth from dst to target
edge_depth = self._bfs_depth(e.dst, target_node)
branch = RankedBranch(
src=e.src,
dst=e.dst,
label=e.label,
score=round(score, 5),
path_count=src_node.path_count_to_target,
depth=edge_depth,
reachability=round(dst_node.reachability_score, 5),
normalized_path_frac=round(path_frac, 5),
divergence_penalty=round(div_penalty, 5),
bottleneck_factor=round(bottleneck, 5),
weight=round(e.weight, 3),
)
scored.append((score, branch))
scored.sort(key=lambda x: x[0], reverse=True)
return [b for _, b in scored]
# ---- helpers -----------------------------------------------------
def identify_sources(self) -> List[str]:
return [nid for nid in self.nodes if not self.incoming[nid]]
def print_report(self, ranked: List[RankedBranch], target: str,
top_n: int = 20) -> None:
n_edges = sum(len(v) for v in self.outgoing.values())
print(f"{'=' * 100}")
print(f" Path-Dependency Influence Ranking")
print(f"{'=' * 100}")
print(f" Target : '{target}'")
print(f" Nodes : {len(self.nodes)} Edges : {n_edges}")
print(f" Sources: {', '.join(self.identify_sources())}")
print()
hdr = (
f"{'#':<4} {'Score':>8} {'Edge':<30} "
f"{'Paths':>8} {'Depth':>5} {'Reach':>8} "
f"{'Div':>6} {'BN':>6} {'W':>5}"
)
print(hdr)
print("-" * len(hdr))
for i, r in enumerate(ranked[:top_n], 1):
edge_str = f"{r.src}→{r.dst}"
if r.label:
edge_str += f" [{r.label}]"
print(
f"{i:<4} {r.score:>8.4f} {edge_str:<30} "
f"{r.path_count:>8} {r.depth:>5} {r.reachability:>8.4f} "
f"{r.divergence_penalty:>6.4f} {r.bottleneck_factor:>6.4f} "
f"{r.weight:>5.3f}"
)
# ---------------------------------------------------------------------------
# Graph generators & loaders
# ---------------------------------------------------------------------------
def generate_dag(num_nodes: int, num_edges: int, seed: int = 42) -> dict:
random.seed(seed)
names = [f"n{i}" for i in range(num_nodes)]
possible = [(names[i], names[j]) for i in range(num_nodes) for j in range(i + 1, num_nodes)]
random.shuffle(possible)
selected = possible[:min(num_edges, len(possible))]
labels = ["approve", "reject", "defer", "escalate", "review", "merge", "split"]
edges = []
for src, dst in selected:
edges.append({
"src": src,
"dst": dst,
"weight": round(random.uniform(0.5, 2.5), 2),
"label": random.choice(labels),
})
return {"edges": edges, "target": names[-1], "nodes": names}
def load_graph(data: dict) -> PathDependencyInfluenceRanker:
ranker = PathDependencyInfluenceRanker()
for e in data.get("edges", []):
ranker.add_edge(e["src"], e["dst"], e.get("weight", 1.0), e.get("label", ""))
return ranker
# ---------------------------------------------------------------------------
# Main entry
# ---------------------------------------------------------------------------
def main() -> int:
parser = argparse.ArgumentParser(
description="Path-Dependency Influence Ranking — recursive graph-traversal scoring"
)
parser.add_argument("--nodes", type=int, default=25)
parser.add_argument("--edges", type=int, default=60)
parser.add_argument("--seed", type=int, default=42)
parser.add_argument("--target", type=str, default=None)
parser.add_argument("--graph-file", type=str, default=None)
parser.add_argument("--top", type=int, default=20)
parser.add_argument("--decay", type=float, default=0.85)
parser.add_argument("--max-depth", type=int, default=200)
parser.add_argument("--export", type=str, default=None)
args = parser.parse_args()
if args.graph_file:
with open(args.graph_file) as f:
data = json.load(f)
target = args.target or data.get("target", "target")
else:
data = generate_dag(args.nodes, args.edges, args.seed)
target = args.target or data["target"]
print(f"[Generated DAG: {len(data['edges'])} edges, "
f"{len(data['nodes'])} nodes, seed={args.seed}]")
print()
ranker = load_graph(data)
ranker.decay_factor = args.decay
ranker.max_depth = args.max_depth
ranked = ranker.compute(target)
ranker.print_report(ranked, target, top_n=args.top)
if args.export:
with open(args.export, "w") as f:
json.dump([r.to_dict() for r in ranked], f, indent=2)
print(f"\n→ Exported {← all inventions · built by the Nowness lab · page generated 02 Aug 2026, 12:40 UTC