| # Phase B3 of autonomous-domain-insights v1.1: vendor-spend
# concentration watch. Two cypher queries (Kuzu 0.6.1 doesn't
# do GROUP BY gracefully alongside attribute projections) joined
# in Python.
vendor_rows = graph.query(
"MATCH (o:Organisation) WHERE o.kind = 'vendor' "
"RETURN o.id AS id, o.name AS name, o.risk_band AS rb "
"ORDER BY o.id"
)
spend_rows = graph.query(
"MATCH (m:Money)-[:PAYS]->(o:Organisation) "
"WHERE o.kind = 'vendor' "
"RETURN o.id AS id, sum(m.amount) AS total"
)
spend_by_vendor = {}
for r in spend_rows:
spend_by_vendor[r["id"]] = float(r["total"] or 0)
total_vendor_spend = 0.0
for v in spend_by_vendor.values():
total_vendor_spend = total_vendor_spend + v
risks = []
proposed_actions = []
active_pauses = 0
fingerprint_parts = []
top_pct = 0.0
for r in vendor_rows:
vid = r["id"]
vname = r["name"] or vid
rb = r["rb"]
spend = spend_by_vendor.get(vid, 0.0)
if total_vendor_spend > 0:
concentration = spend / total_vendor_spend
else:
concentration = 0.0
pct_int = int(concentration * 100)
if concentration > top_pct:
top_pct = concentration
has_pause = len(active_policies_for(
graph,
scope_kind="Organisation",
scope_id=vid,
verdict="freeze",
)) > 0
if has_pause:
active_pauses = active_pauses + 1
is_risk = False
if concentration > 0.12:
is_risk = True
elif rb in ("amber", "red") and concentration > 0.05:
is_risk = True
fingerprint_parts.append(
"(" + vid + "," + str(pct_int) + "," + str(rb or "")
+ "," + str(has_pause) + ")"
)
if is_risk:
risks.append({
"id": vid,
"name": vname,
"rb": rb,
"concentration": concentration,
"pct_int": pct_int,
"has_pause": has_pause,
})
if not has_pause:
slug = vid.lower().replace("org-vendor-", "")
proposed_actions.append({
"id": "pause-vendor-" + slug,
"label": (
"Pause new POs to " + vname
+ " (" + str(pct_int) + "% concentration)"
),
"kind": "policy_set",
"verdict": "freeze",
"decided_on": [vid],
"attributes": {"expiry_days": 14, "scope": "vendor_po"},
"reason": (
vname + " carries " + str(pct_int)
+ "% of total vendor spend (risk_band "
+ (rb or "unknown")
+ ") — request alternate sourcing"
),
})
vendors_tracked = len(vendor_rows)
concentration_risks = len(risks)
if concentration_risks == 0:
headline = "Vendor portfolio diversified"
else:
headline = (
str(concentration_risks)
+ " vendor(s) above concentration threshold — recommend pauses"
)
body = ", ".join(
x["name"] + ": " + str(x["pct_int"]) + "%" for x in risks
)
fp = "sourcing_lead:" + ",".join(fingerprint_parts)
if len(fp) > 256:
fp = fp[:256]
summary = {
"headline": headline,
"body": body,
"kpis": {
"vendors_tracked": vendors_tracked,
"concentration_risks": concentration_risks,
"active_pauses": active_pauses,
"total_vendor_spend_gbp": float(total_vendor_spend),
"top_vendor_pct": float(top_pct),
},
"proposed_actions": proposed_actions,
"fingerprint": fp,
}
|
