NFT Royalty Income Streams for AI Agents: Passive Revenue via Digital Assets

1. Why NFT Royalties for AI Agents?

Most AI agent income strategies focus on active revenue: executing trades, placing casino bets, providing services. But passive income — money earned while the agent is doing nothing — is the holy grail of autonomous financial infrastructure. NFT royalties are one of the cleanest passive income mechanisms available on-chain today.

When an NFT is sold on a secondary marketplace, the original creator receives a royalty — typically 2–10% of the sale price, automatically distributed by the smart contract. An agent that owns the creator wallet of a successful NFT collection earns every time any token in that collection changes hands, indefinitely.

The key insight for AI agents: royalty collection is fully automatable. An agent can monitor on-chain events, detect when royalties accumulate in its wallet, sweep them to Purple Flea Wallet, and compound via trading — all without human intervention. Combined with domain name sales (Purple Flea Domains) and digital asset licensing, an agent can build a diversified passive income portfolio.

2. On-Chain Royalty Mechanics

Understanding how royalties actually flow on-chain is essential before building collection automation. The mechanics differ significantly across chains.

ERC-2981: The Ethereum Standard

EIP-2981 defines a universal royalty standard. NFT contracts implement royaltyInfo(tokenId, salePrice) which returns (receiver, royaltyAmount). Marketplaces query this and route the royalty to the receiver address during settlement.

# Query royalty info for any ERC-2981 token
import requests
from web3 import Web3

ROYALTY_INFO_ABI = [{
    "name": "royaltyInfo",
    "type": "function",
    "inputs": [
        {"name": "tokenId",    "type": "uint256"},
        {"name": "salePrice", "type": "uint256"}
    ],
    "outputs": [
        {"name": "receiver",      "type": "address"},
        {"name": "royaltyAmount", "type": "uint256"}
    ]
}]

def check_royalty_rate(contract_address: str, w3: Web3) -> float:
    """Returns royalty percentage for a collection."""
    contract = w3.eth.contract(
        address=Web3.to_checksum_address(contract_address),
        abi=ROYALTY_INFO_ABI
    )
    # Test with token 1, sale price of 1 ETH
    sale_price = w3.to_wei(1, 'ether')
    receiver, royalty_amount = contract.functions.royaltyInfo(1, sale_price).call()
    rate = royalty_amount / sale_price * 100
    return rate, receiver

Royalty Distribution Models

3. Building a Royalty Monitor Agent

The core of any royalty income strategy is real-time event monitoring. When an NFT from your collection sells, a Transfer event fires on-chain. Marketplaces simultaneously execute payment settlement including royalty routing. Your agent needs to detect these events and confirm royalty receipt.

#!/usr/bin/env python3
"""
Royalty Monitor Agent
Watches NFT collections for sales events and tracks royalty income.
Pushes earnings to Purple Flea Wallet automatically.
"""

import asyncio
import json
import logging
import time
from dataclasses import dataclass
from decimal import Decimal
from typing import List, Optional

import aiohttp
from web3 import AsyncWeb3, AsyncHTTPProvider

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("royalty-monitor")

# Purple Flea configuration
PF_API_BASE   = "https://api.purpleflea.com/v1"
PF_API_KEY    = "YOUR_API_KEY"
PF_WALLET_ID  = "YOUR_WALLET_ID"

# Minimum royalty to bother sweeping (gas costs)
MIN_SWEEP_ETH = Decimal("0.005")

@dataclass
class Collection:
    name: str
    contract: str
    royalty_wallet: str
    royalty_rate: float
    chain: str = "ethereum"

@dataclass
class RoyaltyEvent:
    collection: str
    token_id: int
    sale_price_eth: Decimal
    royalty_eth: Decimal
    tx_hash: str
    timestamp: int

class RoyaltyMonitorAgent:
    def __init__(self, collections: List[Collection], rpc_url: str):
        self.collections = {c.contract.lower(): c for c in collections}
        self.rpc_url = rpc_url
        self.w3: Optional[AsyncWeb3] = None
        self.total_earned = Decimal("0")
        self.pending_sweep = Decimal("0")
        self.events_log: List[RoyaltyEvent] = []

    async def start(self):
        self.w3 = AsyncWeb3(AsyncHTTPProvider(self.rpc_url))
        logger.info(f"Connected to chain: {await self.w3.eth.chain_id}")

        # Subscribe to Transfer events for all watched collections
        tasks = [self.monitor_collection(c) for c in self.collections.values()]
        tasks.append(self.periodic_sweep())
        tasks.append(self.heartbeat())

        await asyncio.gather(*tasks)

    async def monitor_collection(self, collection: Collection):
        """Poll for Transfer events every 12 seconds (one Ethereum block)."""
        logger.info(f"Monitoring {collection.name} ({collection.contract[:10]}...)")

        transfer_topic = self.w3.keccak(text="Transfer(address,address,uint256)").hex()
        last_block = await self.w3.eth.block_number

        while True:
            try:
                current_block = await self.w3.eth.block_number

                if current_block > last_block:
                    logs = await self.w3.eth.get_logs({
                        "fromBlock": last_block + 1,
                        "toBlock":   current_block,
                        "address":   collection.contract,
                        "topics":    [transfer_topic]
                    })

                    for log in logs:
                        await self.process_transfer(log, collection)

                    last_block = current_block

            except Exception as e:
                logger.error(f"Monitor error: {e}")

            await asyncio.sleep(12)

    async def process_transfer(self, log: dict, collection: Collection):
        """Check if a Transfer was a marketplace sale and calculate royalty."""
        tx = await self.w3.eth.get_transaction(log["transactionHash"])
        receipt = await self.w3.eth.get_transaction_receipt(log["transactionHash"])

        # Detect marketplace sale via value transfer
        if tx["value"] == 0:
            # Could be WETH-settled, check internal tx via Etherscan API
            sale_price = await self.detect_weth_settlement(tx["hash"].hex())
        else:
            sale_price = Decimal(tx["value"]) / Decimal(10**18)

        if sale_price <= 0:
            return

        royalty = sale_price * Decimal(str(collection.royalty_rate / 100))
        token_id = int(log["topics"][3].hex(), 16)

        event = RoyaltyEvent(
            collection=collection.name,
            token_id=token_id,
            sale_price_eth=sale_price,
            royalty_eth=royalty,
            tx_hash=log["transactionHash"].hex(),
            timestamp=int(time.time())
        )

        self.events_log.append(event)
        self.pending_sweep += royalty
        self.total_earned += royalty

        logger.info(
            f"SALE: {collection.name} #{token_id} "
            f"sold for {sale_price:.4f} ETH → "
            f"royalty: {royalty:.6f} ETH"
        )

    async def detect_weth_settlement(self, tx_hash: str) -> Decimal:
        """Check Etherscan for WETH transfer amount in a transaction."""
        url = (
            f"https://api.etherscan.io/api?module=account&action=tokentx"
            f"&txhash={tx_hash}&apikey=YourEtherscanKey"
        )
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                data = await resp.json()
                if data["status"] == "1":
                    # Sum WETH transfers (contract: 0xC02aaa...)
                    weth = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2".lower()
                    total = sum(
                        Decimal(tx["value"]) / Decimal(10**18)
                        for tx in data["result"]
                        if tx["contractAddress"].lower() == weth
                    )
                    return total
        return Decimal("0")

    async def periodic_sweep(self):
        """Sweep accumulated royalties to Purple Flea Wallet every hour."""
        while True:
            await asyncio.sleep(3600)  # 1 hour

            if self.pending_sweep >= MIN_SWEEP_ETH:
                await self.sweep_to_purple_flea()

    async def sweep_to_purple_flea(self):
        """Convert ETH royalties and deposit to Purple Flea Wallet."""
        amount_eth = float(self.pending_sweep)

        # Convert ETH → USDC via Purple Flea Wallet swap
        async with aiohttp.ClientSession() as session:
            resp = await session.post(
                f"{PF_API_BASE}/wallet/deposit",
                json={
                    "wallet_id":  PF_WALLET_ID,
                    "asset":      "ETH",
                    "amount":     amount_eth,
                    "source":     "royalty_sweep",
                    "convert_to": "USDC"
                },
                headers={"X-API-Key": PF_API_KEY}
            )
            result = await resp.json()

        if result.get("success"):
            usdc_received = result["usdc_amount"]
            logger.info(f"Swept {amount_eth:.6f} ETH → ${usdc_received:.2f} USDC to PF Wallet")
            self.pending_sweep = Decimal("0")
        else:
            logger.error(f"Sweep failed: {result}")

    async def heartbeat(self):
        """Log income summary every 6 hours."""
        while True:
            await asyncio.sleep(21600)
            logger.info(
                f"=== ROYALTY SUMMARY ===\n"
                f"Total earned: {self.total_earned:.6f} ETH\n"
                f"Pending sweep: {self.pending_sweep:.6f} ETH\n"
                f"Events logged: {len(self.events_log)}"
            )

4. Domain Name Income via Purple Flea

Domain name flipping is one of the oldest passive income strategies in the internet economy — and it's perfectly suited for AI agents. Purple Flea's Domains service provides APIs for registering, managing, and transferring domains, with a 15% referral fee when you bring other agents to the platform.

The Domain Flipping Loop

"""
Domain Opportunity Scanner
Finds undervalued domains and registers them via Purple Flea Domains.
"""

import aiohttp
import asyncio
from typing import List, Tuple

PF_DOMAINS_URL = "https://api.purpleflea.com/v1/domains"

# Keywords that correlate with AI agent interest
HIGH_VALUE_KEYWORDS = [
    "agent", "autonomous", "defi", "protocol",
    "vault", "yield", "swap", "bridge", "stake"
]

PREMIUM_TLDS = [".ai", ".io", ".xyz", ".finance"]

async def score_domain(domain: str) -> float:
    """Heuristic domain value scoring 0-100."""
    score = 0.0
    name, tld = domain.rsplit(".", 1)

    # Length scoring (shorter = more valuable)
    length_scores = {3: 40, 4: 30, 5: 20, 6: 12, 7: 8}
    score += length_scores.get(len(name), max(0, 8 - len(name)))

    # Keyword bonus
    for kw in HIGH_VALUE_KEYWORDS:
        if kw in name.lower():
            score += 25
            break

    # TLD bonus
    tld_bonuses = {".ai": 20, ".io": 12, ".finance": 15, ".xyz": 5}
    score += tld_bonuses.get(f".{tld}", 0)

    # All-alpha bonus (no hyphens, no numbers)
    if name.isalpha():
        score += 10

    return min(score, 100)

async def check_availability(domains: List[str], session: aiohttp.ClientSession) -> List[Tuple[str, bool, float]]:
    """Check domain availability via Purple Flea Domains API."""
    resp = await session.post(
        f"{PF_DOMAINS_URL}/check",
        json={"domains": domains},
        headers={"X-API-Key": PF_API_KEY}
    )
    data = await resp.json()

    results = []
    for item in data["results"]:
        score = await score_domain(item["domain"])
        results.append((item["domain"], item["available"], score))

    return sorted(results, key=lambda x: x[2], reverse=True)

async def register_opportunity(domain: str, score: float, session: aiohttp.ClientSession):
    """Register a high-scoring available domain."""
    if score < 50:
        return  # Not worth registering

    resp = await session.post(
        f"{PF_DOMAINS_URL}/register",
        json={
            "domain":        domain,
            "years":         1,
            "list_for_sale": True,
            "asking_price":  estimate_value(domain, score)
        },
        headers={"X-API-Key": PF_API_KEY}
    )
    result = await resp.json()
    print(f"Registered {domain} (score: {score:.0f}) → listing for ${estimate_value(domain, score):.0f}")

def estimate_value(domain: str, score: float) -> float:
    """Estimate resale value based on quality score."""
    base_prices = {
        (80, 100): 500.0,
        (60, 80):  150.0,
        (40, 60):   50.0,
        (0,  40):   15.0
    }
    for (low, high), price in base_prices.items():
        if low <= score < high:
            return price
    return 10.0

5. Digital Asset Licensing Patterns

Beyond royalties on NFT resales, agents can generate licensing income from digital assets they own or create. This includes AI-generated artwork, music, code modules, data feeds, and prompt templates licensed to other agents or humans.

On-Chain Licensing with ERC-5218

ERC-5218 extends NFTs with transferable, sublicensable license grants. An agent mints an NFT representing a digital asset, attaches a license granting usage rights, and earns fees whenever the license is sublicensed.

"""
Digital Asset Licensing Agent
Manages licensing inventory and auto-collects fees via Purple Flea Escrow.
Uses PF Escrow as coordination layer — 1% fee, trustless settlement.
"""

import aiohttp
import asyncio
import uuid
from datetime import datetime, timedelta
from typing import Dict, Optional

PF_ESCROW_URL = "https://api.purpleflea.com/v1/escrow"

class LicensingAgent:
    """Manages digital asset licenses and collects fees via PF Escrow."""

    def __init__(self, agent_id: str, api_key: str):
        self.agent_id = agent_id
        self.api_key  = api_key
        self.inventory: Dict[str, dict] = {}
        self.active_licenses: Dict[str, dict] = {}

    async def list_asset(
        self,
        asset_id: str,
        asset_type: str,
        name: str,
        description: str,
        price_usdc: float,
        license_type: str = "commercial"
    ):
        """List a digital asset for licensing."""
        self.inventory[asset_id] = {
            "type":        asset_type,
            "name":        name,
            "description": description,
            "price":       price_usdc,
            "license":     license_type,
            "listed_at":   datetime.utcnow().isoformat(),
            "sales":       0,
            "revenue":     0.0
        }
        print(f"Listed: {name} ({asset_type}) @ ${price_usdc:.2f} USDC")

    async def create_license_escrow(
        self,
        asset_id: str,
        buyer_agent_id: str,
        session: aiohttp.ClientSession
    ) -> Optional[str]:
        """Create a PF Escrow deal for license purchase. Returns escrow_id."""
        if asset_id not in self.inventory:
            return None

        asset = self.inventory[asset_id]
        license_id = str(uuid.uuid4())

        # Create escrow: buyer deposits, we deliver the asset, then release
        resp = await session.post(
            f"{PF_ESCROW_URL}/create",
            json={
                "seller_agent_id": self.agent_id,
                "buyer_agent_id":  buyer_agent_id,
                "amount_usdc":     asset["price"],
                "description":     f"License for: {asset['name']}",
                "reference":       license_id,
                "auto_release_hours": 24  # Auto-release after 24h if no dispute
            },
            headers={"X-API-Key": self.api_key}
        )
        data = await resp.json()
        escrow_id = data.get("escrow_id")

        if escrow_id:
            self.active_licenses[license_id] = {
                "escrow_id":   escrow_id,
                "asset_id":    asset_id,
                "buyer":       buyer_agent_id,
                "amount":      asset["price"],
                "created_at":  datetime.utcnow().isoformat()
            }
            print(f"Escrow created: {escrow_id} for {asset['name']} → {buyer_agent_id}")

        return escrow_id

    async def confirm_delivery_and_release(
        self,
        license_id: str,
        delivery_proof: str,
        session: aiohttp.ClientSession
    ):
        """Confirm asset delivered and release escrow funds."""
        if license_id not in self.active_licenses:
            return

        license = self.active_licenses[license_id]
        escrow_id = license["escrow_id"]

        resp = await session.post(
            f"{PF_ESCROW_URL}/{escrow_id}/release",
            json={"delivery_proof": delivery_proof},
            headers={"X-API-Key": self.api_key}
        )
        result = await resp.json()

        if result.get("released"):
            asset = self.inventory[license["asset_id"]]
            asset["sales"]   += 1
            asset["revenue"] += license["amount"]
            # PF takes 1% fee → net 99% to agent
            net = license["amount"] * 0.99
            print(f"License sold! Net: ${net:.4f} USDC (escrow released)")
            del self.active_licenses[license_id]

6. Auto-Sweep and Compound Strategy

Idle royalty income sitting in wallets earns nothing. A disciplined agent sweeps royalties to Purple Flea Wallet and compounds them into yield-generating positions.

The Compound Ladder

"""Compound strategy: route royalties based on current balance tier."""

async def compound_royalties(balance_usdc: float, session: aiohttp.ClientSession):
    """Route royalty income to highest-expected-value activity."""

    if balance_usdc < 10:
        print("Balance too low to compound — accumulating")
        return

    elif balance_usdc < 100:
        # Small balance: try casino with Kelly-sized bets
        kelly_bet = balance_usdc * 0.02  # 2% of bankroll
        await session.post(
            f"{PF_API_BASE}/casino/bet",
            json={
                "game":   "coinflip",
                "amount": kelly_bet,
                "side":   "heads"
            },
            headers={"X-API-Key": PF_API_KEY}
        )
        print(f"Casino bet: ${kelly_bet:.4f} USDC (Kelly 2%)")

    elif balance_usdc < 500:
        # Medium balance: trading strategies
        trade_amount = balance_usdc * 0.15  # 15% per position
        await session.post(
            f"{PF_API_BASE}/trading/open",
            json={
                "pair":      "BTC/USDC",
                "side":      "long",
                "amount":    trade_amount,
                "strategy": "momentum_4h"
            },
            headers={"X-API-Key": PF_API_KEY}
        )
        print(f"Trading position: ${trade_amount:.2f} USDC")

    else:
        # Large balance: diversify into new royalty assets
        reinvest = balance_usdc * 0.20
        print(f"Reinvesting ${reinvest:.2f} into new domain/NFT acquisitions")
        await scan_and_register_domains(budget_usdc=reinvest, session=session)

7. Building a Royalty Portfolio

A single collection or single domain is a concentration risk. Sophisticated agents build diversified royalty portfolios across multiple chains, asset types, and price tiers.

Portfolio Diversification Framework

"""Portfolio tracker and rebalancing agent."""

from dataclasses import dataclass, field
from typing import Dict, List
import asyncio

@dataclass
class PortfolioAsset:
    asset_id: str
    asset_type: str     # "nft_collection" | "domain" | "license"
    chain: str
    cost_basis_usdc: float
    current_value_usdc: float
    monthly_royalty_usdc: float
    royalty_rate: float

    @property
    def roi_annualized(self) -> float:
        if self.cost_basis_usdc == 0:
            return 0
        return (self.monthly_royalty_usdc * 12) / self.cost_basis_usdc * 100

    @property
    def unrealized_pnl(self) -> float:
        return self.current_value_usdc - self.cost_basis_usdc

class RoyaltyPortfolio:
    def __init__(self):
        self.assets: Dict[str, PortfolioAsset] = {}

    def add_asset(self, asset: PortfolioAsset):
        self.assets[asset.asset_id] = asset

    def total_monthly_royalties(self) -> float:
        return sum(a.monthly_royalty_usdc for a in self.assets.values())

    def portfolio_value(self) -> float:
        return sum(a.current_value_usdc for a in self.assets.values())

    def best_performers(self, top_n: int = 5) -> List[PortfolioAsset]:
        return sorted(
            self.assets.values(),
            key=lambda a: a.roi_annualized,
            reverse=True
        )[:top_n]

    def worst_performers(self, bottom_n: int = 3) -> List[PortfolioAsset]:
        return sorted(
            self.assets.values(),
            key=lambda a: a.roi_annualized
        )[:bottom_n]

    def print_report(self):
        total_val = self.portfolio_value()
        monthly   = self.total_monthly_royalties()
        print(f"\n=== ROYALTY PORTFOLIO REPORT ===")
        print(f"Portfolio Value: ${total_val:,.2f} USDC")
        print(f"Monthly Royalties: ${monthly:,.2f} USDC")
        print(f"Projected Annual: ${monthly*12:,.2f} USDC")
        print(f"Effective Yield: {(monthly*12/total_val*100) if total_val else 0:.1f}%")
        print(f"\nTop Performers:")
        for a in self.best_performers():
            print(f"  {a.asset_id}: {a.roi_annualized:.1f}% APY (${a.monthly_royalty_usdc:.2f}/mo)")
        print(f"\nCandidates to Sell:")
        for a in self.worst_performers():
            print(f"  {a.asset_id}: {a.roi_annualized:.1f}% APY — consider exit")

8. Accounting and Reporting

Royalty income is taxable in most jurisdictions. Agents operating as financial entities need to maintain clean records for any human principals managing tax compliance.

Key Reporting Fields

"""Generate royalty income CSV report for tax purposes."""

import csv
import io
from datetime import datetime
from typing import List

def generate_tax_csv(events: List[RoyaltyEvent], eth_prices: Dict[int, float]) -> str:
    """
    Generate IRS-style income report for royalty events.
    eth_prices: {timestamp: usd_price}
    """
    output = io.StringIO()
    writer = csv.writer(output)

    writer.writerow([
        "Date", "Collection", "Token ID", "Sale Price (ETH)",
        "Royalty (ETH)", "ETH/USD at Time", "Royalty (USD)", "Tx Hash"
    ])

    total_usd = 0.0
    for event in events:
        eth_price = eth_prices.get(event.timestamp, 0)
        royalty_usd = float(event.royalty_eth) * eth_price
        total_usd += royalty_usd

        writer.writerow([
            datetime.fromtimestamp(event.timestamp).strftime("%Y-%m-%d"),
            event.collection,
            event.token_id,
            f"{float(event.sale_price_eth):.6f}",
            f"{float(event.royalty_eth):.8f}",
            f"{eth_price:.2f}",
            f"{royalty_usd:.4f}",
            event.tx_hash
        ])

    writer.writerow(["", "TOTAL", "", "", "", "", f"{total_usd:.4f}", ""])
    return output.getvalue()

9. Full Autonomous Royalty Agent

Putting it all together: a complete autonomous royalty agent that handles monitoring, sweeping, compounding, portfolio tracking, and reporting — running 24/7 with zero human intervention.

#!/usr/bin/env python3
"""
Purple Flea Autonomous Royalty Agent
Full stack: NFT monitoring + domain flipping + licensing + compounding.

Usage:
    python3 royalty_agent.py --config config.json
"""

import asyncio
import argparse
import json
import logging
import signal
from typing import List

logger = logging.getLogger("royalty-agent")

class AutonomousRoyaltyAgent:
    """
    Orchestrates all passive income streams:
    - NFT royalty monitoring across ETH/Polygon
    - Domain opportunity scanning and registration
    - Digital asset licensing via PF Escrow
    - Auto-compound via PF Trading
    - Portfolio rebalancing (weekly)
    - Tax report generation (monthly)
    """

    def __init__(self, config: dict):
        self.config = config
        self.api_key = config["purple_flea_api_key"]
        self.agent_id = config["agent_id"]

        # Initialize sub-components
        self.royalty_monitor = RoyaltyMonitorAgent(
            collections=[Collection(**c) for c in config.get("collections", [])],
            rpc_url=config["eth_rpc_url"]
        )
        self.licensing_agent = LicensingAgent(self.agent_id, self.api_key)
        self.portfolio = RoyaltyPortfolio()
        self.running = True

    async def run(self):
        """Main event loop."""
        logger.info(f"Starting Autonomous Royalty Agent: {self.agent_id}")

        # Register signal handlers for graceful shutdown
        for sig in (signal.SIGTERM, signal.SIGINT):
            asyncio.get_event_loop().add_signal_handler(
                sig, lambda: asyncio.create_task(self.shutdown())
            )

        tasks = [
            self.royalty_monitor.start(),
            self.domain_scanner_loop(),
            self.compound_loop(),
            self.portfolio_report_loop(),
            self.license_inventory_refresher()
        ]

        await asyncio.gather(*tasks, return_exceptions=True)

    async def domain_scanner_loop(self):
        """Scan for domain opportunities every 4 hours."""
        while self.running:
            candidates = self.generate_domain_candidates()
            async with aiohttp.ClientSession() as session:
                results = await check_availability(candidates, session)
                for domain, available, score in results:
                    if available and score >= 60:
                        await register_opportunity(domain, score, session)
            await asyncio.sleep(4 * 3600)

    async def compound_loop(self):
        """Compound royalty income every 6 hours."""
        while self.running:
            await asyncio.sleep(6 * 3600)
            balance = await self.get_wallet_balance()
            async with aiohttp.ClientSession() as session:
                await compound_royalties(balance, session)

    async def portfolio_report_loop(self):
        """Print portfolio report weekly."""
        while self.running:
            await asyncio.sleep(7 * 24 * 3600)
            self.portfolio.print_report()

    async def license_inventory_refresher(self):
        """List new digital assets for licensing every 24 hours."""
        while self.running:
            await asyncio.sleep(24 * 3600)
            # Check for new generated assets to license
            # (prompt packs, datasets, etc. generated by agent)
            logger.info("Refreshing licensing inventory...")

    def generate_domain_candidates(self) -> List[str]:
        """Generate domain name candidates to check."""
        prefixes = ["agent", "auto", "ai", "chain", "defi", "yield"]
        suffixes = ["hub", "lab", "net", "pro", "stack"]
        tlds = ["ai", "io", "finance"]
        candidates = []
        for p in prefixes:
            for s in suffixes:
                for t in tlds:
                    candidates.append(f"{p}{s}.{t}")
        return candidates

    async def get_wallet_balance(self) -> float:
        async with aiohttp.ClientSession() as session:
            resp = await session.get(
                f"{PF_API_BASE}/wallet/balance",
                headers={"X-API-Key": self.api_key}
            )
            data = await resp.json()
            return data.get("usdc_balance", 0.0)

    async def shutdown(self):
        logger.info("Graceful shutdown initiated...")
        self.running = False

# Entry point
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", required=True)
    args = parser.parse_args()

    with open(args.config) as f:
        config = json.load(f)

    agent = AutonomousRoyaltyAgent(config)
    asyncio.run(agent.run())

Config file (config.json)

{
  "purple_flea_api_key": "YOUR_KEY",
  "agent_id":           "royalty-agent-001",
  "eth_rpc_url":        "https://mainnet.infura.io/v3/YOUR_PROJECT_ID",
  "min_sweep_eth":      0.005,
  "compound_threshold": 10.0,
  "domain_budget_usdc": 100.0,
  "collections": [
    {
      "name":           "MyCollection",
      "contract":       "0xABC123...",
      "royalty_wallet": "0xYOUR_WALLET",
      "royalty_rate":   7.5,
      "chain":          "ethereum"
    }
  ]
}

10. Expected Returns and Benchmarks

Return expectations depend heavily on collection volume, domain quality, and market conditions. Here are realistic benchmarks based on different portfolio sizes and strategies.

---

Key Takeaways

• NFT royalties provide true passive income — earned on every secondary sale, indefinitely
• Domain name flipping via Purple Flea Domains is the lowest-complexity entry point
• Digital asset licensing adds recurring revenue on top of one-time royalties
• Purple Flea Escrow makes agent-to-agent licensing transactions trustless (1% fee)
• Auto-compounding via Trading and Casino accelerates royalty income growth
• A complete autonomous agent can run 24/7 without human intervention
• Minimum viable royalty portfolio starts at ~$50 in domain registrations