블록체인 거래 그래프와 GNN으로 노드 행동 분석하기

거래 그래프에서 드러나는 노드의 역할

퍼블릭 블록체인의 거래와 계약 상호작용 로그는 주소, 컨트랙트, 검증자 식별자를 노드로 두는 그래프로 표현할 수 있다. 트랜잭션, 이벤트 로그, 토큰 이동, 콜 스택의 호출 관계는 간선이 된다. 이때 간선에는 방향, 가중치, 시간 정보를 함께 담는다.

이 모델은 사기 탐지, 리스크 모니터링, 밸리데이터 간 인과관계 파악처럼 온체인 인텔리전스를 다루는 작업에 쓸 수 있다. 영어 토픽명은 Blockchain Network Analytics: Centrality and Graph Neural Networks다.

그래프는 방향 그래프, 가중 그래프, 시간에 따라 변하는 동적 그래프로 구성할 수 있다. 시간 스냅샷이나 스트리밍 방식으로 변화를 다루며, 동일한 두 노드 사이에 여러 관계가 존재하는 멀티그래프도 허용한다.

연결 구조를 읽는 중심성 지표

연결 수만으로 영향력을 근사하려면 Degree, In Degree, Out Degree를 사용한다. 비용이 낮고 지역적인 연결 특성을 빠르게 반영한다.

경로를 기준으로 노드의 역할을 보려면 Betweenness와 Closeness가 적합하다. 각각 매개 역할과 접근성을 측정하므로 전체 구조를 반영할 수 있지만 계산 비용은 높다. Eigenvector와 PageRank는 연결된 이웃의 연결 품질까지 반영하며, 스팸이나 세탁 패턴을 구분하는 데 유효하다.

지표	성능(복잡도)	확장성	해석 용이성	강점/적합도
In/Out Degree	O(E)	매우 높음	높음	스팸 송신, 펌핑/드레이닝 탐지
PageRank	~O(E) per iter	높음	중간	허브/권위 노드 식별, 세탁 라우팅 가중 탐지
Betweenness(근사)	O(kE)	중간	중간	브리지/릴레이 노드, 라우팅 지점 포착
Closeness	O(EN)	낮음	중간	네트워크 접근성 측정, 확산 속성 평가
Eigenvector	스펙트럴	중간	중간	고품질 이웃 연결 반영, 중심 허브 검출

운영에서는 PageRank와 In/Out Degree를 기본 세트로 두고, Betweenness는 샘플 근사로 활용할 수 있다.

GNN이 구조와 속성을 함께 학습하는 방식

GNN은 메시지 패싱으로 이웃 노드의 특징을 집계한 뒤 노드 임베딩을 만든다. 따라서 그래프 구조와 속성을 동시에 학습할 수 있다.

대표적인 작업은 사기 여부를 가르는 노드 분류, 세탁 경로를 찾는 링크 예측, 비정상 행동을 찾는 이상 탐지다. 정적 GNN은 스냅샷 기반으로 학습하고, 시간적 의존성이 중요하면 TGAT나 TGN을 적용한다.

데이터는 블록, 트랜잭션, 로그, 토큰 전송, 지갑 라벨, 사기 블랙리스트에서 가져와 인제스트, 정규화, 피처링 순서로 처리한다. 이 과정에서는 체인 재구성(reorg), 파이널리티 지연, 중복 및 결측 레코드를 다뤄야 한다.

주소 인덱싱과 에지의 방향·가중치(금액/횟수/가스), 시간 구간 윈도잉이 그래프 구성의 기반이 된다. call과 transfer 같은 멀티리레이션은 분리한다. 저장소는 그래프 DB(Neo4j), 컬럼너(Parquet)+오프라인 PyG를 사용할 수 있으며, 대규모 환경에서는 분산 스파스 텐서를 쓴다.

분석 계층에서는 네트워크X/PageRank 파이프라인과 k-sampling, Sketch 같은 근사화 기법으로 확장성을 확보한다. 중심성 결과를 GNN 입력 피처로 넣으면 해석 가능성도 높아진다. 모델은 GCN, GraphSAGE, GAT를 사용할 수 있고, 대규모 그래프에는 Neighbor Sampling, Cluster-GCN, GraphSAINT를 적용한다. 라벨이 부족할 때는 반지도(Semi-supervised) 및 자기지도(Contrastive) 학습으로 보완한다.

데이터부터 리스크 점수까지의 흐름

⤢✕

체인 데이터를 인제스트한 뒤 그래프와 피처를 만들고, 중심성 및 GNN 점수를 리스크 API로 제공한다. 파이널리티 이전 블록은 보류하며 재구성이 발생하면 롤백과 재계산을 수행한다. 데이터 드리프트와 레이턴시도 알림 대상으로 둔다.

스냅샷 타임스탬프를 고정하고 동일 윈도우에서 중심성과 GNN을 함께 산출해야 결과의 일관성을 지킬 수 있다. 피처 스토어는 버전으로 관리한다. 서빙은 배치 점수화와 스트리밍 알림을 병행하고, 시각화 화면이나 리스크 API를 제공한다. 공격자 적응(컨터 전략)과 데이터 드리프트를 감지해 주기적으로 재학습한다.

온체인 리스크 분석에 적용하는 장면

DeFi·NFT 사기 주소 탐지

최근 30일 거래 그래프를 만든 뒤 PageRank와 OutDegree를 피처로 사용해 피싱·라그풀 라벨의 GraphSAGE 분류 모델을 학습하고, 점수 상위 대상을 알림으로 보낸다. 대량 소액 송금, 풀 라우팅, 분산 인출은 구조적인 패턴으로 포착할 수 있다.

거래소 KYT 심사

신규 지갑의 입출금 그래프에서 중심성, 토큰 다양성, 시간 간격 피처를 만들고 GNN으로 점수화한다. 결과는 심사 큐 자동 분류에 사용한다. 재사용 머니뮬 네트워크를 조기에 차단하고 경보 과다를 줄이는 효과를 기대할 수 있다.

PoS·MEV 검증자와 릴레이 관계

블록 제안과 검증 상호작용을 그래프로 구성한 다음, 브리지 및 리오그 민감 노드의 매개중심성을 평가해 알고리즘 파라미터를 조정한다. 라우팅 탄력성을 높이고 단일 실패점 위험을 완화하는 데 활용한다.

중심성 계산과 GraphSAGE 예시

전제조건은 다음과 같다.

Python 3.10+, pandas 2.x, networkx 3.2+, PyTorch 2.3+, PyG 2.5+
예시 데이터: transactions.csv(from,to,value,block,timestamp,label[0/1] 선택)

중심성 계산(NetworkX):

# python -m pip install pandas networkx
import pandas as pd
import networkx as nx

df = pd.read_csv("transactions.csv")  # columns: from,to,value,block,timestamp
# 최근 30일 필터링 예시(옵션)
# df = df[df["timestamp"] >= df["timestamp"].max() - 30*24*3600]

G = nx.DiGraph()
edges = list(zip(df["from"], df["to"], df["value"]))
G.add_weighted_edges_from(edges)

# 기본 피처
deg_out = dict(G.out_degree(weight=None))
deg_in = dict(G.in_degree(weight=None))

# PageRank (가중치 사용)
pr = nx.pagerank(G, alpha=0.85, weight="weight")

# Betweenness 근사(k 샘플)
btw = nx.betweenness_centrality(G, k=min(500, G.number_of_nodes()), seed=42)

feat = pd.DataFrame({
    "addr": list(G.nodes()),
    "deg_out": [deg_out.get(n, 0) for n in G.nodes()],
    "deg_in": [deg_in.get(n, 0) for n in G.nodes()],
    "pagerank": [pr.get(n, 0.0) for n in G.nodes()],
    "betweenness": [btw.get(n, 0.0) for n in G.nodes()],
})
print(feat.sort_values("pagerank", ascending=False).head(10))

GNN 노드 분류(PyTorch Geometric, GraphSAGE):

# python -m pip install torch torch_geometric pandas numpy scikit-learn
import pandas as pd
import numpy as np
import torch
from torch import nn
from torch_geometric.data import Data
from torch_geometric.nn import SAGEConv
from sklearn.preprocessing import StandardScaler

tx = pd.read_csv("transactions.csv")
# 주소 인덱싱
nodes = pd.Index(pd.unique(tx[["from","to"]].values.ravel()))
nid = {a:i for i,a in enumerate(nodes)}

# edge_index 구성
src = tx["from"].map(nid).values
dst = tx["to"].map(nid).values
edge_index = torch.tensor([src, dst], dtype=torch.long)

# 중심성 피처 로드(앞 단계 feat 기준)
feat = pd.read_csv("centrality_features.csv") if False else None  # 실제로는 위에서 산출
# 데모용 간단 피처: in/out degree 근사
deg_out = np.bincount(src, minlength=len(nodes))
deg_in = np.bincount(dst, minlength=len(nodes))
X = np.vstack([deg_in, deg_out]).T.astype(float)

# 스케일링
scaler = StandardScaler()
X = scaler.fit_transform(X)
x = torch.tensor(X, dtype=torch.float)

# 라벨(있다면 사용), 없으면 더미
labels = np.zeros(len(nodes), dtype=int)
if "label" in tx.columns:
    # 노드 라벨 집계(거래소 라벨 등): 다수결
    lab_map = tx.dropna(subset=["label"]).groupby("from")["label"].max()
    for a, l in lab_map.items():
        labels[nid[a]] = int(l)
y = torch.tensor(labels, dtype=torch.long)

# 학습/검증 분할
num_nodes = len(nodes)
idx = np.arange(num_nodes)
np.random.seed(42)
np.random.shuffle(idx)
train_idx = torch.tensor(idx[: int(0.6*num_nodes)])
val_idx = torch.tensor(idx[int(0.6*num_nodes): int(0.8*num_nodes)])
test_idx = torch.tensor(idx[int(0.8*num_nodes):])

data = Data(x=x, edge_index=edge_index, y=y)

class SAGE(torch.nn.Module):
    def __init__(self, in_ch, hid=64, out_ch=2):
        super().__init__()
        self.conv1 = SAGEConv(in_ch, hid)
        self.conv2 = SAGEConv(hid, out_ch)
        self.dropout = nn.Dropout(0.3)
    def forward(self, x, edge_index):
        x = self.conv1(x, edge_index).relu()
        x = self.dropout(x)
        x = self.conv2(x, edge_index)
        return x

device = "cuda" if torch.cuda.is_available() else "cpu"
model = SAGE(data.num_node_features, 64, 2).to(device)
opt = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-4)
loss_fn = nn.CrossEntropyLoss()

def train_step(idx):
    model.train()
    opt.zero_grad()
    out = model(data.x.to(device), data.edge_index.to(device))
    loss = loss_fn(out[idx].to(device), data.y[idx].to(device))
    loss.backward()
    opt.step()
    return float(loss)

@torch.no_grad()
def eval_step(idx):
    model.eval()
    out = model(data.x.to(device), data.edge_index.to(device))
    pred = out[idx].argmax(dim=-1).cpu()
    acc = (pred == data.y[idx]).float().mean().item()
    return acc

for epoch in range(1, 51):
    loss = train_step(train_idx)
    if epoch % 10 == 0:
        val_acc = eval_step(val_idx)
        print(f"epoch {epoch} loss {loss:.4f} val_acc {val_acc:.3f}")

test_acc = eval_step(test_idx)
print("test_acc:", round(test_acc, 3))

대규모 그래프에서는 NeighborLoader/mini-batch 샘플링, FP16 혼합정밀, 스파스 텐서를 활용한다. 동적 그래프에는 슬라이딩 윈도우 피처와 TGN/TGAT를 적용하며, 라벨 지연은 관측 시점을 고정해 누설을 방지한다. 중심성·규칙 점수와 GNN 확률을 앙상블하고, 최단 경로 서브그래프 시각화로 사례 기반 설명을 제공할 수 있다.

성능과 비용 사이에서 지켜야 할 기준

규칙 기반 방식과 비교해 재현율은 30~~60%p, 정밀도는 10~~30%p 향상 가능하다. 이는 체인과 도메인에 의존하며 최신 정보 확인이 필요하다. 경보 과다는 20~~50% 감소하고, 애널리스트 처리 속도는 2~~5배 개선될 수 있다. 근사 중심성과 샘플링 GNN을 사용하면 스냅샷 10^7 노드/에지 규모에서도 일 단위 재산출이 가능하다.

파이널리티 대기, 재구성 롤백 전략, 버전드 피처 저장은 데이터 정합성을 위해 필요하다. 여기에는 실시간성과 일관성의 트레이드오프가 있다.

라벨이 희소한 환경에서는 반지도 또는 자기지도를 채택하고, 펄스 라벨링과 샤프닝 같은 노이즈 라벨 견고화 기술을 적용한다. 정확도 향상과 계산비용 사이에서는 Betweenness를 근사화하고 GNN에는 이웃 샘플링을 적용하는 방식으로 절충한다.

주소 분할과 체인 홉 회피 같은 적대적 회피에는 경로 기반 피처, 다체인 통합, 메타데이터 결합이 필요하다. 온체인 공개 데이터를 사용하더라도 KYC 데이터를 결합할 때는 최소 수집과 목적 제한 원칙을 준수해야 한다.

초기 구성은 PageRank, Degree, GraphSAGE로 시작할 수 있다. 라벨이 축적되면 동적 GNN과 고급 피처로 분석 범위를 확장한다.