Stream DEX trades

This guide shows how to decode and stream trade data from decentralized exchanges (DEXs) into your data warehouse with a Turbo pipeline. The example writes to PostgreSQL, but any supported sink works.

What you’ll need

A Goldsky account and the CLI installed

Install Goldsky's CLI and log in

Install the Goldsky CLI: For macOS/Linux:
curl https://goldsky.com | sh
For Windows:
npm install -g @goldskycom/cli
Windows users need to have Node.js and npm installed first. Download from nodejs.org if not already installed.
Log into your Project by running:
goldsky login
This opens your browser to sign in (Google, GitHub, SSO, or email). Once you authenticate, the CLI is logged in automatically — there’s no API key to copy or paste.
On a headless or remote machine (or in CI), create an API key on your Project Settings page and pass it directly with goldsky login --token <API_KEY>. Use goldsky login --no-browser to print the login URL instead of opening a browser.
Now that you are logged in, run goldsky to get started:
goldsky

A basic understanding of Turbo pipelines
A destination sink to write your data to. In this example, we will use the PostgreSQL sink

Introduction

Most decentralized exchanges these days are based entirely on the Uniswap protocol or have strong similarities with it.

If you need a high level overview of how Uniswap works you can check out this reference page

With that in mind, we can narrow our focus on identifying events emitted by Uniswap contracts and use them to identify similar events emitted by all DEXs on the chain. There are a number of different events we could track. In this guide we will track the Swap and PoolCreated events as they are arguably two of the most important events to track when wanting to make sense of trading activity in a DEX. For this example implementation, we will choose the base.raw_logs dataset (EVM sources) as the source of our pipeline, but you could choose any other chain for which a raw logs dataset is available. Raw logs need to be decoded for us to be able to identify the events we want to track. For that purpose, we will use the decoding functions to dynamically fetch the ABIs of both the UniswapV3Factory and UniswapV3Pool contracts, since they contain the actual definitions of the PoolCreated and Swap events.

It’s worth mentioning that Uniswap has different versions and it’s possible that some event definitions might differ. In this example we’ll focus on UniswapV3. Depending on the events you are interested in tracking you might want to refine this example accordingly, but the principles explained will stay the same.

Let’s now see all these concepts applied in an example pipeline definition:

Pipeline definition

base-dex-trades.yaml

name: base-dex-trades
resource_size: s

sources:
  base_logs:
    type: dataset
    dataset_name: base.raw_logs
    version: 1.0.0
    start_at: latest

transforms:
  # Fetch the ABI of UniswapV3Factory and use it to decode PoolCreated events
  factory_decoded:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        _gs_log_decode(
          _gs_fetch_abi('https://gist.githubusercontent.com/JavierTrujilloG/7df78272e689bf102cbe97ae86607d94/raw/9733aaa132a2c3e82cccbe5b0681d3270d696c83/UniswapV3Factory-ABI.json', 'raw'),
          topics,
          data
        ) as decoded,
        block_number,
        transaction_hash
      FROM base_logs
      WHERE SPLIT_INDEX(topics, ',', 0) = '0x783cca1c0412dd0d695e784568c96da2e9c22ff989357a2e8b1d9b2b4e6b7118'

  # Fetch the ABI of a UniswapV3Pool and use it to decode Swap events
  pool_decoded:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        _gs_log_decode(
          _gs_fetch_abi('https://gist.githubusercontent.com/JavierTrujilloG/d3d2d80fbfd3415dd8e11aa498bd0909/raw/b8df8303e51ac7ad9ac921f25bfa84936bb4bc63/UniswapV3Pool-ABI.json', 'raw'),
          topics,
          data
        ) as decoded,
        block_number,
        transaction_hash
      FROM base_logs
      WHERE SPLIT_INDEX(topics, ',', 0) = '0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67'

  # Unnest the values from the `decoded` struct to get PoolCreated event data
  factory_clean:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        decoded.event_params as event_params,
        decoded.event_signature as event_signature,
        block_number,
        transaction_hash
      FROM factory_decoded
      WHERE decoded IS NOT NULL
        AND decoded.event_signature = 'PoolCreated'

  # Unnest the values from the `decoded` struct to get Swap event data
  pool_clean:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        decoded.event_params as event_params,
        decoded.event_signature as event_signature,
        block_number,
        transaction_hash
      FROM pool_decoded
      WHERE decoded IS NOT NULL
        AND decoded.event_signature = 'Swap'

sinks:
  poolcreated_events_sink:
    type: postgres
    from: factory_clean
    schema: decoded_events
    table: poolcreated
    secret_name: <YOUR_SECRET>
    primary_key: id

  swaps_event_sink:
    type: postgres
    from: pool_clean
    schema: decoded_events
    table: swaps
    secret_name: <YOUR_SECRET>
    primary_key: id

If you copy and use this configuration file, make sure to update:

Your secret_name. If you already created a secret, you can find it via the CLI command goldsky secret list.
The schema and table you want the data written to. By default it writes to the decoded_events schema.

Let’s deconstruct this pipeline starting at the top:

Filtering by event signature

The first topic of every EVM log (topic0) is the keccak256 hash of the event’s signature, so we can filter raw logs down to just the events we care about before decoding anything. The two events map to these hashes:

PoolCreated (index_topic_1 address token0, index_topic_2 address token1, index_topic_3 uint24 fee, int24 tickSpacing, address pool) maps to 0x783cca1c0412dd0d695e784568c96da2e9c22ff989357a2e8b1d9b2b4e6b7118
Swap (index_topic_1 address sender, index_topic_2 address recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick) maps to 0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67

Since the topics column is a comma-separated string, we use SPLIT_INDEX to extract topic0 and compare it against each hash. You can also write the same filter as topics LIKE '0x783cca1c%', or derive the hash in the query itself with _gs_keccak256.

This pipeline starts at the chain tip (start_at: latest). To also process historical trades, use start_at: earliest. If you can narrow the source with a coarse filter such as a contract address (for example, the factory address for PoolCreated events), add a source-level filter: so fast scan speeds up the backfill. Keep fine-grained filtering, like event signatures, in the transforms.

Next, there are 4 transforms in this pipeline definition which we’ll explain, starting from the top:

Decoding transforms

Transform: factory_decoded

SELECT
  id,
  _gs_log_decode(
    _gs_fetch_abi('https://gist.githubusercontent.com/JavierTrujilloG/7df78272e689bf102cbe97ae86607d94/raw/9733aaa132a2c3e82cccbe5b0681d3270d696c83/UniswapV3Factory-ABI.json', 'raw'),
    topics,
    data
  ) as decoded,
  block_number,
  transaction_hash
FROM base_logs
WHERE SPLIT_INDEX(topics, ',', 0) = '0x783cca1c0412dd0d695e784568c96da2e9c22ff989357a2e8b1d9b2b4e6b7118'

Transform: pool_decoded

SELECT
  id,
  _gs_log_decode(
    _gs_fetch_abi('https://gist.githubusercontent.com/JavierTrujilloG/d3d2d80fbfd3415dd8e11aa498bd0909/raw/b8df8303e51ac7ad9ac921f25bfa84936bb4bc63/UniswapV3Pool-ABI.json', 'raw'),
    topics,
    data
  ) as decoded,
  block_number,
  transaction_hash
FROM base_logs
WHERE SPLIT_INDEX(topics, ',', 0) = '0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67'

The first two transforms fetch the ABIs for UniswapV3Factory and a UniswapV3Pool, allowing us to decode DEX events and filter by PoolCreated and Swap events in the following transforms. As explained in the Decode contract events guide, we first make use of the _gs_fetch_abi function to get each ABI and pass it as the first argument to the _gs_log_decode function, which decodes the log’s topics and data. We store the result in a decoded struct which we unnest in the next transforms.

Event filtering transforms

Transform: factory_clean

SELECT
  id,
  decoded.event_params as event_params,
  decoded.event_signature as event_signature,
  block_number,
  transaction_hash
FROM factory_decoded
WHERE decoded IS NOT NULL
  AND decoded.event_signature = 'PoolCreated'

Transform: pool_clean

SELECT
  id,
  decoded.event_params as event_params,
  decoded.event_signature as event_signature,
  block_number,
  transaction_hash
FROM pool_decoded
WHERE decoded IS NOT NULL
  AND decoded.event_signature = 'Swap'

In the next two transforms we take the result of the previous decoding for each contract and filter by the PoolCreated and Swap events:

id: This is the Goldsky provided id, a string composed of the dataset name, block hash, and log index, which is unique per event. Here’s an example: log_0x60eaf5a2ab37c73cf1f3bbd32fc17f2709953192b530d75aadc521111f476d6c_18
decoded.event_params as event_params: event_params is an array containing the parameters associated with each event. For instance, in the case of Swap events, event_params[1] is the sender. You could use this for further analysis in downstream processing.
decoded.event_signature as event_signature: the decoder outputs the event name as event_signature, excluding its arguments.
WHERE decoded IS NOT NULL: to leave out potential null results from the decoder.
AND decoded.event_signature = 'PoolCreated': we use this value to keep only PoolCreated (or Swap) events. This makes each downstream table single-purpose even though both decode transforms read from the same source.

If you would like to filter by other events like Mint you could easily add them to these queries; for example: WHERE decoded.event_signature IN ('Swap', 'Mint') Both resulting datasets will be used as sources to two different tables at our sink: decoded_events.poolcreated and decoded_events.swaps.

Deploying the pipeline

Assuming we are using the same filename for the pipeline configuration as in this example, we can deploy this pipeline with:

goldsky turbo apply base-dex-trades.yaml

Here’s an example Swap record from our sink:

id	event_params	event_signature	block_number	transaction_hash
log_0x18db9278e431b3bb65c151857448227a649d9f8fe3fd0cdf2b9835eb8c71d8ae_4	0x508fdf90951c1a31faa5dcd119f3b60e0e0e87fb,0x508fdf90951c1a31faa5dcd119f3b60e0e0e87fb,-256654458505550,500000000000000000,3523108129873998835611448265535,631691941157619701,75899	Swap	1472162	0xd8a1b2c1296479f31f048aaf753e16f3d7d908fd17e6697b8850fdf209f080f6

We can see that it corresponds to the Swap event of this transaction:

This concludes our successful deployment of a Turbo pipeline streaming DEX trade events from the Base chain into our database using inline decoders. Congrats!

Conclusion

In this guide, we’ve walked through the process of using Turbo pipelines to decode and stream DEX events, specifically focusing on Swap and PoolCreated events, into a PostgreSQL database. Along the way we have seen an example implementation of how to do inline decoding using the ABIs of factory and pool contracts with the decoding functions. Can’t find what you’re looking for? Reach out to us at support@goldsky.com for help.

Turbo

Guides

Reference

Mirror (legacy)

What you’ll need

Introduction

Pipeline definition

Filtering by event signature

Decoding transforms

Event filtering transforms

Deploying the pipeline

Conclusion

​What you’ll need

​Introduction

​Pipeline definition

​Filtering by event signature

​Decoding transforms

​Event filtering transforms

​Deploying the pipeline

​Conclusion

What you’ll need

Introduction

Pipeline definition

Filtering by event signature

Decoding transforms

Event filtering transforms

Deploying the pipeline

Conclusion