Dưới đây là template hoàn chỉnh của bot trade tự động trên PancakeSwap, dùng Python + Web3, có sẵn các chức năng: Kết nối ví bằng private key. Mua token tự động qua BNB. Theo dõi giá token. Gửi thông báo về Telegram khi giao dịch thành công. Có sẵn stop-loss và take-profit. Cấu trúc thư mục dự án Pgsql Pancake_trade_bot/ ├──. Env ├── config. Py ├── trade_bot. Py ├── abi/ │ ├── pancake_router. Json │ └── erc20. Json . Env Code: PRIVATE_KEY=0xYOUR_PRIVATE_KEY WALLET_ADDRESS=0xYOUR_WALLET_ADDRESS RPC=https://bsc-dataseed.binance.org/ TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN CHAT_ID=YOUR_CHAT_ID config.py Code: from web3 import Web3 import os from dotenv import load_dotenv load_dotenv() WALLET_ADDRESS = Web3.to_checksum_address(os.getenv("WALLET_ADDRESS")) PRIVATE_KEY = os.getenv("PRIVATE_KEY") RPC = os.getenv("RPC") WEB3 = Web3(Web3.HTTPProvider(RPC)) PANCAKE_ROUTER = Web3.to_checksum_address("0x10ED43C718714eb63d5aA57B78B54704E256024E") WBNB = Web3.to_checksum_address("0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c") TARGET_TOKEN = Web3.to_checksum_address("0x...") # thay bằng token cần trade TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") CHAT_ID = os.getenv("CHAT_ID") trade_bot.py Code: import json import requests from config import * from time import time # Load ABI with open("abi/pancake_router.json") as f: router_abi = json.load(f) router = WEB3.eth.contract(address=PANCAKE_ROUTER, abi=router_abi) def notify_telegram(message): url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" requests.post(url, data={"chat_id": CHAT_ID, "text": message}) def buy_token(bnb_amount): amount_in_wei = WEB3.to_wei(bnb_amount, 'ether') path = [WBNB, TARGET_TOKEN] deadline = int(time()) + 60 txn = router.functions.swapExactETHForTokensSupportingFeeOnTransferTokens( 0, path, WALLET_ADDRESS, deadline ).build_transaction({ 'from': WALLET_ADDRESS, 'value': amount_in_wei, 'gas': 300000, 'gasPrice': WEB3.to_wei('5', 'gwei'), 'nonce': WEB3.eth.get_transaction_count(WALLET_ADDRESS) }) signed = WEB3.eth.account.sign_transaction(txn, PRIVATE_KEY) tx_hash = WEB3.eth.send_raw_transaction(signed.rawTransaction) tx_url = f"https://bscscan.com/tx/{WEB3.to_hex(tx_hash)}" print("Đã gửi lệnh mua:", tx_url) notify_telegram(f" Đã mua {bnb_amount} BNB TOKEN\n {tx_url}") # Ví dụ mua 0.01 BNB if __name__ == "__main__": if WEB3.isConnected(): buy_token(0.01) else: print("❌ Không thể kết nối BSC RPC.") ABI Files Lấy ABI của PancakeRouter tại: Link → tab "Contract" → copy JSON "Contract ABI" → lưu thành pancake_router. Json ERC20 ABI dùng để đọc số dư, lấy từ: Link → lưu thành erc20. Json nếu bạn cần quản lý token sau khi mua.
Code: from datetime import datetime def daily_report(): balances = get_balances() prices = get_prices() total_value = 0 lines = [f"\n WALL-E Daily Report - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"] for token, amount in balances.items(): price = prices.get(f"{token}USDT", 1) value = amount * price if token != "USDT" else amount total_value += value lines.append(f"{token}: {amount:4f} (~{value:2f} USDT)") lines.append(f"Tổng giá trị: {total_value:2f} USDT\n") with open("daily_report.txt", "a", encoding="utf-8") as f: f.write("\n".join(lines)) last_report_day = datetime.now().day while True: now = datetime.now() if now.day != last_report_day: daily_report() last_report_day = now.day # Tự động kiểm tra token mới được mua thủ công balances = get_balances() prices = get_prices() for token, amount in balances.items(): if token == "USDT": continue pair = f"{token}/USDT" if pair in order_state: continue symbol = f"{token}USDT" if symbol not in prices: continue price = prices[symbol] value = amount * price if value < 100: continue if pair not in config: config[pair] = { "start_price": round(price, 6), "split_parts": 10, "price_up_percent": 10, "price_down_percent": 10, "take_profit_percent": 30 } settings = config[pair] each_part = float(amount) / settings["split_parts"] base_price = round(settings["start_price"], 6) sell_id = place_order(pair, "SELL", each_part, base_price) buy_price = round(base_price * (1 - settings["price_down_percent"] / 100), 6) buy_id = place_order(pair, "BUY", each_part, buy_price) order_state[pair] = { "each_part": each_part, "sell_order" :(base_price, sell_id), "buy_order" :(buy_price, buy_id), "settings": settings, "base_price": base_price } for pair, state in order_state.items(): symbol = pair.replace("/", "") token, _ = pair.split("/") price_up = state["settings"]["price_up_percent"] / 100 price_down = state["settings"]["price_down_percent"] / 100 take_profit = state["settings"].get("take_profit_percent", 0) / 100 each_part = state["each_part"] prices = get_prices() current_price = prices.get(symbol, 0) base_price = state.get("base_price", current_price) if take_profit > 0 and current_price >= base_price * (1 + take_profit): full_quantity = get_balances().get(token, 0) place_order(pair, "SELL", full_quantity, current_price) report_wallet_status(token) continue last_sell_price, sell_id = state["sell_order"] if sell_id and check_order_status(symbol, sell_id): # đặt ngay lệnh tiếp theo new_sell_price = round(last_sell_price * (1 + price_up), 6) sell_id = place_order(pair, "SELL", each_part, new_sell_price) buy_price = round(last_sell_price * (1 - price_down), 6) buy_id = place_order(pair, "BUY", each_part, buy_price) state["sell_order"] = (new_sell_price, sell_id) state["buy_order"] = (buy_price, buy_id) report_wallet_status(token) last_buy_price, buy_id = state.get("buy_order", (None, None)) if buy_id and check_order_status(symbol, buy_id): sell_price = round(last_buy_price * (1 + price_up), 6) sell_id = place_order(pair, "SELL", each_part, sell_price) buy_price = round(last_buy_price * (1 - price_down), 6) buy_id = place_order(pair, "BUY", each_part, buy_price) state["sell_order"] = (sell_price, sell_id) state["buy_order"] = (buy_price, buy_id) report_wallet_status(token) time.sleep(30)
