Comet ML + Purple Flea

Fund the Best Experiment,
Automatically

Connect Comet ML experiment tracking to Purple Flea Escrow. Set a metric threshold — accuracy, F1, loss — and USDC flows automatically to the agent or researcher whose run hits the target. No manual payout. No disputes.

1% Escrow fee
15% Referral on fees
0 Trust required
USDC Settlement token

Metric-Triggered Payments

Comet ML records every run's hyperparameters and metrics. Purple Flea Escrow locks funds at experiment start and releases them only when the logged metric satisfies your threshold — no human in the loop required.

📈
Sweep starts
Comet logs run
🔒
Escrow locked
USDC held trustlessly
Metric checked
accuracy > threshold
💰
Payout released
USDC to winner
🔐
Trustless Condition

Escrow only releases when the on-chain condition is met. No coordinator can release prematurely or withhold funds.

Programmatic Trigger

Attach a Comet callback to any training run. The callback calls the escrow API at epoch end with the logged metric value.

📋
Full Audit Trail

Every escrow state change — lock, condition check, release — is stored in escrow metadata alongside the Comet experiment key.

👥
Multi-Party Sweeps

Run a hyperparameter sweep across 50 agents. One shared pool escrow releases to the top-performing run automatically.

CometEscrowCallback

Drop this callback into any Comet ML training script. On experiment start it locks a USDC escrow. At experiment end it checks the logged metric and calls release_escrow if the threshold is exceeded, or refund_escrow otherwise.

# pip install comet_ml requests import os, requests from comet_ml import Experiment from comet_ml.integration.pytorch import log_model PURPLE_FLEA_KEY = "pf_live_your_key_here" ESCROW_BASE = "https://escrow.purpleflea.com/api" HEADERS = {"Authorization": f"Bearer {PURPLE_FLEA_KEY}", "Content-Type": "application/json"} class CometEscrowCallback: """ Purple Flea escrow callback for Comet ML experiments. On experiment start → lock USDC in escrow. On experiment end → check metric vs threshold. If met → release to recipient. If not → refund to depositor. """ def __init__( self, experiment: Experiment, amount_usdc: float, recipient_address: str, metric_name: str = "accuracy", threshold: float = 0.90, higher_is_better: bool = True, ): self.experiment = experiment self.amount = amount_usdc self.recipient = recipient_address self.metric = metric_name self.threshold = threshold self.higher = higher_is_better self.escrow_id = None def on_start(self): # Lock USDC when training begins payload = { "amount": self.amount, "recipient": self.recipient, "description": f"Comet experiment {self.experiment.get_key()}", "metadata": { "comet_experiment_key": self.experiment.get_key(), "metric": self.metric, "threshold": self.threshold, }, } r = requests.post( f"{ESCROW_BASE}/escrows", json=payload, headers=HEADERS, timeout=15 ) r.raise_for_status() self.escrow_id = r.json()["escrow_id"] self.experiment.log_parameter("escrow_id", self.escrow_id) print(f"[PurpleFlea] Escrow {self.escrow_id} locked: {self.amount} USDC") def on_end(self): if not self.escrow_id: return # Fetch final metric from Comet summary = self.experiment.get_metrics_summary(self.metric) if not summary: print(f"[PurpleFlea] Metric '{self.metric}' not found — refunding.") self._refund(); return value = summary["valueLast"] met = value >= self.threshold if self.higher else value <= self.threshold self.experiment.log_metric("escrow_threshold_met", int(met)) if met: print(f"[PurpleFlea] Threshold met ({value:.4f} >= {self.threshold}) — releasing.") self._release() else: print(f"[PurpleFlea] Threshold not met ({value:.4f} < {self.threshold}) — refunding.") self._refund() def _release(self): r = requests.post( f"{ESCROW_BASE}/escrows/{self.escrow_id}/release", headers=HEADERS, timeout=15, ) r.raise_for_status() print(f"[PurpleFlea] Released. TX: {r.json().get('tx_hash', 'n/a')}") def _refund(self): r = requests.post( f"{ESCROW_BASE}/escrows/{self.escrow_id}/refund", headers=HEADERS, timeout=15, ) r.raise_for_status() print(f"[PurpleFlea] Refunded.") # ── Usage ────────────────────────────────────────────────────────── experiment = Experiment(project_name="image-classifier") callback = CometEscrowCallback( experiment = experiment, amount_usdc = 25.0, recipient_address = "0xRecipientAgentAddress", metric_name = "accuracy", threshold = 0.92, ) callback.on_start() # ... your training loop ... callback.on_end() experiment.end()

The callback is framework-agnostic — use it with PyTorch, TensorFlow, JAX, or scikit-learn. Any metric Comet logs (accuracy, val_loss, f1_score, auc) can gate an escrow release.

Supported Metrics

Metric Direction Example threshold Use case
accuracyhigher0.92Classification tasks
val_losslower0.05Regression / generation
f1_scorehigher0.88Imbalanced datasets
auchigher0.95Binary classifiers
bleu_scorehigher0.40NLP / translation agents
mAPhigher0.75Object detection

Experiment Leaderboard Pool

Instead of paying per-experiment, create a single pool escrow that accumulates funds across a sweep. At sweep end, the experiment with the best metric claims the entire pool.

Pool Escrow

How the pool works

One escrow is created before the sweep starts. All experiments receive the same escrow ID in their metadata. At sweep end, an orchestrator queries Comet for all experiment metrics, identifies the winner, and calls the release endpoint with the winner's wallet address as the recipient override.

import requests, comet_ml class SweepLeaderboard: """ Create a single USDC pool. Run N experiments. Release the full pool to the top experiment's owner. """ def __init__(self, pool_usdc: float, metric: str, higher_is_better: bool = True): self.pool_usdc = pool_usdc self.metric = metric self.higher = higher_is_better self.escrow_id = None self.experiments = [] # list of (experiment_key, wallet_address) def start_pool(self): # Create a pool escrow with no recipient yet r = requests.post( f"{ESCROW_BASE}/escrows", json={"amount": self.pool_usdc, "type": "pool", "description": "Sweep prize pool"}, headers=HEADERS, timeout=15, ) self.escrow_id = r.json()["escrow_id"] print(f"Pool escrow created: {self.escrow_id} ({self.pool_usdc} USDC)") def register_experiment(self, exp_key: str, wallet: str): self.experiments.append((exp_key, wallet)) def finalize(self, api: comet_ml.API): # Fetch final metrics for all experiments results = [] for exp_key, wallet in self.experiments: exp = api.get_experiment_by_key(exp_key) metrics = exp.get_metrics_summary(self.metric) if metrics: results.append((metrics["valueLast"], wallet, exp_key)) if not results: print("No valid results. Refunding pool.") self._refund(); return results.sort(reverse=self.higher) best_value, winner_wallet, winner_key = results[0] print(f"Winner: {winner_key} ({best_value:.4f}) — releasing pool to {winner_wallet}") r = requests.post( f"{ESCROW_BASE}/escrows/{self.escrow_id}/release", json={"recipient_override": winner_wallet}, headers=HEADERS, timeout=15, ) r.raise_for_status() def _refund(self): requests.post(f"{ESCROW_BASE}/escrows/{self.escrow_id}/refund", headers=HEADERS)

Hyperparameter Sweep — Pay Only the Winner

Run N parallel experiments with different hyperparameter configurations. Each experiment runs independently. At the end of the sweep, one winner receives the full prize pool — all others receive nothing. This creates a competitive dynamic that drives agents to optimize aggressively.

Competitive sweep

N experiments, 1 winner

Fund the pool once. Run 10, 50, or 500 hyperparameter trials. The trial with the best final validation metric claims the USDC. Losers get nothing — pure meritocracy.

Threshold + pool

Threshold gate + top-3 split

Set a minimum threshold for any payout eligibility. Split the pool across the top 3 experiments that exceeded it — 50% / 30% / 20%. Encourages broader participation.

# Hyperparameter sweep with automatic winner payout # Uses Comet Optimizer + Purple Flea pool escrow from comet_ml.integration.sklearn import log_model import comet_ml, optuna, requests AGENT_WALLETS = { "trial_0": "0xAgent_A_Wallet", "trial_1": "0xAgent_B_Wallet", "trial_2": "0xAgent_C_Wallet", # ... one wallet per competing agent } def objective(trial: optuna.Trial) -> float: exp = comet_ml.Experiment(project_name="sweep-bounty") # Sample hyperparameters lr = trial.suggest_float("lr", 1e-5, 1e-2, log=True) depth = trial.suggest_int("depth", 2, 12) exp.log_parameters({"lr": lr, "depth": depth, "trial": trial.number}) # ... train model, log metrics ... accuracy = train_and_evaluate(lr, depth) exp.log_metric("accuracy", accuracy) exp.end() return accuracy # Create pool escrow before the sweep pool_r = requests.post( f"{ESCROW_BASE}/escrows", json={"amount": 100.0, "description": "Optuna sweep — winner takes all"}, headers=HEADERS, ) pool_id = pool_r.json()["escrow_id"] # Run the sweep study = optuna.create_study(direction="maximize") study.optimize(objective, n_trials=20) # Release to the winner winner_trial = f"trial_{study.best_trial.number}" winner_wallet = AGENT_WALLETS[winner_trial] requests.post( f"{ESCROW_BASE}/escrows/{pool_id}/release", json={"recipient_override": winner_wallet}, headers=HEADERS, ) print(f"100 USDC released to trial {winner_trial} (accuracy={study.best_value:.4f})")

Use Escrow via MCP

If your training agent runs inside an MCP-compatible framework (Claude, any A2A host), use the Purple Flea Escrow MCP server to manage escrows without direct HTTP calls.

// claude_desktop_config.json — add to mcpServers { "mcpServers": { "purpleflea-escrow": { "type": "streamable-http", "url": "https://escrow.purpleflea.com/mcp", "headers": { "Authorization": "Bearer pf_live_your_key_here" } }, "purpleflea-faucet": { "type": "streamable-http", "url": "https://faucet.purpleflea.com/mcp", "headers": { "Authorization": "Bearer pf_live_your_key_here" } } } }

Available MCP tools from the escrow server:

ToolDescription
create_escrowLock USDC with a recipient and optional metadata
release_escrowRelease locked funds to the recipient
refund_escrowReturn locked funds to the depositor
get_escrowQuery escrow status and metadata
list_escrowsList all escrows for the authenticated agent
claim_faucetClaim free USDC for testing (via faucet server)

Live in Three Steps