如何跟踪bsc网络中钱包列表的代币交易?
我考虑使用websocket和过滤器功能。我认为有可能使用topics作为过滤器参数的一部分，并仅反映Transfer事件到/从观看地址，这样我的应用程序就不必处理不必要的数据。
但是我做错了什么，不知道如何正确地将钱包列表(或至少只有一个钱包)作为我的过滤函数的参数。怎么做呢?
我有从Transfer事件获取数据的问题，因为我不知道如何解码HexBytes类型。我看到了web3.js的函数，但是没有web3.py的函数。
address_list = ['0x67fdE6D04a82689a59E5188f9B572CBeF53D4763', '...', '...']
web3 = Web3(Web3.WebsocketProvider('wss://bsc.getblock.io/mainnet/?api_key=your_api_key'))
web3_filter = web3.eth.filter({'topics': address_list})
while True:
for event in web3_filter.get_new_entries():
print(web3.toJSON(web3.eth.wait_for_transaction_receipt(event).logs))
我终于找到了解决办法。起初，我使用node.js编写相同的代码，因为web3.js使我更容易理解它的实际工作原理。它有更好的方法命名，更好的文档，等等
回到web.py:
对于获得Transfer事件签名，我使用此代码transferEventSignature = web3.toHex(Web3.sha3(text='Transfer(address,address,uint256)'))
eth_abi库
from web3 import Web3
from eth_abi import encode_abi, decode_abi
from hexbytes import HexBytes
encoded_wallet = (web3.toHex(encode_abi(['address'], [wallet])) # encoding
web3 = Web3(Web3.WebsocketProvider('wss://speedy-nodes-nyc.moralis.io/api-key/bsc/mainnet/ws'))
event_filter = web3.eth.filter({'topics': [transferEventSignature, None, encoded_wallet]}) # setting up a filter with correct parametrs
while True:
for event in event_filter.get_new_entries():
decoded_address = decode_abi(['address'], HexBytes(event.topics[2])) # decoding wallet
value = decode_abi(['uint256'], HexBytes(event.data)) # decoding event.data
tokenContractAddress = event.address
contractInstance = web3.eth.contract(address=tokenContractAddress, abi=jsonAbi) # jsonAbi is standart erc-20 token abi
# I used simplified JSON abi that is only able to read decimals, name and symbol
name = contractInstance.functions.name().call()
decimals = contractInstance.functions.decimals().call()
symbol = contractInstance.functions.symbol().call()
# getting any token information
# doing some useful stuff
GetBlock。IO为我工作，但有时会与网络不同步。我使用这个服务比较成功:https://moralis.io/
我希望有人会发现这是有用的。
下面是我跟踪BEP20令牌交易的代码:
def log_new(event_filter):
for event in event_filter.get_new_entries():
handle_event(event)
def handle_event(event):
receipt = web3.eth.waitForTransactionReceipt(event['transactionHash'])
logs = contract.events.Transfer().processReceipt(receipt)
args = logs[0]['args']
print ("EVENT", event)
hashstr = binascii.b2a_hex(event['transactionHash'])
name = contract.functions.name().call()
decimals = contract.functions.decimals().call()
symbol = contract.functions.symbol().call()
item = {
"from": args["from"],
"to": args.to,
"value": args.value,
"blockNumber":event['blockNumber'],
"transhash": "0x" + hashstr.decode("ascii"),
"timeStamp" : get_block_timestamp(event['blockNumber']),
"tokenSymbol" : symbol, #"KAMPAY"
"decimals" : decimals,
"name" : name
}
print(item)
block_filter = web3.eth.filter({'fromBlock':'latest','address':MYTOKENADDRESS})
while 1:
log_new(block_filter)
time.sleep(1)
bscscanapi不可靠。问题是，API背后的云票价DDoS保护和captcha有时被要求。不幸的是，Python代码没有办法绕过这个captcha检查。
Bscscan提供基本免费使用的api (5 req/sec)
对于事务列表(有不同类型的事务，包括normal)内部,,bep-20等等)你可以用这个
示例用法如下: