"""Reference solution — a financial knowledge graph from scratch. Four pieces, and the reason there are four is the phase's whole argument: * an RDF TRIPLE STORE, because a bank's structural questions are about paths; * an RDFS/OWL REASONER, because "ultimately controlled by" is entailed, not stored; * a SHACL VALIDATOR, because OWL is open-world and cannot tell you a record is INCOMPLETE — which is the question a bank actually asks; * a SPARQL engine, because property paths express in one line what would otherwise be a recursive query nobody can read. Deterministic: sorted bindings, no clock, no randomness. ``python solution.py`` runs the worked example. """ from __future__ import annotations import re from dataclasses import dataclass, field from enum import Enum from typing import Callable, Dict, FrozenSet, Iterable, List, Mapping, Optional, Sequence, Set, Tuple # ====================================================================================== # 1. IRIs, literals and triples # ====================================================================================== @dataclass(frozen=True, order=True) class IRI: """A globally unique identifier. The entire point of RDF: two systems that both say ``fibo-be:LegalEntity`` mean the same thing, without a mapping table.""" value: str def __str__(self) -> str: return self.value @dataclass(frozen=True, order=True) class Literal: """A value, optionally typed. Distinct from an IRI because a literal is never a subject — you can say things about an entity, not about the string "Acme".""" value: str datatype: str = "xsd:string" def __str__(self) -> str: return self.value Term = object # IRI | Literal | Variable @dataclass(frozen=True, order=True) class Variable: name: str def __str__(self) -> str: return f"?{self.name}" @dataclass(frozen=True, order=True) class Triple: subject: IRI predicate: IRI object: Term def __str__(self) -> str: return f"{self.subject} {self.predicate} {self.object} ." # ---- CURIE handling ------------------------------------------------------------------ DEFAULT_PREFIXES: Mapping[str, str] = { "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#", "rdfs": "http://www.w3.org/2000/01/rdf-schema#", "owl": "http://www.w3.org/2002/07/owl#", "sh": "http://www.w3.org/ns/shacl#", "xsd": "http://www.w3.org/2001/XMLSchema#", # FIBO's real namespaces are long; these are faithful in shape. "fibo-be": "https://spec.edmcouncil.org/fibo/ontology/BE/LegalEntities/", "fibo-fnd": "https://spec.edmcouncil.org/fibo/ontology/FND/", "bank": "https://bank.example.ae/ontology/", "ent": "https://bank.example.ae/entity/", } _CURIE_RE = re.compile(r"^([A-Za-z][\w-]*):([^\s]+)$") class PrefixMap: """Expands CURIEs to full IRIs and shortens them back. Kept explicit because a graph that mixes expanded and abbreviated forms of the same IRI has two identifiers for one thing — which silently breaks every join. """ def __init__(self, prefixes: Optional[Mapping[str, str]] = None) -> None: self.prefixes: Dict[str, str] = dict(DEFAULT_PREFIXES) if prefixes: self.prefixes.update(prefixes) def expand(self, curie: str) -> IRI: # A full IRI is not a CURIE, even though "https:..." matches the CURIE shape. # Checking for the scheme separator first is what keeps `https://example.org/x` # from being read as prefix "https" with local name "//example.org/x". if "://" in curie: return IRI(curie) match = _CURIE_RE.match(curie) if not match: return IRI(curie) # a bare name, used as-is prefix, local = match.groups() namespace = self.prefixes.get(prefix) if namespace is None: raise KeyError(f"unknown prefix: {prefix!r}") return IRI(namespace + local) def shorten(self, iri: IRI) -> str: best: Optional[Tuple[str, str]] = None for prefix, namespace in self.prefixes.items(): if iri.value.startswith(namespace): if best is None or len(namespace) > len(self.prefixes[best[0]]): best = (prefix, iri.value[len(namespace):]) return f"{best[0]}:{best[1]}" if best else iri.value # ====================================================================================== # 2. The triple store # ====================================================================================== class Graph: """An in-memory RDF graph with three indexes. A triple store is a *set*, not a list: asserting the same fact twice changes nothing. That idempotence is what makes entailment safe to run repeatedly. """ def __init__(self, prefixes: Optional[PrefixMap] = None) -> None: self.prefixes = prefixes or PrefixMap() self._triples: Set[Triple] = set() self._spo: Dict[IRI, Dict[IRI, Set[Term]]] = {} self._pos: Dict[IRI, Dict[Term, Set[IRI]]] = {} self._osp: Dict[Term, Dict[IRI, Set[IRI]]] = {} # -- writing ------------------------------------------------------------------ def add(self, subject: IRI, predicate: IRI, obj: Term) -> bool: """Returns True if the triple was new — the signal a fixed-point loop needs.""" triple = Triple(subject, predicate, obj) if triple in self._triples: return False self._triples.add(triple) self._spo.setdefault(subject, {}).setdefault(predicate, set()).add(obj) self._pos.setdefault(predicate, {}).setdefault(obj, set()).add(subject) self._osp.setdefault(obj, {}).setdefault(subject, set()).add(predicate) return True def add_curies(self, subject: str, predicate: str, obj: str, *, literal: bool = False, datatype: str = "xsd:string") -> bool: return self.add( self.prefixes.expand(subject), self.prefixes.expand(predicate), Literal(obj, datatype) if literal else self.prefixes.expand(obj)) # -- reading ------------------------------------------------------------------ def __len__(self) -> int: return len(self._triples) def __contains__(self, triple: Triple) -> bool: return triple in self._triples def triples(self) -> List[Triple]: """Sorted, so any dump of the graph is diffable.""" return sorted(self._triples, key=lambda t: (str(t.subject), str(t.predicate), str(t.object))) def match(self, subject: Optional[IRI] = None, predicate: Optional[IRI] = None, obj: Optional[Term] = None) -> List[Triple]: """Pattern match with any position bound or free — the primitive SPARQL sits on. Chooses the most selective index available, which is the only reason to keep three of them. """ if subject is not None: candidates = (Triple(subject, p, o) for p, objects in self._spo.get(subject, {}).items() for o in objects) elif predicate is not None: candidates = (Triple(s, predicate, o) for o, subjects in self._pos.get(predicate, {}).items() for s in subjects) elif obj is not None: candidates = (Triple(s, p, obj) for s, predicates in self._osp.get(obj, {}).items() for p in predicates) else: candidates = iter(self._triples) out = [t for t in candidates if (predicate is None or t.predicate == predicate) and (obj is None or t.object == obj) and (subject is None or t.subject == subject)] return sorted(out, key=lambda t: (str(t.subject), str(t.predicate), str(t.object))) def objects(self, subject: IRI, predicate: IRI) -> List[Term]: return sorted(self._spo.get(subject, {}).get(predicate, set()), key=str) def subjects(self, predicate: IRI, obj: Term) -> List[IRI]: return sorted(self._pos.get(predicate, {}).get(obj, set()), key=str) # ====================================================================================== # 3. The reasoner — RDFS plus a useful OWL subset # ====================================================================================== RDF_TYPE = IRI(DEFAULT_PREFIXES["rdf"] + "type") RDFS_SUBCLASS = IRI(DEFAULT_PREFIXES["rdfs"] + "subClassOf") RDFS_SUBPROPERTY = IRI(DEFAULT_PREFIXES["rdfs"] + "subPropertyOf") RDFS_DOMAIN = IRI(DEFAULT_PREFIXES["rdfs"] + "domain") RDFS_RANGE = IRI(DEFAULT_PREFIXES["rdfs"] + "range") OWL_INVERSE = IRI(DEFAULT_PREFIXES["owl"] + "inverseOf") OWL_TRANSITIVE = IRI(DEFAULT_PREFIXES["owl"] + "TransitiveProperty") OWL_SYMMETRIC = IRI(DEFAULT_PREFIXES["owl"] + "SymmetricProperty") @dataclass(frozen=True) class EntailmentReport: added: int rounds: int def materialize(graph: Graph, *, max_rounds: int = 50) -> EntailmentReport: """Forward-chain to a fixed point. Rules implemented, each with the RDFS/OWL name a reviewer will recognise: rdfs9 type + subClassOf -> type rdfs11 subClassOf transitive rdfs7 property + subPropertyOf -> property rdfs5 subPropertyOf transitive rdfs2 predicate + domain -> subject type rdfs3 predicate + range -> object type owl inverseOf -> the reverse triple owl TransitiveProperty -> composition owl SymmetricProperty -> the reverse triple MATERIALIZATION is a cache. It is correct because RDFS/OWL entailment is MONOTONE: adding facts never retracts a conclusion, so a fixed point reached once stays valid as long as nothing is deleted. Deletion invalidates it, which is why a production store either re-materializes or reasons at query time. """ added_total = 0 rounds = 0 for _ in range(max_rounds): rounds += 1 added = 0 # rdfs11: subClassOf is transitive for triple in graph.match(predicate=RDFS_SUBCLASS): for onward in graph.match(subject=triple.object, predicate=RDFS_SUBCLASS): added += graph.add(triple.subject, RDFS_SUBCLASS, onward.object) # rdfs9: type propagates up the class hierarchy for triple in graph.match(predicate=RDF_TYPE): for superclass in graph.match(subject=triple.object, predicate=RDFS_SUBCLASS): added += graph.add(triple.subject, RDF_TYPE, superclass.object) # rdfs5: subPropertyOf is transitive for triple in graph.match(predicate=RDFS_SUBPROPERTY): for onward in graph.match(subject=triple.object, predicate=RDFS_SUBPROPERTY): added += graph.add(triple.subject, RDFS_SUBPROPERTY, onward.object) # rdfs7: a sub-property's assertions are also the super-property's for sub in graph.match(predicate=RDFS_SUBPROPERTY): if not isinstance(sub.object, IRI): continue for triple in graph.match(predicate=sub.subject): added += graph.add(triple.subject, sub.object, triple.object) # rdfs2 / rdfs3: domain and range imply types for declaration in graph.match(predicate=RDFS_DOMAIN): for triple in graph.match(predicate=declaration.subject): added += graph.add(triple.subject, RDF_TYPE, declaration.object) for declaration in graph.match(predicate=RDFS_RANGE): for triple in graph.match(predicate=declaration.subject): if isinstance(triple.object, IRI): added += graph.add(triple.object, RDF_TYPE, declaration.object) # owl:inverseOf for declaration in graph.match(predicate=OWL_INVERSE): if not isinstance(declaration.object, IRI): continue for triple in graph.match(predicate=declaration.subject): if isinstance(triple.object, IRI): added += graph.add(triple.object, declaration.object, triple.subject) for triple in graph.match(predicate=declaration.object): if isinstance(triple.object, IRI): added += graph.add(triple.object, declaration.subject, triple.subject) # owl:TransitiveProperty — the rule that answers "ultimately controlled by" for declaration in graph.match(predicate=RDF_TYPE, obj=OWL_TRANSITIVE): prop = declaration.subject for first in graph.match(predicate=prop): if not isinstance(first.object, IRI): continue for second in graph.match(subject=first.object, predicate=prop): added += graph.add(first.subject, prop, second.object) # owl:SymmetricProperty for declaration in graph.match(predicate=RDF_TYPE, obj=OWL_SYMMETRIC): prop = declaration.subject for triple in graph.match(predicate=prop): if isinstance(triple.object, IRI): added += graph.add(triple.object, prop, triple.subject) added_total += added if added == 0: break return EntailmentReport(added=added_total, rounds=rounds) # ====================================================================================== # 4. SHACL — closed-world validation over an open-world model # ====================================================================================== class Severity(str, Enum): VIOLATION = "Violation" WARNING = "Warning" INFO = "Info" @dataclass(frozen=True) class PropertyShape: """One constraint on one predicate of a node. This is the piece OWL cannot express. OWL's open-world assumption means "no LEI stated" is UNKNOWN, not missing — so an ontology can never tell you a record is incomplete. SHACL closes the world for validation purposes and answers exactly that. """ path: str # CURIE of the predicate min_count: Optional[int] = None max_count: Optional[int] = None datatype: Optional[str] = None pattern: Optional[str] = None node_kind: Optional[str] = None # "IRI" | "Literal" in_values: Optional[Tuple[str, ...]] = None severity: Severity = Severity.VIOLATION message: str = "" @dataclass(frozen=True) class NodeShape: name: str target_class: str # CURIE properties: Tuple[PropertyShape, ...] = () closed: bool = False # reject undeclared predicates ignored_properties: Tuple[str, ...] = () @dataclass(frozen=True) class Violation: focus_node: str path: str severity: Severity message: str value: Optional[str] = None def __str__(self) -> str: where = f"{self.focus_node} [{self.path}]" return f"{self.severity.value}: {where} {self.message}" + ( f" (value: {self.value})" if self.value else "") @dataclass(frozen=True) class ValidationReport: """SHACL's own output shape: conforms + a list of results.""" results: Tuple[Violation, ...] @property def conforms(self) -> bool: return not any(v.severity is Severity.VIOLATION for v in self.results) def by_severity(self, severity: Severity) -> List[Violation]: return [v for v in self.results if v.severity is severity] def validate(graph: Graph, shapes: Sequence[NodeShape]) -> ValidationReport: """Validate every node targeted by each shape. Note the target selection uses the MATERIALIZED type, so a node typed only by entailment is still validated — which is the interesting interaction between the two halves of this lab and the reason materialization runs first. """ prefixes = graph.prefixes results: List[Violation] = [] for shape in shapes: target = prefixes.expand(shape.target_class) for focus in sorted(graph.subjects(RDF_TYPE, target), key=str): results.extend(_validate_node(graph, focus, shape)) results.sort(key=lambda v: (v.focus_node, v.path, v.message)) return ValidationReport(tuple(results)) def _validate_node(graph: Graph, focus: IRI, shape: NodeShape) -> List[Violation]: prefixes = graph.prefixes out: List[Violation] = [] declared: Set[IRI] = set() for prop in shape.properties: path = prefixes.expand(prop.path) declared.add(path) values = graph.objects(focus, path) short_focus = prefixes.shorten(focus) if prop.min_count is not None and len(values) < prop.min_count: out.append(Violation( short_focus, prop.path, prop.severity, prop.message or f"expected at least {prop.min_count}, found {len(values)}")) if prop.max_count is not None and len(values) > prop.max_count: out.append(Violation( short_focus, prop.path, prop.severity, prop.message or f"expected at most {prop.max_count}, found {len(values)}")) for value in values: rendered = prefixes.shorten(value) if isinstance(value, IRI) else str(value) if prop.node_kind == "IRI" and not isinstance(value, IRI): out.append(Violation(short_focus, prop.path, prop.severity, "expected an IRI", rendered)) continue if prop.node_kind == "Literal" and not isinstance(value, Literal): out.append(Violation(short_focus, prop.path, prop.severity, "expected a literal", rendered)) continue if prop.datatype is not None: if not isinstance(value, Literal) or value.datatype != prop.datatype: out.append(Violation(short_focus, prop.path, prop.severity, f"expected datatype {prop.datatype}", rendered)) continue if prop.pattern is not None: text = value.value if isinstance(value, Literal) else value.value if re.search(prop.pattern, text) is None: out.append(Violation(short_focus, prop.path, prop.severity, f"does not match {prop.pattern}", rendered)) if prop.in_values is not None: text = value.value if isinstance(value, Literal) else prefixes.shorten(value) if text not in prop.in_values: out.append(Violation(short_focus, prop.path, prop.severity, f"not one of {list(prop.in_values)}", rendered)) if shape.closed: ignored = {prefixes.expand(p) for p in shape.ignored_properties} | {RDF_TYPE} for triple in graph.match(subject=focus): if triple.predicate not in declared and triple.predicate not in ignored: out.append(Violation( prefixes.shorten(focus), prefixes.shorten(triple.predicate), Severity.VIOLATION, "predicate is not allowed by a closed shape")) return out # ====================================================================================== # 5. SPARQL — basic graph patterns, OPTIONAL, FILTER and property paths # ====================================================================================== Binding = Dict[str, Term] @dataclass(frozen=True) class PathStep: """One element of a property path. ``modifier`` is "" (exactly one), "+" (one or more), "*" (zero or more). ``inverse`` traverses the predicate backwards — SPARQL's ``^``. Property paths are the reason SPARQL is worth implementing: "ultimately controlled by" is ``bank:controls+``, one token, versus a recursive CTE nobody can read. """ predicate: str modifier: str = "" inverse: bool = False @dataclass(frozen=True) class Pattern: """A triple pattern. Subject/object are a CURIE, a Variable, or a Literal.""" subject: object path: PathStep object: object @dataclass(frozen=True) class Filter: """A boolean over a binding. Kept as a callable so the lab needs no expression parser — the join semantics are the lesson, not the grammar.""" fn: Callable[[Mapping[str, Term]], bool] description: str = "filter" @dataclass(frozen=True) class Optional_: """SPARQL OPTIONAL: a left join. Patterns that fail to match leave their variables unbound rather than eliminating the solution.""" patterns: Tuple[Pattern, ...] @dataclass(frozen=True) class Query: select: Tuple[str, ...] where: Tuple[object, ...] # Pattern | Optional_ | Filter distinct: bool = True order_by: Tuple[str, ...] = () limit: Optional[int] = None def execute(graph: Graph, query: Query) -> List[Dict[str, str]]: """Evaluate a query by successive joins over a list of bindings. The algorithm is the whole of basic SPARQL: start with one empty binding; for each pattern, extend every surviving binding with every way that pattern can match. Shared variable names ARE the join condition — there is no explicit join clause because variable identity is the join. """ bindings: List[Binding] = [{}] for element in query.where: if isinstance(element, Pattern): bindings = _join(graph, bindings, element) elif isinstance(element, Optional_): bindings = _left_join(graph, bindings, element.patterns) elif isinstance(element, Filter): bindings = [b for b in bindings if element.fn(b)] else: # pragma: no cover raise TypeError(f"unsupported query element: {element!r}") prefixes = graph.prefixes rows: List[Dict[str, str]] = [] for binding in bindings: row: Dict[str, str] = {} for name in query.select: value = binding.get(name) if value is None: row[name] = "" elif isinstance(value, IRI): row[name] = prefixes.shorten(value) else: row[name] = str(value) rows.append(row) if query.distinct: seen: Set[Tuple[Tuple[str, str], ...]] = set() unique: List[Dict[str, str]] = [] for row in rows: key = tuple(sorted(row.items())) if key not in seen: seen.add(key) unique.append(row) rows = unique order = query.order_by or query.select rows.sort(key=lambda r: tuple(r.get(name, "") for name in order)) if query.limit is not None: rows = rows[: query.limit] return rows def _resolve(graph: Graph, term: object, binding: Binding) -> Optional[Term]: """A bound value for a term, or None when it is a free variable.""" if isinstance(term, Variable): return binding.get(term.name) if isinstance(term, Literal): return term if isinstance(term, str): return graph.prefixes.expand(term) return term # pragma: no cover def _join(graph: Graph, bindings: Sequence[Binding], pattern: Pattern) -> List[Binding]: out: List[Binding] = [] for binding in bindings: subject = _resolve(graph, pattern.subject, binding) obj = _resolve(graph, pattern.object, binding) for s, o in _walk(graph, pattern.path, subject, obj): extended = dict(binding) if isinstance(pattern.subject, Variable): extended[pattern.subject.name] = s if isinstance(pattern.object, Variable): extended[pattern.object.name] = o out.append(extended) return out def _left_join(graph: Graph, bindings: Sequence[Binding], patterns: Sequence[Pattern]) -> List[Binding]: out: List[Binding] = [] for binding in bindings: extended: List[Binding] = [binding] for pattern in patterns: extended = _join(graph, extended, pattern) out.extend(extended if extended else [binding]) # keep the unmatched row return out def _walk(graph: Graph, step: PathStep, subject: Optional[Term], obj: Optional[Term]) -> List[Tuple[Term, Term]]: """Every (start, end) pair reachable by this path step. ``+`` and ``*`` are breadth-first with a visited set — cycles in an ownership graph are real (circular shareholdings exist) and an unguarded traversal does not terminate. """ predicate = graph.prefixes.expand(step.predicate) def one_hop(start: Term) -> List[Term]: if step.inverse: return list(graph.subjects(predicate, start)) if not isinstance(start, IRI): return [] return list(graph.objects(start, predicate)) def starts() -> List[Term]: if subject is not None: return [subject] if step.inverse: return sorted({t.object for t in graph.match(predicate=predicate)}, key=str) return sorted({t.subject for t in graph.match(predicate=predicate)}, key=str) pairs: List[Tuple[Term, Term]] = [] for start in starts(): if step.modifier == "": ends = one_hop(start) else: ends = [] if step.modifier == "*": ends.append(start) frontier = [start] seen: Set[Term] = {start} if step.modifier == "*" else set() while frontier: current = frontier.pop(0) for nxt in one_hop(current): if nxt in seen: continue seen.add(nxt) ends.append(nxt) frontier.append(nxt) for end in ends: if obj is not None and end != obj: continue pairs.append((start, end)) return sorted(pairs, key=lambda pair: (str(pair[0]), str(pair[1]))) # ====================================================================================== # 6. Graph-grounded retrieval # ====================================================================================== @dataclass(frozen=True) class GraphContext: """What a graph contributes to a prompt: entities, the paths that connect them, and the reason each was included.""" entities: Tuple[str, ...] paths: Tuple[str, ...] rationale: str def expand_neighbourhood(graph: Graph, seed: str, *, hops: int = 1, predicates: Sequence[str] = ()) -> GraphContext: """Neighbourhood expansion: what a vector index cannot do. Similarity retrieves text that RESEMBLES the query. This retrieves entities STRUCTURALLY related to a seed — which is the right operation when the question is "who ultimately controls this counterparty" and the answer is three hops away in a document that shares no vocabulary with the question. """ if hops < 0: raise ValueError("hops must be >= 0") prefixes = graph.prefixes start = prefixes.expand(seed) allowed = {prefixes.expand(p) for p in predicates} if predicates else None reached: Dict[Term, List[str]] = {start: []} frontier: List[Term] = [start] for _ in range(hops): next_frontier: List[Term] = [] for node in frontier: if not isinstance(node, IRI): continue for triple in graph.match(subject=node): if allowed is not None and triple.predicate not in allowed: continue if triple.object in reached: continue reached[triple.object] = reached[node] + [ f"{prefixes.shorten(node)} -{prefixes.shorten(triple.predicate)}-> " f"{prefixes.shorten(triple.object) if isinstance(triple.object, IRI) else triple.object}"] next_frontier.append(triple.object) frontier = next_frontier entities = tuple(sorted( prefixes.shorten(node) if isinstance(node, IRI) else str(node) for node in reached)) paths = tuple(sorted({step for steps in reached.values() for step in steps})) return GraphContext(entities, paths, f"{hops}-hop neighbourhood of {prefixes.shorten(start)}") # ====================================================================================== # Worked example — a FIBO-shaped ownership and obligation graph # ====================================================================================== def build_ontology(graph: Graph) -> None: """A small FIBO-shaped ontology. Faithful in structure, tiny in scope.""" g = graph.add_curies # Class hierarchy g("bank:Bank", "rdfs:subClassOf", "fibo-be:LegalEntity") g("bank:Corporation", "rdfs:subClassOf", "fibo-be:LegalEntity") g("fibo-be:LegalEntity", "rdfs:subClassOf", "fibo-fnd:AutonomousAgent") # Property hierarchy: majority ownership IS control g("bank:majorityOwns", "rdfs:subPropertyOf", "bank:controls") # controls is transitive — the rule that answers "ultimately controlled by" g("bank:controls", "rdf:type", "owl:TransitiveProperty") g("bank:controls", "owl:inverseOf", "bank:controlledBy") g("bank:isCounterpartyOf", "rdf:type", "owl:SymmetricProperty") # Domain and range give free typing g("bank:hasLEI", "rdfs:domain", "fibo-be:LegalEntity") g("bank:majorityOwns", "rdfs:domain", "fibo-be:LegalEntity") g("bank:majorityOwns", "rdfs:range", "fibo-be:LegalEntity") g("bank:obligor", "rdfs:domain", "bank:Obligation") g("bank:obligor", "rdfs:range", "fibo-be:LegalEntity") def build_facts(graph: Graph) -> None: g = graph.add_curies # Ownership chain: Acme <- Northgate <- Meridian g("ent:Meridian", "bank:majorityOwns", "ent:Northgate") g("ent:Northgate", "bank:majorityOwns", "ent:Acme") g("ent:Acme", "rdf:type", "bank:Corporation") g("ent:Northgate", "rdf:type", "bank:Corporation") g("ent:Meridian", "rdf:type", "bank:Corporation") g("ent:Sanctioned", "rdf:type", "bank:Corporation") g("ent:Meridian", "bank:controlledBy", "ent:Sanctioned") # An opaque shell at the top of the chain: no name, no identifier. Exactly the shape # a KYC analyst cares about, and exactly what SHACL is for. g("ent:Opaque", "bank:majorityOwns", "ent:Sanctioned") # Identifiers, of varying quality g("ent:Acme", "bank:hasLEI", "5493001KJTIIGC8Y1R12", literal=True) g("ent:Acme", "bank:legalName", "Acme Trading FZE", literal=True) g("ent:Northgate", "bank:hasLEI", "NOT-AN-LEI", literal=True) g("ent:Northgate", "bank:legalName", "Northgate Holdings Ltd", literal=True) g("ent:Meridian", "bank:legalName", "Meridian Capital", literal=True) # no LEI g("ent:Sanctioned", "bank:legalName", "Sanctioned Entity", literal=True) g("ent:Sanctioned", "bank:hasLEI", "5493009ABCDEFGH12345", literal=True) # An obligation and a counterparty relation g("ent:LoanA", "rdf:type", "bank:Obligation") g("ent:LoanA", "bank:obligor", "ent:Acme") g("ent:LoanA", "bank:amountAED", "250000", literal=True, datatype="xsd:integer") g("ent:Acme", "bank:isCounterpartyOf", "ent:Northgate") g("ent:Sanctioned", "bank:onSanctionsList", "true", literal=True, datatype="xsd:boolean") LEGAL_ENTITY_SHAPE = NodeShape( name="LegalEntityShape", target_class="fibo-be:LegalEntity", properties=( PropertyShape(path="bank:hasLEI", min_count=1, max_count=1, datatype="xsd:string", pattern=r"^[A-Z0-9]{18}[0-9]{2}$", message="a legal entity needs exactly one well-formed LEI"), PropertyShape(path="bank:legalName", min_count=1, max_count=1, datatype="xsd:string"), ), ) OBLIGATION_SHAPE = NodeShape( name="ObligationShape", target_class="bank:Obligation", properties=( PropertyShape(path="bank:obligor", min_count=1, node_kind="IRI"), PropertyShape(path="bank:amountAED", min_count=1, max_count=1, datatype="xsd:integer"), ), ) def main() -> None: # pragma: no cover - narrative output graph = Graph() build_ontology(graph) build_facts(graph) prefixes = graph.prefixes print("=" * 78) print("1. THE GRAPH AS ASSERTED") print("=" * 78) print(f" asserted triples: {len(graph)}") for triple in graph.match(predicate=prefixes.expand("bank:majorityOwns")): print(f" {prefixes.shorten(triple.subject)} majorityOwns " f"{prefixes.shorten(triple.object)}") print(" note: nobody asserted that Meridian controls Acme.") print() print("=" * 78) print("2. ENTAILMENT — WHAT THE GRAPH KNOWS WITHOUT BEING TOLD") print("=" * 78) report = materialize(graph) print(f" {report.added} triples entailed in {report.rounds} rounds " f"-> {len(graph)} total") controls = prefixes.expand("bank:controls") for triple in graph.match(predicate=controls): print(f" {prefixes.shorten(triple.subject)} controls " f"{prefixes.shorten(triple.object)}") print(" Meridian -> Acme is TRANSITIVE closure over a SUB-PROPERTY of controls.") print(" Two rules composed: rdfs7 (majorityOwns => controls) then owl:Transitive.") acme_types = [prefixes.shorten(t) for t in graph.objects(prefixes.expand("ent:Acme"), RDF_TYPE)] print(f" ent:Acme is now typed: {acme_types}") print(" ... Corporation was asserted; LegalEntity and AutonomousAgent were entailed.") print() print("=" * 78) print("3. SHACL — THE QUESTION OWL CANNOT ANSWER") print("=" * 78) result = validate(graph, [LEGAL_ENTITY_SHAPE, OBLIGATION_SHAPE]) print(f" conforms: {result.conforms}") for violation in result.results: print(f" {violation}") print() print(" OWL says 'Meridian has no stated LEI' means UNKNOWN, not missing — the") print(" open-world assumption. It can never report an incomplete record.") print(" SHACL closes the world for validation and reports exactly that.") print(" Note ent:Sanctioned and ent:Meridian were validated because they are") print(" LegalEntity by ENTAILMENT, not by assertion.") print() print("=" * 78) print("4. SPARQL — PROPERTY PATHS EARN THEIR KEEP") print("=" * 78) query = Query( select=("owner", "name"), where=( Pattern(Variable("owner"), PathStep("bank:controls", "+"), "ent:Acme"), Optional_((Pattern(Variable("owner"), PathStep("bank:legalName"), Variable("name")),)), )) print(" SELECT ?owner ?name WHERE { ?owner bank:controls+ ent:Acme .") print(" OPTIONAL { ?owner bank:legalName ?name } }") for row in execute(graph, query): print(f" {row['owner']:<16} {row['name']}") print() print(" the question a bank actually asks:") sanctioned = Query( select=("owner",), where=( Pattern(Variable("owner"), PathStep("bank:controls", "+"), "ent:Acme"), Pattern(Variable("owner"), PathStep("bank:onSanctionsList"), Literal("true", "xsd:boolean")), )) rows = execute(graph, sanctioned) print(" SELECT ?owner WHERE { ?owner bank:controls+ ent:Acme ;") print(" bank:onSanctionsList true }") print(f" -> {[r['owner'] for r in rows] or 'no sanctioned controller'}") print(" (Sanctioned controls Meridian controls Northgate controls Acme:") print(" three hops, one line of SPARQL, and no vector index can answer it.)") print() print(" OPTIONAL is a LEFT join — a missing name does not remove the row:") for row in execute(graph, query): if not row["name"]: print(f" {row['owner']:<16} (no legalName asserted)") print() print(" FILTER, and inverse paths (^):") filtered = Query( select=("obligation", "amount"), where=( Pattern(Variable("obligation"), PathStep("bank:obligor"), "ent:Acme"), Pattern(Variable("obligation"), PathStep("bank:amountAED"), Variable("amount")), Filter(lambda b: int(str(b["amount"])) > 100_000, "amount > 100000"), )) for row in execute(graph, filtered): print(f" {row['obligation']} amount {row['amount']}") inverse = Query( select=("obligation",), where=(Pattern("ent:Acme", PathStep("bank:obligor", inverse=True), Variable("obligation")),)) print(f" ent:Acme ^bank:obligor -> {[r['obligation'] for r in execute(graph, inverse)]}") print() print("=" * 78) print("5. SYMMETRY, INVERSES AND CYCLES") print("=" * 78) sym = graph.match(subject=prefixes.expand("ent:Northgate"), predicate=prefixes.expand("bank:isCounterpartyOf")) print(f" asserted Acme isCounterpartyOf Northgate; entailed the reverse: " f"{[prefixes.shorten(t.object) for t in sym]}") controlled_by = graph.match(subject=prefixes.expand("ent:Acme"), predicate=prefixes.expand("bank:controlledBy")) print(f" inverseOf gave ent:Acme controlledBy " f"{[prefixes.shorten(t.object) for t in controlled_by]}") cyclic = Graph() cyclic.add_curies("bank:controls", "rdf:type", "owl:TransitiveProperty") cyclic.add_curies("ent:A", "bank:controls", "ent:B") cyclic.add_curies("ent:B", "bank:controls", "ent:A") materialize(cyclic) reachable = execute(cyclic, Query( select=("x",), where=(Pattern("ent:A", PathStep("bank:controls", "+"), Variable("x")),))) print(f" circular shareholding A<->B terminates: reachable = " f"{[r['x'] for r in reachable]}") print(" (real ownership graphs contain cycles; an unguarded walk does not return.)") print() print("=" * 78) print("6. GRAPH-GROUNDED RETRIEVAL") print("=" * 78) context = expand_neighbourhood(graph, "ent:Acme", hops=2, predicates=("bank:controlledBy", "bank:legalName", "bank:hasLEI")) print(f" rationale: {context.rationale}") print(f" entities : {list(context.entities)}") for path in context.paths: print(f" {path}") print() print(" A vector index retrieves text that RESEMBLES the question.") print(" This retrieves entities STRUCTURALLY related to it — which is the right") print(" operation when the answer is three hops away in a document that shares no") print(" vocabulary with the question.") if __name__ == "__main__": # pragma: no cover main()