PancakeSwap 트레이딩 봇 2026: Python으로 BSC에서 자동화된 DeFi 전략 구축 — 완전 설정 가이드

x * y = k

# Price of token A in terms of token B
price_a = y / x

# When a swap occurs: (x + dx) * (y - dy) = k
# After 0.25% fee: dx * 0.9975 is what actually enters the pool

# Slippage calculation for a swap
def calculate_min_output(amount_in, reserve_in, reserve_out, slippage_tolerance=0.005):
    """Calculate minimum output with 0.5% slippage tolerance."""
    amount_in_with_fee = amount_in * 9975 // 10000  # 0.25% fee
    numerator = amount_in_with_fee * reserve_out
    denominator = reserve_in + amount_in_with_fee
    expected_output = numerator // denominator
    min_output = int(expected_output * (1 - slippage_tolerance))
    return min_output

# Option A: Public endpoint (rate-limited, NOT for production)
BSC_RPC = "https://bsc-dataseed.binance.org/"

# Option B: QuickNode / Alchemy (recommended for production)
BSC_RPC = "https://docs.chainstack.com/"  # Get your endpoint from Chainstack

# Option C: Self-hosted geth node (maximum reliability)
# geth --config ./config.toml --datadir ./node --http

python -m venv pancakeswap-bot-env
source pancakeswap-bot-env/bin/activate

pip install --upgrade pip
pip install web3==7.6.0 python-dotenv==1.0.1 requests==2.32.3 eth-account==0.13.4

# config.py — all contract addresses and settings
import os
from dotenv import load_dotenv

load_dotenv()

BSC_RPC = os.getenv("BSC_RPC", "https://bsc-dataseed.binance.org/")
PRIVATE_KEY = os.getenv("PRIVATE_KEY")  # 0x-prefixed hex
WALLET_ADDRESS = os.getenv("WALLET_ADDRESS")

# PancakeSwap V2 contracts (verified May 2026)
PANCAKE_ROUTER_V2 = "0x10ED43C718714eb63d5aA57B78B54704E256024E"
PANCAKE_FACTORY_V2 = "0xcA143Ce32Fe78f1f7019d7d551a6402fC5350c73"
PANCAKE_ROUTER_V3 = "0x13f4EA83D0bd40E75C8222255bc855a974568Dd4"
PANCAKE_FACTORY_V3 = "0x0BFbCF9fa4f9C56B0F40a671Ad40E0805A091865"

# Token addresses (BSC mainnet)
WBNB = "0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c"
BUSD = "0xe9e7CEA3DedcA5984780Bafc599bD69ADd087D56"
USDT = "0x55d398326f99059fF775485246999027B3197955"
USDC = "0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d"
CAKE = "0x0E09FaBB73Bd3Ade0a17ECC321fD13a19e81cE82"

# Bot configuration
GAS_LIMIT_SWAP = 300000
GAS_LIMIT_APPROVE = 100000
DEFAULT_SLIPPAGE = 0.005  # 0.5%
MAX_GAS_PRICE_GWEI = 5
MIN_PROFIT_BNB = 0.001    # Minimum profit to execute

# bot/client.py — Web3 connection with retry logic
from web3 import Web3
from web3.middleware import geth_poa_middleware
import config

class BSCClient:
    def __init__(self):
        self.w3 = Web3(Web3.HTTPProvider(config.BSC_RPC))
        # BSC uses PoA consensus — required middleware
        self.w3.middleware_onion.inject(geth_poa_middleware, layer=0)

        if not self.w3.is_connected():
            raise ConnectionError("Failed to connect to BSC node")

        print(f"Connected to BSC. Block: {self.w3.eth.block_number}")
        print(f"Gas price: {self.w3.from_wei(self.w3.eth.gas_price, 'gwei'):.2f} gwei")

        self.account = self.w3.eth.account.from_key(config.PRIVATE_KEY)
        self.address = self.account.address

        # Load PancakeSwap Router contract
        with open("abi/router_v2.json") as f:
            router_abi = f.read()
        self.router = self.w3.eth.contract(
            address=Web3.to_checksum_address(config.PANCAKE_ROUTER_V2),
            abi=router_abi
        )

    def get_balance(self, token_address=None):
        """Get BNB or token balance."""
        if token_address is None:
            return self.w3.from_wei(
                self.w3.eth.get_balance(self.address), "ether"
            )
        token = self.w3.eth.contract(
            address=Web3.to_checksum_address(token_address),
            abi=[{"name":"balanceOf","type":"function","inputs":[{"name":"","type":"address"}],"outputs":[{"name":"","type":"uint256"}],"constant":True}]
        )
        return self.w3.from_wei(token.functions.balanceOf(self.address).call(), "ether")

client = BSCClient()
print(f"BNB Balance: {client.get_balance():.4f} BNB")

# bot/swap.py — swap execution with full safety checks
from web3 import Web3
import config

class PancakeSwapBot:
    def __init__(self, client):
        self.client = client
        self.w3 = client.w3
        self.router = client.router

    def approve_token(self, token_address, spender=None, amount=None):
        """Approve router to spend tokens."""
        spender = spender or config.PANCAKE_ROUTER_V2
        amount = amount or 2**256 - 1  # Max uint256 (unlimited)

        token = self.w3.eth.contract(
            address=Web3.to_checksum_address(token_address),
            abi=[
                {"name":"approve","type":"function","inputs":[{"name":"spender","type":"address"},{"name":"amount","type":"uint256"}],"outputs":[{"name":"","type":"bool"}]},
                {"name":"allowance","type":"function","inputs":[{"name":"owner","type":"address"},{"name":"spender","type":"address"}],"outputs":[{"name":"","type":"uint256"}],"constant":True}
            ]
        )

        # Check existing allowance
        current = token.functions.allowance(self.client.address, spender).call()
        if current >= amount // 2:
            print(f"Token {token_address} already approved")
            return True

        tx = token.functions.approve(
            Web3.to_checksum_address(spender), amount
        ).build_transaction({
            "from": self.client.address,
            "gas": config.GAS_LIMIT_APPROVE,
            "gasPrice": self.w3.eth.gas_price,
            "nonce": self.w3.eth.get_transaction_count(self.client.address),
        })

        signed = self.w3.eth.account.sign_transaction(tx, config.PRIVATE_KEY)
        tx_hash = self.w3.eth.send_raw_transaction(signed.raw_transaction)
        receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)

        print(f"Approval tx: {tx_hash.hex()} — Status: {receipt['status']}")
        return receipt["status"] == 1

    def swap_exact_tokens_for_tokens(
        self,
        amount_in_wei,
        token_in,
        token_out,
        slippage=None,
        deadline_seconds=300
    ):
        """Execute a token swap with slippage protection."""
        slippage = slippage or config.DEFAULT_SLIPPAGE

        # Get expected output
        path = [token_in, config.WBNB, token_out] if token_in != config.WBNB and token_out != config.WBNB else [token_in, token_out]

        amounts_out = self.router.functions.getAmountsOut(
            amount_in_wei, path
        ).call()
        expected_out = amounts_out[-1]
        min_output = int(expected_out * (1 - slippage))

        print(f"Expected output: {self.w3.from_wei(expected_out, 'ether'):.6f}")
        print(f"Min output ({slippage*100:.1f}% slippage): {self.w3.from_wei(min_output, 'ether'):.6f}")

        deadline = self.w3.eth.get_block("latest")["timestamp"] + deadline_seconds

        tx = self.router.functions.swapExactTokensForTokens(
            amount_in_wei,
            min_output,
            path,
            self.client.address,
            deadline
        ).build_transaction({
            "from": self.client.address,
            "gas": config.GAS_LIMIT_SWAP,
            "gasPrice": self.w3.eth.gas_price,
            "nonce": self.w3.eth.get_transaction_count(self.client.address),
        })

        signed = self.w3.eth.account.sign_transaction(tx, config.PRIVATE_KEY)
        tx_hash = self.w3.eth.send_raw_transaction(signed.raw_transaction)
        receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)

        if receipt["status"] == 1:
            print(f"Swap success: {tx_hash.hex()}")
            gas_cost = receipt["gasUsed"] * tx["gasPrice"]
            print(f"Gas cost: {self.w3.from_wei(gas_cost, 'ether'):.6f} BNB")
        else:
            print(f"Swap FAILED: {tx_hash.hex()}")

        return receipt

    def swap_bnb_for_tokens(self, bnb_amount, token_out, slippage=None):
        """Swap BNB for tokens (wraps BNB to WBNB internally)."""
        amount_in_wei = self.w3.to_wei(bnb_amount, "ether")
        path = [config.WBNB, token_out]

        amounts_out = self.router.functions.getAmountsOut(amount_in_wei, path).call()
        min_output = int(amounts_out[-1] * (1 - (slippage or config.DEFAULT_SLIPPAGE)))

        deadline = self.w3.eth.get_block("latest")["timestamp"] + 300

        tx = self.router.functions.swapExactETHForTokens(
            min_output, path, self.client.address, deadline
        ).build_transaction({
            "from": self.client.address,
            "value": amount_in_wei,
            "gas": config.GAS_LIMIT_SWAP,
            "gasPrice": self.w3.eth.gas_price,
            "nonce": self.w3.eth.get_transaction_count(self.client.address),
        })

        signed = self.w3.eth.account.sign_transaction(tx, config.PRIVATE_KEY)
        tx_hash = self.w3.eth.send_raw_transaction(signed.raw_transaction)
        return self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)

# bot/monitor.py — pool monitoring and price tracking
import json
from web3 import Web3
import config

class PoolMonitor:
    def __init__(self, client):
        self.client = client
        self.w3 = client.w3
        with open("abi/factory_v2.json") as f:
            factory_abi = json.load(f)
        with open("abi/pair.json") as f:
            pair_abi = json.load(f)
        self.factory = self.w3.eth.contract(
            address=Web3.to_checksum_address(config.PANCAKE_FACTORY_V2),
            abi=factory_abi
        )
        self.pair_abi = pair_abi

    def get_pair_address(self, token_a, token_b):
        """Get the LP pair address for two tokens."""
        return self.factory.functions.getPair(
            Web3.to_checksum_address(token_a),
            Web3.to_checksum_address(token_b)
        ).call()

    def get_pool_reserves(self, token_a, token_b):
        """Get current reserves and compute price."""
        pair_address = self.get_pair_address(token_a, token_b)
        if pair_address == "0x0000000000000000000000000000000000000000":
            return None

        pair = self.w3.eth.contract(address=pair_address, abi=self.pair_abi)
        reserves = pair.functions.getReserves().call()
        token0 = pair.functions.token0().call()

        if token0 == Web3.to_checksum_address(token_a):
            reserve_a, reserve_b = reserves[0], reserves[1]
        else:
            reserve_a, reserve_b = reserves[1], reserves[0]

        price = reserve_b / reserve_a if reserve_a > 0 else 0

        return {
            "pair_address": pair_address,
            "reserve_a": reserve_a,
            "reserve_b": reserve_b,
            "price_a_per_b": price,
            "price_b_per_a": 1 / price if price > 0 else 0,
            "tvl_approx": reserve_a + reserve_b,
            "block_timestamp": reserves[2]
        }

    def calculate_price_impact(self, token_a, token_b, amount_in_wei):
        """Calculate price impact of a trade."""
        pool = self.get_pool_reserves(token_a, token_b)
        if not pool:
            return None

        reserve_in = pool["reserve_a"]
        reserve_out = pool["reserve_b"]

        amount_in_with_fee = amount_in_wei * 9975 // 10000
        new_reserve_in = reserve_in + amount_in_with_fee
        new_reserve_out = (reserve_in * reserve_out) // new_reserve_in

        amount_out = reserve_out - new_reserve_out
        current_price = reserve_out / reserve_in
        execution_price = amount_out / amount_in_wei if amount_in_wei > 0 else 0

        price_impact = (current_price - execution_price) / current_price if current_price > 0 else 0

        return {
            "amount_out": amount_out,
            "execution_price": execution_price,
            "current_price": current_price,
            "price_impact": price_impact,
            "is_safe": price_impact < 0.01
        }

    def watch_pool(self, token_a, token_b, callback, interval=12):
        """Watch pool and call callback on significant changes."""
        import time
        last_price = None

        while True:
            pool = self.get_pool_reserves(token_a, token_b)
            if pool:
                current_price = pool["price_a_per_b"]
                if last_price and abs(current_price - last_price) / last_price > 0.005:
                    callback({
                        "event": "PRICE_CHANGE",
                        "old_price": last_price,
                        "new_price": current_price,
                        "change_pct": (current_price - last_price) / last_price * 100,
                        "pool": pool
                    })
                last_price = current_price
            time.sleep(interval)

# utils/gas.py — gas optimization and MEV protection
import random

class MEVProtection:
    def __init__(self, client):
        self.client = client
        self.w3 = client.w3

    def calculate_safe_slippage(self, token_in, token_out, amount_in_wei):
        """Dynamic slippage based on pool depth and volatility."""
        monitor = PoolMonitor(self.client)
        impact = monitor.calculate_price_impact(token_in, token_out, amount_in_wei)

        if not impact:
            return 0.02

        base_slippage = impact["price_impact"] * 2
        volatility_buffer = self.estimate_volatility(token_in, token_out)

        safe_slippage = min(base_slippage + volatility_buffer, 0.05)
        return max(safe_slippage, 0.005)

    def estimate_volatility(self, token_a, token_b, blocks=50):
        """Estimate recent price volatility from on-chain data."""
        monitor = PoolMonitor(self.client)
        prices = []

        for i in range(blocks):
            try:
                pool = monitor.get_pool_reserves(token_a, token_b)
                if pool:
                    prices.append(pool["price_a_per_b"])
            except Exception:
                pass

        if len(prices) < 10:
            return 0.01

        import numpy as np
        returns = np.diff(np.log(prices))
        volatility = np.std(returns) * np.sqrt(24 * 3600 / 3)
        return min(volatility, 0.03)

    def generate_private_tx(self, tx_dict):
        """Add randomness to transaction to prevent front-running."""
        base_gas = tx_dict.get("gasPrice", self.w3.eth.gas_price)
        jitter = random.randint(-0.05 * base_gas, 0.05 * base_gas)
        tx_dict["gasPrice"] = base_gas + jitter
        tx_dict["deadline"] = self.w3.eth.get_block("latest")["timestamp"] + 60
        return tx_dict

class PrivateTransactionSender:
    """Send transactions via private mempool to avoid sandwich attacks."""

    def __init__(self, client):
        self.client = client
        self.private_rpcs = [
            "https://bsc.private.rpc.endpoint1",
            "https://bsc.private.rpc.endpoint2",
        ]

    def send_private(self, signed_tx):
        """Try sending via private RPC first, fallback to public."""
        import requests
        for rpc in self.private_rpcs:
            try:
                resp = requests.post(rpc, json={
                    "jsonrpc": "2.0",
                    "method": "eth_sendRawTransaction",
                    "params": [signed_tx.raw_transaction.hex()],
                    "id": 1
                }, timeout=10)
                if resp.status_code == 200:
                    return resp.json()["result"]
            except Exception as e:
                print(f"Private RPC failed: {e}")
                continue

        return self.client.w3.eth.send_raw_transaction(signed_tx.raw_transaction)

# strategies/momentum.py — momentum breakout strategy
import time
from datetime import datetime

class MomentumStrategy:
    def __init__(self, bot, monitor, config_overrides=None):
        self.bot = bot
        self.monitor = monitor
        self.price_history = []
        self.max_history = 20
        self.rsi_period = 14
        self.overbought = 70
        self.oversold = 30

    def calculate_rsi(self, prices, period=14):
        """Calculate Relative Strength Index."""
        if len(prices) < period + 1:
            return 50

        import numpy as np
        deltas = np.diff(prices)
        gains = deltas[deltas > 0]
        losses = -deltas[deltas < 0]

        avg_gain = np.mean(gains[-period:]) if len(gains) > 0 else 0
        avg_loss = np.mean(losses[-period:]) if len(losses) > 0 else 0.001

        rs = avg_gain / avg_loss
        return 100 - (100 / (1 + rs))

    def run(self, token_in, token_out, trade_size_bnb=0.1):
        """Main loop: buy oversold, sell overbought."""
        print(f"Starting momentum strategy on {token_in} -> {token_out}")
        position = 0

        while True:
            pool = self.monitor.get_pool_reserves(token_in, token_out)
            if not pool:
                time.sleep(12)
                continue

            price = pool["price_a_per_b"]
            self.price_history.append(price)
            if len(self.price_history) > self.max_history:
                self.price_history.pop(0)

            rsi = self.calculate_rsi(self.price_history)
            print(f"[{datetime.now()}] Price: {price:.8f}, RSI: {rsi:.1f}")

            if rsi < self.oversold and position == 0:
                print("OVERSOLD — BUY signal")
                receipt = self.bot.swap_bnb_for_tokens(trade_size_bnb, token_out)
                if receipt["status"] == 1:
                    position = 1

            elif rsi > self.overbought and position == 1:
                print("OVERBOUGHT — SELL signal")
                balance = self.bot.client.get_balance(token_out)
                if balance > 0:
                    receipt = self.bot.swap_exact_tokens_for_tokens(
                        self.bot.client.w3.to_wei(balance, "ether"),
                        token_out, config.WBNB
                    )
                    if receipt["status"] == 1:
                        position = 0

            time.sleep(12)

# strategies/arbitrage.py — cross-market arbitrage
import requests

class ArbitrageStrategy:
    def __init__(self, bot, monitor):
        self.bot = bot
        self.monitor = monitor
        self.min_profit_bnb = config.MIN_PROFIT_BNB
        self.binance_api = "https://api.binance.com/api/v3"

    def get_binance_price(self, symbol="BNBUSDT"):
        """Get Binance spot price."""
        try:
            resp = requests.get(
                f"{self.binance_api}/ticker/price",
                params={"symbol": symbol},
                timeout=5
            )
            return float(resp.json()["price"])
        except Exception as e:
            print(f"Binance API error: {e}")
            return None

    def get_pancake_price(self, token_a, token_b):
        """Get PancakeSwap price from pool reserves."""
        pool = self.monitor.get_pool_reserves(token_a, token_b)
        if pool:
            return pool["price_a_per_b"]
        return None

    def find_arbitrage(self):
        """Compare prices and find profitable arbitrage."""
        binance_price = self.get_binance_price("BNBUSDT")
        pancake_price = self.get_pancake_price(config.WBNB, config.BUSD)

        if not binance_price or not pancake_price:
            return None

        diff_pct = abs(pancake_price - binance_price) / binance_price

        if diff_pct > 0.002:
            direction = "BUY_BINANCE_SELL_PANCAKE" if binance_price < pancake_price else "BUY_PANCAKE_SELL_BINANCE"
            return {
                "direction": direction,
                "binance": binance_price,
                "pancake": pancake_price,
                "diff_pct": diff_pct * 100,
                "estimated_profit": diff_pct * 0.1
            }
        return None

    def execute_arbitrage(self, opportunity):
        """Execute arbitrage trade."""
        print(f"Arbitrage found: {opportunity}")
        if opportunity["direction"] == "BUY_PANCAKE_SELL_BINANCE":
            receipt = self.bot.swap_bnb_for_tokens(0.1, config.BUSD)
            if receipt["status"] == 1:
                print("PancakeSwap buy executed — sell on Binance via API")

# strategies/yield_optimizer.py — auto-compound CAKE rewards
import time
import json

class YieldOptimizer:
    def __init__(self, client, bot):
        self.client = client
        self.bot = bot
        self.min_cake_to_harvest = 1.0
        self.compound_interval = 3600

        self.MASTERCHEF_V2 = "0xa5f8C5Dbd5F286960b9d90539899c60F9665A72f"
        with open("abi/masterchef.json") as f:
            masterchef_abi = json.load(f)
        self.masterchef = self.client.w3.eth.contract(
            address=self.MASTERCHEF_V2, abi=masterchef_abi
        )

    def get_pending_cake(self, pid):
        """Get pending CAKE rewards for a farm pool."""
        return self.client.w3.from_wei(
            self.masterchef.functions.pendingCake(pid, self.client.address).call(),
            "ether"
        )

    def harvest(self, pid):
        """Harvest CAKE rewards from a farm."""
        tx = self.masterchef.functions.deposit(
            pid, 0
        ).build_transaction({
            "from": self.client.address,
            "gas": 250000,
            "gasPrice": self.client.w3.eth.gas_price,
            "nonce": self.client.w3.eth.get_transaction_count(self.client.address),
        })

        signed = self.client.w3.eth.account.sign_transaction(tx, config.PRIVATE_KEY)
        tx_hash = self.client.w3.eth.send_raw_transaction(signed.raw_transaction)
        receipt = self.client.w3.eth.wait_for_transaction_receipt(tx_hash)

        if receipt["status"] == 1:
            print(f"Harvested from pool {pid}: {tx_hash.hex()}")
        return receipt

    def compound(self, pid):
        """Harvest CAKE and restake into the farm."""
        self.harvest(pid)
        cake_balance = self.client.get_balance(config.CAKE)
        if cake_balance < self.min_cake_to_harvest:
            print(f"Not enough CAKE to compound: {cake_balance:.4f}")
            return
        print(f"Compounding {cake_balance:.4f} CAKE...")

    def run(self, pid=0):
        """Main loop for auto-compounding."""
        while True:
            pending = self.get_pending_cake(pid)
            print(f"Pending CAKE: {pending:.4f}")
            if pending >= self.min_cake_to_harvest:
                self.compound(pid)
            time.sleep(self.compound_interval)

import asyncio
from web3 import AsyncWeb3

class AsyncBSCBot:
    def __init__(self, rpc_url):
        self.w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc_url))

    async def monitor_multiple_pairs(self, pairs):
        """Monitor multiple pairs concurrently."""
        tasks = [self.check_pair(p) for p in pairs]
        return await asyncio.gather(*tasks)

    async def check_pair(self, pair):
        pool = await self.get_pool_async(pair["token_a"], pair["token_b"])
        return {
            "pair": pair["name"],
            "price": pool["price"],
            "opportunity": self.evaluate(pair, pool)
        }

# utils/alerts.py
import requests

class TelegramAlerter:
    def __init__(self, bot_token, chat_id):
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.base_url = f"https://api.telegram.org/bot{bot_token}"

    def send(self, message, level="INFO"):
        emoji = {"INFO": "ℹ️", "WARNING": "⚠️", "ERROR": "🚨", "PROFIT": "💰"}
        text = f"{emoji.get(level, '')} PancakeBot: {message}"
        requests.post(
            f"{self.base_url}/sendMessage",
            json={"chat_id": self.chat_id, "text": text},
            timeout=5
        )

import sqlite3
from datetime import datetime

class TradeLogger:
    def __init__(self, db_path="trades.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS trades (
                id INTEGER PRIMARY KEY,
                timestamp TEXT,
                strategy TEXT,
                token_in TEXT,
                token_out TEXT,
                amount_in REAL,
                amount_out REAL,
                gas_cost REAL,
                profit REAL,
                tx_hash TEXT
            )
        """)

    def log_trade(self, strategy, token_in, token_out, amount_in, amount_out, gas_cost, profit, tx_hash):
        self.conn.execute("""
            INSERT INTO trades (timestamp, strategy, token_in, token_out, amount_in, amount_out, gas_cost, profit, tx_hash)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (datetime.now().isoformat(), strategy, token_in, token_out, amount_in, amount_out, gas_cost, profit, tx_hash))
        self.conn.commit()

class NonceManager:
    def __init__(self, w3, address):
        self.w3 = w3
        self.address = address
        self._nonce = w3.eth.get_transaction_count(address)

    def next(self):
        nonce = self._nonce
        self._nonce += 1
        return nonce

    def reset(self):
        self._nonce = self.w3.eth.get_transaction_count(self.address)