构建宏观分组邻接掩码以用于资产注意力模型

代码《交易机器学习》

总结

该工具根据资产分组标签和可选的组间连接，构建布尔型资产间邻接矩阵。分配到同一宏观组的资产彼此连接；每条提供的跨组边都会以双向方式连接两组中的所有成员。可选的自连接允许每项资产关注自身。没有映射的资产会被分配到未知组，因此会与其他未映射资产相连。

第二个函数会将邻接矩阵转换为注意力掩码，把不相邻的资产对标记为不允许连接。转换前，代码会检查输入矩阵是否为方阵。这提供了一种将宏观分组先验编码到注意力模型中的简单方法，但它不会根据市场数据估算分组或关系，其效用取决于所提供映射和跨组边的质量与时效性。本文提供的是实现细节，而非实证结果或交易评估。

核心观点

邻接矩阵会为同组资产建立彼此之间的连接。
跨组连接会以对称方式，将一组中的每项资产连接到另一组中的每项资产。
可通过参数选择是否加入自连接。
没有映射的资产归入同一未知组，因此彼此连接。
注意力掩码会将所有不相邻的资产对标记为不允许连接。

# 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

```

在遵守原作品许可的前提下，附作者信息全文展示。许可协议: MIT

此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。