제7장: 크로스-디바이스 vs 크로스-사일로

연합학습의 두 가지 주요 패러다임 비교 및 구현 전략

연합학습의 두 가지 패러다임

연합학습은 크게 두 가지 설정으로 나뉩니다: 크로스-디바이스(Cross-Device)와 크로스-사일로(Cross-Silo). 각각은 다른 규모, 참여자 특성, 기술적 도전과제를 가지며, 적절한 알고리즘과 아키텍처 선택이 필요합니다.

크로스-디바이스 연합학습

특성 및 도전과제

크로스-디바이스 FL은 대규모, 이질적, 불안정한 클라이언트 풀을 다룹니다:

• 대규모: 수백만 개 이상의 잠재적 참여자
• 제한된 리소스: 배터리, 계산, 메모리, 대역폭 제약
• 불안정한 참여: 클라이언트가 무작위로 오프라인/온라인
• 높은 이질성: 다양한 디바이스 유형, OS, 하드웨어
• 비IID 데이터: 사용자마다 매우 다른 데이터 분포

크로스-디바이스 아키텍처

class CrossDeviceFLSystem:
    """
    크로스-디바이스 연합학습 시스템

    수백만 개의 모바일 디바이스를 조율
    """

    def __init__(self, min_clients_per_round=100, max_clients_per_round=1000):
        """
        인자:
            min_clients_per_round: 라운드당 최소 클라이언트 수
            max_clients_per_round: 라운드당 최대 클라이언트 수
        """
        self.min_clients = min_clients_per_round
        self.max_clients = max_clients_per_round
        self.global_model = None
        self.device_registry = {}

    def register_device(self, device_id, device_info):
        """
        디바이스 등록

        인자:
            device_id: 고유 디바이스 식별자
            device_info: {'type': 'mobile', 'os': 'android', 'battery': 0.8, ...}
        """
        self.device_registry[device_id] = {
            'info': device_info,
            'last_seen': time.time(),
            'participation_count': 0,
            'reliability_score': 1.0
        }

    def select_devices_for_round(self):
        """
        라운드에 참여할 디바이스 선택

        반환:
            선택된 디바이스 ID 목록
        """
        # 사용 가능한 디바이스 필터링
        available = []

        for device_id, info in self.device_registry.items():
            # 조건 확인
            device_info = info['info']

            # 배터리 충분한지 (>20%)
            if device_info.get('battery', 0) < 0.2:
                continue

            # WiFi 연결되어 있는지
            if device_info.get('network') != 'wifi':
                continue

            # 충전 중인지
            if not device_info.get('charging', False):
                continue

            available.append(device_id)

        # 무작위 샘플링
        num_to_select = min(len(available), self.max_clients)
        selected = random.sample(available, min(num_to_select, len(available)))

        return selected

    def run_training_round(self, round_num):
        """
        단일 학습 라운드 실행

        인자:
            round_num: 라운드 번호

        반환:
            업데이트된 전역 모델
        """
        print(f"\n=== 라운드 {round_num} 시작 ===")

        # 1. 디바이스 선택
        selected_devices = self.select_devices_for_round()
        print(f"선택된 디바이스: {len(selected_devices)}개")

        if len(selected_devices) < self.min_clients:
            print(f"경고: 최소 클라이언트 수 미달 ({len(selected_devices)} < {self.min_clients})")
            return None

        # 2. 모델 및 설정 전송
        client_updates = []

        for device_id in selected_devices:
            try:
                # 모델 전송 (압축 및 최적화)
                compressed_model = self.compress_model(self.global_model)

                # 디바이스에서 학습
                update = self.train_on_device(device_id, compressed_model)

                if update is not None:
                    client_updates.append(update)

                # 참여 기록
                self.device_registry[device_id]['participation_count'] += 1

            except Exception as e:
                print(f"디바이스 {device_id} 실패: {e}")
                # 신뢰도 점수 감소
                self.device_registry[device_id]['reliability_score'] *= 0.9

        # 3. 집계
        if len(client_updates) >= self.min_clients:
            self.global_model = self.aggregate_updates(client_updates)
            print(f"집계 완료: {len(client_updates)}개 업데이트")
        else:
            print(f"집계 실패: 업데이트 부족 ({len(client_updates)} < {self.min_clients})")

        return self.global_model

    def compress_model(self, model):
        """모델 압축 (양자화, 가지치기 등)"""
        # 간단한 양자화 시뮬레이션
        if model is None:
            return None
        compressed = np.round(model * 100) / 100  # 2자리 정밀도
        return compressed

    def train_on_device(self, device_id, model):
        """디바이스에서 로컬 학습 (시뮬레이션)"""
        # 실제로는 디바이스에서 비동기 실행
        time.sleep(0.01)  # 학습 시뮬레이션
        return np.random.randn(100) * 0.1  # 가상 업데이트

    def aggregate_updates(self, updates):
        """간단한 평균 집계"""
        return np.mean(updates, axis=0)


# 예시 사용
import time
import random

system = CrossDeviceFLSystem(min_clients_per_round=50, max_clients_per_round=200)

# 10,000개 디바이스 등록
for i in range(10000):
    system.register_device(
        device_id=f"device_{i}",
        device_info={
            'type': 'mobile',
            'os': random.choice(['android', 'ios']),
            'battery': random.uniform(0.1, 1.0),
            'network': random.choice(['wifi', '4g', '5g']),
            'charging': random.choice([True, False])
        }
    )

# 초기 모델
system.global_model = np.random.randn(100) * 0.1

# 여러 라운드 실행
for round_num in range(5):
    system.run_training_round(round_num)

디바이스 최적화 기법

제한된 리소스를 효율적으로 사용하기 위한 기법들:

class MobileOptimizer:
    """
    모바일 디바이스를 위한 최적화
    """

    @staticmethod
    def quantize_model(model, num_bits=8):
        """
        모델 양자화로 크기 및 계산 감소

        인자:
            model: 원본 모델 가중치
            num_bits: 양자화 비트 수 (8, 4, 2 등)

        반환:
            양자화된 모델
        """
        # 최소/최대 값 찾기
        min_val = np.min(model)
        max_val = np.max(model)

        # 양자화 범위
        num_levels = 2 ** num_bits

        # 스케일 계산
        scale = (max_val - min_val) / (num_levels - 1)

        # 양자화
        quantized = np.round((model - min_val) / scale)

        # 역양자화 (실제 사용 시)
        dequantized = quantized * scale + min_val

        compression_ratio = 32 / num_bits  # float32 기준
        print(f"압축률: {compression_ratio}x")

        return dequantized, {'scale': scale, 'min': min_val}

    @staticmethod
    def gradient_compression(gradients, top_k_percent=0.1):
        """
        상위 K% 그래디언트만 전송 (스파스 업데이트)

        인자:
            gradients: 원본 그래디언트
            top_k_percent: 전송할 상위 비율

        반환:
            압축된 그래디언트 (스파스 형태)
        """
        # 절대값 기준 정렬
        abs_grads = np.abs(gradients)
        k = int(len(gradients) * top_k_percent)

        # 상위 k 인덱스
        top_k_indices = np.argsort(abs_grads)[-k:]

        # 스파스 표현
        sparse_grads = {
            'indices': top_k_indices,
            'values': gradients[top_k_indices],
            'shape': gradients.shape
        }

        compression_ratio = len(gradients) / k
        print(f"그래디언트 압축률: {compression_ratio}x")

        return sparse_grads

    @staticmethod
    def reconstruct_sparse_gradient(sparse_grads):
        """스파스 그래디언트를 원본 형태로 복원"""
        reconstructed = np.zeros(sparse_grads['shape'])
        reconstructed[sparse_grads['indices']] = sparse_grads['values']
        return reconstructed

    @staticmethod
    def adaptive_local_epochs(device_resources):
        """
        디바이스 리소스에 따라 로컬 에포크 수 조정

        인자:
            device_resources: {'battery': float, 'compute': float, 'network': str}

        반환:
            권장 로컬 에포크 수
        """
        battery = device_resources.get('battery', 0.5)
        compute = device_resources.get('compute', 0.5)  # 0-1 스케일
        network = device_resources.get('network', 'wifi')

        # 기본 에포크
        base_epochs = 5

        # 배터리 낮으면 줄이기
        if battery < 0.3:
            base_epochs = 2
        elif battery < 0.5:
            base_epochs = 3

        # 계산 능력 고려
        if compute < 0.3:
            base_epochs = min(base_epochs, 2)

        # 네트워크 고려
        if network in ['2g', '3g']:
            base_epochs = max(base_epochs - 1, 1)

        return base_epochs


# 예시 사용
optimizer = MobileOptimizer()

# 모델 양자화
model = np.random.randn(10000)
quantized, meta = optimizer.quantize_model(model, num_bits=8)

# 그래디언트 압축
gradients = np.random.randn(10000) * 0.1
sparse = optimizer.gradient_compression(gradients, top_k_percent=0.1)

# 에포크 조정
device_res = {'battery': 0.4, 'compute': 0.6, 'network': '4g'}
epochs = optimizer.adaptive_local_epochs(device_res)
print(f"권장 에포크: {epochs}")

크로스-사일로 연합학습

특성 및 도전과제

크로스-사일로 FL은 소수의 강력한 조직 간 협력을 다룹니다:

• 소규모: 수십 ~ 수백 개 참여자 (병원, 기업 등)
• 풍부한 리소스: 데이터센터 급 계산 및 네트워크
• 안정적 참여: 계획된, 신뢰할 수 있는 참여
• 강한 보안 요구: 민감한 비즈니스/의료 데이터
• 규제 준수: GDPR, HIPAA 등 법적 요구사항

크로스-사일로 아키텍처

class CrossSiloFLSystem:
    """
    크로스-사일로 연합학습 시스템

    수십 개의 조직(병원, 기업) 간 협력
    """

    def __init__(self, participating_organizations):
        """
        인자:
            participating_organizations: 참여 조직 목록
                [{'id': 'hospital_1', 'name': '서울대병원', 'data_size': 10000}, ...]
        """
        self.organizations = {org['id']: org for org in participating_organizations}
        self.global_model = None
        self.audit_log = []

    def run_synchronous_round(self, round_num):
        """
        동기식 학습 라운드

        모든 참여자가 동시에 학습 및 업데이트 전송

        인자:
            round_num: 라운드 번호

        반환:
            업데이트된 전역 모델
        """
        print(f"\n=== 동기식 라운드 {round_num} ===")

        # 1. 모든 조직에 모델 배포
        client_updates = {}

        for org_id, org_info in self.organizations.items():
            print(f"조직 {org_info['name']} 학습 중...")

            # 로컬 학습 (차등 프라이버시 적용)
            update = self.train_with_privacy(
                org_id=org_id,
                model=self.global_model,
                data_size=org_info['data_size']
            )

            client_updates[org_id] = update

            # 감사 로그
            self.audit_log.append({
                'round': round_num,
                'org_id': org_id,
                'timestamp': time.time(),
                'update_norm': np.linalg.norm(update)
            })

        # 2. 가중 집계 (데이터 크기 기반)
        self.global_model = self.weighted_aggregation(client_updates)

        # 3. 검증
        validation_metrics = self.cross_validate(client_updates)
        print(f"검증 메트릭: {validation_metrics}")

        return self.global_model

    def train_with_privacy(self, org_id, model, data_size):
        """
        차등 프라이버시를 사용한 로컬 학습

        인자:
            org_id: 조직 ID
            model: 전역 모델
            data_size: 데이터 크기

        반환:
            프라이버시가 보장된 업데이트
        """
        # 로컬 학습 시뮬레이션
        local_update = np.random.randn(100) * 0.1

        # 차등 프라이버시 적용
        epsilon = 1.0
        delta = 1e-5
        sensitivity = 1.0

        noise_scale = np.sqrt(2 * np.log(1.25 / delta)) * sensitivity / epsilon
        noise = np.random.normal(0, noise_scale, size=local_update.shape)

        private_update = local_update + noise

        return private_update

    def weighted_aggregation(self, client_updates):
        """
        데이터 크기 기반 가중 집계

        인자:
            client_updates: {org_id: update} 딕셔너리

        반환:
            집계된 전역 모델
        """
        total_data = sum(org['data_size'] for org in self.organizations.values())

        aggregated = np.zeros_like(list(client_updates.values())[0])

        for org_id, update in client_updates.items():
            weight = self.organizations[org_id]['data_size'] / total_data
            aggregated += weight * update

        return aggregated

    def cross_validate(self, client_updates):
        """
        교차 검증: 각 조직의 업데이트를 다른 조직 데이터로 검증

        인자:
            client_updates: 클라이언트 업데이트

        반환:
            검증 메트릭
        """
        # 간단한 일관성 검사
        update_norms = [np.linalg.norm(u) for u in client_updates.values()]

        metrics = {
            'mean_norm': np.mean(update_norms),
            'std_norm': np.std(update_norms),
            'consistency_score': 1.0 / (1.0 + np.std(update_norms))
        }

        return metrics

    def generate_compliance_report(self):
        """
        규제 준수 리포트 생성 (GDPR, HIPAA 등)

        반환:
            감사 리포트
        """
        report = f"""
        ===== 연합학습 규제 준수 리포트 =====

        참여 조직 수: {len(self.organizations)}
        총 라운드 수: {len(set(log['round'] for log in self.audit_log))}
        총 업데이트 수: {len(self.audit_log)}

        조직별 참여:
        """

        for org_id, org_info in self.organizations.items():
            participation = sum(1 for log in self.audit_log if log['org_id'] == org_id)
            report += f"\n- {org_info['name']}: {participation} 라운드 참여"

        report += "\n\n프라이버시 보장:"
        report += "\n- 차등 프라이버시 (ε=1.0, δ=1e-5) 적용"
        report += "\n- 개별 데이터 접근 없음"
        report += "\n- 암호화된 통신"

        return report


# 예시 사용
organizations = [
    {'id': 'hospital_1', 'name': '서울대병원', 'data_size': 15000},
    {'id': 'hospital_2', 'name': '연세의료원', 'data_size': 12000},
    {'id': 'hospital_3', 'name': '삼성서울병원', 'data_size': 18000},
    {'id': 'hospital_4', 'name': '아산병원', 'data_size': 14000},
]

silo_system = CrossSiloFLSystem(organizations)
silo_system.global_model = np.random.randn(100) * 0.1

# 동기식 라운드 실행
for round_num in range(5):
    silo_system.run_synchronous_round(round_num)

# 규제 준수 리포트
print(silo_system.generate_compliance_report())

보안 멀티파티 계산 (SMPC)

크로스-사일로 환경에서는 강력한 보안 메커니즘이 필요합니다:

class SecureAggregation:
    """
    비밀 공유를 사용한 안전한 집계

    서버가 개별 업데이트를 보지 못하고 집계만 계산
    """

    def __init__(self, num_clients, threshold):
        """
        인자:
            num_clients: 클라이언트 수
            threshold: 복원에 필요한 최소 공유 수
        """
        self.num_clients = num_clients
        self.threshold = threshold

    def share_update(self, update, client_id):
        """
        클라이언트의 업데이트를 비밀 공유로 분할

        인자:
            update: 모델 업데이트
            client_id: 클라이언트 ID

        반환:
            다른 클라이언트들에게 전송할 공유 목록
        """
        shares = []

        for param in update:
            # 각 파라미터를 비밀 공유로 분할
            param_shares = self.shamir_share(param, self.num_clients, self.threshold)
            shares.append(param_shares)

        return shares

    def shamir_share(self, secret, num_shares, threshold):
        """Shamir 비밀 공유 (간단화)"""
        # 실제 구현은 유한체 연산 사용
        coefficients = [secret] + [np.random.randint(0, 100)
                                   for _ in range(threshold - 1)]

        def poly(x):
            return sum(c * (x ** i) for i, c in enumerate(coefficients))

        shares = [(i, poly(i)) for i in range(1, num_shares + 1)]
        return shares

    def aggregate_shares(self, all_client_shares):
        """
        모든 클라이언트의 공유를 집계

        각 클라이언트는 자신의 공유만 알고 있음
        서버는 집계된 공유만 받음

        인자:
            all_client_shares: 각 클라이언트의 공유 목록

        반환:
            집계된 모델 (복원됨)
        """
        # 공유 합산 (동형 속성)
        num_params = len(all_client_shares[0])
        aggregated_shares = []

        for param_idx in range(num_params):
            # 이 파라미터의 모든 클라이언트 공유
            param_shares = [
                client_shares[param_idx]
                for client_shares in all_client_shares
            ]

            # 공유 합산
            summed_shares = self.sum_shares(param_shares)
            aggregated_shares.append(summed_shares)

        # 집계된 공유를 복원
        aggregated_model = []
        for shares in aggregated_shares:
            value = self.reconstruct(shares[:self.threshold])
            aggregated_model.append(value)

        return np.array(aggregated_model)

    def sum_shares(self, shares_list):
        """여러 공유를 합산"""
        # 같은 x 값의 y 값들을 합산
        summed = {}
        for shares in shares_list:
            for x, y in shares:
                summed[x] = summed.get(x, 0) + y
        return [(x, y) for x, y in summed.items()]

    def reconstruct(self, shares):
        """Lagrange 보간으로 비밀 복원"""
        # 간단화된 구현
        secret = 0
        for i, (xi, yi) in enumerate(shares):
            num = 1
            den = 1
            for j, (xj, yj) in enumerate(shares):
                if i != j:
                    num *= (0 - xj)
                    den *= (xi - xj)
            secret += yi * num // den
        return int(secret)


# 예시 사용
secure_agg = SecureAggregation(num_clients=4, threshold=3)

# 각 클라이언트의 업데이트
client_updates = [
    np.array([10, 20, 30]),
    np.array([15, 25, 35]),
    np.array([12, 22, 32]),
    np.array([18, 28, 38])
]

# 각 클라이언트가 자신의 업데이트를 공유로 분할
all_shares = []
for i, update in enumerate(client_updates):
    shares = secure_agg.share_update(update, client_id=i)
    all_shares.append(shares)

# 서버가 공유를 집계 (개별 업데이트 보지 못함)
aggregated = secure_agg.aggregate_shares(all_shares)
print(f"안전한 집계 결과: {aggregated}")

# 검증: 일반 집계와 비교
normal_avg = np.mean(client_updates, axis=0)
print(f"일반 집계 결과: {normal_avg}")

수직적 연합학습 (Vertical FL)

수평 vs 수직 연합학습

크로스-사일로 환경에서는 특별한 형태인 수직적 FL이 자주 사용됩니다:

• 수평적 FL: 같은 특징, 다른 샘플 (예: 여러 병원의 같은 유형 환자 데이터)
• 수직적 FL: 같은 샘플, 다른 특징 (예: 은행 + 전자상거래 플랫폼이 같은 사용자의 다른 정보 보유)

class VerticalFLSystem:
    """
    수직적 연합학습 시스템

    예: 은행과 전자상거래가 같은 고객의 다른 특징 보유
    """

    def __init__(self, parties):
        """
        인자:
            parties: 참여자 목록
                [{'id': 'bank', 'features': ['income', 'credit_score']},
                 {'id': 'ecommerce', 'features': ['purchase_history', 'browse_time']}]
        """
        self.parties = {p['id']: p for p in parties}
        self.entity_alignment = None  # 공통 사용자 ID

    def entity_resolution(self):
        """
        개체 정렬: 각 참여자의 데이터에서 공통 사용자 찾기

        프라이버시 보호하면서 교집합 계산 (PSI - Private Set Intersection)

        반환:
            공통 사용자 ID 목록
        """
        # 실제로는 암호화된 PSI 프로토콜 사용
        # 여기서는 시뮬레이션

        print("프라이버시 보호 개체 정렬 수행 중...")

        # 시뮬레이션: 공통 사용자 (실제로는 해시 기반 PSI)
        common_users = ['user_1', 'user_2', 'user_3', 'user_4', 'user_5']

        self.entity_alignment = common_users

        print(f"정렬 완료: {len(common_users)}명의 공통 사용자")

        return common_users

    def train_vertical_model(self, epochs=10):
        """
        수직적 연합학습 모델 학습

        각 참여자가 자신의 특징에 대한 부분 모델 학습

        인자:
            epochs: 학습 에포크

        반환:
            학습된 모델 (각 참여자의 부분 모델)
        """
        if self.entity_alignment is None:
            self.entity_resolution()

        print("\n수직적 연합학습 시작")

        # 각 참여자의 부분 모델
        partial_models = {
            party_id: np.random.randn(len(info['features']), 10)  # 10차원 임베딩
            for party_id, info in self.parties.items()
        }

        for epoch in range(epochs):
            # 1. 각 참여자가 자신의 특징으로 전방 전파
            embeddings = {}

            for party_id, model in partial_models.items():
                # 로컬 특징 임베딩 계산
                embedding = np.random.randn(len(self.entity_alignment), 10)
                embeddings[party_id] = embedding

            # 2. 조율자(또는 하나의 참여자)가 임베딩 결합하여 예측
            combined = np.concatenate(list(embeddings.values()), axis=1)

            # 예측 및 손실 계산
            predictions = self.predict(combined)
            loss = self.compute_loss(predictions)

            # 3. 그래디언트 계산 및 역전파
            # 각 참여자는 자신의 부분 그래디언트만 받음
            gradients = self.compute_gradients(combined, loss)

            # 4. 각 참여자가 자신의 모델 업데이트
            start = 0
            for party_id, model in partial_models.items():
                end = start + model.shape[1]
                party_grad = gradients[:, start:end]
                partial_models[party_id] -= 0.01 * party_grad.mean(axis=0)
                start = end

            if epoch % 2 == 0:
                print(f"에포크 {epoch}: 손실 = {loss:.4f}")

        return partial_models

    def predict(self, combined_embedding):
        """결합된 임베딩으로 예측 (시뮬레이션)"""
        return np.random.randn(combined_embedding.shape[0])

    def compute_loss(self, predictions):
        """손실 계산 (시뮬레이션)"""
        return np.random.rand()

    def compute_gradients(self, embedding, loss):
        """그래디언트 계산 (시뮬레이션)"""
        return np.random.randn(*embedding.shape) * 0.1


# 예시 사용
parties = [
    {'id': 'bank', 'features': ['income', 'credit_score', 'loan_history']},
    {'id': 'ecommerce', 'features': ['purchase_count', 'avg_basket', 'return_rate']},
    {'id': 'telco', 'features': ['data_usage', 'call_duration']}
]

vertical_fl = VerticalFLSystem(parties)
models = vertical_fl.train_vertical_model(epochs=10)

패러다임 비교표

특성	크로스-디바이스	크로스-사일로
참여자 수	10⁶ ~ 10⁹	10¹ ~ 10²
참여 패턴	비동기, 불안정	동기, 안정적
데이터 크기	클라이언트당 작음 (KB~MB)	클라이언트당 큼 (GB~TB)
계산 능력	제한적 (모바일)	강력 (서버)
통신	불안정, 제약 많음	안정적, 고대역폭
보안 초점	개인 프라이버시	기업 기밀, 규제 준수
주요 기법	압축, 양자화, 비동기	SMPC, 동형암호, 동기
응용 예시	키보드 예측, 음성 인식	의료 협력, 금융 사기 탐지

하이브리드 접근법

계층적 연합학습

크로스-디바이스와 크로스-사일로를 결합한 계층적 구조:

class HierarchicalFL:
    """
    계층적 연합학습

    레벨 1: 디바이스 → 엣지 서버 (크로스-디바이스)
    레벨 2: 엣지 서버 → 중앙 서버 (크로스-사일로)
    """

    def __init__(self, num_edge_servers, devices_per_edge):
        """
        인자:
            num_edge_servers: 엣지 서버 수
            devices_per_edge: 각 엣지당 디바이스 수
        """
        self.num_edge_servers = num_edge_servers
        self.devices_per_edge = devices_per_edge

        # 엣지 서버 초기화
        self.edge_servers = [
            {'id': f'edge_{i}', 'model': None, 'devices': []}
            for i in range(num_edge_servers)
        ]

        self.global_model = None

    def run_hierarchical_round(self, round_num):
        """
        계층적 학습 라운드

        1단계: 디바이스 → 엣지 (크로스-디바이스 스타일)
        2단계: 엣지 → 중앙 (크로스-사일로 스타일)
        """
        print(f"\n=== 계층적 라운드 {round_num} ===")

        # 1단계: 디바이스-엣지 집계
        edge_models = []

        for edge in self.edge_servers:
            print(f"\n{edge['id']} 로컬 집계 중...")

            # 이 엣지의 디바이스 선택 및 학습
            device_updates = []

            for device_id in range(self.devices_per_edge):
                # 모바일 최적화 기법 사용
                update = self.train_on_mobile_device(
                    device_id,
                    self.global_model
                )

                if update is not None:
                    device_updates.append(update)

            # 엣지에서 로컬 집계
            if device_updates:
                edge['model'] = np.mean(device_updates, axis=0)
                edge_models.append(edge['model'])

            print(f"  {len(device_updates)}개 디바이스 업데이트 집계")

        # 2단계: 엣지-중앙 집계
        print(f"\n중앙 서버 집계 중...")

        if edge_models:
            # 강력한 보안 프로토콜 사용 (크로스-사일로 스타일)
            self.global_model = self.secure_aggregation(edge_models)

            print(f"  {len(edge_models)}개 엣지 모델 집계 완료")

        return self.global_model

    def train_on_mobile_device(self, device_id, model):
        """모바일 디바이스 학습 (시뮬레이션)"""
        # 압축, 양자화 등 모바일 최적화
        return np.random.randn(100) * 0.1

    def secure_aggregation(self, edge_models):
        """엣지 모델의 안전한 집계"""
        # SMPC 또는 동형 암호화 사용
        return np.mean(edge_models, axis=0)


# 예시 사용
hierarchical = HierarchicalFL(num_edge_servers=5, devices_per_edge=1000)
hierarchical.global_model = np.random.randn(100) * 0.1

for round_num in range(3):
    hierarchical.run_hierarchical_round(round_num)