> ## Documentation Index > Fetch the complete documentation index at: https://docs.goldsky.com/llms.txt > Use this file to discover all available pages before exploring further. # Stream DEX trades > Stream decentralized exchange trade events to your database with Turbo pipelines. 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](/turbo-pipelines/sinks/overview) works. ## What you'll need 1. A Goldsky account and the CLI installed 1) Install the Goldsky CLI: **For macOS/Linux:** ```shell theme={"dark"} curl https://goldsky.com | sh ``` **For Windows:** ```shell theme={"dark"} npm install -g @goldskycom/cli ``` Windows users need to have Node.js and npm installed first. Download from [nodejs.org](https://nodejs.org) if not already installed. 2) Log into your Project by running: ```shell theme={"dark"} 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](https://app.goldsky.com/dashboard/settings) page and pass it directly with `goldsky login --token `. Use `goldsky login --no-browser` to print the login URL instead of opening a browser. 3) Now that you are logged in, run `goldsky` to get started: ```shell theme={"dark"} goldsky ``` 2. A basic understanding of [Turbo pipelines](/turbo-pipelines/introduction) 3. A destination sink to write your data to. In this example, we will use the [PostgreSQL sink](/turbo-pipelines/sinks/postgres) ## 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](https://docs.uniswap.org/contracts/v2/concepts/protocol-overview/how-uniswap-works) 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](/turbo-pipelines/sources/evm)) 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](/turbo-pipelines/reference/sql-functions#evm--ethereum-functions) to dynamically fetch the ABIs of both the [UniswapV3Factory](https://basescan.org/address/0x33128a8fc17869897dce68ed026d694621f6fdfd) and [UniswapV3Pool](https://basescan.org/address/0xcccc03b23cd798c06828c377466f267e59bb9739) 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 ```yaml base-dex-trades.yaml expandable theme={"dark"} 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: primary_key: id swaps_event_sink: type: postgres from: pool_clean schema: decoded_events table: swaps secret_name: primary_key: id ``` If you copy and use this configuration file, make sure to update: 1. Your `secret_name`. If you already [created a secret](/platform/secrets), you can find it via the CLI command `goldsky secret list`. 2. 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`](/turbo-pipelines/reference/sql-functions#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`](/turbo-pipelines/reference/sql-functions#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](/turbo-pipelines/sources/evm#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 ```sql Transform: factory_decoded theme={"dark"} 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' ``` ```sql Transform: pool_decoded theme={"dark"} 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](/turbo-pipelines/guides/decoding-contract-events), 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 ```sql Transform: factory_clean theme={"dark"} 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' ``` ```sql Transform: pool_clean theme={"dark"} 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: ```bash theme={"dark"} 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](/turbo-pipelines/reference/sql-functions#evm--ethereum-functions). Can't find what you're looking for? Reach out to us at [support@goldsky.com](mailto:support@goldsky.com) for help. ## Related topics - [Tokenized equities & RWA data layer](/solutions/tokenized-equities.md) - [Solana DEX Trades](/turbo-pipelines/sources/solana-dex-trades.md) - [Stream onchain data with Turbo](/turbo-pipelines/introduction.md) - [ctx.hypercore: trade and deploy on Hyperliquid HyperCore](/compose/context/hypercore.md) - [Securities compliance & market surveillance](/solutions/securities-compliance.md)