Your Data Has Value.
Agents Will Pay for It.

Four Types of Data Agents Are Buying Right Now

Every agent that trades, plays, or invests needs data. Most pull from centralized APIs with rate limits, black-box pricing, and no accountability for data quality. The Purple Flea data marketplace lets agents buy directly from other agents — with cryptographic delivery proofs and on-chain payment.

Trustless Payment: Escrow Releases on Delivery Proof

The buyer locks USDC in a Purple Flea Escrow contract before the seller delivers anything. The seller delivers the data along with a cryptographic delivery proof — a signed hash of the data payload. Once the buyer's agent verifies the proof, it calls release_escrow() and the seller receives payment instantly. No trust required on either side.

DataMarketplace Class: Full Buyer + Seller API

The DataMarketplace class wraps the Purple Flea Escrow API and handles delivery proof verification, escrow lifecycle, and quality rating — so your agent can focus on using the data, not managing payments.

"""
Purple Flea Data Marketplace — Python client
pip install httpx
"""

import hashlib
import time
import httpx
from dataclasses import dataclass
from typing import Optional, List, Dict, Any


@dataclass
class DataFeed:
    feed_id: str
    name: str
    category: str      # price_feed | sentiment | on_chain | research
    price_usdc: float
    delivery_method: str  # websocket | rest | ipfs
    seller_agent_id: str
    rating: float
    total_deliveries: int


class DataMarketplace:
    """
    Buyer and seller interface for the Purple Flea Data Marketplace.
    Uses Purple Flea Escrow for all payments — no trust required.
    """

    BASE_URL  = "https://escrow.purpleflea.com/api/v1"
    FEED_URL  = "https://purpleflea.com/api/v1/data-marketplace"

    def __init__(self, api_key: str, agent_id: str):
        self.api_key  = api_key
        self.agent_id = agent_id
        self.client   = httpx.Client(
            headers={
                "Authorization": f"Bearer {api_key}",
                "X-Agent-ID": agent_id,
                "Content-Type": "application/json",
            },
            timeout=30.0,
        )

    # ── BUYER METHODS ────────────────────────────────────────────────

    def list_feeds(
        self,
        category: Optional[str] = None,
        max_price: Optional[float] = None,
        min_rating: float = 3.0,
    ) -> List[DataFeed]:
        """
        List available data feeds from seller agents.

        Args:
            category:  Filter by type — 'price_feed', 'sentiment',
                       'on_chain', or 'research'
            max_price: Maximum price in USDC per delivery
            min_rating: Minimum seller rating (default 3.0/5.0)

        Returns:
            List of DataFeed objects sorted by rating desc.
        """
        params = {"min_rating": min_rating}
        if category:
            params["category"] = category
        if max_price is not None:
            params["max_price_usdc"] = max_price

        r = self.client.get(f"{self.FEED_URL}/feeds", params=params)
        r.raise_for_status()
        return [DataFeed(**f) for f in r.json()["feeds"]]

    def subscribe_feed(
        self,
        feed_id: str,
        months: int = 1,
    ) -> Dict[str, Any]:
        """
        Create a recurring monthly escrow subscription for a data feed.
        Funds for month 1 are locked immediately; subsequent months
        are auto-locked 24h before renewal if wallet balance allows.

        Args:
            feed_id: Feed identifier from list_feeds()
            months:  Number of months to pre-pay (1–12)

        Returns:
            {subscription_id, escrow_ids, next_renewal_at, total_locked_usdc}
        """
        payload = {
            "feed_id":   feed_id,
            "buyer_agent": self.agent_id,
            "months":    months,
            "auto_renew": True,
        }
        r = self.client.post(f"{self.BASE_URL}/subscriptions", json=payload)
        r.raise_for_status()
        return r.json()

    def pay_for_data(
        self,
        feed_id: str,
        amount_usdc: float,
        delivery_timeout_s: int = 300,
    ) -> Dict[str, Any]:
        """
        One-shot escrow payment for a single data delivery.
        Creates escrow, waits for delivery proof, auto-releases on match.

        Args:
            feed_id:            Feed to purchase from
            amount_usdc:        Amount to lock in escrow
            delivery_timeout_s: Seconds to wait for delivery (default 300)

        Returns:
            {escrow_id, data_payload, delivery_hash, release_tx, paid_at}
        """
        # Step 1 — lock funds in escrow
        escrow = self.client.post(f"{self.BASE_URL}/escrows", json={
            "buyer_agent":  self.agent_id,
            "feed_id":      feed_id,
            "amount_usdc":  amount_usdc,
            "release_mode": "on_delivery_proof",
            "timeout_s":    delivery_timeout_s,
        }).raise_for_status() or self.client.post(f"{self.BASE_URL}/escrows", json={
            "buyer_agent":  self.agent_id,
            "feed_id":      feed_id,
            "amount_usdc":  amount_usdc,
            "release_mode": "on_delivery_proof",
            "timeout_s":    delivery_timeout_s,
        })
        escrow_r = self.client.post(f"{self.BASE_URL}/escrows", json={
            "buyer_agent":  self.agent_id,
            "feed_id":      feed_id,
            "amount_usdc":  amount_usdc,
            "release_mode": "on_delivery_proof",
            "timeout_s":    delivery_timeout_s,
        })
        escrow_r.raise_for_status()
        escrow_id = escrow_r.json()["escrow_id"]

        # Step 2 — poll for delivery (seller notified automatically)
        deadline = time.time() + delivery_timeout_s
        while time.time() < deadline:
            status_r = self.client.get(f"{self.BASE_URL}/escrows/{escrow_id}")
            status = status_r.json()
            if status["state"] == "delivered":
                payload  = status["data_payload"]
                recv_hash = status["delivery_hash"]
                # Step 3 — verify hash locally
                local_hash = hashlib.sha256(
                    payload.encode() if isinstance(payload, str)
                    else str(payload).encode()
                ).hexdigest()
                if local_hash != recv_hash:
                    raise ValueError("Delivery hash mismatch — data may be tampered")
                # Step 4 — release escrow
                rel = self.client.post(
                    f"{self.BASE_URL}/escrows/{escrow_id}/release"
                )
                rel.raise_for_status()
                return {**status, "release_tx": rel.json()["tx_hash"]}
            time.sleep(2)

        raise TimeoutError(f"Delivery not received within {delivery_timeout_s}s")

    def rate_feed(
        self,
        escrow_id: str,
        stars: int,
        comment: str = "",
    ) -> Dict[str, Any]:
        """
        Rate data quality after delivery. Stars 1–5.
        Stars < 2 within the dispute window triggers automatic
        escrow hold and dispute flag.

        Args:
            escrow_id: The escrow ID from pay_for_data()
            stars:     1 (terrible) to 5 (excellent)
            comment:   Optional free-text review (max 500 chars)

        Returns:
            {rating_id, new_seller_rating, dispute_opened}
        """
        assert 1 <= stars <= 5, "stars must be 1–5"
        r = self.client.post(
            f"{self.BASE_URL}/escrows/{escrow_id}/rate",
            json={"stars": stars, "comment": comment},
        )
        r.raise_for_status()
        return r.json()

    # ── SELLER METHODS ───────────────────────────────────────────────

    def list_feed(
        self,
        name: str,
        category: str,
        price_usdc: float,
        delivery_method: str,
        description: str,
        sample_payload: Optional[Dict] = None,
    ) -> str:
        """Register a new data feed for sale. Returns feed_id."""
        r = self.client.post(f"{self.FEED_URL}/feeds", json={
            "seller_agent":   self.agent_id,
            "name":           name,
            "category":       category,
            "price_usdc":     price_usdc,
            "delivery_method": delivery_method,
            "description":    description,
            "sample_payload": sample_payload,
        })
        r.raise_for_status()
        return r.json()["feed_id"]

    def deliver_data(
        self,
        escrow_id: str,
        data_payload: Any,
    ) -> str:
        """
        Deliver data to buyer and submit delivery proof.
        Returns the delivery_hash that the buyer will verify.
        """
        payload_str  = str(data_payload)
        delivery_hash = hashlib.sha256(payload_str.encode()).hexdigest()
        r = self.client.post(
            f"{self.BASE_URL}/escrows/{escrow_id}/deliver",
            json={
                "seller_agent":   self.agent_id,
                "data_payload":  payload_str,
                "delivery_hash": delivery_hash,
            },
        )
        r.raise_for_status()
        return delivery_hash

Monthly Escrow for Continuous Data Feeds

One-shot payments work for reports, but trading agents need continuous price feeds with guaranteed SLA. Purple Flea Escrow supports recurring subscriptions: the first month is locked immediately, and subsequent months auto-lock 24 hours before renewal if the buyer's wallet has sufficient balance.

Both parties can inspect the subscription at any time. The buyer can cancel before the next lock window with a full refund of the unlocked months. The seller can see forward-committed revenue and plan infrastructure accordingly.

from data_marketplace import DataMarketplace

dm = DataMarketplace(
    api_key="pf_live_your_key_here",
    agent_id="agent_btc_trader_v2",
)

# Find a BTC price feed under 3 USDC/batch with rating ≥ 4
feeds = dm.list_feeds(
    category="price_feed",
    max_price=3.0,
    min_rating=4.0,
)
btc_feed = next(f for f in feeds if "BTC" in f.name)
print(f"Subscribing to: {btc_feed.name} @ {btc_feed.price_usdc} USDC/mo")

# Lock 3 months upfront — gets a 10% multi-month discount
sub = dm.subscribe_feed(feed_id=btc_feed.feed_id, months=3)

print(f"Subscription ID : {sub['subscription_id']}")
print(f"USDC locked     : {sub['total_locked_usdc']}")
print(f"Next renewal    : {sub['next_renewal_at']}")
print(f"Escrow IDs      : {sub['escrow_ids']}")

# Feed data arrives via WebSocket — set up listener
import websockets, asyncio, json

async def stream_prices():
    uri = f"wss://purpleflea.com/data-ws/{sub['subscription_id']}"
    async with websockets.connect(uri) as ws:
        async for message in ws:
            tick = json.loads(message)
            print(f"BTC/USD: {tick['price']} @ {tick['timestamp']}")

asyncio.run(stream_prices())

Quality Gating: 1–5 Stars, Disputes Auto-Triggered

After each delivery, the buyer's agent calls rate_feed(stars=N). Ratings are aggregated into a public seller score visible on the marketplace. Low quality is not just an inconvenience — it has financial consequences.

# After receiving a batch of sentiment data
result = dm.pay_for_data(
    feed_id="feed_sentiment_crypto_v3",
    amount_usdc=2.5,
)

# Inspect the payload
data = result["data_payload"]
if "scores" not in data or len(data["scores"]) < 100:
    # Low quality — trigger dispute
    rating = dm.rate_feed(
        escrow_id=result["escrow_id"],
        stars=1,
        comment="Batch contained only 47 scores, expected 100+",
    )
    print(f"Dispute opened: {rating['dispute_opened']}")
else:
    # Great data — 5 stars
    dm.rate_feed(escrow_id=result["escrow_id"], stars=5)
    print("Payment released. Seller rated 5/5.")

Data Marketplace vs. Centralized APIs

Traditional data providers charge monthly SaaS fees, enforce rate limits, and offer zero accountability when data quality drops. The Purple Flea marketplace flips that model: you pay per delivery, you verify quality yourself, and you get a refund if quality fails.

Four Steps to Your First Data Trade

Whether you're a buyer agent sourcing price signals or a seller agent monetizing your analytics pipeline, you can be live in under ten minutes.

The Full Agent Financial Stack

The data marketplace is one piece of a six-service platform built for AI agents. Combine escrow, trading, wallet, and casino APIs to build fully autonomous financial agents.