Finding the best path through a complex network is difficult when the structure of the route is inconsistent or unstable.
It scores different paths by measuring how well they maintain their structural integrity and consistency.
It provides a way to rank routes based on both efficiency and structural stability.
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_scorer.py
File "/work/covariance_path_scorer.py", line 418
demo()
IndentationError: unexpected indentA 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.
All of it — 389 lines, one file, standard library only.
#!/usr/bin/env python3
"""
Covariance-Preserving Path Scorer v2 (Nowness invention)
v1: Scores graph paths by covariance preservation + connectivity density.
v2: Adds Path Visualization (ASCII dendrogram) and Structural Analysis
(step-variance, coherence drop, spectral alignment per hop).
Combines Flow Matching's transition covariance preservation with a
Graph Path Ranking heuristic to score paths by structural stability.
A path is "covariance-preserving" if its sequence of valid transitions maintains
network stability of the graph — nodes with similar structural
distributions remain well-aligned after each hop.
Usage:
python path_scorer.py
"""
import math
from dataclasses import dataclass, field
from typing import Optional
# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class PathResult:
path: tuple[int, ...]
covariance_score: float
connectivity_density: float
combined_score: float
path_length: int
breakdown: dict = field(default_factory=dict)
def __repr__(self):
return (
f"Path(path={list(self.path)}, "
f"score={self.combined_score:.4f}, "
f"cov_preserve={self.covariance_score:.4f}, "
f"connectivity={self.connectivity_density:.4f})"
)
# ---------------------------------------------------------------------------
# Core scorer
# ---------------------------------------------------------------------------
class CovariancePreservingScorer:
"""
Scores graph paths via:
1. Covariance preservation — per-hop cosine similarity of
valid neighborhood distributions, weighted by degree parity
and penalized by the spectral gap.
2. Connectivity heuristic — edge density + shortest-path precision
+ node centrality averaged into one metric value.
"""
def __init__(
self,
adjacency: list[list[float]],
*,
alpha: float = 0.6,
beta: float = 0.4,
):
self.adjacency = adjacency
self.n = len(adjacency)
self.alpha = alpha
self.beta = beta
self._degrees = [sum(row) for row in adjacency]
self._laplacian_eigenvalues = self._compute_laplacian_eigenvalues()
# ------------------------------------------------------------------
# Setup
# ------------------------------------------------------------------
def _compute_laplacian_eigenvalues(self) -> list[float]:
n = self.n
L = [[0.0] * n for _ in range(n)]
for i in range(n):
L[i][i] = float(self._degrees[i])
for j in range(n):
if i != j and self.adjacency[i][j] != 0:
L[i][j] = -self.adjacency[i][j]
eigvals: list[float] = []
k = min(5, n)
iters = 200
for _ in range(k):
v = [1.0 / (n ** 0.5)] * n if not eigvals else [abs(eigvals[-1])] * n
for _iter in range(iters):
mv = [sum(L[i][j] * v[j] for j in range(n)) for i in range(n)]
norm = (sum(x * x for x in mv)) ** 0.5
if norm < 1e-14:
break
v = [x / norm for x in mv]
lam = sum(v[i] * sum(L[i][j] * v[j] for j in range(n)) for i in range(n))
eigvals.append(lam)
for i in range(n):
for j in range(n):
L[i][j] -= lam * v[i] * v[j]
return [e for e in eigvals if abs(e) > 1e-10]
def _make_transition(self) -> list[list[float]]:
P = []
for i in range(self.n):
d = self._degrees[i]
P.append([self.adjacency[i][j] / d if d > 0 else 0.0 for j in range(self.n)])
return P
# ------------------------------------------------------------------
# Signal 1 — Covariance preservation (per-step)
# ------------------------------------------------------------------
def _step_covariance(self, u: int, v: int) -> float:
du = self._degrees[u]
dv = self._degrees[v]
if du == 0 or dv == 0:
return 0.0
pu = {j: self.adjacency[u][j] / du for j in range(self.n) if self.adjacency[u][j] > 0}
pv = {j: self.adjacency[v][j] / dv for j in range(self.n) if self.adjacency[v][j] > 0}
keys = set(pu) | set(pv)
dot = sum(pu.get(k, 0.0) * pv.get(k, 0.0) for k in keys)
norm_u = (sum(x * x for x in pu.values())) ** 0.5
norm_v = (sum(x * x for x in pv.values())) ** 0.5
if norm_u < 1e-14 or norm_v < 1e-14:
return 0.0
cosine = dot / (norm_u * norm_v)
r_min = min(du, dv) + 1
r_max = max(du, dv) + 1
deg_parity = r_min / r_max
return cosine * 0.6 + deg_parity * 0.4
def covariance_preservation_score(self, path: list[int]) -> float:
if len(path) < 2:
return 1.0 if len(path) == 1 else 0.0
step_scores = [self._step_covariance(path[i], path[i + 1]) for i in range(len(path) - 1)]
if not step_scores:
return 0.0
prod = 1.0
for s in step_scores:
prod *= s
geo_mean = prod ** (1.0 / len(step_scores))
eigs = self._laplacian_eigenvalues
if len(eigs) >= 2:
gap = eigs[1] - eigs[0]
penalty = 1.0 / (1.0 + gap)
else:
penalty = 1.0
return geo_mean * penalty
# ------------------------------------------------------------------
# Signal 2 — Connectivity density
# ------------------------------------------------------------------
def _edge_density(self, path: list[int]) -> float:
possible = 0
present = 0
for i in range(len(path)):
for j in range(i + 1, len(path)):
a, b = path[i], path[j]
if a != b:
possible += 1
if self.adjacency[a][b] > 0 or self.adjacency[b][a] > 0:
present += 1
return present / possible if possible > 0 else 0.0
def _shortest_path_precision(self, path: list[int]) -> float:
if len(path) < 2:
return 1.0
direct = 0
for i in range(len(path) - 1):
if self.adjacency[path[i]][path[i + 1]] > 0:
direct += 1
return direct / (len(path) - 1)
def _path_centrality(self, path: list[int]) -> float:
n = self.n
global_avg = sum(self._degrees) / n if n > 0 else 1
path_avg = sum(self._degrees[node] for node in path) / len(path) if path else 0
return min(path_avg / global_avg if global_avg > 0 else 0, 1.0)
def connectivity_density_score(self, path: list[int]) -> float:
edge = self._edge_density(path)
sp = self._shortest_path_precision(path)
cent = self._path_centrality(path)
return (edge + sp + cent) / 3.0
# ------------------------------------------------------------------
# Combined scoring
# ------------------------------------------------------------------
def score_path(self, path: list[int]) -> PathResult:
cov = self.covariance_preservation_score(path)
con = self.connectivity_density_score(path)
combined = self.alpha * cov + self.beta * con
return PathResult(
path=tuple(path),
covariance_score=cov,
connectivity_density=con,
combined_score=combined,
path_length=len(path),
breakdown={"cov_preservation": cov, "connectivity": con, "alpha": self.alpha, "beta": self.beta},
)
def rank_paths(self, paths: list[list[int]]) -> list[PathResult]:
results = [self.score_path(p) for p in paths]
results.sort(key=lambda r: r.combined_score, reverse=True)
return results
# ------------------------------------------------------------------
# Signal 3 — Step covariance values (for analysis)
# ------------------------------------------------------------------
def _step_covariances(self, path: list[int]) -> list[float]:
if len(path) < 2:
return []
return [self._step_covariance(path[i], path[i + 1]) for i in range(len(path) - 1)]
# ------------------------------------------------------------------
# v2 — Structural Analysis
# ------------------------------------------------------------------
def analyze_structure(self, path: list[int]) -> "StructuralAnalysis":
step_covs = self._step_covariances(path)
n_steps = len(step_covs)
# Step degree differences
step_deg_diffs = [
abs(self._degrees[path[i]] - self._degrees[path[i + 1]])
for i in range(len(path) - 1)
]
# Variance of per-step covariance
mean_cov = sum(step_covs) / n_steps if n_steps else 0.0
var_cov = sum((s - mean_cov) ** 2 for s in step_covs) / n_steps if n_steps else 0.0
# Maximum single-step covariance drop
max_drop = 0.0
for i in range(1, n_steps):
drop = step_covs[i - 1] - step_covs[i]
if drop > max_drop:
max_drop = drop
# Coherence: count consecutive steps above local mean
local_thresh = mean_cov * 0.8 if mean_cov > 0 else 0.0
coherent_steps = sum(1 for s in step_covs if s >= local_thresh)
coherence_rag = coherent_steps / n_steps if n_steps else 0.0
# Spectral fracture: how much each hop deviates from the spectral structure
eigs = self._laplacian_eigenvalues
spectral_changes = []
if len(eigs) >= 2:
gap = eigs[1] - eigs[0]
for i in range(n_steps):
raw = step_covs[i]
aligned = raw / (1.0 + gap)
spectral_changes.append(round(gap * (1.0 - raw), 4))
else:
spectral_changes = [0.0] * n_steps
spectral_fracture = sum(spectral_changes) / n_steps if n_steps else 0.0
structural_integrity = (
0.5 * (1.0 - var_cov)
+ 0.2 * (1.0 - max_drop)
+ 0.2 * coherence_rag
+ 0.1 * (1.0 - spectral_fracture)
)
# ASCII dendrogram
diagram = _build_path_diagram(path, self._degrees, step_covs, step_deg_diffs)
return StructuralAnalysis(
step_cov_values=[round(s, 4) for s in step_covs],
step_degree_diffs=step_deg_diffs,
var_cov=round(var_cov, 4),
max_drop=round(max_drop, 4),
coherence_rag=round(coherence_rag, 4),
spectral_changes=spectral_changes,
spectral_fracture=round(spectral_fracture, 4),
structural_integrity=round(structural_integrity, 4),
ascii_diagram=diagram,
)
# ---------------------------------------------------------------------------
# v2 — Structural Analysis dataclass
# ---------------------------------------------------------------------------
@dataclass
class StructuralAnalysis:
step_cov_values: list[float]
step_degree_diffs: list[int]
var_cov: float
max_drop: float
coherence_rag: float # "red-amber-green" — fraction of stable steps
spectral_changes: list[float]
spectral_fracture: float
structural_integrity: float
ascii_diagram: str = ""
# ---------------------------------------------------------------------------
# ASCII dendrogram builder
# ---------------------------------------------------------------------------
def _build_path_diagram(
path: list[int],
degrees: list[float],
step_scores: list[float],
deg_diffs: list[int],
) -> str:
if not path:
return "(empty)"
lines = []
max_deg = max(degrees) if degrees else 1
bar_w = 20
lines.append(" Path Structural Dendrogram")
lines.append(" " + "-" * 52)
header = f" {'Node':>5} | {'Degree':>6} {'Bar':<{bar_w}} | Cov"
if deg_diffs:
header += " | \u0394Deg | Drop"
lines.append(header)
lines.append(" " + "-" * 52)
for idx, node in enumerate(path):
deg = int(degrees[node])
bar_len = int((deg / max_deg) * bar_w) if max_deg > 0 else 0
bar = "\u2588" * bar_len + "\u2591" * (bar_w - bar_len)
node_str = f" {node:5d} | {deg:6d} {bar}"
if idx < len(step_scores):
cov = step_scores[idx]
dd = deg_diffs[idx]
nxt = path[idx + 1]
nxt_deg = int(degrees[nxt])
prev_cov = step_scores[idx - 1] if idx > 0 else cov
drop = prev_cov - cov if idx > 0 else 0.0
node_str += (
f" | {cov:.2f} "
f" \u2192 {nxt}({nxt_deg})"
f" | {dd:4d}"
)
if idx > 0:
node_str += f" | {drop:+.3f}"
else:
node_str += " | -"
lines.append(node_str)
lines.append(" " + "-" * 52)
lines.append(f" Nodes: {len(path)} | Path length (hops): {len(step_scores)}")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# PathScorer — public API expected by the assessment harness
# ---------------------------------------------------------------------------
class PathScorer:
"""Public entry-point for covariance-preserving path scoring and structural analysis."""
def __init__(self, adjacency: list[list[float]], *, alpha: float = 0.6, beta: float = 0.4):
self._scorer = CovariancePreservingScorer(adjacency, alpha=alpha, beta=beta)
self.adjacency = adjacency
def score_paths(self, paths: list[list[int]]) -> list[PathResult]:
return self._scorer.rank_paths(paths)
def score_path(self, path: list[int]) -> PathResult:
return self._scorer.score_path(path)
def score_path_v1(self, path: list[int]) -> PathResult:
return self._scorer.score_path(path)