# Recipes: mentions, polling and datasets

> Ready-to-run examples for brand mentions, cashtag and contract searches, polling without duplicates, following a set of accounts through a list, and collecting a search window to JSON Lines in parallel.

HTML page: https://www.twittrapi.com/docs/recipes

Short, complete programs for the jobs people use TwittrAPI for most. They use `curl` or the [Python SDK](https://www.twittrapi.com/docs/integrations#sdks) with the [simple format](https://www.twittrapi.com/docs/simple-format), and read the key from `TWITTRAPI_API_KEY`. Every page of results is one billed call; calls that fail on our side are refunded and 429 is not charged ([billing](https://www.twittrapi.com/docs/billing)).

## Brand mentions

Mentions come in three shapes: the handle, the name and the domain. Catch all three, drop your own account and plain retweets:

```text
(@acme OR "acme app" OR url:acme.com) -from:acme -filter:nativeretweets
```

Mentions of the last day, English, without job posts and giveaways. The `filters` object adds operators for you:

```bash
curl -G 'https://api.twittrapi.com/search' \
  -H 'X-API-KEY: YOUR_API_KEY' \
  --data-urlencode 'searchItem=(@acme OR "acme app" OR url:acme.com) -from:acme' \
  --data-urlencode 'category=Latest' \
  --data-urlencode 'filters={"lang":"en","noneOfTheseWords":["hiring","giveaway"],"withinTime":"1d"}' \
  --data-urlencode 'format=simple'
```

Common words need context: "acme" alone matches everything, "acme" plus your product words or domain does not. For a campaign hashtag with traction only, use `filters={"hashtags":["AcmeSummer"],"minimumLikesCount":20,"removeReplies":true}`. All operators and filter keys: [search reference](https://www.twittrapi.com/docs/reference/search).

## Cashtags and contract addresses

Latest `$BTC` or `$ETH` tweets with some engagement, no replies:

```bash
curl -G 'https://api.twittrapi.com/search' \
  -H 'X-API-KEY: YOUR_API_KEY' \
  --data-urlencode 'searchItem=$BTC OR $ETH' \
  --data-urlencode 'category=Latest' \
  --data-urlencode 'filters={"minimumLikesCount":5,"removeReplies":true,"noneOfTheseWords":["giveaway","airdrop"]}' \
  --data-urlencode 'format=simple'
```

- Cashtags are plain text in the query: `$BTC`, `$SOL OR $JUP`. The `$` must be URL-encoded (`%24`); `--data-urlencode` and the SDKs do that for you.
- Contract addresses are matched as text. Quote them, `"0x6982508145454ce325ddbe47a25d4ec3d2311933"`, and also search the project handle and domain (`url:example.xyz`).
- Query plus filters can be up to 512 characters. Square brackets are rejected with 422.

## Poll a search without duplicates

Ask for the newest tweets every 30 seconds and pass the highest ID you have as `since_id:` (or `sinceId` in `filters`). Each tweet arrives once, oldest first. The SDK retries 429 and 5xx with backoff. To start from a point in time instead of a tweet you have, turn the date into an ID with the [tweet ID to date converter](https://www.twittrapi.com/tools/tweet-id-to-date).

```python
import time

from twittrapi import TwittrAPI, TwittrAPIError

client = TwittrAPI(response_format="simple")  # key from TWITTRAPI_API_KEY
QUERY = "($BTC OR $ETH) -filter:replies"
newest = None  # highest tweet ID seen so far

while True:
    query = f"{QUERY} since_id:{newest}" if newest else QUERY
    try:
        page = client.search.tweets(query, category="Latest")
    except TwittrAPIError as err:  # 429 and 5xx are already retried by the SDK
        print("error:", err)
        time.sleep(30)
        continue
    tweets = [t for t in page["data"] if not t.get("unavailable")]
    for t in sorted(tweets, key=lambda t: int(t["id"])):  # oldest first
        author = t["author"]["username"] if t["author"] else "?"
        print(t["createdAt"], author, t["text"][:140])
    if tweets:
        newest = max((t["id"] for t in tweets), key=int)
    time.sleep(30)
```

Every poll is one call whether or not it finds something new, so the interval sets the cost. For new tweets from a fixed set of accounts within seconds, without running the loop yourself, see [account monitoring](https://www.twittrapi.com/docs/monitoring).

## Follow a set of accounts with one call

Put the accounts you follow on a public X list. [List tweets](https://www.twittrapi.com/docs/reference/lists-tweets) returns the newest tweets of all members, so one call per poll covers the whole list, however many accounts are on it. The first run prints the current page; after that only new tweets.

```python
import time

from twittrapi import TwittrAPI

client = TwittrAPI(response_format="simple")
LIST_ID = "1234567890"  # a public X list with the accounts you follow
seen: set[str] = set()

while True:
    page = client.lists.tweets(LIST_ID, count=20)
    fresh = [t for t in page["data"] if not t.get("unavailable") and t["id"] not in seen]
    for t in sorted(fresh, key=lambda t: int(t["id"])):
        author = t["author"]["username"] if t["author"] else "?"
        print(t["createdAt"], author, t["text"][:140], t["url"])
    seen.update(t["id"] for t in fresh)  # keep the newest few thousand in production
    time.sleep(15)
```

## Collect a search window to JSON Lines

With the simple format every page carries `nextCursor` and `hasNextPage`, so the loop has a clear end. The SDK's `paginate` follows the cursor for you and retries 429 and 5xx with backoff.

```python
import json

from twittrapi import TwittrAPI

client = TwittrAPI(response_format="simple", max_retries=5)  # key from TWITTRAPI_API_KEY

def collect(query: str, day: str, next_day: str, path: str) -> int:
    """Write every tweet of one day to JSON Lines. Returns the pages fetched (= calls)."""
    pages = 0
    with open(path, "w", encoding="utf-8") as out:
        for page in client.paginate(
            client.search.tweets,
            f"{query} since:{day} until:{next_day}",
            category="Latest",
            count=100,
            max_pages=None,  # follow nextCursor until hasNextPage is false
        ):
            pages += 1
            for t in page["data"]:
                if t.get("unavailable"):
                    continue
                row = {"id": t["id"], "createdAt": t["createdAt"], "lang": t["lang"],
                       "authorId": (t["author"] or {}).get("id"), "text": t["text"]}
                out.write(json.dumps(row, ensure_ascii=False) + "\n")
    return pages
```

- `count` is a hint (1 to 100); X decides the real page size.
- Cursors are X's and can expire after a while. If a run can be interrupted, store the last cursor with your progress and resume soon.

### Going faster: split the time range

Pagination is sequential: page 2 needs the cursor from page 1. At roughly a second per response, one cursor chain gives about one page per second, however high your rate limit is. Split the range into windows (a day, an hour) and run the windows in parallel:

```python
from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta

QUERY = "(climate OR klima) lang:de"
days = [date(2026, 9, 1) + timedelta(d) for d in range(30)]

def run(day: date) -> None:
    path = f"climate_de_{day}.jsonl"
    pages = collect(QUERY, str(day), str(day + timedelta(1)), path)
    print(f"{day}  {pages} pages  → {path}")

with ThreadPoolExecutor(max_workers=25) as pool:  # stays under the included rate limit
    list(pool.map(run, days))
```

Keep the number of workers below your [rate limit](https://www.twittrapi.com/docs/helpful-infos/rate-limits) in requests per second. Higher limits are monthly add-ons you can book for the month of a crawl.