동적 네트워크에서 시간 기반 링크 예측 설계

시간이 바뀌면 연결의 의미도 바뀐다. 동적 네트워크에서는 노드 사이의 관계를 그래프로만 보지 않고, 언제 발생했으며 어떤 순서로 이어졌는지까지 보존해야 미래 연결을 예측할 수 있다. 추천, 사기 탐지, 보안, 지식그래프 갱신처럼 대규모 시스템에 적용할 때는 성능뿐 아니라 안정성과 운영 편의도 함께 맞춰야 한다.

이벤트 시점이 포함된 그래프와 예측 문제

Temporal Graph는 노드와 엣지에 타임스탬프를 붙인 그래프다. 표현 방식은 특정 시점의 상태를 묶어 다루는 스냅샷 모델과, 연속적으로 들어오는 이벤트를 기록하는 이벤트 스트림 모델로 나뉜다.

시간 정보는 절대시간(Unix time), 상대시간(Δt), 주기성(weekday/hour)으로 인코딩할 수 있다. TGAT의 Time Encoding처럼 포지셔널 인코딩을 활용하는 방식도 있다.

시간 기반 링크 예측은 시점 t까지 관찰한 그래프에서 t’>t의 엣지 존재 확률을 추정하는 문제다. 이때 시간 순서를 보존하지 않으면 미래 정보가 학습에 섞이는 누수가 발생한다.

모델은 시간 감쇠 기반 휴리스틱(AA/PA+decay), Temporal GNN(TGAT/TGN/DySAT), 점과정(Hawkes/Neural Hawkes), Temporal GNN과 Hawkes를 결합한 하이브리드로 나눠 볼 수 있다.

검증도 시간축을 따라야 한다. train ≤ val ≤ test 순서로 이벤트 시간을 분리하고 롤링 또는 누적 윈도우를 사용한다. 부정 샘플은 t’ 시점의 미관찰 엣지에서 골라야 하며, 과거나 미래의 엣지가 섞이지 않게 해야 한다. 희소 그래프에서는 AUC-PR을, 추천 문제에서는 Hit@k와 MRR을, 시간 구간별 비교에는 시간 조건부 지표를 함께 쓴다. 미발생 이벤트가 포함되는 검열 데이터 처리도 고려 대상이다.

데이터 시점과 피처를 맞추는 방법

스냅샷은 배치 처리가 단순한 반면, 스트림은 더 미세한 시간 해상도를 다룰 수 있다. 도메인에 따라 두 방식을 혼합할 수 있다.

시간 창은 슬라이딩, 텀블링, 세션 윈도우로 설계한다. 늦게 도착하는 이벤트는 워터마크로 처리한다. As-of join을 이용한 타임 트래블 조인은 시점별 피처 스냅샷을 보장하며, Snapshot Isolation은 이 과정에서의 누수를 막는다.

신규 노드에는 메타·속성 임베딩 또는 사전학습 백오프를 사용해 초기화할 수 있다. 이후 온라인 업데이트로 워밍업한다. 시간 유의적 네거티브 샘플링에서는 동일 시간대와 동일 노드 차수 분포를 유지해 클래스 불균형을 완화한다.

구조와 시간을 함께 학습하는 모델

TGAT은 시간 인코딩과 Attention을, TGN은 메모리와 메시지 패싱을, DySAT은 스냅샷 Self-Attention을 사용한다. 이 계열은 동적 이웃과 시간 정보를 직접 학습한다.

Hawkes는 자가·상호흥분성을, Neural Hawkes는 신경 ODE/RNN을 활용한다. 이벤트 강도 함수로 발생 시점을 예측하는 데 강점이 있다. 구조 표현은 GNN으로 만들고 이벤트 강도는 Hawkes로 다루는 하이브리드는 두 접근의 시너지를 통해 고정밀 링크 예측을 노린다.

수집 계층에서는 Kafka/PubSub 기반 스트리밍 수집, 배치 백필, 늦은 이벤트 보정이 필요하다. 학습은 배치 학습과 인크리멘털 메모리 업데이트를 조합하고, 서비스는 ANN 인덱스와 캐시를 이용해 저지연으로 제공할 수 있다. 운영 단계에서는 개념 변화, 시간 누수, 피처 스토어 버저닝을 관측한다.

재현성을 위해 시계열 시드와 버전을 고정하고 피처 라인리지를 관리한다. 신뢰성은 높아지지만 스토리지 증가는 비용이 된다. 이벤트타임 우선과 스냅샷 격리는 누수를 막는 대신 지연 이벤트 복구를 늦출 수 있다. 노드·시간 파티셔닝과 근사 이웃은 처리량을 높이지만 정확도 저하 가능성이 있다.

결제·추천·침해 탐지에서의 적용

결제 사기 탐지에서는 카드–가맹점–단말기 이분 그래프와 거래 이벤트 스트림을 사용한다. 이벤트타임 피처 스냅샷을 만든 뒤 TGN을 학습하고, 온라인 메모리 업데이트와 실시간 점수를 연결한다. 신규 패턴을 조기에 검출할 수 있으며 AUC-PR은 베이스라인 대비 5~15%p 개선될 수 있다. 다만 효과는 도메인별로 다르다.

대규모 추천에서는 사용자–아이템 상호작용에 클릭·뷰·구매 시점을 포함한다. 윈도우 샘플링, TGAT 임베딩, ANN 서빙, 피드백 루프를 잇는 구성으로 Hit@10/MRR 개선과 콜드스타트 완화를 기대할 수 있다.

보안 침해 탐지에서는 호스트–계정–프로세스 이벤트로 lateral movement 시나리오를 모델링한다. Hawkes 기반 이상 스코어와 Temporal GNN을 결합한 앙상블은 오탐을 줄이고 조사 우선순위 정렬을 자동화하는 데 쓸 수 있다.

접근 방식별 선택 기준

접근	성능	확장성	일관성	안정성	운영 편의
감쇠 휴리스틱(AA/PA+시간가중)	중	상	상	상	상
정적 GNN+시간 피처	중~상	중	중	중	중
Temporal GNN(TGAT/TGN)	상	중	상	중	중
점과정(Hawkes/Neural)	중~상	중	상	중	중
하이브리드(GNN+Hawkes)	상	중	상	중	중하

성능과 일관성은 데이터 품질과 설계에 따라 달라진다. 최신 구현체와 가속기 사용 여부도 확장성 차이를 만든다. 최신 정보 확인이 필요하다.

수집부터 재학습까지 이어지는 흐름

⤢✕

누수 점검에 실패하면 배치를 중단하고 경보를 낸다. Feature Store는 스냅샷 격리와 잠금으로 시점 일관성을 유지한다.

TGN으로 구성한 시간 기반 링크 예측 예시

전제조건은 Python 3.10, torch>=2.2, torch-geometric>=2.5, torch-geometric-temporal>=0.54이다. 단일 머신에서 소규모 데이터를 학습하는 예시이며, 이벤트 스트림을 대상으로 시간 분할과 시간 유의적 네거티브 샘플링, 누수 방지를 다룬다.

# pip install torch torch-geometric torch-geometric-temporal
import math, random, time
import torch
from torch import nn
from torch.optim import Adam
from torch_geometric_temporal.nn.recurrent import TGNMemory
from torch_geometric_temporal.nn.attention import TransformerConv
from torch_geometric.data import Data
from collections import defaultdict

# 1) 합성 이벤트 스트림 생성 (u--v at time t)
num_nodes = 1000
num_events = 20000
random.seed(7)

events = []
t = 0
for _ in range(num_events):
    u = random.randrange(num_nodes)
    v = (u + random.randrange(1, 50)) % num_nodes
    t += random.randrange(1, 5)  # irregular timestamps
    events.append((u, v, t))

# 2) 시간 분할 (train/val/test: 70/15/15)
Tmax = events[-1][2]
t_train = int(Tmax * 0.7)
t_val = int(Tmax * 0.85)

train_ev = [e for e in events if e[2] <= t_train]
val_ev   = [e for e in events if t_train < e[2] <= t_val]
test_ev  = [e for e in events if e[2] > t_val]

# 3) 간단한 시간 유의적 네거티브 샘플링
def sample_negatives(ev_batch, k=1):
    neg = []
    for (u,v,t) in ev_batch:
        for _ in range(k):
            w = random.randrange(num_nodes)
            while w == u or w == v:
                w = random.randrange(num_nodes)
            neg.append((u, w, t))
    return neg

# 4) TGN 구성요소
class TGNSimple(nn.Module):
    def __init__(self, num_nodes, emb_dim=64):
        super().__init__()
        self.memory = TGNMemory(
            num_nodes=num_nodes, raw_message_dim=emb_dim, memory_dim=emb_dim
        )
        self.msg_lin = nn.Linear(emb_dim, emb_dim)
        self.conv = TransformerConv(emb_dim, emb_dim // 2, heads=2)  # output emb_dim
        self.scorer = nn.Bilinear(emb_dim, emb_dim, 1)
        self.node_emb = nn.Embedding(num_nodes, emb_dim)

    def forward(self, edges):
        # edges: list[(u,v,t)]
        if not edges:
            return None
        us = torch.tensor([u for u,_,_ in edges], dtype=torch.long)
        vs = torch.tensor([v for _,v,_ in edges], dtype=torch.long)
        ts = torch.tensor([t for _,_,t in edges], dtype=torch.float)

        # Node states = static emb + memory
        mem = self.memory.get_memory()
        x = self.node_emb.weight + mem  # [N, D]

        # One-hop conv over current batch-induced edges
        edge_index = torch.stack([us, vs], dim=0)
        x_msg = self.conv(x, edge_index)  # [N, D]
        msg = self.msg_lin(x_msg)

        # Update memory at nodes touched by edges with event time
        touched = torch.unique(torch.cat([us, vs]))
        self.memory.update_state(touched, msg[touched], ts.mean())

        # Score
        eu = x[us]
        ev = x[vs]
        s = self.scorer(eu, ev).squeeze(-1)
        return s

    def reset_memory(self):
        self.memory.reset_state()

model = TGNSimple(num_nodes)
opt = Adam(model.parameters(), lr=1e-3)
bce = nn.BCEWithLogitsLoss()

# 5) 배치 학습 루프 (시간 순서 유지, 누수 방지)
def iterate_batches(ev_list, batch_size=512):
    for i in range(0, len(ev_list), batch_size):
        yield ev_list[i: i+batch_size]

def evaluate(ev_list):
    model.reset_memory()
    model.eval()
    y_true, y_score = [], []
    with torch.no_grad():
        for batch in iterate_batches(ev_list):
            neg = sample_negatives(batch, k=1)
            X = batch + neg
            y = [1]*len(batch) + [0]*len(neg)
            s = model(X)
            y_true.extend(y)
            y_score.extend(torch.sigmoid(s).cpu().tolist())
    # Simple AUC-PR proxy: average precision via bucketization
    import numpy as np
    from sklearn.metrics import average_precision_score
    return average_precision_score(y_true, y_score)

# Train
model.train()
model.reset_memory()
for epoch in range(5):
    total = 0.0
    for batch in iterate_batches(train_ev):
        neg = sample_negatives(batch, k=1)
        X = batch + neg
        y = torch.tensor([1]*len(batch) + [0]*len(neg), dtype=torch.float)
        s = model(X)
        loss = bce(s, y)
        opt.zero_grad()
        loss.backward()
        opt.step()
        total += loss.item()
    val_ap = evaluate(val_ev)
    print(f"epoch={epoch} loss={total:.3f} val_AP={val_ap:.4f}")

test_ap = evaluate(test_ev)
print("test_AP=", test_ap)

이 코드는 단순화된 예시다. 실제 환경에서는 시간 인코딩, 이웃 샘플링, 미니배치 메시지 패싱, 메모리 동기화, 고도화된 음성/양성 균형 제어가 필요하다. 학습·검증·테스트 전 구간에서 피처 생성 시점 ≤ 라벨 시점 규칙도 강제해야 한다.

성능과 운영 안정성을 함께 얻기 위한 조건

Temporal GNN이나 점과정을 도입하면 AUC-PR이 5~20% 상대 개선될 수 있으며, 이는 도메인과 데이터 품질에 의존한다. 이벤트타임 기반 서빙과 인크리멘털 업데이트는 의사결정 지연을 줄여 수백 ms 수준을 달성할 수 있다.

스냅샷 격리, 워터마크, 라인리지를 갖추면 재현성과 컴플라이언스를 강화할 수 있다. 실무에서는 휴리스틱에서 Temporal GNN·점과정으로, 필요에 따라 하이브리드로 고도화하면서 피처 스토어 타임 트래블, 시간 분할 검증, 누수 감시를 최소 가용 제품으로 먼저 갖추는 흐름이 적합하다. 이후 데이터 규모와 SLA에 맞춰 샘플링·서빙 최적화와 모델 복합화를 적용한다.