Tiders

https://github.com/yulesa/tiders
Admin
Tiders is an open-source framework that simplifies fetching, transforming, and storing blockchain...
Tokens:10,047
Snippets:84
Trust Score:6.6
Update:6 months ago
Show doc for...
Tokens:
Raw
### Install Tiders and Dependencies

Source: https://context7.com/yulesa/tiders/llms.txt

Install the core Tiders package and optional backend or processing dependencies using pip.

```bash
# Base installation
pip install tiders

# With specific writer backends
pip install tiders[duckdb]
pip install tiders[clickhouse]
pip install tiders[postgresql]
pip install tiders[delta_lake]
pip install tiders[iceberg]
pip install tiders[polars]
pip install tiders[pandas]
pip install tiders[datafusion]

# Install all optional dependencies
pip install tiders[all]
```

--------------------------------

### YAML Pipeline Configuration Example

Source: https://context7.com/yulesa/tiders/llms.txt

Define a Tiders pipeline declaratively in YAML format.

```yaml
# tiders.yaml
project:
  name: erc20_transfers
  description: Fetch ERC-20 Transfer events and write to DuckDB.

# Data provider
provider:
  kind: rpc  # or hypersync, sqd
  url: "${PROVIDER_URL}"  # Environment variable substitution

# Contract ABIs (optional but recommended)
contracts:
  - name: erc20
    address: "0xae78736Cd615f374D3085123A210448E74Fc6393"
    abi: ./erc20.abi.json  # Downloaded via `tiders abi`

# Query definition
query:
  kind: evm
  from_block: 18000000
  to_block: 18000100
  logs:
    - topic0: erc20.Events.Transfer.topic0  # Reference ABI
  fields:
    log: [address, topic0, topic1, topic2, topic3, data, block_number, transaction_hash, log_index]

# Transformation steps
steps:
  - kind: evm_decode_events
    config:
      event_signature: erc20.Events.Transfer.signature  # Reference ABI
      output_table: transfers
      allow_decode_fail: true
      hstack: false
  - kind: hex_encode

# Output destination
writer:
  kind: duckdb
  config:
    path: data/transfers.duckdb
```

--------------------------------

### Tiders CLI Commands

Source: https://context7.com/yulesa/tiders/llms.txt

Examples of using the Tiders CLI to manage pipelines.

```bash
# Run a pipeline from YAML (auto-discovers tiders.yaml in current directory)
tiders start
```

```bash
# Run a specific YAML file
tiders start ./my_pipeline.yaml
```

```bash
# Override block range
tiders start --from-block 18000000 --to-block 18000100
```

```bash
# Use a specific .env file
tiders start --env-file .env.production
```

```bash
# Generate Python script from YAML
tiders codegen ./my_pipeline.yaml -o my_pipeline.py
```

```bash
# Fetch ABI for a contract
tiders abi --address 0xae78736Cd615f374D3085123A210448E74Fc6393 --chain-id 1
```

```bash
# Fetch ABIs declared in YAML
tiders abi --yaml-path ./my_pipeline.yaml
```

--------------------------------

### Ingest Contract Events via YAML

Source: https://github.com/yulesa/tiders/blob/main/examples/polymarket_vol/readme.md

Start the ingestion pipelines for both exchange contracts using the provided YAML configuration files.

```bash
cd examples/polymarket_vol
tiders start polymarket_CTFExchange.yaml
tiders start polymarket_NegRiskCtfExchange.yaml
```

--------------------------------

### Enable Trace Level Logging for Rust Modules

Source: https://github.com/yulesa/tiders/blob/main/README.md

To see logs from Rust modules, set the RUST_LOG environment variable. This example shows how to enable trace level logging.

```bash
RUST_LOG=trace uv run examples/path/to/my/example
```

--------------------------------

### Configure Delta Lake and Iceberg Writers

Source: https://context7.com/yulesa/tiders/llms.txt

Configures writers for Delta Lake and Iceberg table formats, including catalog setup for Iceberg.

```python
from pathlib import Path
from tiders.config import Writer, WriterKind, DeltaLakeWriterConfig, IcebergWriterConfig

# Delta Lake writer
delta_writer = Writer(
    kind=WriterKind.DELTA_LAKE,
    config=DeltaLakeWriterConfig(
        data_uri="data/delta_lake",
        partition_by={"transfers": ["block_number"]},
        storage_options=None,  # Add S3 credentials for cloud storage
        anchor_table=None,
    ),
)
```

```python
# Iceberg writer with SQLite catalog
from pyiceberg.catalog import load_catalog

warehouse_path = Path("data/iceberg/warehouse")
warehouse_path.mkdir(parents=True, exist_ok=True)

catalog = load_catalog(
    "local",
    type="sql",
    uri="sqlite:///data/iceberg/catalog.db",
    warehouse=f"file://{warehouse_path}",
)

iceberg_writer = Writer(
    kind=WriterKind.ICEBERG,
    config=IcebergWriterConfig(
        namespace="blockchain",
        catalog=catalog,
        write_location=f"file://{warehouse_path}",
    ),
)
```

--------------------------------

### Factory Pattern: Two-Stage Indexing (Stage 1)

Source: https://context7.com/yulesa/tiders/llms.txt

Stage 1 of a two-stage indexing pattern: discover child contracts from a factory. This example discovers pools from the Uniswap V3 Factory.

```python
import asyncio
from tiders import run_pipeline
from tiders.config import Pipeline, Step, StepKind, Writer, WriterKind
from tiders.config import EvmDecodeEventsConfig, HexEncodeConfig, EvmTableAliases, DuckdbWriterConfig
from tiders_core.ingest import ProviderConfig, ProviderKind, Query, QueryKind
from tiders_core.ingest import evm

# Stage 1: Discover pools from Uniswap V3 Factory
async def discover_pools():
    provider = ProviderConfig(
        kind=ProviderKind.HYPERSYNC,
        url="https://eth.hypersync.xyz/",
        stop_on_head=True,
    )

    query = Query(
        kind=QueryKind.EVM,
        params=evm.Query(
            from_block=12369621,
            to_block=12370621,
            logs=[evm.LogRequest(
                address=["0x1f98431c8ad98523631ae4a59f267346ea31f984"],  # Factory
                topic0=["0x783cca1c0412dd0d695e784568c96da2e9c22ff989357a2e8b1d9b2b4e6b7118"],  # PoolCreated
            )],
            fields=evm.Fields(
                log=evm.LogFields(
                    block_number=True, transaction_hash=True, log_index=True,
                    address=True, topic0=True, topic1=True, topic2=True, topic3=True, data=True,
                ),
            ),
        ),
    )

    steps = [
        Step(
            kind=StepKind.EVM_DECODE_EVENTS,
            config=EvmDecodeEventsConfig(
                event_signature="PoolCreated(address indexed token0, address indexed token1, uint24 indexed fee, int24 tickSpacing, address pool)",
                output_table="pool_created",
            ),
        ),
        Step(kind=StepKind.HEX_ENCODE, config=HexEncodeConfig()),
    ]

    pipeline = Pipeline(
        provider=provider,
        query=query,
        writer=Writer(kind=WriterKind.DUCKDB, config=DuckdbWriterConfig(path="data/uniswap.duckdb")),
        steps=steps,
    )

    await run_pipeline(pipeline)
```

--------------------------------

### Define Custom Pandas Transformation Function

Source: https://context7.com/yulesa/tiders/llms.txt

Define a custom transformation function for Pandas DataFrames. This example demonstrates calculating normalized value and block date, then aggregating by address.

```python
def transform_with_pandas(
    data: dict[str, pd.DataFrame],
    ctx: Optional[Any]
) -> dict[str, pd.DataFrame]:
    transfers = data["transfers"]

    # Pandas operations
    transfers["value_eth"] = transfers["value"].astype(float) / 1e18
    transfers["block_date"] = pd.to_datetime(transfers["timestamp"], unit="s")

    # Aggregate by address
    summary = transfers.groupby("from").agg({
        "value_eth": "sum",
        "transaction_hash": "count"
    }).reset_index()
    summary.columns = ["address", "total_value", "tx_count"]

    return {"transfers": transfers, "address_summary": summary}
```

--------------------------------

### Define Custom DataFusion SQL Transformation

Source: https://context7.com/yulesa/tiders/llms.txt

Define a SQL transformation function for DataFusion. This example registers DataFrames in the context and executes a SQL query to find top addresses by event count and total value.

```python
def sql_transform(
    ctx: datafusion.SessionContext,
    data: dict[str, datafusion.DataFrame],
    user_ctx: Optional[Any]
) -> dict[str, datafusion.DataFrame]:
    # Register tables in context
    for name, df in data.items():
        ctx.register_dataframe(name, df)

    # Execute SQL query
    result = ctx.sql("""
        SELECT
            address,
            COUNT(*) as event_count,
            SUM(CAST(value AS DOUBLE)) as total_value
        FROM transfers
        GROUP BY address
        ORDER BY total_value DESC
        LIMIT 100
    """)

    return {"top_addresses": result}
```

--------------------------------

### Initialize Environment Configuration

Source: https://github.com/yulesa/tiders/blob/main/examples/polymarket_vol/readme.md

Copy the provided environment template to a local .env file before running the pipeline.

```bash
cp .env.example .env
```

--------------------------------

### Pipeline Initialization

Source: https://context7.com/yulesa/tiders/llms.txt

Initialize a Tiders pipeline with provider, query, writer, steps, and table aliases.

```python
pipeline = Pipeline(
    provider=provider,
    query=query,
    writer=writer,
    steps=steps,
    table_aliases=evm_aliases,
)
```

--------------------------------

### Configure EVM Queries

Source: https://context7.com/yulesa/tiders/llms.txt

Import necessary modules to define EVM-specific query parameters.

```python
from tiders_core.ingest import Query, QueryKind
from tiders_core.ingest import evm
```

--------------------------------

### Sync Environment with uv

Source: https://github.com/yulesa/tiders/blob/main/README.md

After configuring Tiders to use a local tiders-core package, sync the environment using the uv command to apply the changes.

```bash
cd tiders
uv sync
```

--------------------------------

### Configure PyArrow Dataset Writer

Source: https://context7.com/yulesa/tiders/llms.txt

Configures a writer for Parquet files, supporting local storage or S3 buckets with partitioning.

```python
import pyarrow.fs as pa_fs
from tiders.config import Writer, WriterKind, PyArrowDatasetWriterConfig

# Local Parquet files
parquet_writer = Writer(
    kind=WriterKind.PYARROW_DATASET,
    config=PyArrowDatasetWriterConfig(
        base_dir="data/parquet",
        basename_template="part-{i}.parquet",
        partitioning={"transfers": ["block_number"]},
        partitioning_flavor={"transfers": "hive"},
        max_rows_per_file=1_000_000,
        create_dir=True,
    ),
)
```

```python
# Write to S3
s3_writer = Writer(
    kind=WriterKind.PYARROW_DATASET,
    config=PyArrowDatasetWriterConfig(
        base_dir="s3://my-bucket/blockchain-data",
        filesystem=pa_fs.S3FileSystem(region="us-east-1"),
        partitioning={"transfers": ["block_number"]},
    ),
)
```

--------------------------------

### Configure DuckDB Writer (Existing Connection)

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a writer to use an existing DuckDB connection cursor. This provides more control over the database connection.

```python
# Or pass an existing connection for more control
connection = duckdb.connect("data/blockchain.duckdb")
duckdb_writer_with_conn = Writer(
    kind=WriterKind.DUCKDB,
    config=DuckdbWriterConfig(
        connection=connection.cursor(),
    ),
)
```

--------------------------------

### Configure Data Providers

Source: https://context7.com/yulesa/tiders/llms.txt

Define the blockchain data source using HyperSync, SQD, or standard RPC configurations.

```python
from tiders_core.ingest import ProviderConfig, ProviderKind
import os

# HyperSync provider (fastest for EVM chains)
hypersync_provider = ProviderConfig(
    kind=ProviderKind.HYPERSYNC,
    url="https://eth.hypersync.xyz/",
    bearer_token=os.environ.get("BEARER_TOKEN"),
    stop_on_head=True,  # Stop when reaching chain head
)

# SQD provider (supports both EVM and Solana)
sqd_provider = ProviderConfig(
    kind=ProviderKind.SQD,
    url="https://portal.sqd.dev/datasets/ethereum-mainnet",
    stop_on_head=True,
)

# Standard RPC provider
rpc_provider = ProviderConfig(
    kind=ProviderKind.RPC,
    url="https://mainnet.gateway.tenderly.co",
)

# Solana SQD provider
solana_provider = ProviderConfig(
    kind=ProviderKind.SQD,
    url="https://portal.sqd.dev/datasets/solana-mainnet",
)
```

--------------------------------

### Configure ClickHouse Writer

Source: https://context7.com/yulesa/tiders/llms.txt

Configures a ClickHouse writer using either direct connection parameters or an existing async client.

```python
clickhouse_writer = Writer(
    kind=WriterKind.CLICKHOUSE,
    config=ClickHouseWriterConfig(
        host=os.environ.get("CLICKHOUSE_HOST", "localhost"),
        port=int(os.environ.get("CLICKHOUSE_PORT", "8123")),
        username=os.environ.get("CLICKHOUSE_USER", "default"),
        password=os.environ.get("CLICKHOUSE_PASSWORD", ""),
        database="blockchain",
        secure=False,
        engine="MergeTree()",
        order_by={
            "transfers": ["block_number", "log_index"],
            "blocks": ["number"],
        },
        codec={
            "transfers": {
                "data": "ZSTD(3)",
                "transaction_hash": "LZ4",
            }
        },
        skip_index={
            "transfers": [
                ClickHouseSkipIndex(
                    name="idx_address",
                    val="address",
                    type_="bloom_filter",
                    granularity=4,
                )
            ]
        },
        anchor_table="blocks",  # Write this table last for ordering guarantee
        create_tables=True,
    ),
)
```

```python
import clickhouse_connect

async def create_clickhouse_writer():
    client = await clickhouse_connect.get_async_client(
        host="localhost",
        port=8123,
        database="blockchain",
    )
    return Writer(
        kind=WriterKind.CLICKHOUSE,
        config=ClickHouseWriterConfig(client=client),
    )
```

--------------------------------

### Compute Hourly Volume Transformation

Source: https://github.com/yulesa/tiders/blob/main/examples/polymarket_vol/readme.md

Run the transformation script after ingestion to aggregate USDC volume per hour in ClickHouse.

```python
python polymarket_vol_transformation.py
```

--------------------------------

### Configure PostgreSQL Writer

Source: https://context7.com/yulesa/tiders/llms.txt

Configures a PostgreSQL writer using connection parameters or an existing async connection object.

```python
import psycopg
from tiders.config import Writer, WriterKind, PostgresqlWriterConfig

# Configuration with connection parameters
postgresql_writer = Writer(
    kind=WriterKind.POSTGRESQL,
    config=PostgresqlWriterConfig(
        host="localhost",
        port=5432,
        user="postgres",
        password="secret",
        dbname="blockchain",
        schema="public",
        anchor_table=None,
        create_tables=True,
    ),
)
```

```python
# Or with a pre-built connection
async def create_pg_writer():
    conn = await psycopg.AsyncConnection.connect(
        "host=localhost port=5432 dbname=blockchain user=postgres password=secret"
    )
    return Writer(
        kind=WriterKind.POSTGRESQL,
        config=PostgresqlWriterConfig(connection=conn),
    )
```

--------------------------------

### Configure EVM Event Decoding

Source: https://context7.com/yulesa/tiders/llms.txt

Initializes the configuration for decoding raw EVM log entries into structured columns.

```python
from tiders.config import Step, StepKind, EvmDecodeEventsConfig
```

--------------------------------

### Main Execution Flow (Python)

Source: https://context7.com/yulesa/tiders/llms.txt

This function orchestrates the two-stage data pipeline. It first discovers pools, then reads discovered pool addresses from a DuckDB database, and finally calls the function to index events from these pools. Ensure the DuckDB database and the 'pool_created' table exist.

```python
async def main():
    await discover_pools()

    # Read discovered pools from DuckDB
    import duckdb
    conn = duckdb.connect("data/uniswap.duckdb")
    pools = [row[0] for row in conn.execute("SELECT DISTINCT pool FROM pool_created").fetchall()]
    conn.close()

    if pools:
        await index_pool_events(pools)

asyncio.run(main())
```

--------------------------------

### Configure Joining Block Data

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a step to join block metadata like timestamp and number into other tables. This is useful for adding context to logs or other data.

```python
# Join block timestamp and number into logs
join_blocks = Step(
    kind=StepKind.JOIN_BLOCK_DATA,
    config=JoinBlockDataConfig(
        tables=None,  # Join into all tables except blocks
        block_table_name="blocks",
        join_left_on=["block_number"],
        join_blocks_on=["number"],
    ),
)
```

--------------------------------

### Configure Local tiders-core Development

Source: https://github.com/yulesa/tiders/blob/main/README.md

Configure Tiders to use a local tiders-core Python package by adding a path to the pyproject.toml file. This enables editable development of tiders-core.

```toml
[tool.uv.sources]
tiders-core = { path = "../tiders-core/python", editable = true }
```

--------------------------------

### Query Solana (SVM) Instructions

Source: https://context7.com/yulesa/tiders/llms.txt

Configures a query for Solana blockchain data, specifically targeting Jupiter swap instructions with associated block and transaction fields.

```python
from tiders_core.ingest import Query, QueryKind
from tiders_core.ingest.svm import (
    Query as SvmQuery,
    Fields,
    InstructionFields,
    BlockFields,
    TransactionFields,
    InstructionRequest,
)

# Query Jupiter swap instructions on Solana
solana_query = Query(
    kind=QueryKind.SVM,
    params=SvmQuery(
        from_block=330447751,
        to_block=330448751,
        include_all_blocks=True,
        fields=Fields(
            instruction=InstructionFields(
                block_slot=True,
                block_hash=True,
                transaction_index=True,
                instruction_address=True,
                program_id=True,
                a0=True, a1=True, a2=True, a3=True, a4=True,
                a5=True, a6=True, a7=True, a8=True, a9=True,
                data=True,
                error=True,
            ),
            block=BlockFields(
                slot=True,
                hash=True,
                timestamp=True,
            ),
            transaction=TransactionFields(
                block_slot=True,
                block_hash=True,
                transaction_index=True,
                signature=True,
            ),
        ),
        instructions=[
            InstructionRequest(
                program_id=["JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4"],  # Jupiter
                discriminator=["0xe445a52e51cb9a1d40c6cde8260871e2"],
                include_transactions=True,
            )
        ],
    ),
)
```

--------------------------------

### Configure DuckDB Writer (File Path)

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a writer to save data to a DuckDB database file. This is suitable for local development and prototyping.

```python
# Simple file-based configuration
duckdb_writer = Writer(
    kind=WriterKind.DUCKDB,
    config=DuckdbWriterConfig(
        path="data/blockchain.duckdb",
    ),
)
```

--------------------------------

### Execute a Pipeline with run_pipeline

Source: https://context7.com/yulesa/tiders/llms.txt

The primary entry point for running a data pipeline. It requires a configured provider, query, transformation steps, and a writer.

```python
import asyncio
from tiders import run_pipeline
from tiders.config import (
    Pipeline,
    Step,
    StepKind,
    Writer,
    WriterKind,
    DuckdbWriterConfig,
    EvmDecodeEventsConfig,
    HexEncodeConfig,
)
from tiders_core.ingest import ProviderConfig, ProviderKind, Query, QueryKind
from tiders_core.ingest import evm

# Configure the data provider
provider = ProviderConfig(
    kind=ProviderKind.RPC,
    url="https://mainnet.gateway.tenderly.co",
)

# Define the query - what data to fetch
query = Query(
    kind=QueryKind.EVM,
    params=evm.Query(
        from_block=18000000,
        to_block=18000100,
        logs=[evm.LogRequest(
            topic0=["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"]  # Transfer event
        )],
        fields=evm.Fields(
            log=evm.LogFields(
                log_index=True,
                transaction_hash=True,
                block_number=True,
                address=True,
                data=True,
                topic0=True,
                topic1=True,
                topic2=True,
            ),
        ),
    ),
)

# Define transformation steps
steps = [
    Step(
        kind=StepKind.EVM_DECODE_EVENTS,
        config=EvmDecodeEventsConfig(
            event_signature="Transfer(address indexed from, address indexed to, uint256 value)",
            allow_decode_fail=True,
            output_table="transfers",
        ),
    ),
    Step(
        kind=StepKind.HEX_ENCODE,
        config=HexEncodeConfig(),
    ),
]

# Configure the writer
writer = Writer(
    kind=WriterKind.DUCKDB,
    config=DuckdbWriterConfig(path="data/transfers.duckdb"),
)

# Assemble and run the pipeline
pipeline = Pipeline(
    provider=provider,
    query=query,
    writer=writer,
    steps=steps,
)

asyncio.run(run_pipeline(pipeline, pipeline_name="erc20_transfers"))
```

--------------------------------

### Clone Tiders Repositories

Source: https://github.com/yulesa/tiders/blob/main/README.md

Clone the main Tiders repository along with its core and RPC client repositories to set up the development environment.

```bash
git clone https://github.com/yulesa/tiders.git
git clone https://github.com/yulesa/tiders-core.git
git clone https://github.com/yulesa/tiders-rpc-client.git
```

--------------------------------

### Configure ClickHouse Writer

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a writer to save data to ClickHouse for production-scale analytical workloads. Supports configurable ordering keys, codecs, and skip indexes.

```python
import os
from tiders.config import Writer, WriterKind, ClickHouseWriterConfig, ClickHouseSkipIndex
```

--------------------------------

### Multiple Writers Configuration

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a pipeline to write data to multiple destinations simultaneously using a list of writers.

```python
from tiders.config import Pipeline, Writer, WriterKind, DuckdbWriterConfig, PyArrowDatasetWriterConfig

# Write to both DuckDB and Parquet
writers = [
    Writer(
        kind=WriterKind.DUCKDB,
        config=DuckdbWriterConfig(path="data/blockchain.duckdb"),
    ),
    Writer(
        kind=WriterKind.PYARROW_DATASET,
        config=PyArrowDatasetWriterConfig(base_dir="data/parquet"),
    ),
]

pipeline = Pipeline(
    provider=provider,
    query=query,
    writer=writers,  # List of writers
    steps=steps,
)
```

--------------------------------

### Index Events from Discovered Pools (Python)

Source: https://context7.com/yulesa/tiders/llms.txt

This function indexes events from a list of Ethereum pool addresses. It configures a HyperSync provider, defines an EVM query to fetch logs within a block range, and sets up transformation steps for decoding events and encoding them to hex. The output is written to a DuckDB database. Ensure the 'pool_logs' table is available for decoding.

```python
async def index_pool_events(pool_addresses: list[str]):
    provider = ProviderConfig(
        kind=ProviderKind.HYPERSYNC,
        url="https://eth.hypersync.xyz/",
        stop_on_head=True,
    )

    query = Query(
        kind=QueryKind.EVM,
        params=evm.Query(
            from_block=12369621,
            to_block=12370621,
            logs=[evm.LogRequest(address=pool_addresses)],  # All discovered pools
            fields=evm.Fields(
                log=evm.LogFields(
                    block_number=True, transaction_hash=True, log_index=True,
                    address=True, topic0=True, topic1=True, topic2=True, topic3=True, data=True,
                ),
            ),
        ),
    )

    # Decode each event type
    steps = [
        Step(
            kind=StepKind.EVM_DECODE_EVENTS,
            config=EvmDecodeEventsConfig(
                event_signature="Swap(address indexed sender, address indexed recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick)",
                input_table="pool_logs",
                output_table="swaps",
                filter_by_topic0=True,
                allow_decode_fail=True,
            ),
        ),
        Step(kind=StepKind.HEX_ENCODE, config=HexEncodeConfig()),
    ]

    pipeline = Pipeline(
        provider=provider,
        query=query,
        writer=Writer(kind=WriterKind.DUCKDB, config=DuckdbWriterConfig(path="data/uniswap.duckdb")),
        steps=steps,
        table_aliases=EvmTableAliases(logs="pool_logs"),
    )

    await run_pipeline(pipeline)
```

--------------------------------

### ProviderConfig

Source: https://context7.com/yulesa/tiders/llms.txt

Configures the blockchain data source. Supports HyperSync, SQD, and standard RPC nodes.

```APIDOC
## ProviderConfig

### Description
Defines the source of the blockchain data. Supported kinds include HYPERSYNC, SQD, and RPC.

### Parameters
- **kind** (ProviderKind) - Required - The type of provider (e.g., ProviderKind.HYPERSYNC, ProviderKind.SQD, ProviderKind.RPC).
- **url** (str) - Required - The endpoint URL for the provider.
- **bearer_token** (str) - Optional - Authentication token for the provider.
- **stop_on_head** (bool) - Optional - Whether to stop fetching when reaching the chain head.
```

--------------------------------

### Configure Joining EVM Transaction Data

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a step to join EVM transaction data such as gas, sender, and receiver into other tables. This is useful for enriching log data with transaction details.

```python
# Join transaction data (gas, sender, etc.) into logs
join_txs = Step(
    kind=StepKind.JOIN_EVM_TRANSACTION_DATA,
    config=JoinEvmTransactionDataConfig(
        tables=["decoded_logs"],
        tx_table_name="transactions",
        join_left_on=["block_number", "transaction_index"],
        join_transactions_on=["block_number", "transaction_index"],
    ),
)
```

--------------------------------

### run_pipeline

Source: https://context7.com/yulesa/tiders/llms.txt

The main entry point for executing a data pipeline. It streams data from a blockchain provider, applies transformation steps, and writes the results to a configured backend.

```APIDOC
## run_pipeline

### Description
Executes a defined pipeline consisting of a provider, query, transformation steps, and a writer.

### Parameters
- **pipeline** (Pipeline) - Required - The pipeline configuration object containing provider, query, steps, and writer.
- **pipeline_name** (str) - Optional - A unique identifier for the pipeline execution.

### Request Example
```python
import asyncio
from tiders import run_pipeline

# Assemble and run the pipeline
pipeline = Pipeline(
    provider=provider,
    query=query,
    writer=writer,
    steps=steps,
)

asyncio.run(run_pipeline(pipeline, pipeline_name="erc20_transfers"))
```
```

--------------------------------

### Configure Pandas Step in Pipeline

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a Pandas step in the pipeline using a custom runner function. The context is optional and passed to the runner.

```python
pandas_step = Step(
    kind=StepKind.PANDAS,
    config=PandasStepConfig(
        runner=transform_with_pandas,
        context=None,
    ),
)
```

--------------------------------

### Ingest Contract Events via Python

Source: https://github.com/yulesa/tiders/blob/main/examples/polymarket_vol/readme.md

Execute the Python scripts to ingest event data from the exchange contracts.

```python
cd examples/polymarket_vol
python polymarket_ctfexchange.py
python polymarket_neg_risk_ctf_exchange.py
```

--------------------------------

### Decode SVM Instructions

Source: https://context7.com/yulesa/tiders/llms.txt

Decodes raw Solana instruction data using a defined instruction signature.

```python
from tiders.config import Step, StepKind, SvmDecodeInstructionsConfig
from tiders_core.svm_decode import InstructionSignature, ParamInput, DynType, FixedArray

# Define instruction signature for Jupiter swaps
instruction_signature = InstructionSignature(
    discriminator="0xe445a52e51cb9a1d40c6cde8260871e2",
    params=[
        ParamInput(name="Amm", param_type=FixedArray(DynType.U8, 32)),
        ParamInput(name="InputMint", param_type=FixedArray(DynType.U8, 32)),
        ParamInput(name="InputAmount", param_type=DynType.U64),
        ParamInput(name="OutputMint", param_type=FixedArray(DynType.U8, 32)),
        ParamInput(name="OutputAmount", param_type=DynType.U64),
    ],
    accounts_names=[],
)

# Decode Solana instructions
decode_instructions = Step(
    kind=StepKind.SVM_DECODE_INSTRUCTIONS,
    config=SvmDecodeInstructionsConfig(
        instruction_signature=instruction_signature,
        input_table="instructions",
        output_table="jup_swaps_decoded",
        hstack=True,                    # Include original columns
        allow_decode_fail=True,
        filter_by_discriminator=False,
    ),
)
```

--------------------------------

### Define Table Aliases

Source: https://context7.com/yulesa/tiders/llms.txt

Maps default provider table names to custom aliases for use in pipeline transformations.

```python
from tiders.config import Pipeline, EvmTableAliases, SvmTableAliases

# EVM table aliases
evm_aliases = EvmTableAliases(
    blocks="eth_blocks",
    transactions="eth_transactions",
    logs="uniswap_v3_logs",
    traces="eth_traces",
)

# SVM table aliases
svm_aliases = SvmTableAliases(
    blocks="solana_blocks",
    transactions="solana_transactions",
    instructions="jupiter_instructions",
    logs="solana_logs",
)
```

--------------------------------

### Configure DataFusion Step in Pipeline

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a DataFusion step in the pipeline using a custom runner function. The context is optional and passed to the runner.

```python
datafusion_step = Step(
    kind=StepKind.DATAFUSION,
    config=DataFusionStepConfig(
        runner=sql_transform,
        context=None,
    ),
)
```

--------------------------------

### EVM Query Configuration

Source: https://context7.com/yulesa/tiders/llms.txt

Defines the scope and filters for fetching EVM blockchain data.

```APIDOC
## EVM Query Configuration

### Description
Specifies the block ranges, log filters, and field selection for EVM data ingestion.

### Parameters
- **kind** (QueryKind) - Required - Must be QueryKind.EVM.
- **params** (evm.Query) - Required - Configuration object containing from_block, to_block, logs, and fields.
```

--------------------------------

### Query EVM Transfer Events

Source: https://context7.com/yulesa/tiders/llms.txt

Configures a query to fetch specific ERC-20 Transfer events from an EVM contract within a defined block range.

```python
transfer_query = Query(
    kind=QueryKind.EVM,
    params=evm.Query(
        from_block=18000000,
        to_block=18000100,
        logs=[
            evm.LogRequest(
                address=["0xae78736Cd615f374D3085123A210448E74Fc6393"],  # rETH contract
                topic0=["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"],
            )
        ],
        fields=evm.Fields(
            log=evm.LogFields(
                block_number=True,
                block_hash=True,
                transaction_hash=True,
                log_index=True,
                address=True,
                topic0=True,
                topic1=True,
                topic2=True,
                topic3=True,
                data=True,
            ),
        ),
    ),
)
```

--------------------------------

### Decode EVM Events

Source: https://context7.com/yulesa/tiders/llms.txt

Configures event decoding for EVM logs. Use allow_decode_fail to handle malformed logs gracefully.

```python
decode_transfer = Step(
    kind=StepKind.EVM_DECODE_EVENTS,
    config=EvmDecodeEventsConfig(
        event_signature="Transfer(address indexed from, address indexed to, uint256 value)",
        input_table="logs",           # Source table name
        output_table="transfers",     # Destination table name
        allow_decode_fail=True,       # Keep rows that fail to decode (with nulls)
        filter_by_topic0=True,        # Only decode logs matching the event signature
        hstack=False,                 # Only output decoded columns
    ),
)
```

```python
decode_swap = Step(
    kind=StepKind.EVM_DECODE_EVENTS,
    config=EvmDecodeEventsConfig(
        event_signature="Swap(address indexed sender, address indexed recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick)",
        input_table="pool_logs",
        output_table="uniswap_v3_pool_swap",
        allow_decode_fail=True,
        filter_by_topic0=True,
    ),
)

decode_mint = Step(
    kind=StepKind.EVM_DECODE_EVENTS,
    config=EvmDecodeEventsConfig(
        event_signature="Mint(address sender, address indexed owner, int24 indexed tickLower, int24 indexed tickUpper, uint128 amount, uint256 amount0, uint256 amount1)",
        input_table="pool_logs",
        output_table="uniswap_v3_pool_mint",
        allow_decode_fail=True,
        filter_by_topic0=True,
    ),
)
```

--------------------------------

### Query EVM with Block Data

Source: https://context7.com/yulesa/tiders/llms.txt

Configures an EVM query that includes block-level data, such as timestamps, by setting include_blocks to True.

```python
query_with_blocks = Query(
    kind=QueryKind.EVM,
    params=evm.Query(
        from_block=12369621,
        to_block=12370621,
        logs=[
            evm.LogRequest(
                address=["0x1f98431c8ad98523631ae4a59f267346ea31f984"],  # Uniswap V3 Factory
                include_blocks=True,  # Include block data for joining
            )
        ],
        fields=evm.Fields(
            log=evm.LogFields(
                block_number=True,
                transaction_hash=True,
                log_index=True,
                address=True,
                topic0=True,
                topic1=True,
                topic2=True,
                topic3=True,
                data=True,
            ),
            block=evm.BlockFields(
                timestamp=True,
                number=True,
            ),
        ),
    ),
)
```

--------------------------------

### Configure Polars Step in Pipeline

Source: https://context7.com/yulesa/tiders/llms.txt

Configure a Polars step in the pipeline using a custom runner function. The context is optional and passed to the runner.

```python
polars_step = Step(
    kind=StepKind.POLARS,
    config=PolarsStepConfig(
        runner=transform_swaps,
        context=None,  # Optional context passed to runner
    ),
)
```

--------------------------------

### Parse EVM ABI Files

Source: https://context7.com/yulesa/tiders/llms.txt

Extracts event and function signatures, selectors, and topic0 hashes from an ABI JSON file for use in decoding or filtering.

```python
from pathlib import Path
from tiders_core import evm_abi_events, evm_abi_functions

# Load ABI from file
abi_path = Path("./erc20.abi.json")
abi_json = abi_path.read_text()

# Parse events from ABI
events = {
    ev.name: {
        "topic0": ev.topic0,           # Keccak256 hash for filtering
        "signature": ev.signature,      # Full signature for decoding
        "name_snake_case": ev.name_snake_case,
        "selector_signature": ev.selector_signature,
    }
    for ev in evm_abi_events(abi_json)
}

# Parse functions from ABI
functions = {
    fn.name: {
        "selector": fn.selector,        # 4-byte selector
        "signature": fn.signature,
        "name_snake_case": fn.name_snake_case,
        "selector_signature": fn.selector_signature,
    }
    for fn in evm_abi_functions(abi_json)
}

# Use in query
print(f"Transfer topic0: {events['Transfer']['topic0']}")
# Output: 0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef

print(f"Transfer signature: {events['Transfer']['signature']}")
# Output: Transfer(address indexed from, address indexed to, uint256 value)
```

--------------------------------

### Apply Polars Transformations

Source: https://context7.com/yulesa/tiders/llms.txt

Defines a step for custom transformations using Polars DataFrames.

```python
from typing import Any, Optional
import polars as pl
from tiders.config import Step, StepKind, PolarsStepConfig
```

--------------------------------

### Encode Binary Columns

Source: https://context7.com/yulesa/tiders/llms.txt

Converts binary data to string representations using Hex or Base58 encoding.

```python
from tiders.config import Step, StepKind, HexEncodeConfig, Base58EncodeConfig

# Hex encode all binary columns (EVM)
hex_encode_step = Step(
    kind=StepKind.HEX_ENCODE,
    config=HexEncodeConfig(
        tables=None,         # Process all tables (or specify list)
        prefixed=True,       # Include "0x" prefix
    ),
)

# Base58 encode for Solana addresses/signatures
base58_encode_step = Step(
    kind=StepKind.BASE58_ENCODE,
    config=Base58EncodeConfig(
        tables=None,         # Process all tables
    ),
)
```

--------------------------------

### Define Custom Polars Transformation Function

Source: https://context7.com/yulesa/tiders/llms.txt

Use this to define a custom transformation function for Polars DataFrames. It requires joining decoded instructions with block and transaction data, then adding computed columns.

```python
def transform_swaps(
    data: dict[str, pl.DataFrame],
    ctx: Optional[Any]
) -> dict[str, pl.DataFrame]:
    # Join decoded instructions with block and transaction data
    swaps = data["decoded_instructions"]
    blocks = data["blocks"]
    transactions = data["transactions"]

    # Perform joins
    swaps = swaps.join(blocks, left_on="block_slot", right_on="slot")
    swaps = swaps.join(transactions, on=["block_slot", "transaction_index"])

    # Add computed columns
    swaps = swaps.with_columns([
        (pl.col("input_amount") / 1e18).alias("input_amount_normalized"),
        pl.col("timestamp").cast(pl.Datetime).alias("block_time"),
    ])

    return {"decoded_instructions": swaps}
```

--------------------------------

### Cast Column Types

Source: https://context7.com/yulesa/tiders/llms.txt

Casts columns to specific types or applies type-based casting across all columns.

```python
import pyarrow as pa
from tiders.config import Step, StepKind, CastConfig, CastByTypeConfig

# Cast specific columns in a table
cast_step = Step(
    kind=StepKind.CAST,
    config=CastConfig(
        table_name="transfers",
        mappings={
            "value": pa.decimal128(38, 0),
            "block_number": pa.int64(),
        },
        allow_cast_fail=False,
    ),
)

# Cast all Decimal256 to Decimal128 (needed for DuckDB compatibility)
cast_by_type_step = Step(
    name="i256_to_i128",
    kind=StepKind.CAST_BY_TYPE,
    config=CastByTypeConfig(
        from_type=pa.decimal256(76, 0),
        to_type=pa.decimal128(38, 0),
        allow_cast_fail=True,  # Overflowing values become null
    ),
)
```
Tokens:
Raw