import time from web3 import Web3 from web3.exceptions import InvalidAddress, TimeExhausted import logging # For RSI example (commented out) # import numpy as np # === CONFIGURATION === # Blockchain and Wallet Configuration BSC_RPC_URL = "https://bsc-dataseed.bnbchain.org" WALLET_ADDRESS = "" # Replace with your wallet address (e.g., 0xabc123...) PRIVATE_KEY = "" # Replace with your private key (keep secure!) # Token and Pair Addresses IMMORTALITY_ADDR = "0x2bf2141ed175f3236903cf07de33d7324871802d" # IMT token address WBNB_ADDR = "0xbb4cdb9cbd36b01bd1cbaebf2de08d9173bc095c" # WBNB token address PANCAKESWAP_ROUTER_ADDR = "0x10ED43C718714eb63d5aA57B78B54704E256024E" # PancakeSwap router PAIR_ADDRESS = "0xfA56E9AbcaA45207bE5E43cF475Ee061768CA915" # IMT/WBNB pair address # Trading Parameters LOOP_INTERVAL_SECONDS = 60 # Main loop runs every 60 seconds (1 minute) BUY_BNB_AMOUNT = 0.005 # Amount of BNB to spend per buy SELL_IMT_QUANTITY = 10_000_000 # Amount of IMT tokens to sell RPC_SYNC_DELAY_SECONDS = 7 # Delay for RPC synchronization after transactions # ABIs (unchanged from original) PAIR_ABI = [ { "constant": True, "inputs": [], "name": "getReserves", "outputs": [ {"internalType": "uint112", "name": "_reserve0", "type": "uint112"}, {"internalType": "uint112", "name": "_reserve1", "type": "uint112"}, {"internalType": "uint32", "name": "_blockTimestampLast", "type": "uint32"}, ], "stateMutability": "view", "type": "function", } ] ROUTER_ABI = [ { "name": "getAmountsOut", "type": "function", "stateMutability": "view", "inputs": [ {"name": "amountIn", "type": "uint256"}, {"name": "path", "type": "address[]"}, ], "outputs": [{"name": "", "type": "uint256[]"}], }, { "name": "swapExactETHForTokensSupportingFeeOnTransferTokens", "type": "function", "stateMutability": "payable", "inputs": [ {"name": "amountOutMin", "type": "uint256"}, {"name": "path", "type": "address[]"}, {"name": "to", "type": "address"}, {"name": "deadline", "type": "uint256"}, ], "outputs": [], }, { "name": "swapExactTokensForETHSupportingFeeOnTransferTokens", "type": "function", "stateMutability": "nonpayable", "inputs": [ {"name": "amountIn", "type": "uint256"}, {"name": "amountOutMin", "type": "uint256"}, {"name": "path", "type": "address[]"}, {"name": "to", "type": "address"}, {"name": "deadline", "type": "uint256"}, ], "outputs": [], }, ] TOKEN_ABI = [ { "name": "balanceOf", "type": "function", "stateMutability": "view", "inputs": [{"name": "owner", "type": "address"}], "outputs": [{"name": "", "type": "uint256"}], }, { "name": "approve", "type": "function", "stateMutability": "nonpayable", "inputs": [ {"name": "spender", "type": "address"}, {"name": "amount", "type": "uint256"}, ], "outputs": [{"name": "", "type": "bool"}], }, { "name": "allowance", "type": "function", "stateMutability": "view", "inputs": [ {"name": "owner", "type": "address"}, {"name": "spender", "type": "address"}, ], "outputs": [{"name": "", "type": "uint256"}], }, { "name": "decimals", "type": "function", "stateMutability": "view", "inputs": [], "outputs": [{"name": "", "type": "uint8"}], }, ] # === GLOBAL VARIABLES === IMT_DECIMALS = None w3 = None pair_contract = None router_contract = None token_contract = None logger = logging.getLogger(__name__) # For RSI example (commented out) # price_history = [] # List to store historical prices for RSI calculation # === INITIALIZATION === def init(): """Initialize Web3 connection and contract instances.""" global WALLET_ADDRESS, IMMORTALITY_ADDR, WBNB_ADDR, PANCAKESWAP_ROUTER_ADDR, PAIR_ADDRESS, IMT_DECIMALS, w3, pair_contract, router_contract, token_contract logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Connect to BSC w3 = Web3(Web3.HTTPProvider(BSC_RPC_URL)) if not w3.is_connected(): logger.error("Cannot connect to BSC RPC") raise ConnectionError("Cannot connect to BSC RPC") # Validate and checksum addresses try: WALLET_ADDRESS = w3.to_checksum_address(WALLET_ADDRESS) IMMORTALITY_ADDR = w3.to_checksum_address(IMMORTALITY_ADDR) WBNB_ADDR = w3.to_checksum_address(WBNB_ADDR) PANCAKESWAP_ROUTER_ADDR = w3.to_checksum_address(PANCAKESWAP_ROUTER_ADDR) PAIR_ADDRESS = w3.to_checksum_address(PAIR_ADDRESS) except InvalidAddress as e: logger.error(f"Invalid address provided: {e}") raise # Initialize contracts token_contract = w3.eth.contract(address=IMMORTALITY_ADDR, abi=TOKEN_ABI) pair_contract = w3.eth.contract(address=PAIR_ADDRESS, abi=PAIR_ABI) router_contract = w3.eth.contract(address=PANCAKESWAP_ROUTER_ADDR, abi=ROUTER_ABI) # Fetch token decimals try: IMT_DECIMALS = token_contract.functions.decimals().call() logger.info(f"IMT token decimals: {IMT_DECIMALS}") except Exception as e: logger.warning(f"Could not fetch IMT decimals, defaulting to 8. Error: {e}") IMT_DECIMALS = 8 # === TRANSACTION HELPER === def send_tx(fn, value=0): """Send a transaction and wait for confirmation.""" tx_params = { 'from': WALLET_ADDRESS, 'gas': 300000, 'gasPrice': max(int(w3.eth.gas_price * 1.1), w3.to_wei('5', 'gwei')), 'nonce': w3.eth.get_transaction_count(WALLET_ADDRESS), } if value: tx_params['value'] = value try: transaction = fn.build_transaction(tx_params) signed_txn = w3.eth.account.sign_transaction(transaction, private_key=PRIVATE_KEY) tx_hash = w3.eth.send_raw_transaction(signed_txn.raw_transaction) logger.info(f"Transaction sent: {tx_hash.hex()}...") receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=600) if receipt.status == 0: raise Exception(f"Transaction {tx_hash.hex()} failed: {receipt}") logger.info(f"Confirmed in block {receipt.blockNumber}") return tx_hash.hex() except Exception as e: logger.error(f"Transaction error: {e}") return None # === PRICE FETCHING === def get_price(): """Fetch the current IMT price in BNB.""" amount_in_bnb_wei = w3.to_wei(1, 'ether') try: amounts = router_contract.functions.getAmountsOut(amount_in_bnb_wei, [WBNB_ADDR, IMMORTALITY_ADDR]).call() imt_tokens = amounts[1] / (10 ** IMT_DECIMALS) price = 1 / imt_tokens if imt_tokens else float('inf') logger.info(f"Current price: {price:.18f} BNB/IMT") return price except Exception as e: logger.error(f"Price fetch error: {e}") return None # === TRADING LOGIC === def buy_signal(price, balance_bnb, balance_imt): """ Custom buy logic. Modify this function to implement your trading strategy. Parameters: - price (float): Current price of IMT in BNB - balance_bnb (float): Wallet BNB balance in BNB - balance_imt (float): Wallet IMT balance in IMT tokens Returns: - bool: Return True to trigger a buy, False otherwise """ # Original example logic: Buy if price is below a threshold and sufficient BNB balance price_threshold = 0.0000000001 # Example threshold (adjust as needed) if price and price < price_threshold and balance_bnb >= BUY_BNB_AMOUNT: logger.info(f"Buy signal triggered: Price {price:.18f} < {price_threshold}, BNB balance: {balance_bnb}") return True # Alternative example logic: Buy if RSI is below 30 (oversold) """ global price_history if price: price_history.append(price) if len(price_history) > 14: # Need at least 14 periods for RSI prices = np.array(price_history[-14:]) # Last 14 prices deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = np.mean(gains) avg_loss = np.mean(losses) if np.any(losses) else 0.0000000001 # Avoid division by zero rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) logger.info(f"RSI: {rsi:.2f}") if rsi < 30 and balance_bnb >= BUY_BNB_AMOUNT: logger.info(f"Buy signal triggered: RSI {rsi:.2f} < 30, BNB balance: {balance_bnb}") return True else: logger.info(f"Collecting price history: {len(price_history)}/14") """ return False def sell_signal(price, balance_bnb, balance_imt): """ Custom sell logic. Modify this function to implement your trading strategy. Parameters: - price (float): Current price of IMT in BNB - balance_bnb (float): Wallet BNB balance in BNB - balance_imt (float): Wallet IMT balance in IMT tokens Returns: - bool: Return True to trigger a sell, False otherwise """ # Original example logic: Sell if price is above a threshold and sufficient IMT balance price_threshold = 0.0000000002 # Example threshold (adjust as needed) if price and price > price_threshold and balance_imt >= SELL_IMT_QUANTITY: logger.info(f"Sell signal triggered: Price {price:.18f} > {price_threshold}, IMT balance: {balance_imt}") return True # Alternative example logic: Sell if RSI is above 70 (overbought) """ global price_history if price: price_history.append(price) if len(price_history) > 14: # Need at least 14 periods for RSI prices = np.array(price_history[-14:]) # Last 14 prices deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = np.mean(gains) avg_loss = np.mean(losses) if np.any(losses) else 0.0000000001 # Avoid division by zero rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) logger.info(f"RSI: {rsi:.2f}") if rsi > 70 and balance_imt >= SELL_IMT_QUANTITY: logger.info(f"Sell signal triggered: RSI {rsi:.2f} > 70, IMT balance: {balance_imt}") return True else: logger.info(f"Collecting price history: {len(price_history)}/14") """ return False # === TRADING EXECUTION === def buy_with_bnb(bnb_amount): """Execute a buy transaction for IMT using BNB.""" amount_in_wei = w3.to_wei(bnb_amount, 'ether') logger.info(f"Buying IMT with {bnb_amount} BNB...") try: amounts_out = router_contract.functions.getAmountsOut(amount_in_wei, [WBNB_ADDR, IMMORTALITY_ADDR]).call() expected_tokens = amounts_out[-1] min_tokens_out = int(expected_tokens * 0.9 * 0.95) # 90% for fees, 95% for slippage logger.info(f"Expected IMT tokens: {expected_tokens / (10**IMT_DECIMALS)}, Min: {min_tokens_out / (10**IMT_DECIMALS)}") swap_fn = router_contract.functions.swapExactETHForTokensSupportingFeeOnTransferTokens( min_tokens_out, [WBNB_ADDR, IMMORTALITY_ADDR], WALLET_ADDRESS, int(time.time()) + 120 ) return send_tx(swap_fn, value=amount_in_wei) except Exception as e: logger.error(f"Buy error: {e}") return None def sell_exact_imt(imt_quantity): """Execute a sell transaction for IMT tokens.""" imt_units = int(imt_quantity * (10 ** IMT_DECIMALS)) balance = token_contract.functions.balanceOf(WALLET_ADDRESS).call() logger.info(f"Selling {imt_quantity} IMT. Balance: {balance / (10 ** IMT_DECIMALS)} IMT") if balance < imt_units: logger.error("Insufficient IMT balance") return None # Check and approve allowance allowance = token_contract.functions.allowance(WALLET_ADDRESS, PANCAKESWAP_ROUTER_ADDR).call() if allowance < imt_units: logger.info(f"Approving {imt_quantity} IMT for router...") approve_fn = token_contract.functions.approve(PANCAKESWAP_ROUTER_ADDR, imt_units) try: send_tx(approve_fn) time.sleep(RPC_SYNC_DELAY_SECONDS) new_allowance = token_contract.functions.allowance(WALLET_ADDRESS, PANCAKESWAP_ROUTER_ADDR).call() if new_allowance < imt_units: logger.error("Allowance update failed") return None except Exception as e: logger.error(f"Approval error: {e}") return None # Perform swap try: effective_imt_units = int(imt_units * 0.9) amounts_out = router_contract.functions.getAmountsOut(effective_imt_units, [IMMORTALITY_ADDR, WBNB_ADDR]).call() expected_bnb = amounts_out[-1] min_bnb_out = int(expected_bnb * 0.95) # 5% slippage logger.info(f"Effective IMT units: {effective_imt_units / (10**IMT_DECIMALS)}, Expected BNB: {w3.from_wei(expected_bnb, 'ether')}, Min: {w3.from_wei(min_bnb_out, 'ether')}") swap_fn = router_contract.functions.swapExactTokensForETHSupportingFeeOnTransferTokens( imt_units, min_bnb_out, [IMMORTALITY_ADDR, WBNB_ADDR], WALLET_ADDRESS, int(time.time()) + 120 ) return send_tx(swap_fn) except Exception as e: logger.error(f"Sell error: {e}") return None # === MAIN LOOP === def main(): """Main trading loop that runs every LOOP_INTERVAL_SECONDS.""" init() logger.info(f"Starting trading bot - Wallet: {WALLET_ADDRESS}") while True: try: # Fetch current price and balances price = get_price() balance_bnb_wei = w3.eth.get_balance(WALLET_ADDRESS) balance_bnb = w3.from_wei(balance_bnb_wei, 'ether') balance_imt = token_contract.functions.balanceOf(WALLET_ADDRESS).call() / (10 ** IMT_DECIMALS) # Check buy signal if buy_signal(price, balance_bnb, balance_imt): tx_hash = buy_with_bnb(BUY_BNB_AMOUNT) if tx_hash: logger.info(f"Buy transaction successful: {tx_hash}") else: logger.warning("Buy transaction failed") # Check sell signal if sell_signal(price, balance_bnb, balance_imt): tx_hash = sell_exact_imt(SELL_IMT_QUANTITY) if tx_hash: logger.info(f"Sell transaction successful: {tx_hash}") else: logger.warning("Sell transaction failed") except Exception as e: logger.error(f"Error in main loop: {e}") # Wait for the next loop iteration time.sleep(LOOP_INTERVAL_SECONDS) if __name__ == "__main__": main()