1import numpy as np
 2
 3META = {
 4    "name": "graph_landmark_forecast",
 5    "domain": "graph-nn",
 6    "description": "Directed graph node forecasting: predict one-step neighborhood propagation from node signals; landmark signatures are valid structural positional features."
 7}
 8
 9# Fixed directed graph, deliberately with distinct incoming signatures for landmarks.
10def _graph(n=20):
11    rng = np.random.default_rng(731)
12    A = (rng.random((n, n)) < 0.22).astype(np.float32)
13    np.fill_diagonal(A, 0)
14    # Avoid empty incoming neighborhoods and make the graph nondegenerate.
15    for v in range(n):
16        if A[:, v].sum() == 0:
17            A[rng.integers(n), v] = 1
18    return A
19
20
21def get_dataset(seed, n_train, n_test):
22    rng = np.random.default_rng(int(seed) + 1200)
23    A = _graph()
24    n = A.shape[0]
25    deg = A.sum(0, keepdims=True).T
26    P = A / np.maximum(deg, 1.0)
27    total = n_train + n_test
28    # Inputs are independent node signals; targets require the directed graph.
29    x = rng.normal(size=(total, n, 1)).astype(np.float32)
30    y = (0.55*x[:, :, 0] + 0.45*np.einsum('vu,bu->bv', P, x[:, :, 0])).astype(np.float32)
31    return {"xtr": x[:n_train], "ytr": y[:n_train, :, None],
32            "xte": x[n_train:], "yte": y[n_train:, :, None],
33            "task": "regression", "metric": "mse", "out_dim": n,
34            "adjacency": A}