🎯 왜 토큰 스왑 자동화가 필요할까?

🔧 필요한 기술 스택과 준비물

🛠️ 개발 환경 세팅하기

python -m venv swap_automation
source swap_automation/bin/activate  # Windows: swap_automation\Scripts\activate

pip install web3 eth-account python-dotenv requests aiohttp

swap_automation/
├── .env                 # 환경 변수 (개인키, API 키 등)
├── config.py           # 설정 파일
├── wallet_manager.py   # 지갑 관리 모듈
├── swap_engine.py      # 스왑 로직 핵심
├── price_monitor.py    # 가격 모니터링
├── main.py            # 메인 실행 파일
└── utils.py           # 유틸리티 함수들

# .env 파일
PRIVATE_KEY_A=your_wallet_a_private_key_here
PRIVATE_KEY_B=your_wallet_b_private_key_here
INFURA_API_KEY=your_infura_api_key
ETHERSCAN_API_KEY=your_etherscan_api_key
RPC_URL=https://mainnet.infura.io/v3/YOUR_PROJECT_ID

💼 지갑 관리 모듈 구현하기

from web3 import Web3
from eth_account import Account
from dotenv import load_dotenv
import os

load_dotenv()

class WalletManager:
    def __init__(self, rpc_url):
        """
        지갑 매니저 초기화
        rpc_url: 블록체인 노드 연결 주소
        """
        self.w3 = Web3(Web3.HTTPProvider(rpc_url))
        
        # 연결 확인
        if not self.w3.is_connected():
            raise Exception("블록체인 노드 연결 실패! 😢")
        
        print(f"✅ 블록체인 연결 성공! 현재 블록: {self.w3.eth.block_number}")
        
        # 지갑 로드
        self.wallet_a = self._load_wallet(os.getenv('PRIVATE_KEY_A'))
        self.wallet_b = self._load_wallet(os.getenv('PRIVATE_KEY_B'))
    
    def _load_wallet(self, private_key):
        """개인키로부터 지갑 객체 생성"""
        if not private_key:
            raise ValueError("개인키가 설정되지 않았어요!")
        
        account = Account.from_key(private_key)
        return {
            'address': account.address,
            'private_key': private_key,
            'account': account
        }
    
    def get_balance(self, wallet_type='A', token_address=None):
        """
        지갑 잔액 조회
        wallet_type: 'A' 또는 'B'
        token_address: None이면 ETH, 주소 입력시 해당 토큰
        """
        wallet = self.wallet_a if wallet_type == 'A' else self.wallet_b
        
        if token_address is None:
            # ETH 잔액 조회
            balance_wei = self.w3.eth.get_balance(wallet['address'])
            balance_eth = self.w3.from_wei(balance_wei, 'ether')
            return float(balance_eth)
        else:
            # ERC-20 토큰 잔액 조회
            token_contract = self._get_token_contract(token_address)
            balance = token_contract.functions.balanceOf(wallet['address']).call()
            decimals = token_contract.functions.decimals().call()
            return balance / (10 ** decimals)
    
    def _get_token_contract(self, token_address):
        """ERC-20 토큰 컨트랙트 객체 반환"""
        # 표준 ERC-20 ABI (간소화 버전)
        erc20_abi = [
            {
                "constant": True,
                "inputs": [{"name": "_owner", "type": "address"}],
                "name": "balanceOf",
                "outputs": [{"name": "balance", "type": "uint256"}],
                "type": "function"
            },
            {
                "constant": True,
                "inputs": [],
                "name": "decimals",
                "outputs": [{"name": "", "type": "uint8"}],
                "type": "function"
            },
            {
                "constant": False,
                "inputs": [
                    {"name": "_spender", "type": "address"},
                    {"name": "_value", "type": "uint256"}
                ],
                "name": "approve",
                "outputs": [{"name": "", "type": "bool"}],
                "type": "function"
            },
            {
                "constant": True,
                "inputs": [
                    {"name": "_owner", "type": "address"},
                    {"name": "_spender", "type": "address"}
                ],
                "name": "allowance",
                "outputs": [{"name": "", "type": "uint256"}],
                "type": "function"
            }
        ]
        
        return self.w3.eth.contract(
            address=Web3.to_checksum_address(token_address),
            abi=erc20_abi
        )

🔄 스왑 엔진 핵심 로직 구현

from web3 import Web3
from eth_account import Account
import time
import json

class SwapEngine:
    def __init__(self, wallet_manager, router_address, router_abi):
        """
        스왑 엔진 초기화
        wallet_manager: WalletManager 인스턴스
        router_address: Uniswap Router 컨트랙트 주소
        router_abi: Router ABI
        """
        self.wm = wallet_manager
        self.w3 = wallet_manager.w3
        
        # Uniswap Router 컨트랙트
        self.router = self.w3.eth.contract(
            address=Web3.to_checksum_address(router_address),
            abi=router_abi
        )
        
        # WETH 주소 (Wrapped ETH)
        self.weth_address = self.router.functions.WETH().call()
        
        print(f"🦄 Uniswap Router 연결 완료!")
        print(f"WETH 주소: {self.weth_address}")
    
    def approve_token(self, token_address, wallet_type='A', amount=None):
        """
        토큰 사용 승인 (approve)
        DEX가 내 토큰을 사용할 수 있도록 허가하는 과정
        """
        wallet = self.wm.wallet_a if wallet_type == 'A' else self.wm.wallet_b
        token_contract = self.wm._get_token_contract(token_address)
        
        # amount가 None이면 무제한 승인
        if amount is None:
            amount = 2**256 - 1  # 최대값
        else:
            decimals = token_contract.functions.decimals().call()
            amount = int(amount * (10 ** decimals))
        
        # 현재 승인량 확인
        current_allowance = token_contract.functions.allowance(
            wallet['address'],
            self.router.address
        ).call()
        
        if current_allowance >= amount:
            print(f"✅ 이미 충분한 승인량이 있어요: {current_allowance}")
            return True
        
        # approve 트랜잭션 생성
        approve_txn = token_contract.functions.approve(
            self.router.address,
            amount
        ).build_transaction({
            'from': wallet['address'],
            'gas': 100000,
            'gasPrice': self.w3.eth.gas_price,
            'nonce': self.w3.eth.get_transaction_count(wallet['address'])
        })
        
        # 트랜잭션 서명 및 전송
        signed_txn = self.w3.eth.account.sign_transaction(
            approve_txn,
            wallet['private_key']
        )
        
        tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)
        print(f"📝 Approve 트랜잭션 전송: {tx_hash.hex()}")
        
        # 트랜잭션 완료 대기
        receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
        
        if receipt['status'] == 1:
            print(f"✅ Approve 성공!")
            return True
        else:
            print(f"❌ Approve 실패!")
            return False
    
    def get_amounts_out(self, amount_in, path):
        """
        스왑 시 받을 수 있는 토큰 양 계산
        amount_in: 입력 토큰 양
        path: 스왑 경로 [token_in, token_out]
        """
        # 토큰 decimals 가져오기
        token_in_contract = self.wm._get_token_contract(path[0])
        decimals = token_in_contract.functions.decimals().call()
        amount_in_wei = int(amount_in * (10 ** decimals))
        
        # Uniswap에서 예상 출력량 조회
        amounts = self.router.functions.getAmountsOut(
            amount_in_wei,
            path
        ).call()
        
        # 출력 토큰 decimals로 변환
        token_out_contract = self.wm._get_token_contract(path[-1])
        out_decimals = token_out_contract.functions.decimals().call()
        amount_out = amounts[-1] / (10 ** out_decimals)
        
        return amount_out
    
    def execute_swap(self, amount_in, amount_out_min, path, wallet_type='A', deadline_minutes=20):
        """
        실제 스왑 실행
        amount_in: 입력 토큰 양
        amount_out_min: 최소 출력 토큰 양 (슬리피지 보호)
        path: 스왑 경로
        wallet_type: 사용할 지갑
        deadline_minutes: 트랜잭션 유효 시간
        """
        wallet = self.wm.wallet_a if wallet_type == 'A' else self.wm.wallet_b
        
        # 토큰 decimals 처리
        token_in_contract = self.wm._get_token_contract(path[0])
        token_out_contract = self.wm._get_token_contract(path[-1])
        
        in_decimals = token_in_contract.functions.decimals().call()
        out_decimals = token_out_contract.functions.decimals().call()
        
        amount_in_wei = int(amount_in * (10 ** in_decimals))
        amount_out_min_wei = int(amount_out_min * (10 ** out_decimals))
        
        # Deadline 설정 (현재 시간 + 지정 분)
        deadline = int(time.time()) + (deadline_minutes * 60)
        
        print(f"\n🔄 스왑 준비 중...")
        print(f"입력: {amount_in} {path[0][:6]}...")
        print(f"최소 출력: {amount_out_min} {path[-1][:6]}...")
        
        # 스왑 트랜잭션 생성
        swap_txn = self.router.functions.swapExactTokensForTokens(
            amount_in_wei,
            amount_out_min_wei,
            path,
            wallet['address'],
            deadline
        ).build_transaction({
            'from': wallet['address'],
            'gas': 250000,  # 가스 리밋
            'gasPrice': self.w3.eth.gas_price,
            'nonce': self.w3.eth.get_transaction_count(wallet['address'])
        })
        
        # 가스 예측 (선택사항)
        try:
            estimated_gas = self.w3.eth.estimate_gas(swap_txn)
            swap_txn['gas'] = int(estimated_gas * 1.2)  # 20% 여유
            print(f"⛽ 예상 가스: {estimated_gas}")
        except Exception as e:
            print(f"⚠️ 가스 예측 실패, 기본값 사용: {e}")
        
        # 트랜잭션 서명
        signed_txn = self.w3.eth.account.sign_transaction(
            swap_txn,
            wallet['private_key']
        )
        
        # 트랜잭션 전송
        tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)
        print(f"\n📤 스왑 트랜잭션 전송!")
        print(f"TX Hash: {tx_hash.hex()}")
        
        # 트랜잭션 완료 대기
        print("⏳ 블록 확인 대기 중...")
        receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=300)
        
        if receipt['status'] == 1:
            print(f"\n✅ 스왑 성공! 🎉")
            print(f"가스 사용량: {receipt['gasUsed']}")
            print(f"블록 번호: {receipt['blockNumber']}")
            return {
                'success': True,
                'tx_hash': tx_hash.hex(),
                'gas_used': receipt['gasUsed']
            }
        else:
            print(f"\n❌ 스왑 실패! 😢")
            return {
                'success': False,
                'tx_hash': tx_hash.hex()
            }
    
    def swap_with_slippage_protection(self, amount_in, path, wallet_type='A', 
                                     slippage_percent=0.5):
        """
        슬리피지 보호 기능이 있는 스왑
        slippage_percent: 허용 슬리피지 (%)
        """
        # 예상 출력량 계산
        expected_out = self.get_amounts_out(amount_in, path)
        
        # 슬리피지 적용한 최소 출력량
        amount_out_min = expected_out * (1 - slippage_percent / 100)
        
        print(f"\n💱 슬리피지 보호 스왑")
        print(f"예상 출력: {expected_out:.6f}")
        print(f"최소 출력: {amount_out_min:.6f} (슬리피지 {slippage_percent}%)")
        
        # 스왑 실행
        return self.execute_swap(amount_in, amount_out_min, path, wallet_type)

📊 가격 모니터링 시스템 구축

import asyncio
import aiohttp
from datetime import datetime
import json

class PriceMonitor:
    def __init__(self, swap_engine):
        """
        가격 모니터 초기화
        swap_engine: SwapEngine 인스턴스
        """
        self.swap_engine = swap_engine
        self.price_history = {}
        self.alert_conditions = []
        
    def add_alert_condition(self, token_pair, condition_type, threshold, callback):
        """
        가격 알림 조건 추가
        token_pair: (token_a, token_b) 튜플
        condition_type: 'above' 또는 'below'
        threshold: 임계값
        callback: 조건 충족시 실행할 함수
        """
        self.alert_conditions.append({
            'pair': token_pair,
            'type': condition_type,
            'threshold': threshold,
            'callback': callback,
            'triggered': False
        })
        print(f"🔔 알림 조건 추가: {token_pair} {condition_type} {threshold}")
    
    async def get_current_price(self, token_in, token_out, amount=1.0):
        """
        현재 가격 조회 (비동기)
        """
        try:
            price = self.swap_engine.get_amounts_out(
                amount,
                [token_in, token_out]
            )
            return price
        except Exception as e:
            print(f"❌ 가격 조회 실패: {e}")
            return None
    
    async def monitor_price(self, token_in, token_out, interval=10):
        """
        가격 모니터링 (비동기)
        interval: 체크 간격 (초)
        """
        pair_key = f"{token_in[:6]}.../{token_out[:6]}..."
        
        print(f"\n👁️ 가격 모니터링 시작: {pair_key}")
        print(f"체크 간격: {interval}초")
        
        while True:
            try:
                price = await self.get_current_price(token_in, token_out)
                
                if price is not None:
                    timestamp = datetime.now().strftime("%H:%M:%S")
                    
                    # 가격 히스토리 저장
                    if pair_key not in self.price_history:
                        self.price_history[pair_key] = []
                    
                    self.price_history[pair_key].append({
                        'timestamp': timestamp,
                        'price': price
                    })
                    
                    # 최근 100개만 유지
                    if len(self.price_history[pair_key]) > 100:
                        self.price_history[pair_key].pop(0)
                    
                    print(f"[{timestamp}] {pair_key}: {price:.6f}")
                    
                    # 알림 조건 체크
                    await self._check_alert_conditions(
                        (token_in, token_out),
                        price
                    )
                
                await asyncio.sleep(interval)
                
            except Exception as e:
                print(f"❌ 모니터링 오류: {e}")
                await asyncio.sleep(interval)
    
    async def _check_alert_conditions(self, token_pair, current_price):
        """알림 조건 확인 및 실행"""
        for condition in self.alert_conditions:
            if condition['pair'] != token_pair:
                continue
            
            if condition['triggered']:
                continue
            
            should_trigger = False
            
            if condition['type'] == 'above' and current_price > condition['threshold']:
                should_trigger = True
            elif condition['type'] == 'below' and current_price < condition['threshold']:
                should_trigger = True
            
            if should_trigger:
                print(f"\n🚨 알림 조건 충족!")
                print(f"페어: {token_pair}")
                print(f"현재 가격: {current_price}")
                print(f"조건: {condition['type']} {condition['threshold']}")
                
                condition['triggered'] = True
                
                # 콜백 함수 실행
                if asyncio.iscoroutinefunction(condition['callback']):
                    await condition['callback'](token_pair, current_price)
                else:
                    condition['callback'](token_pair, current_price)
    
    def get_price_statistics(self, pair_key):
        """가격 통계 계산"""
        if pair_key not in self.price_history:
            return None
        
        prices = [item['price'] for item in self.price_history[pair_key]]
        
        if not prices:
            return None
        
        return {
            'current': prices[-1],
            'min': min(prices),
            'max': max(prices),
            'avg': sum(prices) / len(prices),
            'change': ((prices[-1] - prices[0]) / prices[0] * 100) if len(prices) > 1 else 0
        }
    
    async def monitor_multiple_pairs(self, pairs, interval=10):
        """
        여러 토큰 페어 동시 모니터링
        pairs: [(token_in, token_out), ...] 리스트
        """
        tasks = []
        for token_in, token_out in pairs:
            task = asyncio.create_task(
                self.monitor_price(token_in, token_out, interval)
            )
            tasks.append(task)
        
        await asyncio.gather(*tasks)

🎮 메인 실행 로직 작성

import asyncio
from wallet_manager import WalletManager
from swap_engine import SwapEngine
from price_monitor import PriceMonitor
import json
import os
from dotenv import load_dotenv

# 환경 변수 로드
load_dotenv()

# Uniswap V2 Router 정보
UNISWAP_ROUTER_ADDRESS = "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D"

# 간소화된 Uniswap Router ABI (필요한 함수만)
UNISWAP_ROUTER_ABI = json.loads('''[
    {
        "inputs": [],
        "name": "WETH",
        "outputs": [{"internalType": "address", "name": "", "type": "address"}],
        "stateMutability": "pure",
        "type": "function"
    },
    {
        "inputs": [
            {"internalType": "uint256", "name": "amountIn", "type": "uint256"},
            {"internalType": "address[]", "name": "path", "type": "address[]"}
        ],
        "name": "getAmountsOut",
        "outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
        "stateMutability": "view",
        "type": "function"
    },
    {
        "inputs": [
            {"internalType": "uint256", "name": "amountIn", "type": "uint256"},
            {"internalType": "uint256", "name": "amountOutMin", "type": "uint256"},
            {"internalType": "address[]", "name": "path", "type": "address[]"},
            {"internalType": "address", "name": "to", "type": "address"},
            {"internalType": "uint256", "name": "deadline", "type": "uint256"}
        ],
        "name": "swapExactTokensForTokens",
        "outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
        "stateMutability": "nonpayable",
        "type": "function"
    }
]''')

# 토큰 주소 (예시 - 실제 주소로 변경 필요)
USDC_ADDRESS = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
DAI_ADDRESS = "0x6B175474E89094C44Da98b954EedeAC495271d0F"

async def auto_swap_on_price_condition(token_pair, current_price):
    """
    가격 조건 충족시 자동 스왑 실행
    """
    print(f"\n🤖 자동 스왑 시작!")
    
    token_in, token_out = token_pair
    amount_to_swap = 100  # 스왑할 양 (실제로는 동적으로 계산)
    
    # 스왑 실행
    result = swap_engine.swap_with_slippage_protection(
        amount_in=amount_to_swap,
        path=[token_in, token_out],
        wallet_type='A',
        slippage_percent=1.0
    )
    
    if result['success']:
        print(f"✅ 자동 스왑 완료! TX: {result['tx_hash']}")
    else:
        print(f"❌ 자동 스왝 실패!")

async def main():
    """메인 실행 함수"""
    print("=" * 60)
    print("🚀 토큰 스왑 자동화 시스템 시작!")
    print("=" * 60)
    
    # 1. 지갑 매니저 초기화
    print("\n📱 지갑 매니저 초기화 중...")
    wallet_manager = WalletManager(os.getenv('RPC_URL'))
    
    # 지갑 잔액 확인
    print(f"\n💰 Wallet A 잔액:")
    eth_balance = wallet_manager.get_balance('A')
    print(f"  ETH: {eth_balance:.4f}")
    
    # 2. 스왑 엔진 초기화
    print("\n🔧 스왑 엔진 초기화 중...")
    global swap_engine
    swap_engine = SwapEngine(
        wallet_manager,
        UNISWAP_ROUTER_ADDRESS,
        UNISWAP_ROUTER_ABI
    )
    
    # 3. 가격 모니터 초기화
    print("\n📊 가격 모니터 초기화 중...")
    price_monitor = PriceMonitor(swap_engine)
    
    # 4. 알림 조건 설정
    print("\n⚙️ 알림 조건 설정 중...")
    
    # 예시: USDC/DAI 가격이 1.01 이상이면 자동 스왑
    price_monitor.add_alert_condition(
        token_pair=(USDC_ADDRESS, DAI_ADDRESS),
        condition_type='above',
        threshold=1.01,
        callback=auto_swap_on_price_condition
    )
    
    # 5. 모니터링 시작
    print("\n👁️ 가격 모니터링 시작!")
    print("Ctrl+C를 눌러 종료하세요.\n")
    
    try:
        # 여러 페어 동시 모니터링
        await price_monitor.monitor_multiple_pairs(
            pairs=[
                (USDC_ADDRESS, DAI_ADDRESS),
                # 필요한 만큼 페어 추가 가능
            ],
            interval=30  # 30초마다 체크
        )
    except KeyboardInterrupt:
        print("\n\n⏹️ 프로그램 종료!")
        
        # 통계 출력
        print("\n📈 가격 통계:")
        for pair_key in price_monitor.price_history:
            stats = price_monitor.get_price_statistics(pair_key)
            if stats:
                print(f"\n{pair_key}:")
                print(f"  현재: {stats['current']:.6f}")
                print(f"  최저: {stats['min']:.6f}")
                print(f"  최고: {stats['max']:.6f}")
                print(f"  평균: {stats['avg']:.6f}")
                print(f"  변동: {stats['change']:.2f}%")

if __name__ == "__main__":
    asyncio.run(main())

⚡ 성능 최적화 전략

def get_optimal_gas_price(self):
    """
    현재 네트워크 상황에 맞는 최적 가스 가격 계산
    """
    # 현재 가스 가격 조회
    current_gas = self.w3.eth.gas_price
    
    # 최근 블록들의 가스 가격 분석
    latest_block = self.w3.eth.block_number
    gas_prices = []
    
    for i in range(5):  # 최근 5개 블록
        block = self.w3.eth.get_block(latest_block - i)
        if 'baseFeePerGas' in block:
            gas_prices.append(block['baseFeePerGas'])
    
    if gas_prices:
        avg_gas = sum(gas_prices) // len(gas_prices)
        # 평균보다 10% 높게 설정 (빠른 처리를 위해)
        optimal_gas = int(avg_gas * 1.1)
    else:
        optimal_gas = current_gas
    
    # 최대 가스 가격 제한 (너무 비싸면 실행 안 함)
    max_gas_price = self.w3.to_wei(100, 'gwei')
    
    return min(optimal_gas, max_gas_price)

class OptimizedRPCManager:
    def __init__(self, rpc_urls):
        """여러 RPC 엔드포인트 관리"""
        self.rpc_urls = rpc_urls
        self.current_index = 0
        self.w3_instances = [Web3(Web3.HTTPProvider(url)) for url in rpc_urls]
        self.cache = {}
        self.cache_ttl = 10  # 캐시 유효 시간 (초)
    
    def get_w3(self):
        """현재 활성 Web3 인스턴스 반환"""
        return self.w3_instances[self.current_index]
    
    def switch_rpc(self):
        """다음 RPC로 전환"""
        self.current_index = (self.current_index + 1) % len(self.rpc_urls)
        print(f"🔄 RPC 전환: {self.rpc_urls[self.current_index]}")
    
    def get_with_cache(self, key, fetch_function, ttl=None):
        """캐싱을 활용한 데이터 조회"""
        import time
        
        if ttl is None:
            ttl = self.cache_ttl
        
        now = time.time()
        
        # 캐시에 있고 유효하면 반환
        if key in self.cache:
            cached_data, cached_time = self.cache[key]
            if now - cached_time < ttl:
                return cached_data
        
        # 캐시가 없거나 만료되면 새로 조회
        data = fetch_function()
        self.cache[key] = (data, now)
        return data

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

class AsyncSwapSystem:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=5)
    
    async def fetch_multiple_prices(self, token_pairs):
        """여러 토큰 페어의 가격을 동시에 조회"""
        async with aiohttp.ClientSession() as session:
            tasks = []
            for token_in, token_out in token_pairs:
                task = self.fetch_price_async(session, token_in, token_out)
                tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return results
    
    async def fetch_price_async(self, session, token_in, token_out):
        """비동기 가격 조회"""
        try:
            # API 호출 또는 블록체인 조회
            # 실제 구현은 사용하는 DEX에 따라 다름
            price = await self.get_price_from_api(session, token_in, token_out)
            return {
                'pair': (token_in, token_out),
                'price': price,
                'success': True
            }
        except Exception as e:
            return {
                'pair': (token_in, token_out),
                'error': str(e),
                'success': False
            }
    
    async def parallel_swap_execution(self, swap_tasks):
        """여러 스왑을 병렬로 실행"""
        results = await asyncio.gather(*swap_tasks, return_exceptions=True)
        
        successful = [r for r in results if r.get('success')]
        failed = [r for r in results if not r.get('success')]
        
        print(f"✅ 성공: {len(successful)}개")
        print(f"❌ 실패: {len(failed)}개")
        
        return results

🛡️ 보안과 리스크 관리

class RiskManager:
    def __init__(self, max_daily_loss=1000, max_single_trade=100):
        """
        리스크 매니저 초기화
        max_daily_loss: 일일 최대 손실 한도 (USD)
        max_single_trade: 단일 거래 최대 금액 (USD)
        """
        self.max_daily_loss = max_daily_loss
        self.max_single_trade = max_single_trade
        self.daily_pnl = 0  # 일일 손익
        self.trade_history = []
        self.emergency_stop = False
    
    def check_trade_allowed(self, trade_amount_usd):
        """거래 실행 가능 여부 확인"""
        # 긴급 정지 상태 확인
        if self.emergency_stop:
            print("🚨 긴급 정지 상태! 거래 불가")
            return False
        
        # 단일 거래 금액 확인
        if trade_amount_usd > self.max_single_trade:
            print(f"⚠️ 거래 금액 초과: {trade_amount_usd} > {self.max_single_trade}")
            return False
        
        # 일일 손실 한도 확인
        if self.daily_pnl < -self.max_daily_loss:
            print(f"🛑 일일 손실 한도 도달: {self.daily_pnl}")
            self.emergency_stop = True
            return False
        
        return True
    
    def record_trade(self, trade_result):
        """거래 결과 기록"""
        self.trade_history.append(trade_result)
        
        if 'pnl' in trade_result:
            self.daily_pnl += trade_result['pnl']
        
        # 이상 패턴 감지
        self._detect_anomalies()
    
    def _detect_anomalies(self):
        """비정상적인 거래 패턴 감지"""
        if len(self.trade_history) < 5:
            return
        
        # 최근 5개 거래 분석
        recent_trades = self.trade_history[-5:]
        failed_count = sum(1 for t in recent_trades if not t.get('success'))
        
        # 연속 실패가 3번 이상이면 경고
        if failed_count >= 3:
            print("🚨 경고: 연속 거래 실패 감지!")
            print("시스템 점검이 필요합니다.")
            # 여기서 텔레그램 알림 등을 보낼 수 있음
    
    def reset_daily_stats(self):
        """일일 통계 초기화 (매일 자정에 실행)"""
        print(f"\n📊 일일 리포트")
        print(f"총 거래: {len(self.trade_history)}건")
        print(f"손익: ${self.daily_pnl:.2f}")
        
        self.daily_pnl = 0
        self.trade_history = []
        self.emergency_stop = False

def calculate_dynamic_slippage(self, token_pair, base_slippage=0.5):
    """
    시장 상황에 따라 동적으로 슬리피지 계산
    """
    # 최근 가격 변동성 계산
    price_history = self.get_recent_prices(token_pair, minutes=10)
    
    if len(price_history) < 2:
        return base_slippage
    
    # 표준편차로 변동성 측정
    import statistics
    volatility = statistics.stdev(price_history)
    avg_price = statistics.mean(price_history)
    
    # 변동성 비율
    volatility_ratio = (volatility / avg_price) * 100
    
    # 변동성이 높으면 슬리피지 증가
    if volatility_ratio > 2.0:
        adjusted_slippage = base_slippage * 2
        print(f"⚠️ 높은 변동성 감지: {volatility_ratio:.2f}%")
        print(f"슬리피지 조정: {base_slippage}% → {adjusted_slippage}%")
    elif volatility_ratio > 1.0:
        adjusted_slippage = base_slippage * 1.5
    else:
        adjusted_slippage = base_slippage
    
    # 최대 슬리피지 제한
    return min(adjusted_slippage, 5.0)

📱 실전 배포와 운영

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 애플리케이션 코드 복사
COPY . .

# 환경 변수 설정
ENV PYTHONUNBUFFERED=1

# 실행
CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'

services:
  swap-bot:
    build: .
    container_name: token-swap-automation
    restart: unless-stopped
    env_file:
      - .env
    volumes:
      - ./logs:/app/logs
    networks:
      - swap-network

  # 모니터링 도구 (선택사항)
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    networks:
      - swap-network

networks:
  swap-network:
    driver: bridge

import logging
from logging.handlers import RotatingFileHandler
import json
from datetime import datetime

class SwapLogger:
    def __init__(self, log_file='swap_bot.log'):
        """로거 초기화"""
        self.logger = logging.getLogger('SwapBot')
        self.logger.setLevel(logging.INFO)
        
        # 파일 핸들러 (최대 10MB, 5개 백업)
        file_handler = RotatingFileHandler(
            log_file,
            maxBytes=10*1024*1024,
            backupCount=5
        )
        
        # 콘솔 핸들러
        console_handler = logging.StreamHandler()
        
        # 포맷 설정
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        
        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)
    
    def log_swap(self, swap_data):
        """스왑 실행 로그"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'type': 'SWAP',
            'data': swap_data
        }
        self.logger.info(json.dumps(log_entry))
    
    def log_price(self, token_pair, price):
        """가격 로그"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'type': 'PRICE',
            'pair': token_pair,
            'price': price
        }
        self.logger.info(json.dumps(log_entry))
    
    def log_error(self, error_msg, exception=None):
        """에러 로그"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'type': 'ERROR',
            'message': error_msg,
            'exception': str(exception) if exception else None
        }
        self.logger.error(json.dumps(log_entry))

import requests
import asyncio

class TelegramNotifier:
    def __init__(self, bot_token, chat_id):
        """
        텔레그램 알림 초기화
        bot_token: 봇 토큰 (@BotFather에서 발급)
        chat_id: 채팅방 ID
        """
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.api_url = f"https://api.telegram.org/bot{bot_token}"
    
    async def send_message(self, message, parse_mode='HTML'):
        """메시지 전송"""
        url = f"{self.api_url}/sendMessage"
        data = {
            'chat_id': self.chat_id,
            'text': message,
            'parse_mode': parse_mode
        }
        
        try:
            response = requests.post(url, data=data)
            return response.json()
        except Exception as e:
            print(f"텔레그램 전송 실패: {e}")
            return None
    
    async def notify_swap_success(self, swap_result):
        """스왑 성공 알림"""
        message = f"""
🎉 <b>스왑 성공!</b>

💱 거래 해시: <code>{swap_result['tx_hash'][:10]}...</code>
⛽ 가스 사용: {swap_result['gas_used']}
💰 금액: {swap_result.get('amount', 'N/A')}

✅ 거래가 성공적으로 완료되었습니다!
        """
        await self.send_message(message)
    
    async def notify_error(self, error_msg):
        """에러 알림"""
        message = f"""
🚨 <b>에러 발생!</b>

❌ {error_msg}

시스템을 확인해주세요.
        """
        await self.send_message(message)
    
    async def notify_daily_report(self, stats):
        """일일 리포트"""
        message = f"""
📊 <b>일일 거래 리포트</b>

📈 총 거래: {stats['total_trades']}건
✅ 성공: {stats['successful']}건
❌ 실패: {stats['failed']}건
💵 총 손익: ${stats['pnl']:.2f}

계속 열심히 일하고 있어요! 💪
        """
        await self.send_message(message)

🎓 고급 전략과 최적화

class ArbitrageEngine:
    def __init__(self, dex_list):
        """
        차익거래 엔진
        dex_list: [(dex_name, router_address, router_abi), ...]
        """
        self.dexes = {}
        for name, address, abi in dex_list:
            self.dexes[name] = {
                'router': self.w3.eth.contract(address=address, abi=abi)
            }
    
    async def find_arbitrage_opportunities(self, token_a, token_b, amount=1.0):
        """
        차익거래 기회 탐색
        """
        opportunities = []
        
        # 모든 DEX 페어에서 가격 조회
        prices = {}
        for dex_name, dex_info in self.dexes.items():
            try:
                price = await self.get_price_from_dex(
                    dex_name,
                    token_a,
                    token_b,
                    amount
                )
                prices[dex_name] = price
            except Exception as e:
                print(f"❌ {dex_name} 가격 조회 실패: {e}")
        
        # 가격 차이 분석
        if len(prices) < 2:
            return opportunities
        
        dex_names = list(prices.keys())
        for i in range(len(dex_names)):
            for j in range(i + 1, len(dex_names)):
                dex1, dex2 = dex_names[i], dex_names[j]
                price1, price2 = prices[dex1], prices[dex2]
                
                # 가격 차이 계산
                if price1 > price2:
                    profit_percent = ((price1 - price2) / price2) * 100
                    buy_dex, sell_dex = dex2, dex1
                    buy_price, sell_price = price2, price1
                else:
                    profit_percent = ((price2 - price1) / price1) * 100
                    buy_dex, sell_dex = dex1, dex2
                    buy_price, sell_price = price1, price2
                
                # 수익성 있는 기회만 저장 (가스비 고려)
                if profit_percent > 1.0:  # 1% 이상 차이
                    opportunities.append({
                        'buy_dex': buy_dex,
                        'sell_dex': sell_dex,
                        'buy_price': buy_price,
                        'sell_price': sell_price,
                        'profit_percent': profit_percent,
                        'estimated_profit': (sell_price - buy_price) * amount
                    })
        
        # 수익률 높은 순으로 정렬
        opportunities.sort(key=lambda x: x['profit_percent'], reverse=True)
        
        return opportunities
    
    async def execute_arbitrage(self, opportunity, amount):
        """
        차익거래 실행
        """
        print(f"\n💎 차익거래 실행!")
        print(f"매수: {opportunity['buy_dex']} @ {opportunity['buy_price']}")
        print(f"매도: {opportunity['sell_dex']} @ {opportunity['sell_price']}")
        print(f"예상 수익: {opportunity['profit_percent']:.2f}%")
        
        # 1단계: 저렴한 DEX에서 매수
        buy_result = await self.swap_on_dex(
            opportunity['buy_dex'],
            token_a,
            token_b,
            amount
        )
        
        if not buy_result['success']:
            print("❌ 매수 실패!")
            return {'success': False}
        
        # 2단계: 비싼 DEX에서 매도
        sell_result = await self.swap_on_dex(
            opportunity['sell_dex'],
            token_b,
            token_a,
            buy_result['amount_out']
        )
        
        if not sell_result['success']:
            print("⚠️ 매도 실패! 토큰이 묶였습니다.")
            return {'success': False, 'stuck_tokens': buy_result['amount_out']}
        
        # 실제 수익 계산
        actual_profit = sell_result['amount_out'] - amount
        print(f"✅ 차익거래 완료! 실제 수익: {actual_profit:.6f}")
        
        return {
            'success': True,
            'profit': actual_profit,
            'buy_tx': buy_result['tx_hash'],
            'sell_tx': sell_result['tx_hash']
        }

class FlashLoanArbitrage:
    def __init__(self, aave_pool_address, aave_pool_abi):
        """
        플래시론 차익거래 엔진
        Aave 프로토콜 사용
        """
        self.aave_pool = self.w3.eth.contract(
            address=aave_pool_address,
            abi=aave_pool_abi
        )
    
    def create_flashloan_contract(self):
        """
        플래시론 실행 스마트 컨트랙트 배포
        (Solidity 코드를 미리 작성해야 함)
        """
        # 실제로는 Solidity로 작성한 컨트랙트를 배포해야 해
        # 여기서는 개념만 설명
        
        flashloan_contract_code = """
        // SPDX-License-Identifier: MIT
        pragma solidity ^0.8.0;
        
        import "@aave/core-v3/contracts/flashloan/base/FlashLoanSimpleReceiverBase.sol";
        
        contract FlashLoanArbitrage is FlashLoanSimpleReceiverBase {
            address payable owner;
            
            constructor(address _addressProvider) 
                FlashLoanSimpleReceiverBase(IPoolAddressesProvider(_addressProvider)) 
            {
                owner = payable(msg.sender);
            }
            
            function executeOperation(
                address asset,
                uint256 amount,
                uint256 premium,
                address initiator,
                bytes calldata params
            ) external override returns (bool) {
                // 1. 차익거래 로직 실행
                // 2. 수익 확인
                // 3. 대출금 + 수수료 상환
                
                uint256 amountOwed = amount + premium;
                IERC20(asset).approve(address(POOL), amountOwed);
                
                return true;
            }
            
            function requestFlashLoan(address _token, uint256 _amount) public {
                address receiverAddress = address(this);
                address asset = _token;
                uint256 amount = _amount;
                bytes memory params = "";
                uint16 referralCode = 0;
                
                POOL.flashLoanSimple(
                    receiverAddress,
                    asset,
                    amount,
                    params,
                    referralCode
                );
            }
        }
        """
        
        print("📝 플래시론 컨트랙트 배포 필요")
        print("Remix나 Hardhat을 사용해서 배포하세요!")

class MEVStrategy:
    def __init__(self, flashbots_relay_url):
        """
        MEV 전략 엔진
        Flashbots를 통해 프라이빗 트랜잭션 전송
        """
        self.flashbots_url = flashbots_relay_url
    
    async def detect_profitable_tx(self, pending_tx):
        """
        수익성 있는 트랜잭션 감지
        """
        # 대기 중인 트랜잭션 분석
        if self.is_large_swap(pending_tx):
            # 큰 스왑 트랜잭션 발견
            opportunity = await self.calculate_sandwich_opportunity(pending_tx)
            
            if opportunity['profitable']:
                return opportunity
        
        return None
    
    def is_large_swap(self, tx):
        """큰 스왑 트랜잭션인지 확인"""
        # 트랜잭션 데이터 파싱
        # Uniswap swapExactTokensForTokens 함수 호출인지 확인
        # 금액이 충분히 큰지 확인
        pass
    
    async def send_bundle_to_flashbots(self, transactions):
        """
        Flashbots를 통해 트랜잭션 번들 전송
        """
        bundle = {
            'txs': transactions,
            'blockNumber': self.w3.eth.block_number + 1,
            'minTimestamp': 0,
            'maxTimestamp': 0
        }
        
        # Flashbots API 호출
        # 실제 구현은 Flashbots 문서 참고
        print("📦 Flashbots 번들 전송")

🔍 디버깅과 문제 해결

class DebugHelper:
    def __init__(self, w3):
        self.w3 = w3
    
    def analyze_failed_transaction(self, tx_hash):
        """
        실패한 트랜잭션 분석
        """
        print(f"\n🔍 트랜잭션 분석: {tx_hash}")
        
        # 트랜잭션 정보 조회
        tx = self.w3.eth.get_transaction(tx_hash)
        receipt = self.w3.eth.get_transaction_receipt(tx_hash)
        
        print(f"\n📋 기본 정보:")
        print(f"  From: {tx['from']}")
        print(f"  To: {tx['to']}")
        print(f"  Value: {self.w3.from_wei(tx['value'], 'ether')} ETH")
        print(f"  Gas Price: {self.w3.from_wei(tx['gasPrice'], 'gwei')} Gwei")
        print(f"  Gas Limit: {tx['gas']}")
        
        print(f"\n📊 실행 결과:")
        print(f"  Status: {'성공' if receipt['status'] == 1 else '실패'}")
        print(f"  Gas Used: {receipt['gasUsed']}")
        print(f"  Block: {receipt['blockNumber']}")
        
        # 실패 원인 추정
        if receipt['status'] == 0:
            print(f"\n❌ 실패 원인 분석:")
            
            # 가스 부족
            if receipt['gasUsed'] == tx['gas']:
                print("  • 가스 부족 (Out of Gas)")
                print("  → 가스 리밋을 높여보세요")
            
            # 트랜잭션 시뮬레이션으로 상세 에러 확인
            try:
                self.w3.eth.call(tx, receipt['blockNumber'] - 1)
            except Exception as e:
                print(f"  • 에러 메시지: {str(e)}")
    
    def simulate_transaction(self, tx_params):
        """
        트랜잭션 시뮬레이션 (실제 전송 전 테스트)
        """
        print("\n🧪 트랜잭션 시뮬레이션...")
        
        try:
            # eth_call로 시뮬레이션
            result = self.w3.eth.call(tx_params)
            print("✅ 시뮬레이션 성공!")
            print(f"예상 결과: {result.hex()}")
            
            # 가스 예측
            estimated_gas = self.w3.eth.estimate_gas(tx_params)
            print(f"예상 가스: {estimated_gas}")
            
            return True
        except Exception as e:
            print(f"❌ 시뮬레이션 실패: {str(e)}")
            return False
    
    def check_token_approval(self, token_address, owner, spender):
        """토큰 승인 상태 확인"""
        token = self.w3.eth.contract(
            address=token_address,
            abi=ERC20_ABI
        )
        
        allowance = token.functions.allowance(owner, spender).call()
        decimals = token.functions.decimals().call()
        
        print(f"\n🔐 승인 상태:")
        print(f"  Owner: {owner}")
        print(f"  Spender: {spender}")
        print(f"  Allowance: {allowance / (10 ** decimals)}")
        
        return allowance

📚 추가 학습 리소스

🎬 마무리하며