Filter for all ETH-USDC swap prices on Uniswap

Suppose the user wants to write an operator that outputs a collection of prices for each swap of a specific pair (e.g., ETH-USDC on Uniswap, considering a collection of all transaction receipt logs in all Ethereum blockchain blocks. This can be accomplished using the filter operator.

To do so, we specify the contract address of the ETH-USDC pair and filter the log collection, keeping only the logs in the output stream that contain data about the ETH-USDC swap:

const ETH_USDC_PAIR_CONTRACT_ADDRESS = "CONTRACT_ADDRESS"
export collection filtered_receipts_logs = @input("ETHEREUM_RECEIPTS_LOGS")
    .filter(log => log.address == ETH_USDC_PAIR_CONTRACT_ADDRESS)

Next, the user can create custom operators, parse_swap_logs and calc_swap_token0_price. The parse_swap_logs operator takes a stream of filtered logs and deserializes the data from log.data into a structure with fields representing the Swap event: sender, amount0In, amount1In, amount0Out, amount1Out, and to. The output stream, containing the collection of swaps, is then passed as input to the calc_swap_token0_price operator, which calculates the price of the first token in terms of the second token for each swap. Both operators utilize the map operator to apply the computation to all instances in the collection:

operator parse_swap_logs<T> {
input logs: T
output logs
.map(closure_parse_swap_logs)
}

operator calc_swap_token0_price<T> {
input swaps: T
output swaps
.map(swap => swap.amount1Out / swap.amount0In)
}

export collection token0_prices = filtered_receipts_logs
	.parse_swap_logs()
.calc_swap_token0_price()