Adjazenzmasken für Makrogruppen in der Asset-Attention erstellen
Zusammenfassung
Dieses Hilfsprogramm erstellt aus Asset-Gruppen und optionalen Verknüpfungen zwischen Gruppen eine boolesche Adjazenzmatrix. Assets derselben Makrogruppe werden miteinander verbunden. Jede angegebene gruppenübergreifende Kante verbindet alle Assets der beiden Gruppen in beide Richtungen. Optionale Selbstverbindungen ermöglichen jedem Asset, sich selbst zu berücksichtigen. Assets ohne Zuordnung werden einer unbekannten Gruppe zugeteilt und daher mit anderen nicht zugeordneten Assets verbunden.
Eine zweite Funktion wandelt die Adjazenzmatrix in eine Attention-Maske um, indem sie nicht benachbarte Paare als unzulässig markiert. Der Code prüft vor der Umwandlung, ob die Eingabematrix quadratisch ist. Damit lässt sich eine Makrogruppenstruktur einfach als Vorwissen in ein Attention-Modell einbringen. Das Programm leitet Gruppen oder Beziehungen jedoch nicht aus Marktdaten ab, und sein Nutzen hängt von Qualität und Aktualität der bereitgestellten Zuordnungen und gruppenübergreifenden Verbindungen ab. Das Dokument beschreibt die Implementierung, nicht empirische Ergebnisse oder eine Trading-Auswertung.
Kernaussagen
- Assets derselben Gruppe erhalten in der Adjazenzmatrix gegenseitige Verbindungen.
- Gruppenübergreifende Verbindungen verknüpfen jedes Asset der einen Gruppe symmetrisch mit jedem Asset der anderen.
- Ein Parameter legt fest, ob Selbstverbindungen enthalten sind.
- Nicht zugeordnete Assets teilen eine unbekannte Gruppe und werden deshalb miteinander verbunden.
- Die Attention-Maske markiert jedes nicht benachbarte Asset-Paar als unzulässig.
Schlagwörter
Volltext
# graph.py
```py
"""Macroeconomic graph prior utilities.
Builds an asset-level adjacency matrix from macro group labels and
cross-group edges for adjacency-masked attention in DeePM.
"""
from __future__ import annotations
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True, slots=True)
class MacroGraph:
"""Asset-level macro graph."""
assets: list[str]
groups: list[str]
adjacency: np.ndarray # Boolean (N, N)
def build_macro_adjacency(
*,
assets: Sequence[str],
asset_to_group: dict[str, str],
cross_group_edges: Iterable[tuple[str, str]] = (),
include_self_loops: bool = True,
) -> MacroGraph:
"""Build a boolean adjacency matrix from group labels and group edges.
Parameters
----------
assets: Asset identifiers matching the price panel columns.
asset_to_group: Mapping from asset -> macro group label.
cross_group_edges: Undirected (group_a, group_b) edges.
include_self_loops: If True, sets A[i,i] = True.
"""
assets_list = [str(a) for a in assets]
groups = [str(asset_to_group.get(a, "UNKNOWN")) for a in assets_list]
n = len(assets_list)
adj = np.zeros((n, n), dtype=bool)
if include_self_loops:
np.fill_diagonal(adj, True)
group_to_indices: dict[str, list[int]] = {}
for i, g in enumerate(groups):
group_to_indices.setdefault(g, []).append(i)
for indices in group_to_indices.values():
idx = np.array(indices, dtype=int)
adj[np.ix_(idx, idx)] = True
for g1, g2 in cross_group_edges:
idx1 = group_to_indices.get(g1, [])
idx2 = group_to_indices.get(g2, [])
if not idx1 or not idx2:
continue
a = np.array(idx1, dtype=int)
b = np.array(idx2, dtype=int)
adj[np.ix_(a, b)] = True
adj[np.ix_(b, a)] = True
return MacroGraph(assets=assets_list, groups=groups, adjacency=adj)
def adjacency_to_attn_mask(adjacency: np.ndarray) -> np.ndarray:
"""Convert boolean adjacency to attention mask (True = disallowed)."""
if adjacency.ndim != 2 or adjacency.shape[0] != adjacency.shape[1]:
raise ValueError("adjacency must be square (N,N)")
return ~adjacency
```Vollständig mit Quellenangabe unter der Lizenz der Quelle angezeigt. Lizenz: MIT
Diese Zusammenfassung wurde vom Research-Agenten von Stratmill anhand des Originals verfasst; sie ist keine Kopie der Quelle.