qbiocode.apps.quvine.baselines.gat module

QuVINE GAT Variants: controlled downstream probes for quantum/classical diffusion features.

This module replaces the earlier GCN-MF pathway with a cleaner GAT pathway:

feature_builder(G) -> X_variant -> same GAT encoder -> embeddings / logits

Important design choice:

The GAT is NOT quantum. Quantum content enters only through the input representation, e.g. QCal-Heat/QCal-Poly features calibrated to CTQW/DTQW targets, or direct CTQW/DTQW embeddings computed elsewhere.

Supported input variants:

raw_structural : scalable structural node features provided : user-provided features rwr : classical random-walk-with-restart diffusion features heat_fixed : classical heat kernel features with user-provided t poly_fixed : classical polynomial features with user-provided coeffs heat_qcal_ctqw : heat kernel calibrated to CTQW targets poly_qcal_ctqw : polynomial filter calibrated to CTQW targets heat_qcal_rwr : heat kernel calibrated to RWR/classical-walk targets poly_qcal_rwr : polynomial filter calibrated to RWR/classical-walk targets direct_ctqw : direct CTQW embedding/features provided by user direct_dtqw : direct DTQW embedding/features provided by user

Dependencies beyond the previous GCN-MF file:

Required: PyTorch, NumPy, SciPy, NetworkX

Optional: scikit-learn only for StandardScaler if normalize_structural_features=True; code falls back to internal standardization when sklearn is absent.

No PyTorch Geometric is required. This file implements a sparse edge-index GAT using only PyTorch index_add/scatter_reduce operations.

Summary

Classes:

EdgeIndexGATLayer	Sparse edge-index GAT layer using only PyTorch.
GATConfig	GATConfig(hidden_dim: 'int' = 64, output_dim: 'int' = 128, num_layers: 'int' = 2, heads: 'int' = 4, dropout: 'float' = 0.5, attention_dropout: 'float' = 0.2, negative_slope: 'float' = 0.2, residual: 'bool' = True)
GATEncoder	GAT encoder returning node embeddings.
GATNodeClassifier	GAT encoder plus linear classifier.
TrainConfig	TrainConfig(epochs: 'int' = 200, lr: 'float' = 0.005, weight_decay: 'float' = 0.0005, patience: 'int' = 25, edge_batch_size: 'int' = 8192, val_edge_fraction: 'float' = 0.1, random_state: 'int' = 42, device: 'str' = 'cpu', verbose: 'bool' = False)

Functions:

align_features	Align feature matrix to nodelist.
apply_heat_filter
apply_polynomial_filter
apply_rwr_filter	Random walk with restart / PPR-style feature diffusion.
as_numpy
build_gat_input_features	Construct input features for a GAT variant with optional train/test separation.
build_normalized_laplacian
build_structural_features	Scalable structural node features for graphs without attributes.
calibrate_heat_kernel	Fit heat time to target distributions.
calibrate_polynomial_filter	Fit monomial polynomial coefficients to target distributions.
edge_index_from_graph	Return directed edge_index [2, E_dir], including both directions for undirected G.
edge_scores_dot
generate_gat_embedding	Generate unsupervised GAT embeddings for one feature variant.
generate_gat_embedding_by_method_name	Generate GAT embedding using standardized method names.
generate_multiple_gat_embeddings	Generate embeddings for several variants using identical kwargs/configs.
get_direct_walk_features	Fetch direct CTQW/DTQW features computed by the SGNS/walk pipeline.
get_nodelist	Return the node order to use, validating that it belongs to G.
heat_taylor_coeffs	Classical fixed polynomial baseline approximating exp(-tL).
make_edge_batch
require_torch	Check if PyTorch is available and raise informative error if not.
row_normalize
sample_negative_edges
scipy_sparse_to_torch_coo
set_seed
split_edges
standardize_columns	Standardize columns with optional train-only statistics.
train_gat_link_reconstruction	Train a GAT encoder with edge reconstruction and return embeddings.
train_gat_node_classifier	Train GAT directly for node classification and return embeddings plus metadata.

Reference

class GATConfig(hidden_dim=64, output_dim=128, num_layers=2, heads=4, dropout=0.5, attention_dropout=0.2, negative_slope=0.2, residual=True)

Bases: object

hidden_dim: int = 64

output_dim: int = 128

num_layers: int = 2

heads: int = 4

dropout: float = 0.5

attention_dropout: float = 0.2

negative_slope: float = 0.2

residual: bool = True

class TrainConfig(epochs=200, lr=0.005, weight_decay=0.0005, patience=25, edge_batch_size=8192, val_edge_fraction=0.1, random_state=42, device='cpu', verbose=False)

Bases: object

epochs: int = 200

lr: float = 0.005

weight_decay: float = 0.0005

patience: int = 25

edge_batch_size: int = 8192

val_edge_fraction: float = 0.1

random_state: int = 42

device: str = 'cpu'

verbose: bool = False

require_torch()

Check if PyTorch is available and raise informative error if not.

Raises:

TorchNotAvailableError – If PyTorch is not installed

Return type:

None

set_seed(seed)

Return type:

None

get_nodelist(G, nodelist=None)

Return the node order to use, validating that it belongs to G.

Every consumer of this ordering indexes G by these ids – degree views, Laplacian construction, per-node metric dicts – so an id that is not in the graph has no meaningful feature row. Rejecting it here names the offending ids; letting it through produced either a silently zeroed feature column or an inscrutable numpy shape error several frames deeper.

Raises:

ValueError – if nodelist contains ids absent from G, or duplicates.

Return type:

List

as_numpy(X)

Return type:

ndarray

row_normalize(X, eps=1e-12)

Return type:

ndarray

standardize_columns(X, train_mask=None, eps=1e-12)

Standardize columns with optional train-only statistics.

Parameters:

• X (ndarray) – Feature matrix [N, d]

• train_mask (Optional[ndarray]) – Boolean mask for training nodes [N]. If None, uses all nodes.

• eps (float) – Small constant for numerical stability

Return type:

ndarray

Returns:

Standardized features [N, d]

scipy_sparse_to_torch_coo(A, device='cpu')

Return type:

Tensor

edge_index_from_graph(G, nodelist, add_self_loops=True, device='cpu')

Return directed edge_index [2, E_dir], including both directions for undirected G.

Return type:

Tensor

build_normalized_laplacian(G, nodelist, normalized=True, weight='weight')

Return type:

csr_matrix

build_structural_features(G, nodelist=None, train_mask=None, normalize=True)

Scalable structural node features for graphs without attributes.

Parameters:

• G (Graph) – NetworkX graph

• nodelist (Optional[Sequence]) – Ordered list of nodes (default: G.nodes())

• train_mask (Optional[ndarray]) – Boolean mask [N] indicating training nodes. If None, uses all nodes (acceptable for transductive tasks).

• normalize (bool) – Whether to standardize features

Returns:

[degree, log_degree, clustering, core_number, pagerank,

log_triangles, avg_neighbor_degree, local_degree_fraction]

Return type:

Feature matrix [N, 8] with columns

Note

For transductive node classification, train_mask can be None since the full graph structure is known. For inductive tasks or to prevent test leakage, always provide train_mask.

Degenerate graphs:

Any structural column that networkx cannot compute for this graph class (clustering and triangles on multigraphs, k-core with self-loops, pagerank that fails to converge) falls back to zeros – pagerank to the uniform distribution – and logs the reason at DEBUG. The feature matrix always has the documented width; it is never partially built.

align_features(features, nodelist, feature_nodes=None)

Align feature matrix to nodelist.

If feature_nodes is None, features are assumed already ordered as nodelist.

Return type:

ndarray

apply_heat_filter(L, X, t)

Return type:

ndarray

apply_polynomial_filter(L, X, coeffs)

Return type:

ndarray

apply_rwr_filter(G, X, nodelist, alpha=0.15, steps=50, tol=1e-06)

Random walk with restart / PPR-style feature diffusion.

Iteration: Z_{k+1} = alpha X + (1-alpha) P Z_k, where P = D^{-1} A is row-stochastic. This is classical, not quantum.

Return type:

ndarray

calibrate_heat_kernel(L, targets, node_to_idx, t_grid=None, loss='l2')

Fit heat time to target distributions.

targets can be CTQW targets, DTQW targets, or classical RWR targets. The routine is agnostic: it simply matches target probabilities on sampled node sets.

Return type:

Tuple[float, float]

calibrate_polynomial_filter(L, targets, node_to_idx, K=4, ridge=1e-05)

Fit monomial polynomial coefficients to target distributions.

Critical bug fix relative to the earlier code: we do NOT column-normalize the polynomial basis during fitting unless that same normalization is also used at application time. Ridge handles conditioning.

Return type:

ndarray

heat_taylor_coeffs(t, K)

Classical fixed polynomial baseline approximating exp(-tL).

Return type:

ndarray

get_direct_walk_features(direct_features, key, nodelist, feature_nodes=None)

Fetch direct CTQW/DTQW features computed by the SGNS/walk pipeline.

This function intentionally does not recompute CTQW/DTQW. Your attached pipeline trains SGNS embeddings from corpora produced by BaseWalker and then stores embeddings in an EmbeddingStore; pass the resulting array here.

Return type:

ndarray

class EdgeIndexGATLayer(in_dim, out_dim, heads=4, dropout=0.5, attention_dropout=0.2, negative_slope=0.2, concat=True, residual=True)

Bases: Module

Sparse edge-index GAT layer using only PyTorch.

This layer computes attention over incoming neighbors for each destination node. edge_index[0] = source, edge_index[1] = destination.

reset_parameters()

Return type:

None

forward(x, edge_index)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Return type:

Tensor

class GATEncoder(input_dim, config)

Bases: Module

GAT encoder returning node embeddings.

forward(x, edge_index)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Return type:

Tensor

class GATNodeClassifier(input_dim, num_classes, config)

Bases: Module

GAT encoder plus linear classifier.

forward(x, edge_index)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Return type:

Tensor

get_embeddings(x, edge_index)

Return type:

Tensor

sample_negative_edges(n_nodes, existing_edges, num_samples, rng, max_attempt_factor=50)

Return type:

List[Tuple[int, int]]

split_edges(edges, val_fraction, seed)

Return type:

Tuple[List[Tuple[int, int]], List[Tuple[int, int]]]

edge_scores_dot(z, edge_index_2xn)

Return type:

Tensor

make_edge_batch(pos_edges, existing_edges, n_nodes, batch_size, rng, device)

Return type:

Tuple[Tensor, Tensor]

train_gat_link_reconstruction(G, X, nodelist=None, gat_config=None, train_config=None, train_graph_for_message_passing=None)

Train a GAT encoder with edge reconstruction and return embeddings.

Return type:

Tuple[ndarray, Dict]

train_gat_node_classifier(G, X, y, train_mask, val_mask, nodelist=None, gat_config=None, train_config=None, test_mask=None)

Train GAT directly for node classification and return embeddings plus metadata.

Return type:

Tuple[ndarray, Dict]

build_gat_input_features(G, variant, nodelist=None, train_mask=None, base_features=None, base_feature_nodes=None, direct_features=None, direct_feature_nodes=None, ctqw_targets=None, dtqw_targets=None, rwr_targets=None, heat_t=None, poly_coeffs=None, poly_K=4, poly_ridge=1e-05, rwr_alpha=0.15, rwr_steps=50, normalize_laplacian=True, t_grid=None)

Construct input features for a GAT variant with optional train/test separation.

Parameters:

• G (Graph) – NetworkX graph

• variant (str) – Feature variant name

• nodelist (Optional[Sequence]) – Ordered list of nodes

• train_mask (Optional[ndarray]) – Boolean mask [N] for training nodes. If None, uses all nodes (acceptable for transductive tasks). Prevents test leakage.

• base_features (Union[ndarray, Tensor, None]) – Pre-computed base features (if provided, assumed pre-normalized)

• ... (other args as before)

Returns:

Feature matrix and metadata dict

Return type:

(X, meta)

generate_gat_embedding(G, variant='raw', nodelist=None, base_features=None, base_feature_nodes=None, direct_features=None, direct_feature_nodes=None, ctqw_targets=None, dtqw_targets=None, rwr_targets=None, heat_t=None, poly_coeffs=None, poly_K=4, poly_ridge=1e-05, rwr_alpha=0.15, rwr_steps=50, normalize_laplacian=True, t_grid=None, gat_config=None, train_config=None, train_graph_for_message_passing=None)

Generate unsupervised GAT embeddings for one feature variant.

Return type:

Tuple[ndarray, Dict]

generate_multiple_gat_embeddings(G, variants, **kwargs)

Generate embeddings for several variants using identical kwargs/configs.

Return type:

Tuple[Dict[str, ndarray], Dict[str, Dict]]

generate_gat_embedding_by_method_name(G, method_name, embedding_dim=128, nodelist=None, base_features=None, ctqw_targets=None, dtqw_targets=None, rwr_targets=None, heat_t=None, poly_K=4, rwr_alpha=0.15, gat_config=None, train_config=None, **kwargs)

Generate GAT embedding using standardized method names.

This function maps the 12 standardized GAT method names to the existing variant system in generate_gat_embedding.

Supported methods:

• gat_baseline: Raw structural features

• gat_heat: Heat kernel filter only

• gat_poly: Polynomial filter only

• gat_rwr: RWR walk only

• gat_ctqw: CTQW walk only (direct features)

• gat_dtqw: DTQW walk only (direct features)

• gat_rwr_heat: RWR + heat filter

• gat_rwr_poly: RWR + polynomial filter

• gat_ctqw_heat: CTQW + heat filter (quantum calibrated)

• gat_ctqw_poly: CTQW + polynomial filter (quantum calibrated)

• gat_dtqw_heat: DTQW + heat filter (quantum calibrated)

• gat_dtqw_poly: DTQW + polynomial filter (quantum calibrated)

Parameters:

• G (Graph) – NetworkX graph

• method_name (str) – Standardized method name (e.g., ‘gat_baseline’, ‘gat_ctqw_heat’)

• embedding_dim (int) – Output embedding dimension

• nodelist (Optional[Sequence]) – Ordered list of nodes

• base_features (Union[ndarray, Tensor, None]) – Pre-computed base features

• ctqw_targets (Optional[Sequence[Mapping]]) – CTQW calibration targets

• dtqw_targets (Optional[Sequence[Mapping]]) – DTQW calibration targets

• rwr_targets (Optional[Sequence[Mapping]]) – RWR calibration targets

• heat_t (Optional[float]) – Heat kernel time parameter (for fixed variants)

• poly_K (int) – Polynomial degree

• rwr_alpha (float) – RWR restart probability

• gat_config (Optional[GATConfig]) – GAT model configuration

• train_config (Optional[TrainConfig]) – Training configuration

• **kwargs – Additional arguments

Return type:

ndarray

Returns:

Node embeddings [N, embedding_dim]

Example

>>> G = nx.karate_club_graph()
>>> emb = generate_gat_embedding_by_method_name(G, 'gat_baseline', embedding_dim=64)
>>> print(emb.shape)  # (34, 64)

Raises:

ValueError – If method_name is not recognized