关于PancakeRouter: EXPIRED问题

主要代码是下面,但是执行后出现这个问题,不知道是哪里的问题,希望得到解答,谢谢

            bnb = trader.get_bnb_balance(address)
            if bnb > 0 and trader.can_buy(bnb, wallet=address):
                result = trader.buy(address, key, bnb)
import time
from typing import Optional

from loguru import logger
from web3 import Web3
from web3.contract import ContractFunction, Contract
from web3.types import Wei, TxParams

class Trader:
    max_approval_hex = f"0x{64 * 'f'}"
    max_approval_int = int(max_approval_hex, 16)
    max_approval_check_hex = f"0x{15 * '0'}{49 * 'f'}"
    max_approval_check_int = int(max_approval_check_hex, 16)

    def __init__(self, web3: Web3, router_address, router_abi, token_contract: Contract, token_address):
        self.web3 = web3
        self.router_address = router_address
        self.router_contract = self.web3.eth.contract(address=router_address, abi=router_abi)
        self.token_contract = token_contract
        self.token_address = token_address
        self.symbol = self.token_contract.functions.symbol().call()
        self.decimals = 10 ** self.token_contract.functions.decimals().call()
        self.wbnb_address = self.router_contract.functions.WETH().call()

    def approve(self, wallet, private_key) -> None:
        """Give an router max approval of a token."""
        approve_function = self.token_contract.functions.approve(self.router_address, self.max_approval_int)
        logger.info(f"Approving {self.symbol}...")
        tx = self._build_and_send_tx(approve_function, wallet, private_key)
        receipt = self.web3.eth.waitForTransactionReceipt(tx, timeout=6000)
        logger.info(f'Approved: {receipt}')
        # Add extra sleep to let tx propagate correctly
        time.sleep(1)

    def _is_approved(self, wallet) -> bool:
        """Check to see if the exchange and token is approved."""
        amount = self.token_contract.functions.allowance(wallet, self.router_address).call()
        if amount >= self.max_approval_check_int:
            return True
        return False

    def get_bnb_balance(self, wallet, in_ether: bool = False):
        """Get the balance of BNB in a wallet."""
        balance = self.web3.eth.getBalance(wallet)
        return Web3.fromWei(balance, 'ether') if in_ether else balance

    def get_token_balance(self, wallet, formatted: bool = False) -> int:
        """Get the balance of a token in a wallet."""
        balance: int = self.token_contract.functions.balanceOf(wallet).call()
        return balance / self.decimals if formatted else balance

    @staticmethod
    def _deadline() -> int:
        """Get a predefined deadline. 10min by default (same as the Uniswap SDK)."""
        return int(time.time()) + 15 * 60

    def can_buy(self, bnb, wallet=None, tx_fee=None) -> bool:
        if not tx_fee and wallet:
            buy_function = self._swap_eth_for_tokens(wallet)
            gas_limit = self.estimate_gas(buy_function, wallet, bnb)
            gas_price = self.web3.eth.gas_price
            tx_fee = self._calc_tx_fee(gas_limit, gas_price)
        return bnb - (tx_fee * 6) > 0

    def can_sell(self, wallet, amount):
        sell_function = self._swap_tokens_for_eth(wallet, amount)
        gas_limit = self.estimate_gas(sell_function, wallet)
        gas_price = self.web3.eth.gas_price
        tx_fee = self._calc_tx_fee(gas_limit, gas_price)
        bnb_balance = self.get_bnb_balance(wallet)
        print(tx_fee)
        return bnb_balance - tx_fee > 0

    @staticmethod
    def estimate_gas(function: ContractFunction, address_from, value: Wei = Wei(0)) -> Wei:
        return Wei(function.estimateGas({'from': address_from, 'value': value}) + 20000)

    def _get_tx_params(self, function: ContractFunction, address_from: str, value: Wei = Wei(0)) -> TxParams:
        """Get generic transaction parameters."""
        gas_limit = self.estimate_gas(function, address_from, value)
        gas_price = self.web3.eth.gas_price
        if value > 0:
            tx_fee = self._calc_tx_fee(gas_limit, gas_price)
            if not self.can_buy(value, tx_fee=tx_fee):
                raise Exception('BNB balance is insufficient for [BUY]')
            value -= tx_fee * 6
        return {
            'from': address_from,
            'value': value,
            "gas": gas_limit,
            'gasPrice': gas_price,
            "nonce": self.web3.eth.getTransactionCount(address_from)
        }

    @staticmethod
    def _calc_tx_fee(gas_limit: Wei, gas_price: Wei):
        return gas_limit * gas_price

    @staticmethod
    def wei_to_eth(wei):
        return Web3.fromWei(wei, 'ether')

    def _build_and_send_tx(self, function: ContractFunction, address_from, private_key,
                           tx_params: Optional[TxParams] = None):
        """Build and send a transaction."""
        if not tx_params:
            tx_params = self._get_tx_params(function, address_from)
        transaction = function.buildTransaction(tx_params)
        signed_txn = self.web3.eth.account.sign_transaction(
            transaction, private_key=private_key
        )
        return self.web3.eth.sendRawTransaction(signed_txn.rawTransaction)

    def _swap_eth_for_tokens(self, wallet):
        return self.router_contract.functions.swapExactETHForTokens(
            0,
            [self.wbnb_address, self.token_address],
            wallet,
            self._deadline()
        )

    def buy(self, wallet, private_key, bnb_amount) -> dict:
        try:
            buy_function = self._swap_eth_for_tokens(wallet)
            balance_before = self.get_token_balance(wallet, formatted=True)
            before = time.time()
            tx_params = self._get_tx_params(buy_function, wallet, bnb_amount)
            bnb_used = self.wei_to_eth(tx_params["value"])
            logger.info(f'[BUY] using {wallet} {self.symbol} token for {bnb_used} BNB')
            tx_hash = self._build_and_send_tx(buy_function, wallet, private_key, tx_params)
            tx_hash_hex = str(self.web3.toHex(tx_hash))
            receipt = self.web3.eth.wait_for_transaction_receipt(tx_hash)

            after = time.time()
            after_balance = self.get_token_balance(wallet, formatted=True)
            after_bnb_balance = self.get_bnb_balance(wallet, in_ether=True)
            if receipt.status == 1:
                logger.info(
                    f"[BUY] Successfully bought {after_balance - balance_before} {self.symbol} for {bnb_used} BNB - TX HASH: {tx_hash_hex}")
                logger.info(f'Time spend: {after - before} secs')
                tx_fee = self.wei_to_eth(bnb_amount) - (after_bnb_balance + bnb_used)
                logger.info(f'Tx fee: {tx_fee}')
                logger.info(f'Current BNB balance: {after_bnb_balance}')
                logger.info(f'Current {self.symbol} balance: {after_balance}\n')
            else:
                logger.error(f'Transaction failed: {receipt}')
            return {'tx': tx_hash_hex, 'status': receipt.status, 'bnb': bnb_used,
                    'amount': after_balance - balance_before}
        except Exception as e:
            logger.exception(f"[ERROR] while [BUY]: {e}")

    def _swap_tokens_for_eth(self, wallet, amount):
        return self.router_contract.functions.swapExactTokensForETHSupportingFeeOnTransferTokens(
            amount,
            0,
            [self.token_address, self.wbnb_address],
            wallet,
            self._deadline()
        )

    def sell(self, wallet, private_key, amount):
        try:
            if amount <= 0:
                raise Exception('Invalid token amount or token balance is insufficient')
            if not self._is_approved(wallet):
                self.approve(wallet, private_key)
            logger.info(f'[SELL] using {wallet} {amount / self.decimals} {self.symbol} tokens for BNB')
            sell_function = self._swap_tokens_for_eth(wallet, amount)

            before_bnb_balance = self.get_bnb_balance(wallet, in_ether=True)
            before = time.time()
            tx_params = self._get_tx_params(sell_function, wallet)
            tx_hash = self._build_and_send_tx(sell_function, wallet, private_key, tx_params)
            tx_hash_hex = str(self.web3.toHex(tx_hash))
            receipt = self.web3.eth.wait_for_transaction_receipt(tx_hash)

            after = time.time()
            after_balance = self.get_token_balance(wallet, formatted=True)
            after_bnb_balance = self.get_bnb_balance(wallet, in_ether=True)
            if receipt.status == 1:
                logger.info(
                    f"[SELL] Successfully sold {amount / self.decimals} {self.symbol} for {after_bnb_balance - before_bnb_balance} BNB - TX HASH: {tx_hash_hex}")
                logger.info(f'Time spend: {after - before} secs')
                logger.info(f'Current BNB balance: {after_bnb_balance}')
                logger.info(f'Current {self.symbol} balance: {after_balance}\n')
            else:
                logger.error(f'Transaction failed: {receipt}')
            return {'tx': tx_hash_hex, 'status': receipt.status, 'bnb': after_bnb_balance - before_bnb_balance,
                    'amount': amount / self.decimals}

        except Exception as e:
            logger.error(f"[ERROR] while [SELL]: {e}")

image.png

请先 登录 后评论

最佳答案 2022-07-26 17:17

_deadline 时间设置大大大大一点

请先 登录 后评论

其它 0 个回答

  • 1 关注
  • 0 收藏,2646 浏览
  • hunterliy 提出于 2022-07-26 16:19