macrocosmos 3.1.0

The official Python SDK for Macrocosmos

Macrocosmos Python SDK

The official Python SDK for Macrocosmos.

Installation

Using pip

pip install macrocosmos

Using uv

uv add macrocosmos

Usage

For a comprehensive overview of available functionality and integration patterns, refer to the Macrocosmos SDK guide.

SN13 OnDemandAPI

SN13 is focused on large-scale data collection. With the OnDemandAPI, you can run precise, real-time queries against platforms like X (Twitter), Reddit and YouTube.

As of the latest data-universe release:

• Users may select two post-filtering modes via the keyword_mode parameter:
  • "any": Returns posts that contain any combination of the listed keywords.
  • "all": Returns posts that contain all of the keywords (default).
• For Reddit requests, the first keyword in the list corresponds to the requested subreddit, and subsequent keywords are treated as normal.
• For YouTube requests, only one of the following should be applied: One username (corresponding to YouTube channel name) or one keyword (corresponding to one YouTube video URL).
• URL mode is mutually exclusive with usernames and keywords fields. If url is provided, usernames and keywords must be empty or omitted.

Use the synchronous Sn13Client to query historical or current data based on users, keywords, and time range.

Query Example

import macrocosmos as mc

client = mc.Sn13Client(api_key="<your-api-key>", app_name="my_app")

response = client.sn13.OnDemandData(
    source='X',                 # or 'Reddit'
    usernames=["@nasa"],        # Optional, up to 5 users
    keywords=["galaxy"],        # Optional, up to 5 keywords
    start_date='2025-04-15',    # Defaults to 24h range if not specified
    end_date='2025-05-15',      # Defaults to current time if not specified
    limit=1000,                 # Optional, up to 1000 results
    keyword_mode='any'          # Optional, "any" or "all"
)

print(response)

Gravity

Gravity is a decentralized data collection platform powered by Subnet 13 (Data Universe) on the Bittensor network. You can read more about this subnet on the Macrocosmos Data Universe page.

Use the synchronous GravityClient or asynchronous AsyncGravityClient for creating and monitoring data collection tasks. See the examples/gravity_workflow_example.py for a complete working example of a data collection CLI you can use for your next big project or to plug right into your favorite data product.

Creating a Gravity Task for Data Collection

Gravity tasks will immediately be registered on the network for miners to start working on your job. The job will stay registered for 7 days. After which, it will automatically generate a dataset of the data that was collected and an email will be sent to the email address you specify.

import macrocosmos as mc

client = mc.GravityClient(api_key="<your-api-key>", app_name="my_app")

gravity_tasks = [
    {"topic": "#ai", "platform": "x"},
    {"topic": "r/MachineLearning", "platform": "reddit"},
]

response =  client.gravity.CreateGravityTask(
    gravity_tasks=gravity_tasks, name="My First Gravity Task"
)

# Print the gravity task ID
print(response)

Get the status of a Gravity Task and its Crawlers

If you wish to get further information about the crawlers, you can use the include_crawlers flag or make separate GetCrawler() calls since returning in bulk can be slow.

import macrocosmos as mc

client = mc.GravityClient(api_key="<your-api-key>", app_name="my_app")

response = client.gravity.GetGravityTasks(gravity_task_id="<your-gravity-task-id>", include_crawlers=False)

# Print the details about the gravity task and crawler IDs
print(response)

Build Dataset

If you do not want to wait 7-days for your data, you can request it earlier. Add a notification to get notified when the build is complete or you can monitor the status by calling GetDataset(). Once the dataset is built, the gravity task will be de-registered. Calling CancelDataset() will cancel a build in-progress or, if it's already complete, will purge the created dataset.

import macrocosmos as mc

client = mc.GravityClient(api_key="<your-api-key>", app_name="my_app")

response = client.gravity.BuildDataset(
    crawler_id="<your-crawler-id>"
)

# Print the dataset ID
print(response)

Build All Datasets

If a Gravity task was launched with multiple crawlers, you can build multiple datasets simultaneously with this function call. Simply supply the the maximum rows to retrieve from each crawler's dataset via build_crawlers_config (you can obtain the crawler IDs via the GetGravityTasks call above). Example below for a two-crawler gravity task:

import macrocosmos as mc

client = mc.GravityClient(api_key="<your-api-key>", app_name="my_app")

response = client.gravity.BuildAllDatasets(
    gravity_task_id="your-gravity-task-id",
    build_crawlers_config=[
        {
            "crawler_id": f"crawler-0-your-gravity-task-id",
            "max_rows": 100,
        },
        {
            "crawler_id": f"crawler-1-your-gravity-task-id",
            "max_rows": 200,
        },
        ],
    )

# Prints the gravity task ID for the datasets built
print(response)