Snowsight — Getting Started
The Avalanche Mempool Service
UPDATE
The Snowsight beta program has ended, so the authentication portion of this guide is out of date. See THIS post for post-launch changes.
…
Hopefully you’ve checked out the new Snowsight mempool service being built by Chainsight.
Some users on the Degen Code Discord told me that they didn’t get any Twitter DMs after following the instructions, so I’m posting the Telegram info below. It’s public knowledge anyway, so I doubt they will take issue with me sharing this here.
This post will bring a lot of things together and it will be quite technical, so set aside some time to work through everything.
Snowsight
The Snowsight service provides three very exciting features for MEV searchers:
They run a network of Avalanche validators, which are nodes with an associated AVAX stake. The validators on Avalanche are the only ones allowed to see the mempool, so they have a huge advantage when it comes to securing arbitrage opportunities. Only searchers with access to the mempool can secure same-block arbitrage. Searchers with read-only node access can only secure post-block arbitrage.
The websocket connection provides pre-decoded transactions. This is different from the Fantom mempool results we found in the previous post. Instead of simply publishing a transaction hash, Snowsight publishes a completely decoded transaction that you can analyze locally without having to re-query the RPC.
They have recently introduced a validator relay, which allows us to send transactions directly to their validator pools. This gives us an extreme advantage because it reduces the latency involved with transaction propagation (the time between submitting a transaction and it reaching a validator for mining and synchronization with the rest of the blockchain).
The service is for coders, by coders, and they don’t hold your hand. They provide a stream of data over a websocket and a relay to submit your transactions, that’s it!
What transactions you choose to submit and what MEV you choose to capture is entirely up to you.
Lesson Outline
Snowsight is currently in beta, is not user-friendly, and requires some deep technical knowledge to run. There’s almost zero documentation, but I’ve been poking at this thing for a few days and I’m ready to take you from zero to a working bot that watches the Snowsight websocket, filters out interesting transactions, keeps track of your current subscription credits and periodically submits top-up payments to keep the subscription active 24/7.
Along the way you’ll also learn a few things about mixing synchronous and asynchronous code in Python.
Let’s go!
Prerequisites
I have converted this post to a standalone free lesson to get more exposure to Snowsight. I think it’s a very useful service and democratizing access to Avalanche MEV is a noble goal.
If you’re not a regular subscriber here, you’ll need to set up a few things:
Install Python (ideally version 3.9 for best compatibility with packages)
Install Brownie, a smart contract development tool written in Python
Install Websockets version 10.3 (or newer)
Create a dedicated account within Brownie (with associated seed phrase and address) and fund it with some AVAX
Regular readers here know how to do these things already, so subscribe if you’re not feeling confident in the steps above and read through the archives to learn how to do all of this.
Authentication
The first thing we need to do is review how Snowsight manages authentication. The websocket expects a message to be sent in JSON format with a key-value pair named 'signed_key' and the value equal to a hexadecimal signature of a specific message, signed by your private key.
This allows Snowsight to confirm that your external address has recently called a function of its smart contract (Snowtrace link) called pay() with an appropriate amount of AVAX to pay for your subscription. The current price is set at 1 Wei per block, which is basically free. I expect this value will increase in the future, so enjoy it while it’s cheap!
We have access to an account-specific method in Brownie called sign_defunct_message(), which we use to generate our hex signature.
>>> import brownie
>>> degenbot = brownie.accounts.load('degenbot')
>>> signed_msg = degenbot.sign_defunct_message("Sign this message to authenticate your wallet with Snowsight.")
>>> signed_msg.signature.hex()
'0x60de661f0c424d4920125af9a35c22ee72b96c501f37113da0743a24ceefd81878334acb4bb19833ce8c5672d2ce86afa9a37b21ea488cd8a14b1cb8364720cf1b'This produces an EIP-191 compatible signed message, which Snowsight can use to ensure that a given message was signed with the private key for a known public address.
Project: CRA-WAVAX Watcher
We will build a bot with a simple goal: monitor the Avalanche mempool for swaps between WAVAX and CRA from users using the TraderJoe router contract (likely the web front-end). These tokens are fairly active so you’ll get a change to see a new transaction several times a minute.
We will pass the input data for each Snowsight transaction through the web3 library, decode it using the router contract’s ABI to get the raw inputs, and display them.
Pending Transaction Watcher
Now let’s write an async function to connect to the websocket, send our authentication message, and start receiving messages.
async def watch_pending_transactions():
signed_message = degenbot.sign_defunct_message(
"Sign this message to authenticate your wallet with Snowsight."
)
async for websocket in websockets.connect(
uri="ws://avax.chainsight.dev:8589",
ping_timeout=None,
):
try:
await websocket.send(
json.dumps({"signed_key": signed_message.signature.hex()})
)
resp = json.loads(await websocket.recv())
print(resp)
# if unauthenticted, pay the contract
# and restart the loop
if "unauthenticated" in resp["status"]:
pay_sync()
continue
if resp["status"] == "authenticated":
while True:
tx_message = json.loads(
await asyncio.wait_for(
websocket.recv(),
timeout=30,
)
)
tx_to = tx_message["to"].lower()
# ignore the TX if it's not
# on our watchlist
if tx_to not in [address.lower() for address in ROUTERS.keys()]:
continue
else:
func, params = brownie.web3.eth.contract(
address=brownie.web3.toChecksumAddress(tx_to),
abi=ROUTERS[tx_to]["abi"],
).decode_function_input(tx_message["input"])
# Print all TX with a 'path' argument,
# including the function name and
# inputs
if params.get("path") in [
[cra.address, wavax.address],
[wavax.address, cra.address],
]:
print()
print("*** Pending CRA-WAVAX swap! ***")
print(func.fn_name)
print(params)
except (websockets.WebSocketException, asyncio.exceptions.TimeoutError) as e:
print(e)
print("reconnecting...")
except Exception as e:
print(f"Exception in watch_pending_transactions: {e}")
continueThere are some references to addresses, token objects, and some functions that don’t exist (yet). So please review it and see if you can figure out what it does before proceeding.
Pay Function
We need a function that will pay the Snowsight contract, which has several values that are adjusted by the team. There is a minimum block payment, a maximum block payment, and a payment per block. We are most gas-efficient when we buy as many blocks as possible, so before submitting the payment we calculate how much credit we can buy by comparing the credit we already have to the maximum payment.
It will attempt to run until the websocket disconnects us with an “unauthenticated” response, then we will pay it again. It’s a simple example and can be improved later.
Since we’re working with a lot of async code, I mark each synchronous function with a _sync suffix. This is a nice reminder that all code inside it will block the asyncio event loop. This means that concurrent functions (like the pending transaction monitor function) will not run until this function completes. For something like a payment, this is OK since it rarely runs and likely needs to complete before the websocket will give us more data.
In the future when we have multiple arbitrage calculations going on, multiple transactions being sent, it will be important to manage these synchronous functions inside of a separate thread or process pool to keep them from blocking our time-sensitive tasks.
def pay_sync():
"""
A synchronous function that retrieves data from the Snowsight contract, calculates the maximum payment,
and submits a payment to the Chainsight contract.
"""
snowsight_contract = brownie.Contract(SNOWSIGHT_CONTRACT_ADDRESS)
block_payment = snowsight_contract.paymentPerBlock() * (
brownie.chain.height
+ snowsight_contract.maximumPaymentBlocks()
- snowsight_contract.payments(degenbot.address)[-1]
)
snowsight_contract.pay(
{
"from": degenbot,
"value": min(
block_payment,
snowsight_contract.calculateMaxPayment(),
),
"priority_fee": 0,
}
)Setup and Event Loop
Before we do any of this, we need to define our contract addresses, build a dictionary of relevant router information, tell Brownie how to connect to our network, and build our helper objects.
These statements are declared outside of all functions, which means the variables and objects are in the global scope. This means all functions have equal access to them without having to pass them around as arguments.
This is not the best design practice, but it is simple and low-distraction. Please don’t rely on global variables to make multi-function programs work correctly.
BROWNIE_NETWORK = "moralis-avax-main-websocket"
BROWNIE_ACCOUNT = "degenbot"
# Contract addresses
SNOWSIGHT_CONTRACT_ADDRESS = "0xD9B1ee4AE46d4fe51Eeaf644107f53A37F93352f"
CRA_CONTRACT_ADDRESS = "0xA32608e873F9DdEF944B24798db69d80Bbb4d1ed"
WAVAX_CONTRACT_ADDRESS = "0xB31f66AA3C1e785363F0875A1B74E27b85FD66c7"
TRADERJOE_LP_CRA_WAVAX_ADDRESS = "0x140cac5f0e05cbec857e65353839fddd0d8482c1"
# Change this to your Snowtrace API key!
SNOWTRACE_API_KEY = "[redacted]"
ROUTERS = {
"0x60aE616a2155Ee3d9A68541Ba4544862310933d4".lower(): {
"name": "TraderJoe",
"abi": [],
},
}
os.environ["SNOWTRACE_TOKEN"] = SNOWTRACE_API_KEY
try:
network.connect(BROWNIE_NETWORK)
except:
sys.exit(
"Could not connect! Verify your Brownie network settings using 'brownie networks list'"
)
try:
degenbot = accounts.load(BROWNIE_ACCOUNT)
except:
sys.exit(
"Could not load account! Verify your Brownie account settings using 'brownie accounts list'"
)
print("\nContracts loaded:")
cra = Erc20Token(address=CRA_CONTRACT_ADDRESS)
wavax = Erc20Token(address=WAVAX_CONTRACT_ADDRESS)
for address in ROUTERS.keys():
ROUTERS[address]["abi"] = brownie.Contract.from_explorer(address).abi
traderjoe_lp_cra_wavax = LiquidityPool(
address=TRADERJOE_LP_CRA_WAVAX_ADDRESS,
name="TraderJoe",
tokens=[cra, wavax],
)
lps = [
traderjoe_lp_cra_wavax,
]
asyncio.run(main())Finally notice the use of asyncio.run(), which will take the async function main() and run it directly. This is necessary because asyncio can only execute functions in parallel when they are declared with the async keyword. An async function does not behave like a regular function, so you cannot call it directly. It must be run directly using asyncio.run() or managed by the event loop with an await keyword.
Async Event Loop
Now we add our pending transaction watcher to the event loop using asyncio.create_task(), then capture the results using asyncio.gather(). The advantage of this structure is that you can add other async functions inside gather() later, and run everything in parallel.
async def main():
await asyncio.gather(
asyncio.create_task(watch_pending_transactions()),
)Complete Source Code
snowsight_example.py
import asyncio
import json
import websockets
import os
import sys
import time
import requests
from brownie import accounts, network, Contract
from degenbot import *
from pprint import pprint
def pay_sync():
"""
A synchronous function that retrieves data from the Snowsight contract, calculates the maximum payment,
and submits a payment to the Chainsight contract.
"""
snowsight_contract = brownie.Contract.from_explorer(SNOWSIGHT_CONTRACT_ADDRESS)
block_payment = snowsight_contract.paymentPerBlock() * (
brownie.chain.height
+ snowsight_contract.maximumPaymentBlocks()
- snowsight_contract.payments(degenbot.address)[-1]
)
snowsight_contract.pay(
{
"from": degenbot,
"value": min(
block_payment,
snowsight_contract.calculateMaxPayment(),
),
"priority_fee": 0,
}
)
async def watch_pending_transactions():
signed_message = degenbot.sign_defunct_message(
"Sign this message to authenticate your wallet with Snowsight."
)
async for websocket in websockets.connect(
uri="ws://avax.chainsight.dev:8589",
ping_timeout=None,
):
try:
await websocket.send(
json.dumps({"signed_key": signed_message.signature.hex()})
)
resp = json.loads(await websocket.recv())
print(resp)
# if we're currently unauthenticted, pay the contract and restart the loop
if "unauthenticated" in resp["status"]:
pay_sync()
continue
if resp["status"] == "authenticated":
while True:
tx_message = json.loads(
await asyncio.wait_for(
websocket.recv(),
timeout=30,
)
)
# message keys:
# ['from', 'gas', 'gasPrice', 'maxFeePerGas', 'maxPriorityFeePerGas', 'hash', 'input', 'nonce', 'to', 'value', 'txType']
tx_to = tx_message["to"].lower()
# ignore the TX if it's not on our watchlist
if tx_to not in [address.lower() for address in ROUTERS.keys()]:
continue
else:
func, params = brownie.web3.eth.contract(
address=brownie.web3.toChecksumAddress(tx_to),
abi=ROUTERS[tx_to]["abi"],
).decode_function_input(tx_message["input"])
# Print all TX with a 'path' argument, including the function name and inputs
if params.get("path") in [
[cra.address, wavax.address],
[wavax.address, cra.address],
]:
print()
print("*** Pending CRA-WAVAX swap! ***")
print(func.fn_name)
print(params)
except (websockets.WebSocketException, asyncio.exceptions.TimeoutError) as e:
print(e)
print("reconnecting...")
except Exception as e:
print(f"Exception in watch_pending_transactions: {e}")
continue
async def main():
await asyncio.create_task(watch_pending_transactions())
# await asyncio.gather(
# asyncio.create_task(watch_pending_transactions()),
# )
BROWNIE_NETWORK = "moralis-avax-main-websocket"
BROWNIE_ACCOUNT = "degenbot"
# Contract addresses
SNOWSIGHT_CONTRACT_ADDRESS = "0xD9B1ee4AE46d4fe51Eeaf644107f53A37F93352f"
CRA_CONTRACT_ADDRESS = "0xA32608e873F9DdEF944B24798db69d80Bbb4d1ed"
WAVAX_CONTRACT_ADDRESS = "0xB31f66AA3C1e785363F0875A1B74E27b85FD66c7"
TRADERJOE_LP_CRA_WAVAX_ADDRESS = "0x140cac5f0e05cbec857e65353839fddd0d8482c1"
# Change this to your Snowtrace API key!
SNOWTRACE_API_KEY = "[redacted]"
ROUTERS = {
"0x60aE616a2155Ee3d9A68541Ba4544862310933d4".lower(): {
"name": "TraderJoe",
"abi": [],
},
}
os.environ["SNOWTRACE_TOKEN"] = SNOWTRACE_API_KEY
try:
network.connect(BROWNIE_NETWORK)
except:
sys.exit(
"Could not connect! Verify your Brownie network settings using 'brownie networks list'"
)
try:
# account object needs be accessible from other functions
degenbot = accounts.load(BROWNIE_ACCOUNT)
except:
sys.exit(
"Could not load account! Verify your Brownie account settings using 'brownie accounts list'"
)
print("\nContracts loaded:")
cra = Erc20Token(address=CRA_CONTRACT_ADDRESS)
wavax = Erc20Token(address=WAVAX_CONTRACT_ADDRESS)
for address in ROUTERS.keys():
ROUTERS[address]["abi"] = brownie.Contract.from_explorer(address).abi
traderjoe_lp_cra_wavax = LiquidityPool(
address=TRADERJOE_LP_CRA_WAVAX_ADDRESS,
name="TraderJoe",
tokens=[cra, wavax],
)
lps = [
traderjoe_lp_cra_wavax,
]
asyncio.run(main())Run and Test
After getting this all configured (adjust your networks and API keys as needed), run the program and start watching the stream of interesting data.
(.venv) devil@hades:~/bots$ python3 snowsight_example.py
Enter password for "degenbot":
Contracts loaded:
• CRA (CRA)
• WAVAX (Wrapped AVAX)
TraderJoe
• Token 0: CRA - Reserves: 4683686888683365171537888
• Token 1: WAVAX - Reserves: 22808241671816956903653
{'status': 'authenticated', 'endBlock': 14123795, 'timeLeftEstimateDays': '0.44081017'}
*** Pending CRA-WAVAX swap! ***
swapExactTokensForAVAX
{'amountIn': 90150000000000000000,
'amountOutMin': 435502292506301990,
'deadline': 1651346504908,
'path': ['0xA32608e873F9DdEF944B24798db69d80Bbb4d1ed',
'0xB31f66AA3C1e785363F0875A1B74E27b85FD66c7'],
'to': '0xB16d3fAc987dFEB80f78e5013f0b3F3fac870Cc2'}
*** Pending CRA-WAVAX swap! ***
swapExactTokensForAVAX
{'amountIn': 90150000000000000000,
'amountOutMin': 435502292506301990,
'deadline': 1651346504908,
'path': ['0xA32608e873F9DdEF944B24798db69d80Bbb4d1ed',
'0xB31f66AA3C1e785363F0875A1B74E27b85FD66c7'],
'to': '0xB16d3fAc987dFEB80f78e5013f0b3F3fac870Cc2'}
*** Pending CRA-WAVAX swap! ***
swapExactTokensForAVAX
{'amountIn': 46912500000000000000,
'amountOutMin': 226621231884186146,
'deadline': 1651346528125,
'path': ['0xA32608e873F9DdEF944B24798db69d80Bbb4d1ed',
'0xB31f66AA3C1e785363F0875A1B74E27b85FD66c7'],
'to': '0x19a9982a26143e4b6801818Ac20F3b2a81c8DB34'}
*** Pending CRA-WAVAX swap! ***
swapExactTokensForAVAX
{'amountIn': 46912500000000000000,
'amountOutMin': 226621231884186146,
'deadline': 1651346528125,
'path': ['0xA32608e873F9DdEF944B24798db69d80Bbb4d1ed',
'0xB31f66AA3C1e785363F0875A1B74E27b85FD66c7'],
'to': '0x19a9982a26143e4b6801818Ac20F3b2a81c8DB34'}
[...]Moving Forward
In the next lesson I will review how to use the Snowsight relay to submit transactions directly to their validator pool instead of submitting the old-fashioned (slow) way. This requires some more focused attention and use of the web3py library, directly building and signing raw transactions instead of using the Brownie / RPC abstraction.
Sorry for my newbie doubt. I gest the following error...
File "snowsight_example.py", line 152, in <module>
cra = Erc20Token(address=CRA_CONTRACT_ADDRESS)
NameError: name 'Erc20Token' is not defined