From 34edef0aa3def16a0250abef0eccab29f26b110e Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 06:41:20 -0400 Subject: [PATCH 1/9] feat(core): public vocab accessors + GraphError for the entity graph Co-Authored-By: Claude Fable 5 --- src/energex/core/connectors/ercot.py | 6 +++++ src/energex/core/exceptions.py | 4 ++++ src/energex/core/symbology.py | 12 ++++++++++ tests/test_graph_vocab.py | 35 ++++++++++++++++++++++++++++ 4 files changed, 57 insertions(+) create mode 100644 tests/test_graph_vocab.py diff --git a/src/energex/core/connectors/ercot.py b/src/energex/core/connectors/ercot.py index 37b52a6..0b346ff 100644 --- a/src/energex/core/connectors/ercot.py +++ b/src/energex/core/connectors/ercot.py @@ -61,6 +61,12 @@ ) +def settlement_points() -> frozenset[str]: + """The canonical tradeable settlement points (hubs + load zones) this + connector ingests. Public so the entity graph never reaches into privates.""" + return _SETTLEMENT_POINTS + + def _is_retryable(exc: BaseException) -> bool: """Retry only transient failures: transport/timeout errors and 5xx responses. Never retry 4xx (bad creds, bad request) — retrying those just hammers ERCOT's auth/API and risks diff --git a/src/energex/core/exceptions.py b/src/energex/core/exceptions.py index c75d4fc..6731910 100644 --- a/src/energex/core/exceptions.py +++ b/src/energex/core/exceptions.py @@ -62,5 +62,9 @@ class SymbologyError(EnergexError): """Raised when an instrument_id cannot be resolved or its mode is inconsistent.""" +class GraphError(EnergexError): + """Raised on Neo4j entity-graph connection or sync failures.""" + + class PartitionError(EnergexError): """Raised when a Dagster partition key cannot be mapped to a valid_time range.""" diff --git a/src/energex/core/symbology.py b/src/energex/core/symbology.py index 6af1419..dc1d7e3 100644 --- a/src/energex/core/symbology.py +++ b/src/energex/core/symbology.py @@ -138,3 +138,15 @@ def mode_for_library(library: str) -> str: return LIBRARY_MODE[library] except KeyError as exc: raise SymbologyError(f"unknown library {library!r}") from exc + + +def instruments() -> list[str]: + """The static instrument_id universe (a copy). Rule-based power ids + (EIA930.*/ERCOT.*) are not enumerable here — discover them from the store + and route via power_prefixes().""" + return list(_TABLE) + + +def power_prefixes() -> dict[str, tuple[str, str]]: + """prefix -> (library, revision_mode) for the rule-based power namespace (a copy).""" + return dict(_POWER_PREFIX) diff --git a/tests/test_graph_vocab.py b/tests/test_graph_vocab.py new file mode 100644 index 0000000..706e2b4 --- /dev/null +++ b/tests/test_graph_vocab.py @@ -0,0 +1,35 @@ +"""Public vocabulary accessors the entity graph consumes (never private dicts).""" + +from energex.core import symbology +from energex.core.connectors import ercot +from energex.core.exceptions import EnergexError, GraphError + + +def test_symbology_instruments_enumerates_static_table(): + ids = symbology.instruments() + assert "EIA.NG.STORAGE.LOWER48" in ids + assert "FRED.WTI.SPOT" in ids + assert "NOAA.HDD.TEXAS" in ids + # rule-based power ids are NOT in the static universe + assert not any(i.startswith("EIA930.") for i in ids) + # accessor returns a copy: mutating it must not corrupt the table + ids.clear() + assert symbology.instruments() + + +def test_symbology_power_prefixes_maps_prefix_to_library_and_mode(): + prefixes = symbology.power_prefixes() + assert prefixes["EIA930.D"] == ("power.demand", "degenerate") + assert prefixes["ERCOT.SPP"] == ("power.lmp", "bitemporal_merge") + prefixes.clear() + assert symbology.power_prefixes() + + +def test_ercot_settlement_points_public_accessor(): + points = ercot.settlement_points() + assert "HB_NORTH" in points and "LZ_HOUSTON" in points + assert len(points) == 13 + + +def test_graph_error_is_energex_error(): + assert issubclass(GraphError, EnergexError) From d27244191d82a4ac40a829f70221fd49889b0e3a Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 06:43:24 -0400 Subject: [PATCH 2/9] feat(core): entity-graph plan builder (pure derivation from vocabularies) Co-Authored-By: Claude Fable 5 --- src/energex/core/graph.py | 451 ++++++++++++++++++++++++++++++++++++++ tests/test_graph_plan.py | 98 +++++++++ 2 files changed, 549 insertions(+) create mode 100644 src/energex/core/graph.py create mode 100644 tests/test_graph_plan.py diff --git a/src/energex/core/graph.py b/src/energex/core/graph.py new file mode 100644 index 0000000..e0299a8 --- /dev/null +++ b/src/energex/core/graph.py @@ -0,0 +1,451 @@ +"""Neo4j entity graph (phase 9): the what/who/connected layer over the catalog. + +Numbers live in ArcticDB; this module mirrors IDENTITY only — instrument_ids +(symbology), the entities those ids imply (balancing authorities, ERCOT +settlement points, NOAA regions, commodities, fuel types) and a small curated +set of cross-domain edges. Writes are idempotent MERGE upserts stamped with +first_seen/last_seen knowledge times, so the graph restores independently of +the store of record (operations doc) and never misrepresents when an entity +entered the catalog. + +Pure by construction: the neo4j driver is imported lazily inside +create_driver() only; plan building and Cypher generation are side-effect-free, +and sync/query helpers accept any object exposing execute_query() (duck-typed) +so tests inject fakes and installs without the `graph` extra never break. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +from energex.core import symbology +from energex.core.config import Neo4jConfig +from energex.core.connectors import ercot +from energex.core.connectors.weather import REGIONS +from energex.core.exceptions import GraphError + +# Label -> unique key property. The only labels sync will ever create; Cypher +# labels cannot be parameterized, so they are always drawn from this dict. +KEY_PROPERTY: dict[str, str] = { + "Source": "name", + "Library": "name", + "Instrument": "instrument_id", + "Market": "code", + "BalancingAuthority": "code", + "SettlementPoint": "code", + "Region": "code", + "Commodity": "code", + "FuelType": "code", +} + +# Relationship types traversed by /graph/related. IN_LIBRARY and FROM_SOURCE are +# deliberately excluded: they are catalog plumbing, and traversing through those +# hub nodes would relate every instrument to every other one. +SEMANTIC_RELS: tuple[str, ...] = ( + "MEASURES", + "IN_MARKET", + "OPERATES", + "WEATHER_PROXY_FOR", + "GENERATES", + "FUELS", +) + +# instrument_id prefix -> connector source string (each connector's _SOURCE). +_PREFIX_SOURCE: dict[str, str] = { + "EIA930": "eia930", + "ERCOT": "ercot", + "EIA": "eia", + "FRED": "fred", + "NOAA": "noaa", + "CME": "yfinance", +} + +# library -> instrument kind (measurement family). Keys mirror symbology.LIBRARY_MODE. +_LIBRARY_KIND: dict[str, str] = { + "fundamentals.eia": "fundamentals", + "weather": "degree_days", + "prices.spot": "spot", + "prices.intraday": "intraday", + "prices.futures": "futures", + "power.demand": "demand", + "power.demand_forecast": "demand_forecast", + "power.generation": "net_generation", + "power.interchange": "interchange", + "power.generation_by_fuel": "generation_by_fuel", + "power.lmp": "lmp", + "power.load": "load", + "power.dalmp": "dalmp", +} + +_COMMODITIES: dict[str, str] = { + "WTI": "WTI crude oil", + "BRENT": "Brent crude oil", + "NATGAS": "Natural gas", +} + +# Static instrument -> the commodity it measures. EIA.PET.CRUDE.STOCKS maps to +# WTI as the US crude benchmark its stocks move. +_STATIC_COMMODITY: dict[str, str] = { + "FRED.WTI.SPOT": "WTI", + "FRED.BRENT.SPOT": "BRENT", + "FRED.HENRYHUB.SPOT": "NATGAS", + "CME.CL.FRONT": "WTI", + "CME.BZ.FRONT": "BRENT", + "CME.NG.FRONT": "NATGAS", + "CME.CL.CLF26": "WTI", + "CME.CL.CLG26": "WTI", + "EIA.NG.STORAGE.LOWER48": "NATGAS", + "EIA.PET.CRUDE.STOCKS": "WTI", +} + +# EIA930 instrument prefixes per BA (all five families exist for every BA). +_EIA930_PREFIXES: tuple[str, ...] = ( + "EIA930.D", + "EIA930.DF", + "EIA930.NG", + "EIA930.TI", + "EIA930.GEN_FUEL", +) + + +@dataclass(frozen=True) +class GraphNode: + label: str + key: str + properties: Mapping[str, str] = field(default_factory=dict) + + +@dataclass(frozen=True) +class GraphEdge: + rel_type: str + src: tuple[str, str] # (label, key) + dst: tuple[str, str] + + +@dataclass(frozen=True) +class GraphPlan: + nodes: tuple[GraphNode, ...] + edges: tuple[GraphEdge, ...] + + +@dataclass(frozen=True) +class ObservedEntities: + """Store-discovered vocabulary (the repo deliberately hardcodes no BA/fuel + lists): BA codes from power.demand symbols, fuel types per BA from + power.generation_by_fuel frames. Uppercase codes.""" + + balancing_authorities: tuple[str, ...] = () + fuel_types_by_ba: Mapping[str, tuple[str, ...]] = field(default_factory=dict) + + +@dataclass(frozen=True) +class GraphSyncResult: + nodes_by_label: dict[str, int] + edges_by_type: dict[str, int] + + +class _PlanBuilder: + def __init__(self) -> None: + self._nodes: dict[tuple[str, str], dict[str, str]] = {} + self._edges: dict[tuple[str, tuple[str, str], tuple[str, str]], None] = {} + + def node(self, label: str, key: str, **props: str) -> tuple[str, str]: + merged = self._nodes.setdefault((label, key), {}) + merged.update(props) + return (label, key) + + def edge(self, rel_type: str, src: tuple[str, str], dst: tuple[str, str]) -> None: + self._edges[(rel_type, src, dst)] = None + + def build(self) -> GraphPlan: + nodes = tuple( + GraphNode(label=label, key=key, properties=dict(props)) + for (label, key), props in sorted(self._nodes.items()) + ) + edges = tuple( + GraphEdge(rel_type=rel, src=src, dst=dst) for (rel, src, dst) in sorted(self._edges) + ) + return GraphPlan(nodes=nodes, edges=edges) + + +def _source_for(instrument_id: str) -> str: + head = instrument_id.split(".", 1)[0] + # EIA930 shares the EIA head only lexically; exact head match keeps them apart. + return _PREFIX_SOURCE[head] + + +def _instrument(builder: _PlanBuilder, instrument_id: str) -> tuple[str, str]: + library, symbol = symbology.resolve(instrument_id) + node = builder.node("Instrument", instrument_id, symbol=symbol, kind=_LIBRARY_KIND[library]) + builder.edge("IN_LIBRARY", node, builder.node("Library", library)) + builder.edge("FROM_SOURCE", node, builder.node("Source", _source_for(instrument_id))) + return node + + +def build_entity_graph(observed: ObservedEntities | None = None) -> GraphPlan: + """Derive the full entity catalog. Pure: no I/O, no driver.""" + observed = observed or ObservedEntities() + b = _PlanBuilder() + + for library, mode in symbology.LIBRARY_MODE.items(): + b.node("Library", library, revision_mode=mode) + for source in sorted(set(_PREFIX_SOURCE.values())): + b.node("Source", source) + for code, name in _COMMODITIES.items(): + b.node("Commodity", code, name=name) + + market = b.node("Market", "ERCOT", name="Electric Reliability Council of Texas") + for point in sorted(ercot.settlement_points()): + kind = "hub" if point.startswith("HB_") else "load_zone" + b.edge("IN_MARKET", b.node("SettlementPoint", point, kind=kind), market) + b.edge("MEASURES", _instrument(b, f"ERCOT.SPP.{point}"), ("SettlementPoint", point)) + b.edge("MEASURES", _instrument(b, f"ERCOT.DASPP.{point}"), ("SettlementPoint", point)) + # ERCOT.LOAD.ERCOT's tail is the market-wide aggregate, not a settlement point. + b.edge("MEASURES", _instrument(b, "ERCOT.LOAD.ERCOT"), market) + + for nclimdiv_code, token in REGIONS.items(): + b.node("Region", token, nclimdiv_code=nclimdiv_code) + + for instrument_id in symbology.instruments(): + node = _instrument(b, instrument_id) + if instrument_id.startswith("NOAA.HDD."): + token = instrument_id.rpartition(".")[2] + b.edge("MEASURES", node, ("Region", token)) + commodity = _STATIC_COMMODITY.get(instrument_id) + if commodity is not None: + b.edge("MEASURES", node, ("Commodity", commodity)) + + # ERCO is seeded statically: it is the join point between the EIA-930 BA + # universe and the ERCOT nodal universe, whether or not it was observed yet. + bas = sorted(set(observed.balancing_authorities) | {"ERCO"}) + for ba in bas: + ba_node = b.node("BalancingAuthority", ba) + for prefix in _EIA930_PREFIXES: + b.edge("MEASURES", _instrument(b, f"{prefix}.{ba}"), ba_node) + + fuel_types: set[str] = set() + for ba, fuels in sorted(observed.fuel_types_by_ba.items()): + for fuel in sorted(set(fuels)): + fuel_types.add(fuel) + b.edge("GENERATES", ("BalancingAuthority", ba), b.node("FuelType", fuel)) + + # Curated cross-domain edges. + b.edge("OPERATES", ("BalancingAuthority", "ERCO"), market) + b.edge("WEATHER_PROXY_FOR", ("Region", "TEXAS"), market) + if "NG" in fuel_types: # gas-fired generation links power to the gas complex + b.edge("FUELS", ("Commodity", "NATGAS"), ("FuelType", "NG")) + + return b.build() + + +def constraint_statements() -> list[str]: + """One uniqueness constraint per label key; IF NOT EXISTS keeps it idempotent.""" + return [ + ( + f"CREATE CONSTRAINT {label.lower()}_{key}_unique IF NOT EXISTS " + f"FOR (n:{label}) REQUIRE n.{key} IS UNIQUE" + ) + for label, key in KEY_PROPERTY.items() + ] + + +def plan_to_cypher(plan: GraphPlan, *, synced_at: datetime) -> list[tuple[str, dict[str, Any]]]: + """Batched idempotent upserts: one UNWIND+MERGE statement per node label and + per (rel_type, src_label, dst_label). Labels/rel-types come only from our + fixed vocabulary (Cypher cannot parameterize them); values are parameters.""" + statements: list[tuple[str, dict[str, Any]]] = [] + + by_label: dict[str, list[GraphNode]] = {} + for node in plan.nodes: + if node.label not in KEY_PROPERTY: + raise GraphError(f"unknown label {node.label!r}") + by_label.setdefault(node.label, []).append(node) + for label, nodes in by_label.items(): + key = KEY_PROPERTY[label] + node_rows = [{"key": n.key, "props": dict(n.properties)} for n in nodes] + statements.append( + ( + f"UNWIND $rows AS row\n" + f"MERGE (n:{label} {{{key}: row.key}})\n" + f"ON CREATE SET n.first_seen = $synced_at\n" + f"SET n += row.props, n.last_seen = $synced_at", + {"rows": node_rows, "synced_at": synced_at}, + ) + ) + + by_shape: dict[tuple[str, str, str], list[GraphEdge]] = {} + for edge in plan.edges: + by_shape.setdefault((edge.rel_type, edge.src[0], edge.dst[0]), []).append(edge) + for (rel_type, src_label, dst_label), edges in by_shape.items(): + if src_label not in KEY_PROPERTY or dst_label not in KEY_PROPERTY: + raise GraphError(f"unknown label in edge {rel_type!r}") + src_key, dst_key = KEY_PROPERTY[src_label], KEY_PROPERTY[dst_label] + edge_rows = [{"src": e.src[1], "dst": e.dst[1]} for e in edges] + statements.append( + ( + f"UNWIND $rows AS row\n" + f"MERGE (a:{src_label} {{{src_key}: row.src}})\n" + f"MERGE (b:{dst_label} {{{dst_key}: row.dst}})\n" + f"MERGE (a)-[r:{rel_type}]->(b)\n" + f"ON CREATE SET r.first_seen = $synced_at\n" + f"SET r.last_seen = $synced_at", + {"rows": edge_rows, "synced_at": synced_at}, + ) + ) + return statements + + +def sync_graph( + driver: Any, + plan: GraphPlan, + *, + synced_at: datetime | None = None, + database: str | None = None, +) -> GraphSyncResult: + """Idempotent MERGE upsert of the whole plan. ``driver`` is duck-typed + (anything with execute_query). Errors are wrapped redacted — exception type + only, never the message, which may embed a connection URI.""" + stamp = synced_at or datetime.now(timezone.utc) + try: + for statement in constraint_statements(): + driver.execute_query(statement, parameters_={}, database_=database) + for query, params in plan_to_cypher(plan, synced_at=stamp): + driver.execute_query(query, parameters_=params, database_=database) + except GraphError: + raise + except Exception as exc: + raise GraphError(f"entity-graph sync failed: {type(exc).__name__}") from None + + nodes_by_label: dict[str, int] = {} + for node in plan.nodes: + nodes_by_label[node.label] = nodes_by_label.get(node.label, 0) + 1 + edges_by_type: dict[str, int] = {} + for edge in plan.edges: + edges_by_type[edge.rel_type] = edges_by_type.get(edge.rel_type, 0) + 1 + return GraphSyncResult(nodes_by_label=nodes_by_label, edges_by_type=edges_by_type) + + +def _redact_uri(uri: str) -> str: + """Strip any userinfo from a bolt/neo4j URI before it can reach a log line.""" + scheme, sep, rest = uri.partition("://") + if sep and "@" in rest: + rest = rest.rsplit("@", 1)[1] + return f"{scheme}{sep}{rest}" + + +def create_driver(cfg: Neo4jConfig) -> Any: + """The ONLY place the neo4j driver is imported (lazily): installs without the + `graph` extra can import this module freely and get a GraphError only when + they actually try to connect.""" + try: + from neo4j import GraphDatabase + except ImportError as exc: + raise GraphError("neo4j driver not installed — install the 'graph' extra") from exc + password = cfg.password.get_secret_value() if cfg.password else "" + try: + # 5s connect cap keeps optional-graph startups snappy when no server runs. + driver = GraphDatabase.driver(cfg.uri, auth=(cfg.user, password), connection_timeout=5.0) + driver.verify_connectivity() + except Exception as exc: + raise GraphError( + f"Neo4j connection failed (uri={_redact_uri(cfg.uri)}): {type(exc).__name__}" + ) from None + return driver + + +def _clean_value(value: Any) -> Any: + if hasattr(value, "to_native"): # neo4j.time.DateTime and friends + value = value.to_native() + if isinstance(value, datetime): + return value.isoformat() + return value + + +def _clean_props(props: Mapping[str, Any]) -> dict[str, Any]: + return {k: _clean_value(v) for k, v in props.items()} + + +def _keyed_case() -> str: + """CASE arm extracting each label's key property (labels are code-fixed).""" + return " ".join(f"WHEN '{label}' THEN n.{key}" for label, key in KEY_PROPERTY.items()) + + +def list_entities( + driver: Any, *, label: str | None = None, database: str | None = None +) -> list[dict[str, Any]]: + """Catalog listing for the S2 seam: [{label, key, properties}, ...].""" + if label is not None and label not in KEY_PROPERTY: + raise GraphError(f"unknown label {label!r}") + match = f"MATCH (n:{label})" if label else "MATCH (n)" + query = ( + f"{match}\n" + f"RETURN labels(n)[0] AS label,\n" + f" CASE labels(n)[0] {_keyed_case()} END AS key,\n" + f" properties(n) AS properties\n" + f"ORDER BY label, key" + ) + try: + result = driver.execute_query(query, parameters_={}, database_=database) + except Exception as exc: + raise GraphError(f"entity-graph query failed: {type(exc).__name__}") from None + return [ + { + "label": rec["label"], + "key": rec["key"], + "properties": _clean_props(rec["properties"]), + } + for rec in result.records + ] + + +def related_instruments( + driver: Any, + instrument_id: str, + *, + depth: int = 2, + database: str | None = None, +) -> dict[str, Any] | None: + """Neighbors of an instrument through SEMANTIC_RELS only (never through the + Library/Source hubs). depth 1 = the entities it measures; 2 adds sibling + instruments and adjacent entities. Returns None when the id is not in the + graph.""" + if depth not in (1, 2, 3): + raise GraphError(f"depth must be 1..3, got {depth!r}") + try: + exists = driver.execute_query( + "MATCH (i:Instrument {instrument_id: $iid}) RETURN i.instrument_id", + parameters_={"iid": instrument_id}, + database_=database, + ) + if not exists.records: + return None + rels = "|".join(SEMANTIC_RELS) + # depth is validated above; rel types and labels are code-fixed vocabulary. + query = ( + f"MATCH (i:Instrument {{instrument_id: $iid}})\n" + f"MATCH (i)-[:{rels}*1..{depth}]-(n)\n" + f"WHERE n <> i\n" + f"WITH DISTINCT n\n" + f"RETURN labels(n)[0] AS label,\n" + f" CASE labels(n)[0] {_keyed_case()} END AS key,\n" + f" properties(n) AS properties\n" + f"ORDER BY label, key" + ) + result = driver.execute_query(query, parameters_={"iid": instrument_id}, database_=database) + except Exception as exc: + raise GraphError(f"entity-graph query failed: {type(exc).__name__}") from None + return { + "instrument_id": instrument_id, + "depth": depth, + "related": [ + { + "label": rec["label"], + "key": rec["key"], + "properties": _clean_props(rec["properties"]), + } + for rec in result.records + ], + } diff --git a/tests/test_graph_plan.py b/tests/test_graph_plan.py new file mode 100644 index 0000000..6c3b837 --- /dev/null +++ b/tests/test_graph_plan.py @@ -0,0 +1,98 @@ +"""build_entity_graph: pure derivation of the entity catalog from vocabularies.""" + +from energex.core import graph + + +def _plan(observed=None): + return graph.build_entity_graph(observed or graph.ObservedEntities()) + + +def _node(plan, label, key): + return next((n for n in plan.nodes if n.label == label and n.key == key), None) + + +def _edges(plan, rel_type): + return [(e.src, e.dst) for e in plan.edges if e.rel_type == rel_type] + + +def test_static_catalog_nodes_exist_without_observations(): + plan = _plan() + assert _node(plan, "Library", "power.lmp").properties["revision_mode"] == "bitemporal_merge" + assert _node(plan, "Source", "fred") is not None + assert _node(plan, "Market", "ERCOT") is not None + assert _node(plan, "SettlementPoint", "HB_NORTH").properties["kind"] == "hub" + assert _node(plan, "SettlementPoint", "LZ_HOUSTON").properties["kind"] == "load_zone" + assert _node(plan, "Region", "TEXAS").properties["nclimdiv_code"] == "041" + assert _node(plan, "Commodity", "NATGAS") is not None + # ERCO is seeded statically: it is the EIA-930 <-> ERCOT join point + assert _node(plan, "BalancingAuthority", "ERCO") is not None + + +def test_static_instruments_routed_via_symbology(): + plan = _plan() + wti = _node(plan, "Instrument", "FRED.WTI.SPOT") + assert wti.properties["symbol"] == "wti_spot" + assert (("Instrument", "FRED.WTI.SPOT"), ("Library", "prices.spot")) in _edges( + plan, "IN_LIBRARY" + ) + assert (("Instrument", "FRED.WTI.SPOT"), ("Source", "fred")) in _edges(plan, "FROM_SOURCE") + assert (("Instrument", "FRED.WTI.SPOT"), ("Commodity", "WTI")) in _edges(plan, "MEASURES") + + +def test_ercot_static_instruments_and_market_aggregate(): + plan = _plan() + assert (("Instrument", "ERCOT.SPP.HB_NORTH"), ("SettlementPoint", "HB_NORTH")) in _edges( + plan, "MEASURES" + ) + assert _node(plan, "Instrument", "ERCOT.DASPP.LZ_WEST") is not None + # ERCOT.LOAD.ERCOT measures the MARKET aggregate, not a settlement point + assert (("Instrument", "ERCOT.LOAD.ERCOT"), ("Market", "ERCOT")) in _edges(plan, "MEASURES") + assert (("SettlementPoint", "HB_NORTH"), ("Market", "ERCOT")) in _edges(plan, "IN_MARKET") + + +def test_observed_bas_produce_all_five_eia930_families(): + plan = _plan(graph.ObservedEntities(balancing_authorities=("MISO",))) + for prefix, kind in [ + ("EIA930.D", "demand"), + ("EIA930.DF", "demand_forecast"), + ("EIA930.NG", "net_generation"), + ("EIA930.TI", "interchange"), + ("EIA930.GEN_FUEL", "generation_by_fuel"), + ]: + node = _node(plan, "Instrument", f"{prefix}.MISO") + assert node is not None and node.properties["kind"] == kind + assert (("Instrument", "EIA930.D.MISO"), ("BalancingAuthority", "MISO")) in _edges( + plan, "MEASURES" + ) + assert ( + ("Instrument", "EIA930.GEN_FUEL.MISO"), + ("Library", "power.generation_by_fuel"), + ) in _edges(plan, "IN_LIBRARY") + + +def test_fuel_types_and_generates_edges(): + plan = _plan( + graph.ObservedEntities( + balancing_authorities=("ERCO",), fuel_types_by_ba={"ERCO": ("NG", "WND")} + ) + ) + assert _node(plan, "FuelType", "WND") is not None + assert (("BalancingAuthority", "ERCO"), ("FuelType", "NG")) in _edges(plan, "GENERATES") + # curated cross-domain edge appears only when fuel NG is observed + assert (("Commodity", "NATGAS"), ("FuelType", "NG")) in _edges(plan, "FUELS") + assert not _edges(_plan(), "FUELS") + + +def test_curated_cross_domain_edges(): + plan = _plan() + assert (("BalancingAuthority", "ERCO"), ("Market", "ERCOT")) in _edges(plan, "OPERATES") + assert (("Region", "TEXAS"), ("Market", "ERCOT")) in _edges(plan, "WEATHER_PROXY_FOR") + + +def test_plan_is_deduplicated_and_all_edge_endpoints_exist(): + plan = _plan(graph.ObservedEntities(balancing_authorities=("ERCO", "MISO"))) + keys = [(n.label, n.key) for n in plan.nodes] + assert len(keys) == len(set(keys)) + node_set = set(keys) + for e in plan.edges: + assert e.src in node_set and e.dst in node_set From ee2afafa0d29cd76018ee68c8ad2220b36536c49 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 06:44:03 -0400 Subject: [PATCH 3/9] test(core): entity-graph sync, driver factory, and read-query contracts Co-Authored-By: Claude Fable 5 --- tests/test_graph_sync.py | 122 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 tests/test_graph_sync.py diff --git a/tests/test_graph_sync.py b/tests/test_graph_sync.py new file mode 100644 index 0000000..2dab606 --- /dev/null +++ b/tests/test_graph_sync.py @@ -0,0 +1,122 @@ +"""sync_graph / create_driver / read queries against a duck-typed fake driver.""" + +from datetime import datetime, timezone +from types import SimpleNamespace + +import pytest + +from energex.core import graph +from energex.core.config import Neo4jConfig +from energex.core.exceptions import GraphError + +SYNCED_AT = datetime(2026, 7, 15, 12, 0, tzinfo=timezone.utc) + + +class FakeDriver: + def __init__(self, results=None, fail=False): + self.calls: list[tuple[str, dict, str | None]] = [] + self._results = results or {} + self._fail = fail + + def execute_query(self, query_, parameters_=None, database_=None, **_kw): + if self._fail: + raise RuntimeError("bolt://user:secret@host exploded") + self.calls.append((query_, dict(parameters_ or {}), database_)) + for needle, records in self._results.items(): + if needle in query_: + return SimpleNamespace(records=records, summary=None, keys=[]) + return SimpleNamespace(records=[], summary=None, keys=[]) + + def close(self): + pass + + +def _sync(driver): + plan = graph.build_entity_graph() + return graph.sync_graph(driver, plan, synced_at=SYNCED_AT) + + +def test_sync_runs_constraints_then_merges_and_counts(): + driver = FakeDriver() + result = _sync(driver) + queries = [q for q, _, _ in driver.calls] + n_constraints = sum("CREATE CONSTRAINT" in q for q in queries) + assert n_constraints == len(graph.KEY_PROPERTY) + # constraints run before any MERGE batch + first_merge = next(i for i, q in enumerate(queries) if "UNWIND" in q) + assert all("CREATE CONSTRAINT" in q for q in queries[:first_merge]) + # every batch is an idempotent MERGE; nothing uses bare CREATE + assert all("CREATE (" not in q for q in queries) + assert result.nodes_by_label["SettlementPoint"] == 13 + assert result.edges_by_type["IN_MARKET"] == 13 + + +def test_sync_stamps_first_and_last_seen(): + driver = FakeDriver() + _sync(driver) + merges = [(q, p) for q, p, _ in driver.calls if "UNWIND" in q] + assert merges + for q, params in merges: + assert "ON CREATE SET" in q and "first_seen" in q and "last_seen" in q + assert params["synced_at"] == SYNCED_AT + + +def test_sync_failure_is_wrapped_and_redacted(): + with pytest.raises(GraphError) as exc_info: + _sync(FakeDriver(fail=True)) + assert "secret" not in str(exc_info.value) + assert "RuntimeError" in str(exc_info.value) + + +def test_create_driver_without_neo4j_extra_raises_graph_error(monkeypatch): + import builtins + + real_import = builtins.__import__ + + def no_neo4j(name, *args, **kwargs): + if name == "neo4j" or name.startswith("neo4j."): + raise ImportError("No module named 'neo4j'") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", no_neo4j) + with pytest.raises(GraphError, match="graph.*extra"): + graph.create_driver(Neo4jConfig()) + + +def test_list_entities_validates_label_and_cleans_records(): + records = [ + { + "label": "Instrument", + "key": "FRED.WTI.SPOT", + "properties": {"instrument_id": "FRED.WTI.SPOT", "first_seen": SYNCED_AT}, + } + ] + driver = FakeDriver(results={"MATCH": records}) + rows = graph.list_entities(driver, label="Instrument") + assert rows[0]["key"] == "FRED.WTI.SPOT" + assert rows[0]["properties"]["first_seen"] == SYNCED_AT.isoformat() + with pytest.raises(GraphError, match="unknown label"): + graph.list_entities(driver, label="DropAllTables") + + +def test_related_instruments_unknown_returns_none_and_depth_validated(): + driver = FakeDriver() # no records -> instrument not found + assert graph.related_instruments(driver, "NOPE.X") is None + with pytest.raises(GraphError, match="depth"): + graph.related_instruments(driver, "FRED.WTI.SPOT", depth=9) + + +def test_related_instruments_returns_cleaned_neighbors(): + exists = [{"instrument_id": "ERCOT.SPP.HB_NORTH"}] + related = [ + {"label": "SettlementPoint", "key": "HB_NORTH", "properties": {"code": "HB_NORTH"}}, + { + "label": "Instrument", + "key": "ERCOT.DASPP.HB_NORTH", + "properties": {"instrument_id": "ERCOT.DASPP.HB_NORTH"}, + }, + ] + driver = FakeDriver(results={"RETURN i.instrument_id": exists, "DISTINCT": related}) + out = graph.related_instruments(driver, "ERCOT.SPP.HB_NORTH", depth=2) + assert out["instrument_id"] == "ERCOT.SPP.HB_NORTH" + assert {r["key"] for r in out["related"]} == {"HB_NORTH", "ERCOT.DASPP.HB_NORTH"} From 4d65c71043dede41c8b46b7558e1c0d13a7d3076 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 06:46:18 -0400 Subject: [PATCH 4/9] feat(orchestration): Neo4jResource + entity_graph asset, check, daily schedule Co-Authored-By: Claude Fable 5 --- src/energex/orchestration/definitions.py | 7 +- src/energex/orchestration/graph.py | 147 +++++++++++++++++++++++ src/energex/orchestration/resources.py | 37 ++++++ tests/test_definitions_load.py | 2 + tests/test_graph_asset.py | 120 ++++++++++++++++++ 5 files changed, 310 insertions(+), 3 deletions(-) create mode 100644 src/energex/orchestration/graph.py create mode 100644 tests/test_graph_asset.py diff --git a/src/energex/orchestration/definitions.py b/src/energex/orchestration/definitions.py index be56520..416b8ac 100644 --- a/src/energex/orchestration/definitions.py +++ b/src/energex/orchestration/definitions.py @@ -4,15 +4,16 @@ from energex.orchestration.assets import ASSETS from energex.orchestration.checks import CHECKS +from energex.orchestration.graph import GRAPH_ASSETS, GRAPH_CHECKS, GRAPH_SCHEDULES from energex.orchestration.reconcile import RECONCILE_ASSETS from energex.orchestration.resources import RESOURCES from energex.orchestration.schedules import SCHEDULES from energex.orchestration.sensors import SENSORS defs = dg.Definitions( - assets=[*ASSETS, *RECONCILE_ASSETS], - asset_checks=CHECKS, - schedules=SCHEDULES, + assets=[*ASSETS, *RECONCILE_ASSETS, *GRAPH_ASSETS], + asset_checks=[*CHECKS, *GRAPH_CHECKS], + schedules=[*SCHEDULES, *GRAPH_SCHEDULES], sensors=SENSORS, resources=RESOURCES, ) diff --git a/src/energex/orchestration/graph.py b/src/energex/orchestration/graph.py new file mode 100644 index 0000000..5ed6c21 --- /dev/null +++ b/src/energex/orchestration/graph.py @@ -0,0 +1,147 @@ +"""Entity-graph sync (phase 9): project the instrument catalog into Neo4j. + +Discovery is store-driven (the repo deliberately hardcodes no BA/fuel lists): +balancing authorities come from ``power.demand`` symbols and fuel types from a +bounded tail read of ``power.generation_by_fuel``. The sync is an idempotent +MERGE upsert; ArcticDB remains the store of record and the graph never carries +a number. Missing power libraries degrade to the static catalog. +""" + +from typing import Any + +# arcticdb MUST load before pandas/pyarrow (phase-0 AWS-SDK load-order hazard). +import arcticdb # noqa: F401 +import dagster as dg + +from energex.core import graph, symbology +from energex.core.exceptions import SymbologyError +from energex.orchestration.resources import ArcticDBResource, Neo4jResource + +# Rows per generation_by_fuel tail read: ~hourly x ~10 fuels x 3 days, rounded up. +_FUEL_TAIL_ROWS = 1_000 + + +def _observed_bas(arctic: ArcticDBResource) -> tuple[str, ...]: + """BA codes = power.demand symbols (lowercased BA codes), uppercased back.""" + try: + symbols = arctic.get_library("power.demand").list_symbols() + except Exception: + return () + return tuple(sorted(s.upper() for s in symbols)) + + +def _observed_fuel_types(arctic: ArcticDBResource) -> dict[str, tuple[str, ...]]: + """fuel_type values seen in the recent tail of each generation_by_fuel symbol.""" + try: + lib = arctic.get_library("power.generation_by_fuel") + symbols = lib.list_symbols() + except Exception: + return {} + out: dict[str, tuple[str, ...]] = {} + for symbol in symbols: + if symbol.endswith("__vintages"): + continue + try: + frame = lib.tail(symbol, _FUEL_TAIL_ROWS).data + except Exception: + continue + if "fuel_type" not in frame.columns or frame.empty: + continue + fuels = tuple(sorted(str(f) for f in frame["fuel_type"].dropna().unique())) + if fuels: + out[symbol.upper()] = fuels + return out + + +@dg.asset( + name="entity_graph", + group_name="graph", + compute_kind="neo4j", + description=( + "Idempotent MERGE sync of the entity catalog (instruments, balancing " + "authorities, ERCOT settlement points, NOAA regions, commodities, fuel " + "types) into Neo4j. References instrument_ids; never owns a number." + ), +) +def entity_graph( + context: dg.AssetExecutionContext, arctic: ArcticDBResource, neo4j: Neo4jResource +) -> dg.MaterializeResult: + bas = _observed_bas(arctic) + fuels = _observed_fuel_types(arctic) + plan = graph.build_entity_graph( + graph.ObservedEntities(balancing_authorities=bas, fuel_types_by_ba=fuels) + ) + result = graph.sync_graph(neo4j.driver, plan) + nodes_total = sum(result.nodes_by_label.values()) + edges_total = sum(result.edges_by_type.values()) + context.log.info( + "entity graph synced: %d nodes, %d edges (%d BAs)", nodes_total, edges_total, len(bas) + ) + fuel_universe: set[str] = set() + for fuel_list in fuels.values(): + fuel_universe.update(fuel_list) + return dg.MaterializeResult( + metadata={ + "nodes_total": nodes_total, + "edges_total": edges_total, + "nodes_by_label": dg.MetadataValue.json(result.nodes_by_label), + "edges_by_type": dg.MetadataValue.json(result.edges_by_type), + "balancing_authorities": len(bas), + "fuel_types": len(fuel_universe), + } + ) + + +@dg.asset_check( + asset="entity_graph", + name="entity_graph_instruments_resolve", + description=( + "Every Instrument node read back from the graph must resolve through " + "core.symbology — catches catalog drift between the graph and routing." + ), +) +def entity_graph_instruments_resolve( + context: dg.AssetCheckExecutionContext, neo4j: Neo4jResource +) -> dg.AssetCheckResult: + rows = graph.list_entities(neo4j.driver, label="Instrument") + if not rows: + return dg.AssetCheckResult( + passed=False, metadata={"reason": "no Instrument nodes in the graph"} + ) + unresolvable: list[str] = [] + for row in rows: + try: + symbology.resolve(row["key"]) + except SymbologyError: + unresolvable.append(row["key"]) + return dg.AssetCheckResult( + passed=not unresolvable, + metadata={ + "instruments": len(rows), + "unresolvable": dg.MetadataValue.json(unresolvable[:20]), + }, + ) + + +_entity_graph_job = dg.define_asset_job( + "entity_graph_job", selection=dg.AssetSelection.assets(entity_graph) +) + + +# Daily catalog refresh; 06:10 NY avoids the :20-:35 ingestion window. +@dg.schedule( + job=_entity_graph_job, + cron_schedule="10 6 * * *", + execution_timezone="America/New_York", + name="entity_graph_schedule", + default_status=dg.DefaultScheduleStatus.RUNNING, +) +def entity_graph_schedule( + context: dg.ScheduleEvaluationContext, +) -> dg.RunRequest: + return dg.RunRequest() + + +GRAPH_ASSETS: list[Any] = [entity_graph] +GRAPH_CHECKS: list[Any] = [entity_graph_instruments_resolve] +GRAPH_SCHEDULES: list[Any] = [entity_graph_schedule] diff --git a/src/energex/orchestration/resources.py b/src/energex/orchestration/resources.py index c9f9667..5f43d7a 100644 --- a/src/energex/orchestration/resources.py +++ b/src/energex/orchestration/resources.py @@ -77,6 +77,38 @@ def client(self) -> httpx.Client: return httpx.Client(timeout=self.timeout) +class Neo4jResource(ConfigurableResource): + """Opens the entity-graph driver in ``setup_for_execution``. Credentials come + from Dagster ``EnvVar`` (kept out of the UI and run logs); connection failures + surface redacted via ``core.graph.create_driver``. The graph is optional + (compose ``full`` profile only): a failed connect fails the RUN, never + Definitions load.""" + + uri: str + user: str + password: str + + _driver: Any = PrivateAttr(default=None) + + def setup_for_execution(self, context) -> None: # noqa: ARG002 (dagster hook signature) + from pydantic import SecretStr + + from energex.core import graph + from energex.core.config import Neo4jConfig + + cfg = Neo4jConfig(uri=self.uri, user=self.user, password=SecretStr(self.password)) + self._driver = graph.create_driver(cfg) + + def teardown_after_execution(self, context) -> None: # noqa: ARG002 + if self._driver is not None: + self._driver.close() + self._driver = None + + @property + def driver(self) -> Any: + return self._driver + + RESOURCES: dict[str, object] = { "arctic": ArcticDBResource( endpoint=EnvVar("MINIO_ENDPOINT"), @@ -86,4 +118,9 @@ def client(self) -> httpx.Client: secret_key=EnvVar("MINIO_SECRET_KEY"), ), "http": HttpResource(), + "neo4j": Neo4jResource( + uri=EnvVar("NEO4J_URI"), + user=EnvVar("NEO4J_USER"), + password=EnvVar("NEO4J_PASSWORD"), + ), } diff --git a/tests/test_definitions_load.py b/tests/test_definitions_load.py index eaf7ea3..c4f3e0f 100644 --- a/tests/test_definitions_load.py +++ b/tests/test_definitions_load.py @@ -25,6 +25,7 @@ def test_definitions_builds_with_intraday_slice(): assert "ercot_rt_spp" in asset_keys assert "ercot_dam_spp" in asset_keys assert "ercot_load" in asset_keys + assert "entity_graph" in asset_keys # asset_checks MUST be wired explicitly (spec §5.6); key by check name. check_keys = {key.name for key in repo.asset_checks_defs_by_key} @@ -38,6 +39,7 @@ def test_definitions_builds_with_intraday_slice(): assert "ercot_rt_spp_pass_quality_gate" in check_keys assert "ercot_dam_spp_pass_quality_gate" in check_keys assert "ercot_load_pass_quality_gate" in check_keys + assert "entity_graph_instruments_resolve" in check_keys def test_dagster_definitions_validate_cli(): diff --git a/tests/test_graph_asset.py b/tests/test_graph_asset.py new file mode 100644 index 0000000..f0710ce --- /dev/null +++ b/tests/test_graph_asset.py @@ -0,0 +1,120 @@ +"""entity_graph asset: store-driven discovery -> pure plan -> idempotent sync.""" + +from types import SimpleNamespace + +import dagster as dg +import pandas as pd + + +class FakeDriver: + def __init__(self, results=None): + self.calls = [] + self._results = results or {} + + def execute_query(self, query_, parameters_=None, database_=None, **_kw): + self.calls.append((query_, dict(parameters_ or {}), database_)) + for needle, records in self._results.items(): + if needle in query_: + return SimpleNamespace(records=records, summary=None, keys=[]) + return SimpleNamespace(records=[], summary=None, keys=[]) + + def close(self): + pass + + +class FakeLib: + def __init__(self, symbols=(), frames=None): + self._symbols = list(symbols) + self._frames = frames or {} + + def list_symbols(self): + return list(self._symbols) + + def tail(self, symbol, n): + return SimpleNamespace(data=self._frames[symbol].tail(n)) + + +class FakeArctic: + """Stands in for ArcticDBResource: get_library(name) only.""" + + def __init__(self, libs): + self._libs = libs + + def get_library(self, name): + return self._libs[name] + + +class FakeNeo4j: + def __init__(self, driver): + self.driver = driver + + +def _gen_fuel_frame(fuels): + now = pd.Timestamp("2026-07-14", tz="UTC") + return pd.DataFrame( + { + "instrument_id": ["EIA930.GEN_FUEL.ERCO"] * len(fuels), + "valid_time": [now] * len(fuels), + "fuel_type": list(fuels), + "value": [1.0] * len(fuels), + } + ) + + +def test_entity_graph_syncs_discovered_universe(): + from energex.orchestration.graph import entity_graph + + arctic = FakeArctic( + { + "power.demand": FakeLib(symbols=["erco", "miso"]), + "power.generation_by_fuel": FakeLib( + symbols=["erco"], frames={"erco": _gen_fuel_frame(["NG", "WND"])} + ), + } + ) + driver = FakeDriver() + result = entity_graph(dg.build_asset_context(), arctic=arctic, neo4j=FakeNeo4j(driver)) + assert isinstance(result, dg.MaterializeResult) + assert result.metadata["balancing_authorities"] == 2 + assert result.metadata["fuel_types"] == 2 + assert any("UNWIND" in q for q, _, _ in driver.calls) + + +def test_entity_graph_degrades_to_static_catalog_when_store_empty(): + from energex.orchestration.graph import entity_graph + + class EmptyArctic: + def get_library(self, name): + raise RuntimeError("no such library") + + driver = FakeDriver() + result = entity_graph(dg.build_asset_context(), arctic=EmptyArctic(), neo4j=FakeNeo4j(driver)) + # static catalog still syncs: 13 settlement points among the node batches + assert result.metadata["nodes_total"] > 40 + assert result.metadata["balancing_authorities"] == 0 + + +def test_integrity_check_flags_unresolvable_instruments(): + from energex.orchestration.graph import entity_graph_instruments_resolve + + good = {"label": "Instrument", "key": "FRED.WTI.SPOT", "properties": {}} + bad = {"label": "Instrument", "key": "BOGUS.NOPE", "properties": {}} + result = entity_graph_instruments_resolve( + dg.build_asset_context(), neo4j=FakeNeo4j(FakeDriver(results={"MATCH": [good, bad]})) + ) + assert isinstance(result, dg.AssetCheckResult) + assert result.passed is False + + ok = entity_graph_instruments_resolve( + dg.build_asset_context(), neo4j=FakeNeo4j(FakeDriver(results={"MATCH": [good]})) + ) + assert ok.passed is True + + +def test_integrity_check_fails_on_empty_graph(): + from energex.orchestration.graph import entity_graph_instruments_resolve + + result = entity_graph_instruments_resolve( + dg.build_asset_context(), neo4j=FakeNeo4j(FakeDriver()) + ) + assert result.passed is False From 13f5296401c750b97bdcff7b43b0f7995d35e4ff Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 06:48:16 -0400 Subject: [PATCH 5/9] feat(service): /graph/entities + /graph/related on the S2 seam (optional, 503-degrading) Co-Authored-By: Claude Fable 5 --- src/energex/service/readapi.py | 62 +++++++++++++++++- tests/test_graph_api.py | 114 +++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 3 deletions(-) create mode 100644 tests/test_graph_api.py diff --git a/src/energex/service/readapi.py b/src/energex/service/readapi.py index 1de52aa..63ef3fc 100644 --- a/src/energex/service/readapi.py +++ b/src/energex/service/readapi.py @@ -17,6 +17,7 @@ import json import logging import os +import threading from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import Any @@ -27,9 +28,9 @@ from fastapi import Depends, FastAPI, Header, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware -from energex.core import storage, symbology +from energex.core import graph, storage, symbology from energex.core.config import get_settings -from energex.core.exceptions import SymbologyError +from energex.core.exceptions import GraphError, SymbologyError logger = logging.getLogger(__name__) @@ -114,6 +115,21 @@ def _require_api_key(x_api_key: str | None = Header(default=None)) -> None: raise HTTPException(status_code=401, detail="missing or invalid API key") +def _get_graph_driver(app: FastAPI) -> Any: + """Lazily connect the entity-graph driver on first use (the neo4j service is + optional and may start after the api). Serialized by a lock: endpoints run in + FastAPI's threadpool. 503 when the graph is genuinely unreachable.""" + if app.state.neo4j is None: + with app.state.neo4j_lock: + if app.state.neo4j is None: + try: + app.state.neo4j = graph.create_driver(get_settings().neo4j) + except GraphError as exc: + logger.warning("entity graph unavailable: %s", exc) + raise HTTPException(status_code=503, detail="entity graph unavailable") from exc + return app.state.neo4j + + def create_app() -> FastAPI: @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: @@ -127,10 +143,18 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: except Exception as exc: # pragma: no cover - defensive logger.error("ArcticDB connection failed: %s", type(exc).__name__) raise + # The entity graph is OPTIONAL: never fail startup on it. Connect lazily + # so a late-starting neo4j container is picked up by the first /graph/* + # request rather than requiring an api restart. + app.state.neo4j = None + app.state.neo4j_lock = threading.Lock() logger.info("energex S2 read API started") try: yield finally: + if app.state.neo4j is not None: + app.state.neo4j.close() + app.state.neo4j = None app.state.arctic = None logger.info("energex S2 read API stopped") @@ -158,7 +182,15 @@ def healthz() -> dict[str, Any]: latest = _latest_as_of(ac) except Exception: # pragma: no cover - health must not fail on the latest probe latest = None - return {"status": "ok", "libraries": libraries, "latest_as_of": latest} + return { + "status": "ok", + "libraries": libraries, + "latest_as_of": latest, + # Cached state only (no connect attempt): the compose healthcheck + # budget is 5s and the graph is optional. True after the first + # successful /graph/* call. + "graph": app.state.neo4j is not None, + } @app.get("/libraries", dependencies=[Depends(_require_api_key)]) def libraries() -> list[str]: @@ -213,6 +245,30 @@ def curve( ) from exc return _records(df) + @app.get("/graph/entities", dependencies=[Depends(_require_api_key)]) + def graph_entities(label: str | None = Query(default=None)) -> list[dict[str, Any]]: + if label is not None and label not in graph.KEY_PROPERTY: + raise HTTPException(status_code=404, detail=f"unknown label: {label!r}") + driver = _get_graph_driver(app) + try: + return graph.list_entities(driver, label=label) + except GraphError as exc: + raise HTTPException(status_code=503, detail="entity graph unavailable") from exc + + @app.get("/graph/related", dependencies=[Depends(_require_api_key)]) + def graph_related( + instrument_id: str = Query(...), + depth: int = Query(default=2, ge=1, le=3), + ) -> dict[str, Any]: + driver = _get_graph_driver(app) + try: + result = graph.related_instruments(driver, instrument_id, depth=depth) + except GraphError as exc: + raise HTTPException(status_code=503, detail="entity graph unavailable") from exc + if result is None: + raise HTTPException(status_code=404, detail=f"unknown instrument: {instrument_id!r}") + return result + return app diff --git a/tests/test_graph_api.py b/tests/test_graph_api.py new file mode 100644 index 0000000..8c661fb --- /dev/null +++ b/tests/test_graph_api.py @@ -0,0 +1,114 @@ +"""S2 /graph endpoints: lazy driver, 503 degradation, catalog + related queries.""" + +from types import SimpleNamespace + +import pytest + +pytest.importorskip("fastapi") +from fastapi.testclient import TestClient # noqa: E402 + +from energex.core import graph as core_graph # noqa: E402 +from energex.core.exceptions import GraphError # noqa: E402 +from energex.service.readapi import create_app # noqa: E402 + + +class FakeDriver: + def __init__(self, results=None): + self.calls = [] + self._results = results or {} + self.closed = False + + def execute_query(self, query_, parameters_=None, database_=None, **_kw): + params = dict(parameters_ or {}) + self.calls.append((query_, params, database_)) + for needle, records in self._results.items(): + if needle in query_: + if callable(records): # parameter-sensitive results + records = records(params) + return SimpleNamespace(records=records, summary=None, keys=[]) + return SimpleNamespace(records=[], summary=None, keys=[]) + + def close(self): + self.closed = True + + +@pytest.fixture +def client(monkeypatch, arctic_uri): + monkeypatch.setenv("ENERGEX_ARCTIC_URI", arctic_uri) + return TestClient(create_app()) + + +def _install_driver(monkeypatch, driver): + monkeypatch.setattr(core_graph, "create_driver", lambda cfg: driver) + + +def test_graph_unavailable_returns_503_and_healthz_false(monkeypatch, client): + def boom(cfg): + raise GraphError("Neo4j connection failed: ServiceUnavailable") + + monkeypatch.setattr(core_graph, "create_driver", boom) + with client as c: + assert c.get("/healthz").json()["graph"] is False + response = c.get("/graph/entities") + assert response.status_code == 503 + assert "unavailable" in response.json()["detail"] + + +def test_graph_entities_lists_catalog(monkeypatch, client): + records = [ + { + "label": "SettlementPoint", + "key": "HB_NORTH", + "properties": {"code": "HB_NORTH", "kind": "hub"}, + } + ] + _install_driver(monkeypatch, FakeDriver(results={"MATCH": records})) + with client as c: + response = c.get("/graph/entities", params={"label": "SettlementPoint"}) + assert response.status_code == 200 + assert response.json()[0]["key"] == "HB_NORTH" + # after a successful graph call, healthz reports the graph as up + assert c.get("/healthz").json()["graph"] is True + + +def test_graph_entities_unknown_label_404(monkeypatch, client): + _install_driver(monkeypatch, FakeDriver()) + with client as c: + assert c.get("/graph/entities", params={"label": "Nope"}).status_code == 404 + + +def test_graph_related_contract(monkeypatch, client): + def exists(params): + if params["iid"] == "ERCOT.SPP.HB_NORTH": + return [{"instrument_id": "ERCOT.SPP.HB_NORTH"}] + return [] + + related = [{"label": "SettlementPoint", "key": "HB_NORTH", "properties": {"code": "HB_NORTH"}}] + _install_driver( + monkeypatch, FakeDriver(results={"RETURN i.instrument_id": exists, "DISTINCT": related}) + ) + with client as c: + response = c.get( + "/graph/related", params={"instrument_id": "ERCOT.SPP.HB_NORTH", "depth": 1} + ) + assert response.status_code == 200 + body = response.json() + assert body["instrument_id"] == "ERCOT.SPP.HB_NORTH" + assert body["related"][0]["key"] == "HB_NORTH" + + assert c.get("/graph/related", params={"instrument_id": "NOPE.X"}).status_code == 404 + assert ( + c.get( + "/graph/related", + params={"instrument_id": "ERCOT.SPP.HB_NORTH", "depth": 7}, + ).status_code + == 422 + ) + + +def test_driver_closed_on_shutdown(monkeypatch, client): + driver = FakeDriver() + _install_driver(monkeypatch, driver) + with client as c: + c.get("/graph/entities") + assert driver.closed is True From dfb951262331014855e5634c50bdc13e07c265cd Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 06:48:55 -0400 Subject: [PATCH 6/9] feat(wiring): graph extra + NEO4J env for the api, CI graph-gate, env-sync note Co-Authored-By: Claude Fable 5 --- .env.example | 2 ++ .github/workflows/ci.yml | 18 ++++++++++++++++++ docker-compose.yml | 7 ++++++- 3 files changed, 26 insertions(+), 1 deletion(-) diff --git a/.env.example b/.env.example index f432e91..02d66af 100644 --- a/.env.example +++ b/.env.example @@ -77,6 +77,8 @@ NEO4J_USER=neo4j NEO4J_PASSWORD=change-me-neo4j # Compose-only: neo4j container auth, in user/password form. +# NOTE: keep NEO4J_AUTH's password in sync with NEO4J_PASSWORD above — the server +# reads this one, clients read the other; rotating only one fails auth at graph-run time. NEO4J_AUTH=neo4j/change-me-neo4j # ============================================================================= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 060bff1..945b83a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -119,6 +119,24 @@ jobs: tests/test_connector_weather.py tests/test_readapi.py + graph-gate: + name: Entity graph gate + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + with: + enable-cache: true + - name: entity-graph suite (offline; fake driver, no Neo4j service) + run: > + uv run --extra graph --extra storage --extra quality --extra service + --extra orchestration --extra dev pytest -q + tests/test_graph_vocab.py + tests/test_graph_plan.py + tests/test_graph_sync.py + tests/test_graph_asset.py + tests/test_graph_api.py + secrets: name: Secret scan (gitleaks) runs-on: ubuntu-latest diff --git a/docker-compose.yml b/docker-compose.yml index 50f67ca..5d7473a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -5,7 +5,7 @@ services: build: context: . args: - EXTRAS: "--extra service --extra storage" + EXTRAS: "--extra service --extra storage --extra graph" image: ghcr.io/oldhero5/energex:api platform: linux/amd64 container_name: energex-api @@ -24,6 +24,11 @@ services: # Scoped ArcticDB service account created by minio-init (NOT MinIO root); read-only API. MINIO_ACCESS_KEY: ${ARCTIC_ACCESS_KEY:?set ARCTIC_ACCESS_KEY in .env} MINIO_SECRET_KEY: ${ARCTIC_SECRET_KEY:?set ARCTIC_SECRET_KEY in .env} + # Entity graph (optional): /graph/* endpoints 503 until the `full` profile's + # neo4j is reachable; the api starts and serves series data regardless. + NEO4J_URI: bolt://neo4j:7687 + NEO4J_USER: neo4j + NEO4J_PASSWORD: ${NEO4J_PASSWORD:?set NEO4J_PASSWORD in .env} TZ: UTC ports: - "8000:8001" From 353b9b05e3f5cddf0f523fa6b0f279fe0a9fc0a5 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 06:50:56 -0400 Subject: [PATCH 7/9] docs: entity-graph documentation + S2 contract additions Co-Authored-By: Claude Fable 5 --- README.md | 10 ++- website/docs/architecture.md | 11 ++- website/docs/entity-graph.md | 129 +++++++++++++++++++++++++++ website/docs/frontend-integration.md | 14 ++- website/docs/quickstart.md | 3 +- website/sidebars.ts | 1 + 6 files changed, 159 insertions(+), 9 deletions(-) create mode 100644 website/docs/entity-graph.md diff --git a/README.md b/README.md index 628d428..458fdc4 100644 --- a/README.md +++ b/README.md @@ -74,8 +74,12 @@ sources ──> Connector ──> quality gate ──> ArcticDB (MinIO) > **yfinance is dev-only and unscheduled.** Yahoo frequently blocks programmatic > access, so a schedule would only fire failing runs. The asset stays manual. -An optional **Neo4j** entity graph references these instruments by symbol but never -owns the numbers. +An optional **Neo4j entity graph** mirrors the *identity* layer — instruments, +balancing authorities, ERCOT hubs and load zones, NOAA regions, commodities, and +observed fuel types — with a daily idempotent MERGE sync (`entity_graph` asset). It +references instruments by symbol but never owns the numbers, and it powers the +read API's `/graph/entities` and `/graph/related` discovery endpoints. See the +[entity graph docs](website/docs/entity-graph.md). ## Quickstart @@ -94,7 +98,7 @@ Then open: - **Dagster UI** — http://localhost:3000 (assets, schedules, run history, backfills) - **MinIO console** — http://localhost:9001 (the ArcticDB object store) - **Neo4j browser** — http://localhost:7474 -- **Read API (S2)** — http://localhost:8000 (`/series`, `/curve`, `/symbols`, `/libraries`, `/healthz`) +- **Read API (S2)** — http://localhost:8000 (`/series`, `/curve`, `/symbols`, `/libraries`, `/graph/entities`, `/graph/related`, `/healthz`) Four schedules run by default and keep the store current with no manual intervention: EIA gas storage (Thursday), EIA petroleum status (Wednesday), FRED spot (weekday diff --git a/website/docs/architecture.md b/website/docs/architecture.md index 1d5eb69..083caec 100644 --- a/website/docs/architecture.md +++ b/website/docs/architecture.md @@ -51,9 +51,10 @@ RT/DA SPP + load · FRED · │ EIA fundamentals · NOAA ┌──────────────────┴───────────────────┐ ▼ ▼ Dagster (assets · schedules · checks · reconcile) S2 read API (FastAPI) - │ - ▼ - S4 frontend / S3 agent + │ │ + ▼ (daily, idempotent MERGE) ▼ + Neo4j entity graph (what/who/connected S4 frontend / S3 agent + — never the numbers) ``` 1. A **connector** (`energex.core.connectors`) fetches a window from a source and returns a @@ -74,6 +75,10 @@ EIA fundamentals · NOAA ┌───────── 5. The **S2 read API** (`energex.service.readapi`) serves the store point-in-time over HTTP — the only contract the frontend consumes. See [Frontend Integration](./frontend-integration.md). +6. The **[entity graph](./entity-graph.md)** (`energex.core.graph` + a daily Dagster + sync) mirrors the *identity* layer — instruments, balancing authorities, settlement + points, regions, commodities — into Neo4j. It references `instrument_id`s and never + owns a number. ## Symbology — the single router diff --git a/website/docs/entity-graph.md b/website/docs/entity-graph.md new file mode 100644 index 0000000..39893d1 --- /dev/null +++ b/website/docs/entity-graph.md @@ -0,0 +1,129 @@ +--- +id: entity-graph +title: Entity Graph (Neo4j) +sidebar_label: Entity Graph +--- + +# Entity Graph + +Every number in Energex lives in ArcticDB, keyed by `instrument_id`. The **entity +graph** is the *what/who/connected* layer around those ids: a small Neo4j catalog of +the entities the instrument namespace implies — balancing authorities, ERCOT +settlement points, NOAA climate regions, commodities, fuel types — plus a curated set +of cross-domain edges. It answers questions the time-series store cannot: + +- *What instruments exist for ERCOT, and how do they relate?* (discovery for the + frontend cockpit and watchlists) +- *"Houston power prices"* → `ERCOT.SPP.HB_HOUSTON`, `ERCOT.SPP.LZ_HOUSTON`, + `ERCOT.DASPP.…` (entity grounding for the S3 agent) +- *Which balancing authorities generate wind? What weather series proxies the ERCOT + footprint?* (cross-domain navigation) + +The invariant, stated in `core/symbology.py` since phase 1: **the graph references +instrument_ids and never owns a number.** Nodes and edges carry identity plus +`first_seen`/`last_seen` knowledge stamps — no values, no ArcticDB version integers — +so the graph [restores independently](./operations.md) of the store of record. + +## Data model + +| Label | Key | Notable properties | +| --- | --- | --- | +| `Instrument` | `instrument_id` | `symbol`, `kind` (demand, lmp, spot, …) | +| `Library` | `name` | `revision_mode` | +| `Source` | `name` (eia, eia930, ercot, fred, noaa, yfinance) | — | +| `Market` | `code` (`ERCOT`) | `name` | +| `BalancingAuthority` | `code` (EIA-930 respondent, e.g. `ERCO`) | — | +| `SettlementPoint` | `code` (e.g. `HB_NORTH`) | `kind`: `hub` \| `load_zone` | +| `Region` | `code` (NOAA nClimDiv token, e.g. `TEXAS`) | `nclimdiv_code` | +| `Commodity` | `code` (`WTI`, `BRENT`, `NATGAS`) | `name` | +| `FuelType` | `code` (observed EIA-930 fueltype, e.g. `NG`, `WND`) | — | + +Relationships: + +- `(Instrument)-[:IN_LIBRARY]->(Library)` and `(Instrument)-[:FROM_SOURCE]->(Source)` + — catalog plumbing, routed through `symbology.resolve`. +- `(Instrument)-[:MEASURES]->(BalancingAuthority | SettlementPoint | Region | Commodity | Market)` + — the entity a series is *about*. `ERCOT.LOAD.ERCOT` measures the **market + aggregate**, not a settlement point. +- `(SettlementPoint)-[:IN_MARKET]->(Market)` — the 13 canonical tradeable points. +- `(BalancingAuthority)-[:GENERATES]->(FuelType)` — observed generation-by-fuel mix. +- Curated cross-domain edges: `(ERCO)-[:OPERATES]->(ERCOT)` (the EIA-930 ↔ ERCOT + nodal join point), `(TEXAS)-[:WEATHER_PROXY_FOR]->(ERCOT)` (nClimDiv Texas is the + documented ERCOT footprint), and `(NATGAS)-[:FUELS]->(NG)` when gas-fired + generation is observed. + +Deliberately **not** modeled: BA↔BA interchange pairs (only *total* interchange is +ingested), per-vintage lineage (that is the ArcticDB vintage index's job), and the +~17k non-tradeable ERCOT nodes (no data behind them). + +## How it syncs + +`energex.core.graph` is a pure module: `build_entity_graph(observed)` derives a +`GraphPlan` from the symbology tables, the ERCOT settlement-point set, and the NOAA +region map, plus **store-observed** vocabulary — balancing authorities from +`power.demand` symbols and fuel types from a bounded tail read of +`power.generation_by_fuel`. The repo hardcodes no BA or fuel list anywhere; the graph +discovers them. The plan compiles to batched `UNWIND … MERGE` Cypher (one statement +per label / relationship shape), stamped `first_seen` on create and `last_seen` on +every sync — running it twice is a no-op apart from `last_seen`. + +The Dagster side (`energex.orchestration.graph`) wires that into an `entity_graph` +asset (group `graph`) with a `Neo4jResource`, an `entity_graph_instruments_resolve` +asset check (every `Instrument` node read back must resolve through +`core.symbology` — catches drift), and a daily `entity_graph_schedule` at 06:10 ET. + +The neo4j driver itself is imported lazily in exactly one function +(`core.graph.create_driver`), so installs without the `graph` extra — and every test +environment — never import it. Tests run against a duck-typed fake driver; no Neo4j +service exists in CI. + +## Querying it + +Through the S2 read API (the only surface the frontend consumes): + +| Method & path | Query params | Returns | +|---|---|---| +| `GET /graph/entities` | `label?` | catalog nodes `[{label, key, properties}, …]` | +| `GET /graph/related` | `instrument_id`, `depth?` (1–3, default 2) | connected entities + sibling instruments | + +`/graph/related` traverses **semantic** relationships only (`MEASURES`, `IN_MARKET`, +`OPERATES`, `WEATHER_PROXY_FOR`, `GENERATES`, `FUELS`) — never through the +`Library`/`Source` hub nodes, which would relate everything to everything. Depth 1 +returns the entities an instrument measures; depth 2 adds sibling instruments and +adjacent entities. + +The graph is **optional**: the api starts and serves all series endpoints with no +Neo4j running (`dev`/`api` compose profiles have none), `/graph/*` returns **503** +until the `full` profile's neo4j is reachable, and the driver connects lazily on the +first graph request — no api restart needed after neo4j comes up. `/healthz` carries +a `graph: bool` reflecting the cached connection state. + +Or directly in the Neo4j browser (http://localhost:7474): + +```cypher +// what relates to the ERCOT North Hub? +MATCH (i:Instrument {instrument_id: "ERCOT.SPP.HB_NORTH"})-[r]-(n) +RETURN i, r, n; + +// which balancing authorities generate wind? +MATCH (ba:BalancingAuthority)-[:GENERATES]->(:FuelType {code: "WND"}) +RETURN ba.code ORDER BY ba.code; +``` + +## Knowledge-time honesty + +The graph is a **current-state catalog**, not a bitemporal store — it has no `as_of` +parameter by design. What it does carry is provenance: every node and relationship +records `first_seen` (when the entity entered the catalog) and `last_seen` (the most +recent sync that still observed it), so consumers can always tell how fresh a catalog +fact is. Anything time-series-shaped stays in ArcticDB behind `read_as_of`. + +## Operations + +- Runs under the `full` compose profile only (`neo4j:5.26.0-community`, bolt 7687, + browser 7474, 512m heap + 512m pagecache). +- Keep `NEO4J_AUTH` (server, compose-only) and `NEO4J_PASSWORD` (clients) in sync — + rotating one without the other fails auth at graph-run time. +- Back up via `neo4j-admin database dump` of the `neo4j-data` volume (see + [Operations](./operations.md)); the graph restores independently of ArcticDB, and a + lost graph is fully rebuilt by one `entity_graph` materialization. diff --git a/website/docs/frontend-integration.md b/website/docs/frontend-integration.md index d4d6dec..5ecbcf7 100644 --- a/website/docs/frontend-integration.md +++ b/website/docs/frontend-integration.md @@ -57,11 +57,20 @@ never leaks the future. | Method & path | Query params | Returns | |---|---|---| -| `GET /healthz` | — | `{status, libraries, latest_as_of}` (cheap probe; never auth-gated) | +| `GET /healthz` | — | `{status, libraries, latest_as_of, graph}` (cheap probe; never auth-gated) | | `GET /libraries` | — | `["power.lmp", "power.demand", …]` — the ArcticDB libraries present | | `GET /symbols` | `library` (required) | symbols in a library (the `*__vintages` sidecars are hidden) | | `GET /series` | `library`, `symbol` (required); `as_of`, `start`, `end` (optional) | `read_as_of` rows as JSON records | | `GET /curve` | `commodity` (required); `as_of` (optional) | the assembled forward curve as JSON records | +| `GET /graph/entities` | `label` (optional) | [entity-graph](./entity-graph.md) catalog nodes `[{label, key, properties}, …]` | +| `GET /graph/related` | `instrument_id` (required); `depth` (optional, 1–3) | connected entities + sibling instruments for discovery | + +The `/graph/*` endpoints serve the **current-state** [entity graph](./entity-graph.md) +(no `as_of` — catalog facts carry `first_seen`/`last_seen` provenance instead) and are +**optional**: they return `503` when the `full` profile's Neo4j is not running, while +every series endpoint keeps working. `healthz.graph` reflects the cached graph +connection state. Like every other data endpoint, they are gated by the optional API +key (see [Auth](#auth-optional-api-key)). ### `GET /healthz` @@ -151,7 +160,8 @@ for the full bitemporal model. ## Auth (optional API key) Auth is **opt-in**. When the `ENERGEX_READ_API_KEY` environment variable is set, every -data endpoint (`/libraries`, `/symbols`, `/series`, `/curve`) requires a matching +data endpoint (`/libraries`, `/symbols`, `/series`, `/curve`, `/graph/entities`, +`/graph/related`) requires a matching `X-API-Key` header, compared in constant time. `/healthz` stays open so probes keep working. When the variable is **unset** the API is open and logs a startup warning. A missing or invalid key returns **`401`**. diff --git a/website/docs/quickstart.md b/website/docs/quickstart.md index ae9ec33..fa562ce 100644 --- a/website/docs/quickstart.md +++ b/website/docs/quickstart.md @@ -62,7 +62,7 @@ account, never the MinIO root. The full annotated list lives in | `dagster-postgres` | — | Dagster instance storage | | `minio` | 9000 / 9001 | ArcticDB object store + web console | | `minio-init` | — | One-shot: creates the bucket and scoped service account | -| `neo4j` | 7474 / 7687 | Optional entity graph | +| `neo4j` | 7474 / 7687 | Optional [entity graph](./entity-graph.md) (synced daily by the `entity_graph` asset) | The `api` service is the **only** contract the separate, private frontend consumes — see [Frontend Integration](./frontend-integration.md). @@ -85,6 +85,7 @@ primary power feeds lead the cadence; oil/gas/weather follow as supporting conte | FRED spot | Daily (weekday mornings) | WTI / Brent / Henry Hub benchmark spot | | EIA fundamentals | Weekly (gas Thu, crude Wed) | Lower-48 gas storage, crude stocks ex-SPR | | NOAA degree days | Monthly | HDD/CDD by US region | +| Entity graph | Daily 06:10 ET | Neo4j [entity-graph](./entity-graph.md) catalog sync (instruments, BAs, hubs) | To verify it is alive, open the Dagster UI and confirm the schedules show as running, or trigger a single asset run from the asset graph and watch it land in MinIO. Full detail diff --git a/website/sidebars.ts b/website/sidebars.ts index d7b1073..f09a35c 100644 --- a/website/sidebars.ts +++ b/website/sidebars.ts @@ -9,6 +9,7 @@ const sidebars: SidebarsConfig = { 'data-sources-connectors', 'storage-point-in-time', 'orchestration', + 'entity-graph', 'deployment', 'operations', 'testing', From 839451b8483abf6e135fb45c795e21ec49ea95ba Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Wed, 15 Jul 2026 07:14:56 -0400 Subject: [PATCH 8/9] fix(graph): apply confirmed adversarial-review findings - fuel-only BAs become full plan nodes (never bare edge-MERGE nodes) - schedule probes neo4j reachability and skips instead of failing daily in profiles without the optional server - readapi backs off 15s between failed graph connects so an unreachable neo4j cannot stall the threadpool - _redact_uri strips userinfo from scheme-less URIs too - tests: neo4j temporal-type cleaning, query-time 503s, Cypher shape for label=None, connect backoff, schedule skip/fire - docs: note that GENERATES fuel mix is under-reported pending the generation-by-fuel storage dedup fix (flagged separately) Co-Authored-By: Claude Fable 5 --- src/energex/core/graph.py | 16 ++++++++++----- src/energex/orchestration/graph.py | 15 +++++++++++--- src/energex/service/readapi.py | 15 +++++++++++++- tests/test_graph_api.py | 31 ++++++++++++++++++++++++++++ tests/test_graph_asset.py | 29 ++++++++++++++++++++++++++ tests/test_graph_plan.py | 13 ++++++++++++ tests/test_graph_sync.py | 33 +++++++++++++++++++++++++++++- website/docs/entity-graph.md | 7 +++++++ 8 files changed, 149 insertions(+), 10 deletions(-) diff --git a/src/energex/core/graph.py b/src/energex/core/graph.py index e0299a8..aaa506a 100644 --- a/src/energex/core/graph.py +++ b/src/energex/core/graph.py @@ -220,7 +220,11 @@ def build_entity_graph(observed: ObservedEntities | None = None) -> GraphPlan: # ERCO is seeded statically: it is the join point between the EIA-930 BA # universe and the ERCOT nodal universe, whether or not it was observed yet. - bas = sorted(set(observed.balancing_authorities) | {"ERCO"}) + # fuel_types_by_ba keys are unioned in: the two observed inputs come from + # different libraries (different assets), so a BA can appear in one and not + # the other — every GENERATES endpoint must be a full plan node, never a + # bare node silently MERGEd into existence by an edge statement. + bas = sorted(set(observed.balancing_authorities) | set(observed.fuel_types_by_ba) | {"ERCO"}) for ba in bas: ba_node = b.node("BalancingAuthority", ba) for prefix in _EIA930_PREFIXES: @@ -329,11 +333,13 @@ def sync_graph( def _redact_uri(uri: str) -> str: - """Strip any userinfo from a bolt/neo4j URI before it can reach a log line.""" + """Strip any userinfo from a bolt/neo4j URI before it can reach a log line + (scheme-less strings like ``user:pass@host:7687`` included).""" scheme, sep, rest = uri.partition("://") - if sep and "@" in rest: - rest = rest.rsplit("@", 1)[1] - return f"{scheme}{sep}{rest}" + host = rest if sep else scheme + if "@" in host: + host = host.rsplit("@", 1)[1] + return f"{scheme}{sep}{host}" if sep else host def create_driver(cfg: Neo4jConfig) -> Any: diff --git a/src/energex/orchestration/graph.py b/src/energex/orchestration/graph.py index 5ed6c21..3979360 100644 --- a/src/energex/orchestration/graph.py +++ b/src/energex/orchestration/graph.py @@ -14,7 +14,8 @@ import dagster as dg from energex.core import graph, symbology -from energex.core.exceptions import SymbologyError +from energex.core.config import Neo4jConfig +from energex.core.exceptions import GraphError, SymbologyError from energex.orchestration.resources import ArcticDBResource, Neo4jResource # Rows per generation_by_fuel tail read: ~hourly x ~10 fuels x 3 days, rounded up. @@ -128,7 +129,10 @@ def entity_graph_instruments_resolve( ) -# Daily catalog refresh; 06:10 NY avoids the :20-:35 ingestion window. +# Daily catalog refresh; 06:10 NY avoids the :20-:35 ingestion window. The graph +# is optional (neo4j runs only under the `full` compose profile), so the tick +# probes reachability and SKIPS instead of firing a guaranteed-failing run in +# profiles without a neo4j server. @dg.schedule( job=_entity_graph_job, cron_schedule="10 6 * * *", @@ -138,7 +142,12 @@ def entity_graph_instruments_resolve( ) def entity_graph_schedule( context: dg.ScheduleEvaluationContext, -) -> dg.RunRequest: +) -> dg.RunRequest | dg.SkipReason: + try: + driver = graph.create_driver(Neo4jConfig()) + except GraphError: + return dg.SkipReason("neo4j unreachable; entity-graph sync skipped (optional service)") + driver.close() return dg.RunRequest() diff --git a/src/energex/service/readapi.py b/src/energex/service/readapi.py index 63ef3fc..116360b 100644 --- a/src/energex/service/readapi.py +++ b/src/energex/service/readapi.py @@ -18,6 +18,7 @@ import logging import os import threading +import time from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import Any @@ -36,6 +37,11 @@ VINTAGE_SUFFIX = "__vintages" +# Minimum seconds between graph connect attempts. create_driver blocks up to ~5s +# against an unreachable server; without a backoff, a burst of /graph/* requests +# would queue threadpool workers on the connect lock and stall series endpoints. +GRAPH_RETRY_SECONDS = 15.0 + def _resolve_arctic_uri() -> str: """Resolve the Arctic URI: an explicit ``ENERGEX_ARCTIC_URI`` (tests / lmdb) wins; @@ -118,13 +124,19 @@ def _require_api_key(x_api_key: str | None = Header(default=None)) -> None: def _get_graph_driver(app: FastAPI) -> Any: """Lazily connect the entity-graph driver on first use (the neo4j service is optional and may start after the api). Serialized by a lock: endpoints run in - FastAPI's threadpool. 503 when the graph is genuinely unreachable.""" + FastAPI's threadpool. After a failed connect, further attempts fail fast for + GRAPH_RETRY_SECONDS so an unreachable graph cannot stall the threadpool. + 503 when the graph is genuinely unreachable.""" if app.state.neo4j is None: with app.state.neo4j_lock: if app.state.neo4j is None: + now = time.monotonic() + if now < app.state.neo4j_next_retry: + raise HTTPException(status_code=503, detail="entity graph unavailable") try: app.state.neo4j = graph.create_driver(get_settings().neo4j) except GraphError as exc: + app.state.neo4j_next_retry = now + GRAPH_RETRY_SECONDS logger.warning("entity graph unavailable: %s", exc) raise HTTPException(status_code=503, detail="entity graph unavailable") from exc return app.state.neo4j @@ -148,6 +160,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # request rather than requiring an api restart. app.state.neo4j = None app.state.neo4j_lock = threading.Lock() + app.state.neo4j_next_retry = 0.0 logger.info("energex S2 read API started") try: yield diff --git a/tests/test_graph_api.py b/tests/test_graph_api.py index 8c661fb..9db314e 100644 --- a/tests/test_graph_api.py +++ b/tests/test_graph_api.py @@ -112,3 +112,34 @@ def test_driver_closed_on_shutdown(monkeypatch, client): with client as c: c.get("/graph/entities") assert driver.closed is True + + +def test_failed_connect_backs_off_without_reattempting(monkeypatch, client): + calls = {"n": 0} + + def boom(cfg): + calls["n"] += 1 + raise GraphError("Neo4j connection failed: ServiceUnavailable") + + monkeypatch.setattr(core_graph, "create_driver", boom) + with client as c: + assert c.get("/graph/entities").status_code == 503 + # within GRAPH_RETRY_SECONDS the second request fails fast: no new attempt + assert c.get("/graph/entities").status_code == 503 + assert calls["n"] == 1 + + +class BrokenQueryDriver(FakeDriver): + """Connects fine, then every query blows up (e.g. neo4j died mid-flight).""" + + def execute_query(self, query_, parameters_=None, database_=None, **_kw): + raise RuntimeError("bolt connection reset") + + +def test_query_time_failure_returns_503_on_both_endpoints(monkeypatch, client): + _install_driver(monkeypatch, BrokenQueryDriver()) + with client as c: + response = c.get("/graph/entities") + assert response.status_code == 503 + response = c.get("/graph/related", params={"instrument_id": "FRED.WTI.SPOT"}) + assert response.status_code == 503 diff --git a/tests/test_graph_asset.py b/tests/test_graph_asset.py index f0710ce..d42d481 100644 --- a/tests/test_graph_asset.py +++ b/tests/test_graph_asset.py @@ -118,3 +118,32 @@ def test_integrity_check_fails_on_empty_graph(): dg.build_asset_context(), neo4j=FakeNeo4j(FakeDriver()) ) assert result.passed is False + + +def test_schedule_skips_when_neo4j_unreachable(monkeypatch): + from energex.core import graph as core_graph + from energex.core.exceptions import GraphError + from energex.orchestration.graph import entity_graph_schedule + + def boom(cfg): + raise GraphError("Neo4j connection failed: ServiceUnavailable") + + monkeypatch.setattr(core_graph, "create_driver", boom) + result = entity_graph_schedule(dg.build_schedule_context()) + assert isinstance(result, dg.SkipReason) + + +def test_schedule_fires_and_closes_probe_driver_when_reachable(monkeypatch): + from energex.core import graph as core_graph + from energex.orchestration.graph import entity_graph_schedule + + class ProbeDriver: + closed = False + + def close(self): + ProbeDriver.closed = True + + monkeypatch.setattr(core_graph, "create_driver", lambda cfg: ProbeDriver()) + result = entity_graph_schedule(dg.build_schedule_context()) + assert isinstance(result, dg.RunRequest) + assert ProbeDriver.closed is True diff --git a/tests/test_graph_plan.py b/tests/test_graph_plan.py index 6c3b837..062280a 100644 --- a/tests/test_graph_plan.py +++ b/tests/test_graph_plan.py @@ -96,3 +96,16 @@ def test_plan_is_deduplicated_and_all_edge_endpoints_exist(): node_set = set(keys) for e in plan.edges: assert e.src in node_set and e.dst in node_set + + +def test_fuel_only_ba_becomes_a_full_node_with_instruments(): + # The two observed inputs come from different libraries (different assets), so + # a BA can appear in fuel_types_by_ba without being in balancing_authorities. + # It must still become a full plan node (provenance stamps + instruments), not + # a bare node silently MERGEd into existence by the GENERATES edge statement. + plan = _plan(graph.ObservedEntities(fuel_types_by_ba={"MISO": ("COL",)})) + assert _node(plan, "BalancingAuthority", "MISO") is not None + assert _node(plan, "Instrument", "EIA930.D.MISO") is not None + node_set = {(n.label, n.key) for n in plan.nodes} + for e in plan.edges: + assert e.src in node_set and e.dst in node_set diff --git a/tests/test_graph_sync.py b/tests/test_graph_sync.py index 2dab606..b6037da 100644 --- a/tests/test_graph_sync.py +++ b/tests/test_graph_sync.py @@ -83,12 +83,27 @@ def no_neo4j(name, *args, **kwargs): graph.create_driver(Neo4jConfig()) +class FakeNeo4jDateTime: + """Mimics neo4j.time.DateTime: exposes to_native() -> datetime.""" + + def __init__(self, dt): + self._dt = dt + + def to_native(self): + return self._dt + + def test_list_entities_validates_label_and_cleans_records(): records = [ { "label": "Instrument", "key": "FRED.WTI.SPOT", - "properties": {"instrument_id": "FRED.WTI.SPOT", "first_seen": SYNCED_AT}, + # exercise the real driver contract: temporal props arrive as + # neo4j.time.DateTime-like objects, not python datetimes + "properties": { + "instrument_id": "FRED.WTI.SPOT", + "first_seen": FakeNeo4jDateTime(SYNCED_AT), + }, } ] driver = FakeDriver(results={"MATCH": records}) @@ -99,6 +114,22 @@ def test_list_entities_validates_label_and_cleans_records(): graph.list_entities(driver, label="DropAllTables") +def test_list_entities_cypher_shape_for_label_and_none(): + driver = FakeDriver() + graph.list_entities(driver, label="Market") + graph.list_entities(driver) # label=None must match ALL nodes, not :None + labelled, unlabelled = driver.calls[0][0], driver.calls[1][0] + assert labelled.startswith("MATCH (n:Market)") + assert unlabelled.startswith("MATCH (n)\n") + assert "None" not in unlabelled + + +def test_redact_uri_strips_userinfo_with_and_without_scheme(): + assert graph._redact_uri("bolt://user:secret@host:7687") == "bolt://host:7687" + assert graph._redact_uri("user:secret@host:7687") == "host:7687" + assert graph._redact_uri("bolt://host:7687") == "bolt://host:7687" + + def test_related_instruments_unknown_returns_none_and_depth_validated(): driver = FakeDriver() # no records -> instrument not found assert graph.related_instruments(driver, "NOPE.X") is None diff --git a/website/docs/entity-graph.md b/website/docs/entity-graph.md index 39893d1..cf7586c 100644 --- a/website/docs/entity-graph.md +++ b/website/docs/entity-graph.md @@ -47,6 +47,13 @@ Relationships: aggregate**, not a settlement point. - `(SettlementPoint)-[:IN_MARKET]->(Market)` — the 13 canonical tradeable points. - `(BalancingAuthority)-[:GENERATES]->(FuelType)` — observed generation-by-fuel mix. + :::caution + Currently **under-reported**: the degenerate write path deduplicates + generation-by-fuel rows on timestamp alone, so only one fuel per (BA, hour) + survives in storage today. The discovery reads whatever the store has and heals + automatically once that storage fix lands; until then expect one or two fuels per + BA, not the full mix. + ::: - Curated cross-domain edges: `(ERCO)-[:OPERATES]->(ERCOT)` (the EIA-930 ↔ ERCOT nodal join point), `(TEXAS)-[:WEATHER_PROXY_FOR]->(ERCOT)` (nClimDiv Texas is the documented ERCOT footprint), and `(NATGAS)-[:FUELS]->(NG)` when gas-fired From 1576dc70140d08e5bafb50cb446e511cdf1d088a Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Fri, 17 Jul 2026 05:49:46 -0400 Subject: [PATCH 9/9] docs(graph): update fuel-mix caveat for the in-flight storage dedup fix Co-Authored-By: Claude Fable 5 --- website/docs/entity-graph.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/website/docs/entity-graph.md b/website/docs/entity-graph.md index cf7586c..edec3e3 100644 --- a/website/docs/entity-graph.md +++ b/website/docs/entity-graph.md @@ -47,12 +47,12 @@ Relationships: aggregate**, not a settlement point. - `(SettlementPoint)-[:IN_MARKET]->(Market)` — the 13 canonical tradeable points. - `(BalancingAuthority)-[:GENERATES]->(FuelType)` — observed generation-by-fuel mix. - :::caution - Currently **under-reported**: the degenerate write path deduplicates - generation-by-fuel rows on timestamp alone, so only one fuel per (BA, hour) - survives in storage today. The discovery reads whatever the store has and heals - automatically once that storage fix lands; until then expect one or two fuels per - BA, not the full mix. + :::note + The degenerate write path now keys dedup on (timestamp, `fuel_type`), so the full + fuel mix survives storage. Hours ingested **before** that fix still carry a single + arbitrary fuel each; re-materializing `eia930_generation_by_fuel` over the affected + window (EIA-930 retains history) backfills the full mix, and the discovery heals + automatically as it reads whatever the store has. ::: - Curated cross-domain edges: `(ERCO)-[:OPERATES]->(ERCOT)` (the EIA-930 ↔ ERCOT nodal join point), `(TEXAS)-[:WEATHER_PROXY_FOR]->(ERCOT)` (nClimDiv Texas is the