Locy Use Case: Probabilistic Risk Scoring¶

Evaluate vendor reliability by combining independent quality signals with MNOR (noisy-OR failure) and MPROD (joint reliability).

This notebook uses schema-first mode (recommended): labels, edge types, and typed properties are defined before ingest.

How To Read This Notebook¶

Step 1 initializes an isolated local database.
Step 2 defines schema (the recommended production path).
Step 3 seeds a minimal graph for this use case.
Step 4 declares Locy rules and query statements.
Steps 5-6 evaluate and inspect command/query outputs.
Step 7 tells you what to look for in the results.

1) Setup¶

Creates a temporary database directory so the example is reproducible and leaves no state behind.

import os
import shutil
import tempfile
from pprint import pprint

import uni_db

DB_DIR = tempfile.mkdtemp(prefix="uni_locy_")
print("DB_DIR:", DB_DIR)

db = uni_db.Uni.open(DB_DIR)
session = db.session()

DB_DIR: /tmp/uni_locy_9gwyweq7

2) Define Schema (Recommended)¶

Define labels, property types, and edge types before inserting data.

(
    db.schema()
    .label("Vendor")
        .property("name", "string")
    .done()
    .label("Component")
        .property("name", "string")
    .done()
    .label("QualitySignal")
        .property("name", "string")
        .property("pass_rate", "float64")
    .done()
    .edge_type("SUPPLIES", ["Vendor"], ["Component"])
    .done()
    .edge_type("HAS_SIGNAL", ["Component"], ["QualitySignal"])
    .done()
    .apply()
)

print('Schema created')

Schema created

3) Seed Graph Data¶

Insert only the entities/relationships needed for this scenario so rule behavior stays easy to inspect.

tx = session.tx()
tx.execute("CREATE (:Vendor {name: 'ReliaCorp'})")
tx.execute("CREATE (:Vendor {name: 'QuickParts'})")
tx.execute("CREATE (:Vendor {name: 'BudgetSupply'})")
tx.execute("CREATE (:Component {name: 'Sensor'})")
tx.execute("CREATE (:Component {name: 'Motor'})")
tx.execute("CREATE (:Component {name: 'Controller'})")
tx.execute("CREATE (:Component {name: 'Battery'})")
tx.execute("CREATE (:QualitySignal {name: 'Thermal Test', pass_rate: 0.95})")
tx.execute("CREATE (:QualitySignal {name: 'Vibration Test', pass_rate: 0.90})")
tx.execute("CREATE (:QualitySignal {name: 'Voltage Tolerance', pass_rate: 0.85})")
tx.execute("CREATE (:QualitySignal {name: 'Humidity Test', pass_rate: 0.92})")
tx.execute("CREATE (:QualitySignal {name: 'Load Test', pass_rate: 0.88})")
tx.execute("CREATE (:QualitySignal {name: 'EMC Test', pass_rate: 0.75})")
tx.execute("CREATE (:QualitySignal {name: 'Cycle Life', pass_rate: 0.80})")
tx.execute("CREATE (:QualitySignal {name: 'Drop Test', pass_rate: 0.70})")
tx.execute("MATCH (v:Vendor {name:'ReliaCorp'}), (c:Component {name:'Sensor'}) CREATE (v)-[:SUPPLIES]->(c)")
tx.execute("MATCH (v:Vendor {name:'ReliaCorp'}), (c:Component {name:'Motor'}) CREATE (v)-[:SUPPLIES]->(c)")
tx.execute("MATCH (v:Vendor {name:'QuickParts'}), (c:Component {name:'Motor'}) CREATE (v)-[:SUPPLIES]->(c)")
tx.execute("MATCH (v:Vendor {name:'QuickParts'}), (c:Component {name:'Controller'}) CREATE (v)-[:SUPPLIES]->(c)")
tx.execute("MATCH (v:Vendor {name:'BudgetSupply'}), (c:Component {name:'Controller'}) CREATE (v)-[:SUPPLIES]->(c)")
tx.execute("MATCH (v:Vendor {name:'BudgetSupply'}), (c:Component {name:'Battery'}) CREATE (v)-[:SUPPLIES]->(c)")
tx.execute("MATCH (c:Component {name:'Sensor'}), (s:QualitySignal {name:'Thermal Test'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.execute("MATCH (c:Component {name:'Sensor'}), (s:QualitySignal {name:'Vibration Test'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.execute("MATCH (c:Component {name:'Motor'}), (s:QualitySignal {name:'Voltage Tolerance'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.execute("MATCH (c:Component {name:'Motor'}), (s:QualitySignal {name:'Humidity Test'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.execute("MATCH (c:Component {name:'Controller'}), (s:QualitySignal {name:'Load Test'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.execute("MATCH (c:Component {name:'Controller'}), (s:QualitySignal {name:'EMC Test'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.execute("MATCH (c:Component {name:'Battery'}), (s:QualitySignal {name:'Cycle Life'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.execute("MATCH (c:Component {name:'Battery'}), (s:QualitySignal {name:'Drop Test'}) CREATE (c)-[:HAS_SIGNAL]->(s)")
tx.commit()
print('Seeded graph data')

Seeded graph data

4) Locy Program¶

CREATE RULE defines derived relations. QUERY ... WHERE ... RETURN ... reads from those relations.

program = r'''
CREATE RULE component_failure_risk AS
MATCH (c:Component)-[:HAS_SIGNAL]->(s:QualitySignal)
FOLD risk = MNOR(1.0 - s.pass_rate)
YIELD KEY c, risk

CREATE RULE vendor_reliability AS
MATCH (v:Vendor)-[:SUPPLIES]->(c:Component)
WHERE c IS component_failure_risk
FOLD reliability = MPROD(1.0 - risk)
YIELD KEY v, reliability

QUERY component_failure_risk RETURN c.name AS component, risk
QUERY vendor_reliability RETURN v.name AS vendor, reliability
'''
print(program)

CREATE RULE component_failure_risk AS
MATCH (c:Component)-[:HAS_SIGNAL]->(s:QualitySignal)
FOLD risk = MNOR(1.0 - s.pass_rate)
YIELD KEY c, risk

CREATE RULE vendor_reliability AS
MATCH (v:Vendor)-[:SUPPLIES]->(c:Component)
WHERE c IS component_failure_risk
FOLD reliability = MPROD(1.0 - risk)
YIELD KEY v, reliability

QUERY component_failure_risk RETURN c.name AS component, risk
QUERY vendor_reliability RETURN v.name AS vendor, reliability

5) Evaluate Locy Program¶

Run the program, then inspect materialization stats (iterations, strata, and executed queries).

out = session.locy(program)

print("Derived relations:", list(out.derived.keys()))
stats = out.stats
print("Iterations:", stats.total_iterations)
print("Strata:", stats.strata_evaluated)
print("Queries executed:", stats.queries_executed)

Derived relations: ['component_failure_risk', 'vendor_reliability']
Iterations: 0
Strata: 2
Queries executed: 0

6) Inspect Command Results¶

Each command result can contain rows; this is the easiest way to verify your rule outputs and query projections.

print("Derived relation snapshots:")
for rel_name, rel_rows in out.derived.items():
    print(f"\\n{rel_name}: {len(rel_rows)} row(s)")
    pprint(rel_rows)

if out.command_results:
    print("\\nCommand results:")
for i, cmd in enumerate(out.command_results, start=1):
    print(f"\\nCommand #{i}:", cmd.command_type)
    rows = getattr(cmd, 'rows', None)
    if rows is not None:
        pprint(rows)
if not out.command_results:
    print("\\nNo QUERY/EXPLAIN/ABDUCE command outputs in this program.")

Derived relation snapshots:
\ncomponent_failure_risk: 4 row(s)
[{'c': Node(id=3, labels=["Component"], properties={'name': 'Sensor'}),
  'risk': 0.14500000000000002},
 {'c': Node(id=4, labels=["Component"], properties={'name': 'Motor'}),
  'risk': 0.21799999999999997},
 {'c': Node(id=5, labels=["Component"], properties={'name': 'Controller'}),
  'risk': 0.33999999999999997},
 {'c': Node(id=6, labels=["Component"], properties={'name': 'Battery'}),
  'risk': 0.44000000000000006}]
\nvendor_reliability: 3 row(s)
[{'reliability': 0.66861,
  'v': Node(id=0, labels=["Vendor"], properties={'name': 'ReliaCorp'})},
 {'reliability': 0.51612,
  'v': Node(id=1, labels=["Vendor"], properties={'name': 'QuickParts'})},
 {'reliability': 0.3696,
  'v': Node(id=2, labels=["Vendor"], properties={'name': 'BudgetSupply'})}]
\nCommand results:
\nCommand #1: query
[{'component': 'Sensor', 'risk': 0.14500000000000002},
 {'component': 'Motor', 'risk': 0.21799999999999997},
 {'component': 'Controller', 'risk': 0.33999999999999997},
 {'component': 'Battery', 'risk': 0.44000000000000006}]
\nCommand #2: query
[{'reliability': 0.66861, 'vendor': 'ReliaCorp'},
 {'reliability': 0.51612, 'vendor': 'QuickParts'},
 {'reliability': 0.3696, 'vendor': 'BudgetSupply'}]

7) What To Expect¶

Use these checks to validate output after evaluation: - Component risk ordering: Battery > Controller > Motor > Sensor (lower pass rates → higher risk). - Vendor reliability ordering: ReliaCorp > QuickParts > BudgetSupply. - MNOR values stay in [0, 1] — noisy-OR never exceeds 1.0 even with many signals. - MPROD values decrease with more components — each additional component can only reduce joint reliability. - Two query result blocks should appear: one for component_failure_risk, one for vendor_reliability.

8) Cleanup¶

Delete the temporary database directory created in setup.

shutil.rmtree(DB_DIR, ignore_errors=True)
print("Cleaned up", DB_DIR)

Cleaned up /tmp/uni_locy_9gwyweq7