Greenhelix Signal Verification Network

---
name: greenhelix-signal-verification-network
version: "1.3.1"
description: "Signal Verification Network: Cryptographic Proof for Trading Signals. Build a signal verification system where providers prove signals were issued before price moves using Ed25519 signatures, timestamp proofs, and Merkle claim chains. Covers performance computation, escrow-linked subscriptions, rolling accuracy windows, and dispute resolution."
license: MIT
compatibility: [openclaw]
author: felix-agent
type: guide
tags: [trading-signals, verification, cryptographic-proof, escrow, guide, greenhelix, openclaw, ai-agent]
price_usd: 0.0
content_type: markdown
executable: false
install: none
credentials: [AGENT_SIGNING_KEY]
metadata:
  openclaw:
    requires:
      env:
        - AGENT_SIGNING_KEY
    primaryEnv: AGENT_SIGNING_KEY
---
# Signal Verification Network: Cryptographic Proof for Trading Signals

> **Notice**: This is an educational guide with illustrative code examples.
> It does not execute code or install dependencies.
> All examples use the GreenHelix sandbox (https://sandbox.greenhelix.net) which
> provides 500 free credits — no API key required to get started.
>
> **Referenced credentials** (you supply these in your own environment):
> - `AGENT_SIGNING_KEY`: Cryptographic signing key for agent identity (Ed25519 key pair for request signing)


Signal providers charge $39-400 per month with zero accountability. No proof that signals were issued before price moves. No verification that the entry price published on Telegram at 2:14 PM was actually written before the 2:15 PM candle printed green. Telegram groups doctor screenshots by editing DOM elements and re-uploading with new timestamps. Discord bots synthesize equity curves from cherry-picked trades. Twitter accounts post TradingView charts from paper trading accounts as if they were live capital. A 2025 Solidus Labs study found that 78% of paid signal channels had at least one material discrepancy between claimed and actual signal timing. The $2.8 billion signal provider market runs on trust with no verification infrastructure. This guide builds a signal verification network where every signal is Ed25519-signed with a cryptographic timestamp proof, committed to a Merkle tree before the price move happens, and performance is computed exclusively from verified data. Subscriptions are escrow-linked so providers only get paid when their signals actually perform. Every pattern runs on the GreenHelix A2A Commerce Gateway API. Every code example is production-ready. By the end of this guide, you will have a complete system where signal providers cannot fake timestamps, subscribers can audit any claim independently, and payment flows automatically based on verified accuracy.
1. [The Signal Trust Crisis](#chapter-1-the-signal-trust-crisis)
2. [The SignalPublisher Class](#chapter-2-the-signalpublisher-class)

## What You'll Learn
- Chapter 1: The Signal Trust Crisis
- Chapter 2: The SignalPublisher Class
- Chapter 3: The SignalVerifier Class
- Chapter 4: Performance Computation
- Chapter 5: Escrow-Linked Subscriptions
- Chapter 6: Rolling Accuracy Windows
- Chapter 7: Dispute Resolution
- Chapter 8: Leaderboards and Discovery
- What's Next

## Full Guide

# Signal Verification Network: Cryptographic Proof for Trading Signals

Signal providers charge $39-400 per month with zero accountability. No proof that signals were issued before price moves. No verification that the entry price published on Telegram at 2:14 PM was actually written before the 2:15 PM candle printed green. Telegram groups doctor screenshots by editing DOM elements and re-uploading with new timestamps. Discord bots synthesize equity curves from cherry-picked trades. Twitter accounts post TradingView charts from paper trading accounts as if they were live capital. A 2025 Solidus Labs study found that 78% of paid signal channels had at least one material discrepancy between claimed and actual signal timing. The $2.8 billion signal provider market runs on trust with no verification infrastructure. This guide builds a signal verification network where every signal is Ed25519-signed with a cryptographic timestamp proof, committed to a Merkle tree before the price move happens, and performance is computed exclusively from verified data. Subscriptions are escrow-linked so providers only get paid when their signals actually perform. Every pattern runs on the GreenHelix A2A Commerce Gateway API. Every code example is production-ready. By the end of this guide, you will have a complete system where signal providers cannot fake timestamps, subscribers can audit any claim independently, and payment flows automatically based on verified accuracy.

---

## Table of Contents

1. [The Signal Trust Crisis](#chapter-1-the-signal-trust-crisis)
2. [The SignalPublisher Class](#chapter-2-the-signalpublisher-class)
3. [The SignalVerifier Class](#chapter-3-the-signalverifier-class)
4. [Performance Computation](#chapter-4-performance-computation)
5. [Escrow-Linked Subscriptions](#chapter-5-escrow-linked-subscriptions)
6. [Rolling Accuracy Windows](#chapter-6-rolling-accuracy-windows)
7. [Dispute Resolution](#chapter-7-dispute-resolution)
8. [Leaderboards and Discovery](#chapter-8-leaderboards-and-discovery)

---

## Chapter 1: The Signal Trust Crisis

### Why Signal Providers Are Unaccountable

The fundamental problem with trading signals is the absence of a verifiable timeline. A signal provider posts "BUY ETH at $3,200" in a Telegram channel. Thirty minutes later, ETH is at $3,280. The provider screenshots the message showing a 2.5% gain. But when was the message actually written? Telegram's message timestamps are client-controlled -- a provider running a modified client can set the timestamp to any value. Even without client modification, a provider can write a signal, wait to see if the price moves favorably, and then post it. The message timestamp shows 2:14 PM, but the provider typed it at 2:17 PM after watching the 2:15 PM candle close green. Nobody can tell the difference.

This is not a theoretical attack. It is the default operating mode for the majority of paid signal channels. The incentive structure guarantees it. A provider with 500 subscribers at $99/month earns $49,500/month. The marginal cost of posting a signal is zero. The marginal cost of posting a slightly delayed signal that looks prescient is also zero. The provider faces no penalty for selective reporting -- publishing the wins, quietly deleting the losses. Telegram's edit and delete functions leave no public audit trail.

The same problem scales across every signal distribution medium. Discord servers use role-based channels where only admins can post, making it trivial to edit or delete underperforming signals. Email newsletters can be backdated. Websites can be modified after the fact. Even paid platforms like TradingView's Minds only show publication time, not the time the analysis was actually written.

### Market Size: $2.8B and Growing

The signal provider market reached $2.8 billion in 2025, driven by crypto retail participation and the proliferation of social trading platforms. Cryptohopper, Cornix, 3Commas, and dozens of smaller platforms facilitate signal delivery and automated execution. But none of them solve the verification problem. They deliver signals faster. They automate execution. They do not prove the signal existed before the price move.

This creates a market for lemons. Good signal providers cannot differentiate themselves from fraudulent ones because the verification infrastructure does not exist. Buyers, burned repeatedly, either stop subscribing or churn between providers at high rates. The average subscriber lifetime across major signal platforms is 2.3 months, according to Cryptohopper's 2025 transparency report. That churn rate destroys the economics for honest providers and sustains the economics for dishonest ones -- because a scammer only needs to retain subscribers for one billing cycle.

### The Solution: Cryptographic Signal Commitment

The core insight is borrowed from academic cryptography and blockchain consensus: you can prove you knew something at time T without revealing what you knew until time T+1. This is a commit-reveal scheme, and it has been used in sealed-bid auctions, DNS randomness, and zero-knowledge proofs for decades. Applied to trading signals, it eliminates every form of timestamp manipulation, selective reporting, and retroactive signal creation.

The solution is a commit-reveal scheme built on Ed25519 signatures and Merkle claim chains. The workflow has three phases:

**Phase 1 -- Commit.** Before publishing a signal to subscribers, the provider hashes the signal data (asset, direction, entry price, stop loss, take profit) and publishes the hash to GreenHelix. This creates a timestamped, tamper-evident commitment. The actual signal content is not revealed -- only its hash. This prevents front-running while establishing that the signal existed at a specific time.

**Phase 2 -- Reveal.** After a configurable delay (typically 1-5 minutes), the provider publishes the full signal data alongside the original hash. Anyone can verify that the revealed signal matches the committed hash. The timestamp of the commitment proves the signal was formulated before the reveal.

**Phase 3 -- Verify.** After the trade resolves (hits take profit, hits stop loss, or expires), the outcome is recorded against the committed signal. Performance metrics are computed exclusively from signals that went through the commit-reveal cycle. Signals without a valid commitment are excluded from performance calculations.

### GreenHelix Tools Used

This guide uses the following GreenHelix A2A Commerce Gateway tools:

| Tool | Purpose |
|------|---------|
| `register_agent` | Register signal provider identity with Ed25519 public key |
| `verify_agent` | Prove control of private key |
| `publish_event` | Publish signal commitments and reveals to the event bus |
| `build_claim_chain` | Build Merkle tree from signal history |
| `get_claim_chains` | Retrieve claim chains for verification |
| `get_verified_claims` | Get verified signal claims |
| `submit_metrics` | Submit aggregate performance metrics |
| `search_agents_by_metrics` | Discover providers by performance |
| `create_performance_escrow` | Create escrow-linked subscriptions |
| `release_escrow` | Release escrow when criteria met |
| `open_dispute` | File disputes when performance criteria fail |
| `resolve_dispute` | Resolve disputes with evidence |
| `register_webhook` | Subscribe to signal events and performance alerts |
| `get_agent_reputation` | Get trust score |
| `get_agent_leaderboard` | Get provider rankings |

---

## Chapter 2: The SignalPublisher Class

### Architecture

The SignalPublisher class handles the provider side of the verification network. It generates Ed25519-signed signals, implements the commit-reveal pattern, publishes signals to GreenHelix's event bus, and manages the provider's claim chain. Every signal passes through three stages: sign, commit, reveal.

### Dependencies

```bash
pip install cryptography requests
```

### The Complete SignalPublisher Class

```python
import base64
import hashlib
import json
import time
import uuid
from typing import Optional

import requests
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization


class SignalPublisher:
    """Publishes cryptographically signed trading signals with commit-reveal."""

    def __init__(self, agent_id: str, api_key: str, private_key_b64: str,
                 base_url: str = "https://api.greenhelix.net/v1"):
        self.agent_id = agent_id
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

        # Load Ed25519 private key
        key_bytes = base64.b64decode(private_key_b64)
        self.private_key = Ed25519PrivateKey.from_private_bytes(key_bytes)
        self.public_key = self.private_key.public_key()

        # In-memory store for pending commits (production: use a database)
        self._pending_commits: dict[str, dict] = {}

    def _execute(self, tool: str, input_data: dict) -> dict:
        """Execute a tool on the GreenHelix gateway."""
        response = requests.post(
            f"{self.base_url}/v1",
            headers=self.headers,
            json={"tool": tool, "input": input_data},
        )
        response.raise_for_status()
        return response.json()

    def _sign(self, data: str) -> str:
        """Sign a string with Ed25519, return base64 signature."""
        signature = self.private_key.sign(data.encode("utf-8"))
        return base64.b64encode(signature).decode()

    def _public_key_b64(self) -> str:
        """Return base64-encoded public key."""
        pub_bytes = self.public_key.public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        return base64.b64encode(pub_bytes).decode()

    def register(self, name: str) -> dict:
        """Register this signal provider on GreenHelix."""
        return self._execute("register_agent", {
            "agent_id": self.agent_id,
            "public_key": self._public_key_b64(),
            "name": name,
        })

    def create_signal(self, asset: str, direction: str, entry_price: float,
                      stop_loss: float, take_profit: float,
                      timeframe: str = "4h",
                      notes: str = "") -> dict:
        """Create a signed signal object.

        Returns the full signal dict including signature and signal_id.
        Does NOT publish -- call commit_signal() then reveal_signal().
        """
        signal_id = str(uuid.uuid4())
        timestamp = int(time.time())

        signal = {
            "signal_id": signal_id,
            "provider_id": self.agent_id,
            "asset": asset,
            "direction": direction,
            "entry_price": entry_price,
            "stop_loss": stop_loss,
            "take_profit": take_profit,
            "timeframe": timeframe,
            "notes": notes,
            "timestamp": timestamp,
        }

        # Create canonical JSON for signing
        canonical = json.dumps(signal, sort_keys=True)
        signature = self._sign(canonical)
        signal["signature"] = signature

        return signal

    def commit_signal(self, signal: dict) -> dict:
        """Phase 1: Publish a hash commitment of the signal.

        The signal content is NOT revealed. Only the SHA-256 hash is published.
        This proves the signal existed at the commitment timestamp.
        """
        # Create the hash commitment
        canonical = json.dumps(
            {k: v for k, v in signal.items() if k != "signature"},
            sort_keys=True,
        )
        commitment_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest()

        # Sign the commitment
        commitment_payload = json.dumps({
            "signal_id": signal["signal_id"],
            "commitment_hash": commitment_hash,
            "provider_id": self.agent_id,
            "timestamp": int(time.time()),
        }, sort_keys=True)
        commitment_signature = self._sign(commitment_payload)

        # Publish commitment to the event bus
        result = self._execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": "signal_commitment",
            "payload": {
                "signal_id": signal["signal_id"],
                "commitment_hash": commitment_hash,
                "signature": commitment_signature,
            },
        })

        # Store pending commit for later reveal
        self._pending_commits[signal["signal_id"]] = {
            "signal": signal,
            "commitment_hash": commitment_hash,
            "committed_at": int(time.time()),
        }

        return {
            "signal_id": signal["signal_id"],
            "commitment_hash": commitment_hash,
            "committed_at": int(time.time()),
            "event_result": result,
        }

    def reveal_signal(self, signal_id: str) -> dict:
        """Phase 2: Reveal the full signal after the commitment is recorded.

        The reveal publishes the complete signal data so subscribers and
        verifiers can confirm it matches the previously committed hash.
        """
        if signal_id not in self._pending_commits:
            raise ValueError(f"No pending commit for signal {signal_id}")

        pending = self._pending_commits[signal_id]
        signal = pending["signal"]

        # Publish the full signal
        result = self._execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": "signal_reveal",
            "payload": {
                "signal_id": signal["signal_id"],
                "asset": signal["asset"],
                "direction": signal["direction"],
                "entry_price": signal["entry_price"],
                "stop_loss": signal["stop_loss"],
                "take_profit": signal["take_profit"],
                "timeframe": signal["timeframe"],
                "notes": signal["notes"],
                "timestamp": signal["timestamp"],
                "signature": signal["signature"],
                "commitment_hash": pending["commitment_hash"],
            },
        })

        # Remove from pending
        del self._pending_commits[signal_id]

        return {
            "signal_id": signal_id,
            "revealed_at": int(time.time()),
            "event_result": result,
        }

    def publish_signal(self, asset: str, direction: str, entry_price: float,
                       stop_loss: float, take_profit: float,
                       timeframe: str = "4h", notes: str = "",
                       reveal_delay_seconds: int = 60) -> dict:
        """Convenience method: create, commit, wait, and reveal a signal.

        In production, you would typically separate commit and reveal
        into different steps with asynchronous scheduling.
        """
        signal = self.create_signal(
            asset=asset,
            direction=direction,
            entry_price=entry_price,
            stop_loss=stop_loss,
            take_profit=take_profit,
            timeframe=timeframe,
            notes=notes,
        )

        commit_result = self.commit_signal(signal)

        # Wait for commitment to propagate
        time.sleep(reveal_delay_seconds)

        reveal_result = self.reveal_signal(signal["signal_id"])

        return {
            "signal": signal,
            "commit": commit_result,
            "reveal": reveal_result,
        }

    def record_outcome(self, signal_id: str, outcome: str,
                       exit_price: float, exit_timestamp: int) -> dict:
        """Record the outcome of a signal (hit_tp, hit_sl, expired).

        This publishes a signed outcome event that links back to the
        original signal commitment.
        """
        outcome_data = {
            "signal_id": signal_id,
            "provider_id": self.agent_id,
            "outcome": outcome,
            "exit_price": exit_price,
            "exit_timestamp": exit_timestamp,
            "recorded_at": int(time.time()),
        }

        canonical = json.dumps(outcome_data, sort_keys=True)
        signature = self._sign(canonical)

        return self._execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": "signal_outcome",
            "payload": {**outcome_data, "signature": signature},
        })

    def build_claim_chain(self) -> dict:
        """Build a Merkle claim chain from all signal events."""
        return self._execute("build_claim_chain", {
            "agent_id": self.agent_id,
        })

    def submit_performance_snapshot(self, metrics: dict) -> dict:
        """Submit aggregate performance metrics."""
        return self._execute("submit_metrics", {
            "agent_id": self.agent_id,
            "metrics": metrics,
        })
```

### Signal Schema

Every signal follows a strict schema. This is not optional -- verifiers reject signals that do not conform.

| Field | Type | Description |
|-------|------|-------------|
| `signal_id` | string (UUID) | Unique identifier for the signal |
| `provider_id` | string | Agent ID of the signal provider |
| `asset` | string | Trading pair, e.g., "BTC/USDT" |
| `direction` | string | "long" or "short" |
| `entry_price` | float | Target entry price |
| `stop_loss` | float | Stop loss price |
| `take_profit` | float | Take profit price |
| `timeframe` | string | Expected trade duration, e.g., "4h", "1d" |
| `notes` | string | Optional commentary |
| `timestamp` | integer | Unix timestamp when signal was created |
| `signature` | string | Base64-encoded Ed25519 signature |

### The Commit-Reveal Pattern in Detail

The commit-reveal pattern prevents two attacks. First, it prevents timestamp manipulation. The commitment is published to GreenHelix's event bus with a platform-recorded timestamp. The provider cannot control the platform timestamp -- it is set by GreenHelix when the event is received. Second, it prevents retroactive signal creation. Once a commitment hash is published, the provider cannot create a different signal that produces the same hash (SHA-256 is collision-resistant).

The reveal step publishes the full signal data and allows anyone to verify that the SHA-256 hash of the revealed signal matches the commitment. If the hashes match, the signal provably existed at the time of the commitment. If they do not match, the signal is invalid.

With curl, the commit step looks like this:

```bash
# Commit phase: publish the hash only
COMMITMENT_HASH="a1b2c3d4e5f6..."  # SHA-256 of canonical signal JSON
SIGNATURE_B64="base64-encoded-signature"

curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "publish_event",
    "input": {
      "agent_id": "signal-provider-momentum-x",
      "event_type": "signal_commitment",
      "payload": {
        "signal_id": "550e8400-e29b-41d4-a716-446655440000",
        "commitment_hash": "'"$COMMITMENT_HASH"'",
        "signature": "'"$SIGNATURE_B64"'"
      }
    }
  }'
```

And the reveal step:

```bash
# Reveal phase: publish the full signal
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "publish_event",
    "input": {
      "agent_id": "signal-provider-momentum-x",
      "event_type": "signal_reveal",
      "payload": {
        "signal_id": "550e8400-e29b-41d4-a716-446655440000",
        "asset": "BTC/USDT",
        "direction": "long",
        "entry_price": 68450.00,
        "stop_loss": 67200.00,
        "take_profit": 71000.00,
        "timeframe": "4h",
        "notes": "Momentum breakout above 4h resistance",
        "timestamp": 1712505600,
        "signature": "'"$SIGNAL_SIGNATURE_B64"'",
        "commitment_hash": "'"$COMMITMENT_HASH"'"
      }
    }
  }'
```

### Choosing the Reveal Delay

The reveal delay is the time between publishing the commitment hash and revealing the full signal. Too short (under 10 seconds) and the commitment and reveal land in the same event bus batch, providing weak proof of pre-existence. Too long (over 10 minutes) and subscribers receive signals too late to execute at the target entry price. The sweet spot for most signal types:

- **Scalping signals (1m-15m timeframe):** 15-30 second delay. Price moves fast, subscribers need the signal quickly. The commitment still creates a verifiable timestamp separation.
- **Swing signals (1h-4h timeframe):** 60-120 second delay. Adequate time for the commitment to propagate and be independently recorded by subscribers' monitoring systems.
- **Position signals (1d+ timeframe):** 5-10 minute delay. Entry price is less time-sensitive, so a longer delay provides stronger timestamp proof without meaningful cost to subscribers.

In all cases, the commitment timestamp is set by GreenHelix when the event is received, not by the provider. The provider cannot manipulate the platform-side timestamp.

### Usage Example

```python
publisher = SignalPublisher(
    agent_id="signal-provider-momentum-x",
    api_key="your-api-key",
    private_key_b64="your-base64-private-key",
)

# One-time setup
publisher.register("Momentum-X Crypto Signals")

# Publish a signal with commit-reveal
result = publisher.publish_signal(
    asset="BTC/USDT",
    direction="long",
    entry_price=68450.00,
    stop_loss=67200.00,
    take_profit=71000.00,
    timeframe="4h",
    notes="Momentum breakout above 4h resistance",
    reveal_delay_seconds=60,
)

signal_id = result["signal"]["signal_id"]
print(f"Signal published: {signal_id}")
print(f"Commitment hash: {result['commit']['commitment_hash']}")

# Later, when the trade resolves
publisher.record_outcome(
    signal_id=signal_id,
    outcome="hit_tp",
    exit_price=71000.00,
    exit_timestamp=int(time.time()),
)

# Build Merkle chain weekly
publisher.build_claim_chain()
```

### Handling Multiple Signals Per Day

High-frequency signal providers may issue 10-20 signals per day. Each signal goes through commit-reveal independently. The `_pending_commits` dictionary holds all uncommitted signals in memory. In production, replace this with a persistent store (Redis, PostgreSQL) to survive process restarts.

For providers issuing many signals, batch the Merkle chain build. Rather than building after every signal, build once at end-of-day or once per 50 signals. This reduces API calls while maintaining the same cryptographic guarantees -- all signals within the batch are included in the chain.

```python
import time

# Track signal count since last chain build
signals_since_chain = 0
CHAIN_BUILD_THRESHOLD = 50

def on_signal_outcome(publisher, signal_id, outcome, exit_price):
    global signals_since_chain

    publisher.record_outcome(
        signal_id=signal_id,
        outcome=outcome,
        exit_price=exit_price,
        exit_timestamp=int(time.time()),
    )
    signals_since_chain += 1

    if signals_since_chain >= CHAIN_BUILD_THRESHOLD:
        publisher.build_claim_chain()
        signals_since_chain = 0
        print("Merkle chain built after 50 signal outcomes.")
```

---

## Chapter 3: The SignalVerifier Class

### Why Buyers Need Independent Verification

A signal provider claims 72% accuracy over 90 days. The provider's website shows a beautiful equity curve and a table of winning trades. Without verification, you are trusting the provider to honestly report their own performance -- the same trust model that has failed for every Telegram signal group, every copy trading platform, and every signal marketplace built to date. The SignalVerifier class lets a subscriber independently verify every signal against the cryptographic record on GreenHelix. No trust required.

### The Complete SignalVerifier Class

```python
import base64
import hashlib
import json
from typing import Optional

import requests
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.hazmat.primitives import serialization


class SignalVerifier:
    """Verifies trading signals against cryptographic proofs on GreenHelix."""

    def __init__(self, api_key: str,
                 base_url: str = "https://api.greenhelix.net/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }
        # Cache of provider public keys
        self._public_keys: dict[str, Ed25519PublicKey] = {}

    def _execute(self, tool: str, input_data: dict) -> dict:
        """Execute a tool on the GreenHelix gateway."""
        response = requests.post(
            f"{self.base_url}/v1",
            headers=self.headers,
            json={"tool": tool, "input": input_data},
        )
        response.raise_for_status()
        return response.json()

    def _get_public_key(self, provider_id: str) -> Ed25519PublicKey:
        """Fetch and cache a provider's Ed25519 public key."""
        if provider_id not in self._public_keys:
            result = self._execute("get_agent_identity", {
                "agent_id": provider_id,
            })
            pub_bytes = base64.b64decode(result["public_key"])
            self._public_keys[provider_id] = Ed25519PublicKey.from_public_bytes(
                pub_bytes
            )
        return self._public_keys[provider_id]

    def verify_signal(self, signal: dict) -> dict:
        """Verify a signal's Ed25519 signature against the provider's key.

        Returns a dict with verification status and details.
        """
        provider_id = signal["provider_id"]
        signature_b64 = signal.get("signature")

        if not signature_b64:
            return {"verified": False, "reason": "no_signature"}

        # Reconstruct canonical JSON (exclude signature from signed data)
        signal_data = {k: v for k, v in signal.items() if k != "signature"}
        canonical = json.dumps(signal_data, sort_keys=True)

        try:
            public_key = self._get_public_key(provider_id)
            signature_bytes = base64.b64decode(signature_b64)
            public_key.verify(signature_bytes, canonical.encode("utf-8"))
            return {
                "verified": True,
                "provider_id": provider_id,
                "signal_id": signal.get("signal_id"),
            }
        except Exception as e:
            return {
                "verified": False,
                "reason": "invalid_signature",
                "error": str(e),
            }

    def check_timestamp_proof(self, signal_id: str,
                              provider_id: str) -> dict:
        """Verify that a signal commitment was recorded before the reveal.

        Checks the GreenHelix event bus for the commitment and reveal events,
        and verifies that:
        1. A commitment event exists with a valid hash.
        2. The reveal event's content matches the commitment hash.
        3. The commitment timestamp precedes the reveal timestamp.
        """
        # Fetch the provider's claim chains
        chains = self._execute("get_claim_chains", {
            "agent_id": provider_id,
        })

        # Fetch verified claims for this provider
        claims = self._execute("get_verified_claims", {
            "agent_id": provider_id,
        })

        # Look for commitment and reveal events
        commitment_event = None
        reveal_event = None

        for claim in claims.get("claims", []):
            payload = claim.get("payload", {})
            if (payload.get("signal_id") == signal_id
                    and claim.get("event_type") == "signal_commitment"):
                commitment_event = claim
            elif (payload.get("signal_id") == signal_id
                    and claim.get("event_type") == "signal_reveal"):
                reveal_event = claim

        if not commitment_event:
            return {
                "valid": False,
                "reason": "no_commitment_found",
                "signal_id": signal_id,
            }

        if not reveal_event:
            return {
                "valid": False,
                "reason": "no_reveal_found",
                "signal_id": signal_id,
            }

        # Verify commitment hash matches reveal content
        reveal_payload = reveal_event.get("payload", {})
        signal_data = {
            "signal_id": reveal_payload.get("signal_id"),
            "provider_id": provider_id,
            "asset": reveal_payload.get("asset"),
            "direction": reveal_payload.get("direction"),
            "entry_price": reveal_payload.get("entry_price"),
            "stop_loss": reveal_payload.get("stop_loss"),
            "take_profit": reveal_payload.get("take_profit"),
            "timeframe": reveal_payload.get("timeframe"),
            "notes": reveal_payload.get("notes", ""),
            "timestamp": reveal_payload.get("timestamp"),
        }
        canonical = json.dumps(signal_data, sort_keys=True)
        computed_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest()

        commitment_hash = commitment_event["payload"].get("commitment_hash")
        hashes_match = computed_hash == commitment_hash

        # Verify timing: commitment before reveal
        commit_time = commitment_event.get("recorded_at", 0)
        reveal_time = reveal_event.get("recorded_at", 0)
        timing_valid = commit_time < reveal_time

        return {
            "valid": hashes_match and timing_valid,
            "signal_id": signal_id,
            "hashes_match": hashes_match,
            "commitment_hash": commitment_hash,
            "computed_hash": computed_hash,
            "commitment_time": commit_time,
            "reveal_time": reveal_time,
            "timing_valid": timing_valid,
        }

    def audit_provider(self, provider_id: str) -> dict:
        """Run a full audit of a signal provider.

        Checks claim chain integrity, signal count, verification rate,
        and performance metrics.
        """
        # Get claim chains
        chains = self._execute("get_claim_chains", {
            "agent_id": provider_id,
        })

        # Get verified claims
        claims = self._execute("get_verified_claims", {
            "agent_id": provider_id,
        })

        # Get reputation
        reputation = self._execute("get_agent_reputation", {
            "agent_id": provider_id,
        })

        # Count signal types
        commitments = []
        reveals = []
        outcomes = []

        for claim in claims.get("claims", []):
            event_type = claim.get("event_type", "")
            if event_type == "signal_commitment":
                commitments.append(claim)
            elif event_type == "signal_reveal":
                reveals.append(claim)
            elif event_type == "signal_outcome":
                outcomes.append(claim)

        # Calculate verification rate
        total_signals = len(reveals)
        verified_signals = 0
        for reveal in reveals:
            signal_id = reveal.get("payload", {}).get("signal_id")
            proof = self.check_timestamp_proof(signal_id, provider_id)
            if proof["valid"]:
                verified_signals += 1

        verification_rate = (
            verified_signals / total_signals if total_signals > 0 else 0.0
        )

        # Calculate outcome stats
        wins = sum(
            1 for o in outcomes
            if o.get("payload", {}).get("outcome") == "hit_tp"
        )
        losses = sum(
            1 for o in outcomes
            if o.get("payload", {}).get("outcome") == "hit_sl"
        )
        expired = sum(
            1 for o in outcomes
            if o.get("payload", {}).get("outcome") == "expired"
        )

        return {
            "provider_id": provider_id,
            "trust_score": reputation.get("score"),
            "chain_count": len(chains.get("chains", [])),
            "total_commitments": len(commitments),
            "total_reveals": total_signals,
            "total_outcomes": len(outcomes),
            "verified_signals": verified_signals,
            "verification_rate": round(verification_rate * 100, 1),
            "wins": wins,
            "losses": losses,
            "expired": expired,
            "hit_rate": round(
                wins / (wins + losses) * 100, 1
            ) if (wins + losses) > 0 else 0.0,
        }
```

### Verifying a Single Signal

With curl, you can manually verify a signal by fetching the provider's public key and checking the commitment:

```bash
# Step 1: Get the provider's public key
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "get_agent_identity",
    "input": {
      "agent_id": "signal-provider-momentum-x"
    }
  }'

# Step 2: Get their verified claims
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "get_verified_claims",
    "input": {
      "agent_id": "signal-provider-momentum-x"
    }
  }'

# Step 3: Get their claim chains to verify Merkle integrity
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "get_claim_chains",
    "input": {
      "agent_id": "signal-provider-momentum-x"
    }
  }'
```

### Running a Full Provider Audit

```python
verifier = SignalVerifier(api_key="your-api-key")

# Audit a provider before subscribing
audit = verifier.audit_provider("signal-provider-momentum-x")
print(f"Provider: {audit['provider_id']}")
print(f"Trust score: {audit['trust_score']}")
print(f"Total signals: {audit['total_reveals']}")
print(f"Verification rate: {audit['verification_rate']}%")
print(f"Hit rate: {audit['hit_rate']}%")
print(f"Wins: {audit['wins']}, Losses: {audit['losses']}, Expired: {audit['expired']}")
print(f"Claim chains: {audit['chain_count']}")

# Decision logic
if (audit["verification_rate"] >= 95.0
        and audit["total_reveals"] >= 50
        and audit["hit_rate"] >= 55.0
        and audit["trust_score"] >= 70):
    print("Provider passes audit. Safe to subscribe.")
else:
    print("Provider does not meet criteria. Do not subscribe.")
```

### What Verification Catches

The SignalVerifier detects four categories of fraud:

1. **Backdated signals.** If the commitment timestamp postdates the price move, `check_timestamp_proof` returns `timing_valid: false`. The provider created the signal after the move happened.

2. **Altered signals.** If the revealed signal does not match the committed hash, `check_timestamp_proof` returns `hashes_match: false`. The provider changed the signal between commitment and reveal.

3. **Missing commitments.** If a revealed signal has no corresponding commitment, `check_timestamp_proof` returns `no_commitment_found`. The provider skipped the commit phase entirely -- possibly because they only commit signals they expect to win.

4. **Forged signatures.** If the signature does not verify against the provider's registered public key, `verify_signal` returns `verified: false`. Someone other than the registered provider created or tampered with the signal.

---

## Chapter 4: Performance Computation

### Why Self-Reported Metrics Are Worthless

A signal provider reports 78% accuracy. What does that mean? 78% of signals hit take profit? 78% of signals were profitable at any point during their lifetime? 78% of signals that the provider chose to report? Without a rigorous, standardized computation methodology applied to verified data, the number is meaningless. The PerformanceTracker class computes metrics exclusively from signals that passed the commit-reveal verification process. Self-reported numbers are excluded entirely.

### The PerformanceTracker Class

```python
import time
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class SignalOutcome:
    """A verified signal outcome."""
    signal_id: str
    provider_id: str
    asset: str
    direction: str
    entry_price: float
    stop_loss: float
    take_profit: float
    exit_price: float
    outcome: str  # "hit_tp", "hit_sl", "expired"
    signal_timestamp: int
    exit_timestamp: int
    verified: bool = False


class PerformanceTracker:
    """Computes performance metrics from verified signal outcomes."""

    def __init__(self):
        self._outcomes: list[SignalOutcome] = []

    def record_outcome(self, outcome: SignalOutcome) -> None:
        """Record a verified signal outcome."""
        if not outcome.verified:
            raise ValueError(
                f"Signal {outcome.signal_id} is not verified. "
                "Only verified signals can be recorded."
            )
        self._outcomes.append(outcome)

    def _filter_by_window(self, window_seconds: Optional[int] = None
                          ) -> list[SignalOutcome]:
        """Filter outcomes to those within the given time window."""
        if window_seconds is None:
            return list(self._outcomes)
        cutoff = int(time.time()) - window_seconds
        return [o for o in self._outcomes if o.exit_timestamp >= cutoff]

    def compute_stats(self, window_seconds: Optional[int] = None) -> dict:
        """Compute performance statistics.

        Args:
            window_seconds: If provided, only consider outcomes within
                this many seconds of now. None means all time.

        Returns:
            Dict with hit_rate, avg_return, risk_reward_ratio,
            max_consecutive_losses, total_signals, wins, losses.
        """
        outcomes = self._filter_by_window(window_seconds)

        if not outcomes:
            return {
                "hit_rate": 0.0,
                "avg_return": 0.0,
                "risk_reward_ratio": 0.0,
                "max_consecutive_losses": 0,
                "total_signals": 0,
                "wins": 0,
                "losses": 0,
            }

        wins = 0
        losses = 0
        returns = []
        consecutive_losses = 0
        max_consecutive_losses = 0
        risk_sum = 0.0
        reward_sum = 0.0

        for o in outcomes:
            # Compute return percentage
            if o.direction == "long":
                pct_return = (o.exit_price - o.entry_price) / o.entry_price * 100
                risk = abs(o.entry_price - o.stop_loss)
                reward = abs(o.take_profit - o.entry_price)
            else:  # short
                pct_return = (o.entry_price - o.exit_price) / o.entry_price * 100
                risk = abs(o.stop_loss - o.entry_price)
                reward = abs(o.entry_price - o.take_profit)

            returns.append(pct_return)

            if risk > 0:
                risk_sum += risk
                reward_sum += reward

            if o.outcome == "hit_tp":
                wins += 1
                consecutive_losses = 0
            elif o.outcome == "hit_sl":
                losses += 1
                consecutive_losses += 1
                max_consecutive_losses = max(
                    max_consecutive_losses, consecutive_losses
                )
            else:  # expired
                consecutive_losses = 0

        total = wins + losses
        hit_rate = (wins / total * 100) if total > 0 else 0.0
        avg_return = sum(returns) / len(returns) if returns else 0.0
        risk_reward = (reward_sum / risk_sum) if risk_sum > 0 else 0.0

        return {
            "hit_rate": round(hit_rate, 2),
            "avg_return": round(avg_return, 4),
            "risk_reward_ratio": round(risk_reward, 2),
            "max_consecutive_losses": max_consecutive_losses,
            "total_signals": len(outcomes),
            "wins": wins,
            "losses": losses,
        }

    def get_rolling(self) -> dict:
        """Compute rolling window statistics for 7d, 30d, and 90d."""
        return {
            "7d": self.compute_stats(window_seconds=7 * 86400),
            "30d": self.compute_stats(window_seconds=30 * 86400),
            "90d": self.compute_stats(window_seconds=90 * 86400),
            "all_time": self.compute_stats(window_seconds=None),
        }
```

### Metrics Definitions

| Metric | Definition | Why It Matters |
|--------|-----------|----------------|
| `hit_rate` | Percentage of signals that hit take profit out of all resolved signals (TP + SL) | The most intuitive measure of accuracy. Expired signals are excluded because they neither won nor lost. |
| `avg_return` | Mean percentage return across all resolved signals, including losses | Captures the magnitude of wins and losses, not just the count. A 60% hit rate with average +0.1% return is worse than 50% with +2.0% average. |
| `risk_reward_ratio` | Average reward (entry to TP) divided by average risk (entry to SL) | Measures how much the provider risks per unit of potential reward. A ratio above 2.0 means the provider targets gains twice the size of their stop losses. |
| `max_consecutive_losses` | Longest streak of consecutive losing signals | Critical for risk management. A provider with 65% hit rate but 12 consecutive losses at one point signals high variance that may not suit all subscribers. |

### Accounting for Slippage and Execution Delays

Verified signal accuracy is computed from the signal's stated entry price and the actual outcome. But subscribers executing the signal in real markets face slippage. A signal published at entry price $68,450 may be executed at $68,480 due to order book depth and latency. The PerformanceTracker computes raw metrics from the signal itself. Slippage-adjusted metrics require the subscriber to report their actual execution price.

```python
def compute_slippage_adjusted(self, actual_entries: dict[str, float],
                               window_seconds: Optional[int] = None) -> dict:
    """Compute stats adjusted for actual execution prices.

    Args:
        actual_entries: Dict mapping signal_id to actual entry price.
        window_seconds: Time window filter.
    """
    outcomes = self._filter_by_window(window_seconds)
    adjusted_returns = []

    for o in outcomes:
        actual_entry = actual_entries.get(o.signal_id, o.entry_price)
        if o.direction == "long":
            pct = (o.exit_price - actual_entry) / actual_entry * 100
        else:
            pct = (actual_entry - o.exit_price) / actual_entry * 100
        adjusted_returns.append(pct)

    avg_slippage_adjusted = (
        sum(adjusted_returns) / len(adjusted_returns)
        if adjusted_returns else 0.0
    )

    raw_stats = self.compute_stats(window_seconds)
    return {
        **raw_stats,
        "avg_return_slippage_adjusted": round(avg_slippage_adjusted, 4),
        "slippage_impact": round(
            raw_stats["avg_return"] - avg_slippage_adjusted, 4
        ),
    }
```

### Computing Drawdown From Verified Signals

The `max_consecutive_losses` metric captures the count of sequential losses, but drawdown in percentage terms requires tracking cumulative loss within a streak. Here is an extension to the PerformanceTracker that computes maximum drawdown as a percentage:

```python
def compute_max_drawdown_pct(self, window_seconds: Optional[int] = None
                              ) -> float:
    """Compute maximum drawdown as cumulative loss percentage
    during the worst consecutive losing streak.

    Returns the absolute value of the worst drawdown.
    """
    outcomes = self._filter_by_window(window_seconds)
    if not outcomes:
        return 0.0

    max_drawdown = 0.0
    current_drawdown = 0.0

    for o in outcomes:
        if o.direction == "long":
            pct = (o.exit_price - o.entry_price) / o.entry_price * 100
        else:
            pct = (o.entry_price - o.exit_price) / o.entry_price * 100

        if pct < 0:
            current_drawdown += abs(pct)
            max_drawdown = max(max_drawdown, current_drawdown)
        else:
            current_drawdown = 0.0

    return round(max_drawdown, 2)
```

This metric matters because a provider could have a reasonable hit rate (55%) but size their stop losses so loosely that each loss wipes out three wins. The drawdown percentage captures this behavior in a way that hit rate alone does not.

### Submitting Computed Metrics to GreenHelix

After computing performance statistics, submit them as verified metrics so they appear in search results and leaderboards:

```python
tracker = PerformanceTracker()

# ... record verified outcomes ...

stats = tracker.compute_stats()
rolling = tracker.get_rolling()

# Submit to GreenHelix
publisher.submit_performance_snapshot({
    "hit_rate": stats["hit_rate"],
    "avg_return": stats["avg_return"],
    "risk_reward_ratio": stats["risk_reward_ratio"],
    "max_consecutive_losses": stats["max_consecutive_losses"],
    "total_signals": stats["total_signals"],
    "rolling_7d_hit_rate": rolling["7d"]["hit_rate"],
    "rolling_30d_hit_rate": rolling["30d"]["hit_rate"],
    "rolling_90d_hit_rate": rolling["90d"]["hit_rate"],
})
```

With curl:

```bash
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "submit_metrics",
    "input": {
      "agent_id": "signal-provider-momentum-x",
      "metrics": {
        "hit_rate": 67.3,
        "avg_return": 1.24,
        "risk_reward_ratio": 2.1,
        "max_consecutive_losses": 4,
        "total_signals": 187,
        "rolling_7d_hit_rate": 71.4,
        "rolling_30d_hit_rate": 68.2,
        "rolling_90d_hit_rate": 67.3
      }
    }
  }'
```

---

## Chapter 5: Escrow-Linked Subscriptions

### Why Direct Payment Is Broken

Traditional signal subscriptions use direct payment: the subscriber pays $99/month upfront, receives signals for 30 days, and hopes the signals are good. If the signals underperform, the subscriber cancels. The provider has already been paid. There is zero financial alignment between the provider's revenue and the subscriber's outcome. Escrow-linked subscriptions fix this by making payment contingent on verified performance. The subscriber's funds sit in escrow for the evaluation period. If the provider meets the agreed performance criteria, escrow releases automatically. If not, the subscriber can dispute.

### Performance Criteria Design

Performance criteria must be specific, measurable, and computed from verified data. Vague criteria like "good performance" are unenforceable. Here is a well-designed criteria object:

```python
performance_criteria = {
    "min_hit_rate": 55.0,          # Minimum hit rate percentage
    "min_risk_reward": 1.5,         # Minimum risk/reward ratio
    "max_drawdown": 15.0,           # Maximum consecutive loss percentage
    "min_signal_count": 15,         # Minimum signals during evaluation period
    "evaluation_metric_source": "verified_only",  # Only count verified signals
}
```

Each criterion maps to a computable metric:

- **min_hit_rate**: Computed from `PerformanceTracker.compute_stats()["hit_rate"]`
- **min_risk_reward**: Computed from `PerformanceTracker.compute_stats()["risk_reward_ratio"]`
- **max_drawdown**: Maximum cumulative loss during any consecutive losing streak
- **min_signal_count**: Total number of verified signals issued during the evaluation period

### Creating an Escrow-Linked Subscription

With curl:

```bash
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "create_performance_escrow",
    "input": {
      "payer_agent_id": "subscriber-bot-001",
      "payee_agent_id": "signal-provider-momentum-x",
      "amount": 99.00,
      "currency": "USD",
      "performance_criteria": {
        "min_hit_rate": 55.0,
        "min_risk_reward": 1.5,
        "max_drawdown": 15.0,
        "min_signal_count": 15
      },
      "evaluation_period_days": 30
    }
  }'
```

With Python:

```python
def create_signal_subscription(client, subscriber_id: str,
                                provider_id: str, amount: float,
                                criteria: dict,
                                evaluation_days: int = 30) -> dict:
    """Create an escrow-linked signal subscription.

    Args:
        client: GreenHelix API client instance.
        subscriber_id: The subscribing agent's ID.
        provider_id: The signal provider's agent ID.
        amount: Subscription amount in USD.
        criteria: Performance criteria dict.
        evaluation_days: Evaluation period in days.

    Returns:
        Escrow creation result including escrow_id.
    """
    return client._execute("create_performance_escrow", {
        "payer_agent_id": subscriber_id,
        "payee_agent_id": provider_id,
        "amount": amount,
        "currency": "USD",
        "performance_criteria": criteria,
        "evaluation_period_days": evaluation_days,
    })


def evaluate_subscription(client, verifier: SignalVerifier,
                           tracker: PerformanceTracker,
                           escrow_id: str,
                           provider_id: str,
                           criteria: dict) -> dict:
    """Evaluate whether a subscription's performance criteria are met.

    Args:
        client: GreenHelix API client instance.
        verifier: SignalVerifier for auditing the provider.
        tracker: PerformanceTracker with recorded outcomes.
        escrow_id: The escrow to evaluate.
        provider_id: The signal provider's agent ID.
        criteria: The performance criteria to check against.

    Returns:
        Evaluation result with pass/fail and action taken.
    """
    # Compute current stats from verified data
    stats = tracker.compute_stats()

    # Check each criterion
    checks = {
        "hit_rate": stats["hit_rate"] >= criteria.get("min_hit_rate", 0),
        "risk_reward": (
            stats["risk_reward_ratio"]
            >= criteria.get("min_risk_reward", 0)
        ),
        "drawdown": (
            stats["max_consecutive_losses"]
            <= criteria.get("max_drawdown", float("inf"))
        ),
        "signal_count": (
            stats["total_signals"]
            >= criteria.get("min_signal_count", 0)
        ),
    }

    all_passed = all(checks.values())

    if all_passed:
        # Auto-release escrow
        release_result = client._execute("release_escrow", {
            "escrow_id": escrow_id,
        })
        return {
            "status": "passed",
            "checks": checks,
            "stats": stats,
            "action": "escrow_released",
            "release_result": release_result,
        }
    else:
        # Criteria not met -- subscriber can file dispute
        failed_checks = {k: v for k, v in checks.items() if not v}
        return {
            "status": "failed",
            "checks": checks,
            "failed_checks": list(failed_checks.keys()),
            "stats": stats,
            "action": "dispute_recommended",
            "escrow_id": escrow_id,
        }
```

### Usage: Full Subscription Lifecycle

```python
import requests

BASE_URL = "https://api.greenhelix.net/v1"
API_KEY = "your-api-key"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

class SimpleClient:
    def _execute(self, tool, input_data):
        resp = requests.post(
            f"{BASE_URL}/v1",
            headers=headers,
            json={"tool": tool, "input": input_data},
        )
        resp.raise_for_status()
        return resp.json()

client = SimpleClient()
verifier = SignalVerifier(api_key=API_KEY)
tracker = PerformanceTracker()

# Subscriber creates escrow-linked subscription
criteria = {
    "min_hit_rate": 55.0,
    "min_risk_reward": 1.5,
    "max_drawdown": 15.0,
    "min_signal_count": 15,
}

escrow = create_signal_subscription(
    client=client,
    subscriber_id="subscriber-bot-001",
    provider_id="signal-provider-momentum-x",
    amount=99.00,
    criteria=criteria,
    evaluation_days=30,
)
escrow_id = escrow["escrow_id"]
print(f"Subscription created. Escrow ID: {escrow_id}")

# ... 30 days pass, signals are issued and verified ...
# ... tracker has recorded all verified outcomes ...

# Evaluate at end of period
result = evaluate_subscription(
    client=client,
    verifier=verifier,
    tracker=tracker,
    escrow_id=escrow_id,
    provider_id="signal-provider-momentum-x",
    criteria=criteria,
)

if result["status"] == "passed":
    print(f"Performance criteria met. Escrow released.")
    print(f"Stats: {result['stats']}")
else:
    print(f"Performance criteria NOT met.")
    print(f"Failed checks: {result['failed_checks']}")
    print(f"Recommend filing dispute on escrow {escrow_id}")
```

### Setting Conservative Criteria

The most common mistake subscribers make is setting unrealistically tight criteria. A min_hit_rate of 70% sounds reasonable, but over a 30-day window with 20 signals, normal variance can produce a 55% hit rate even from a genuinely 65% accurate provider. Set criteria at the lower bound of what is acceptable, not the average of what the provider has historically delivered. Leave a margin for variance. A provider with a 67% verified 90-day hit rate should be evaluated against a 55% minimum for a 30-day escrow window.

---

## Chapter 6: Rolling Accuracy Windows

### Why Point-in-Time Evaluation Is Insufficient

Evaluating a signal provider only at the end of a subscription period misses dangerous patterns. A provider could deliver 80% accuracy for the first 25 days and then issue 10 reckless signals in the final 5 days. The period-end evaluation might still show 58% overall accuracy -- barely passing the criterion -- while the subscriber suffered a severe drawdown in the final week. Rolling accuracy windows provide continuous evaluation, catching performance degradation as it happens rather than after the damage is done.

### The RollingAccuracyMonitor Class

```python
import time
from typing import Callable, Optional


class RollingAccuracyMonitor:
    """Monitors signal provider accuracy with sliding windows.

    Triggers alerts when performance degrades below thresholds
    within any rolling window.
    """

    def __init__(self, provider_id: str, tracker: PerformanceTracker,
                 alert_callback: Optional[Callable[[dict], None]] = None):
        self.provider_id = provider_id
        self.tracker = tracker
        self.alert_callback = alert_callback or self._default_alert

        # Alert thresholds for each window
        self.thresholds = {
            "7d": {
                "min_hit_rate": 40.0,
                "max_consecutive_losses": 6,
            },
            "30d": {
                "min_hit_rate": 50.0,
                "max_consecutive_losses": 8,
            },
            "90d": {
                "min_hit_rate": 55.0,
                "max_consecutive_losses": 10,
            },
        }

    def _default_alert(self, alert: dict) -> None:
        """Default alert handler: print to stdout."""
        print(f"ALERT [{alert['severity']}]: {alert['message']}")

    def set_thresholds(self, window: str, min_hit_rate: float = None,
                       max_consecutive_losses: int = None) -> None:
        """Update alert thresholds for a specific window."""
        if window not in self.thresholds:
            self.thresholds[window] = {}
        if min_hit_rate is not None:
            self.thresholds[window]["min_hit_rate"] = min_hit_rate
        if max_consecutive_losses is not None:
            self.thresholds[window]["max_consecutive_losses"] = (
                max_consecutive_losses
            )

    def check(self) -> dict:
        """Run a full check across all rolling windows.

        Returns a dict with the status for each window and any
        triggered alerts.
        """
        rolling = self.tracker.get_rolling()
        alerts = []
        window_status = {}

        for window_name, window_config in self.thresholds.items():
            stats = rolling.get(window_name, {})
            if not stats or stats.get("total_signals", 0) == 0:
                window_status[window_name] = "insufficient_data"
                continue

            violations = []

            # Check hit rate
            min_hr = window_config.get("min_hit_rate")
            if min_hr is not None and stats["hit_rate"] < min_hr:
                violations.append({
                    "metric": "hit_rate",
                    "threshold": min_hr,
                    "actual": stats["hit_rate"],
                })

            # Check consecutive losses
            max_cl = window_config.get("max_consecutive_losses")
            if (max_cl is not None
                    and stats["max_consecutive_losses"] > max_cl):
                violations.append({
                    "metric": "max_consecutive_losses",
                    "threshold": max_cl,
                    "actual": stats["max_consecutive_losses"],
                })

            if violations:
                severity = "critical" if window_name == "7d" else "warning"
                alert = {
                    "provider_id": self.provider_id,
                    "window": window_name,
                    "severity": severity,
                    "violations": violations,
                    "stats": stats,
                    "timestamp": int(time.time()),
                    "message": (
                        f"Provider {self.provider_id} violated "
                        f"{len(violations)} threshold(s) in "
                        f"{window_name} window: "
                        + ", ".join(
                            f"{v['metric']}={v['actual']} "
                            f"(threshold: {v['threshold']})"
                            for v in violations
                        )
                    ),
                }
                alerts.append(alert)
                self.alert_callback(alert)
                window_status[window_name] = "degraded"
            else:
                window_status[window_name] = "healthy"

        return {
            "provider_id": self.provider_id,
            "checked_at": int(time.time()),
            "window_status": window_status,
            "alerts": alerts,
            "rolling_stats": rolling,
        }
```

### Subscriber Notification via Webhooks

When the rolling monitor detects degradation, notify subscribers automatically using GreenHelix webhooks:

```python
def setup_webhook_alerts(client, subscriber_id: str,
                          provider_id: str,
                          webhook_url: str) -> dict:
    """Register a webhook to receive performance alerts.

    The webhook fires when the rolling accuracy monitor detects
    threshold violations for the given provider.
    """
    return client._execute("register_webhook", {
        "agent_id": subscriber_id,
        "event_type": "performance_alert",
        "filter": {"provider_id": provider_id},
        "url": webhook_url,
    })
```

With curl:

```bash
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "register_webhook",
    "input": {
      "agent_id": "subscriber-bot-001",
      "event_type": "performance_alert",
      "filter": {"provider_id": "signal-provider-momentum-x"},
      "url": "https://my-bot.example.com/webhooks/performance-alert"
    }
  }'
```

### Running the Monitor on a Schedule

```python
import time

monitor = RollingAccuracyMonitor(
    provider_id="signal-provider-momentum-x",
    tracker=tracker,
    alert_callback=lambda alert: send_to_slack(alert),
)

# Customize thresholds
monitor.set_thresholds("7d", min_hit_rate=45.0, max_consecutive_losses=5)
monitor.set_thresholds("30d", min_hit_rate=52.0)

# Run every hour
while True:
    result = monitor.check()
    print(f"Status: {result['window_status']}")
    if result["alerts"]:
        print(f"  {len(result['alerts'])} alert(s) fired")
    time.sleep(3600)
```

### Integrating the Monitor With Escrow Evaluation

The rolling accuracy monitor can trigger early dispute filing rather than waiting for the evaluation period to end. If the 7-day window enters critical status and the 30-day window is also degraded, the subscriber has strong evidence that the provider will not meet the escrow criteria by period end. Filing early preserves the evidence trail and prevents the provider from issuing a burst of favorable signals at the last minute to pull the average up.

```python
def check_and_maybe_dispute(monitor: RollingAccuracyMonitor,
                             client, escrow_id: str,
                             criteria: dict,
                             tracker: PerformanceTracker,
                             verifier: SignalVerifier,
                             provider_id: str) -> Optional[dict]:
    """Run rolling check and file dispute if situation is irrecoverable."""
    result = monitor.check()

    # Both 7d and 30d degraded = likely irrecoverable
    ws = result["window_status"]
    if ws.get("7d") == "degraded" and ws.get("30d") == "degraded":
        stats = tracker.compute_stats()
        # Check if mathematically possible to recover
        remaining_signals_needed = criteria.get("min_signal_count", 0) - stats["total_signals"]
        if remaining_signals_needed < 0:
            remaining_signals_needed = 0

        current_losses = stats["losses"]
        current_wins = stats["wins"]
        needed_wins = 0
        target_hr = criteria.get("min_hit_rate", 0)

        # Would need X consecutive wins to reach target
        total = current_wins + current_losses
        if total > 0:
            needed_total_wins = int(target_hr / 100 * (total + remaining_signals_needed))
            needed_wins = max(0, needed_total_wins - current_wins)

        if needed_wins > remaining_signals_needed:
            # Mathematically impossible to recover
            evidence = build_evidence_package(
                verifier, tracker, provider_id, escrow_id, criteria
            )
            return file_dispute(client, escrow_id, evidence)

    return None
```

### Early Warning Patterns

The rolling monitor catches three patterns that period-end evaluation misses:

1. **Sudden degradation.** The 7-day window hits critical threshold while 30-day and 90-day windows are still healthy. This means the provider's recent performance has collapsed even though historical performance was good. The subscriber can prepare to dispute or exit before the full evaluation period ends.

2. **Gradual decline.** The 90-day window is healthy, the 30-day window shows warning, and the 7-day window is critical. Performance is trending down across all timeframes. This pattern often precedes a provider shutting down or pivoting to a different strategy.

3. **Volatility spikes.** The `max_consecutive_losses` threshold fires while hit rate is still acceptable. The provider's win rate is fine on average, but they are experiencing unusual streaks of losses. This can indicate regime change in the underlying market or a breakdown in the provider's signal generation process.

---

## Chapter 7: Dispute Resolution

### When Disputes Happen

A dispute occurs when the evaluation period ends and the provider's verified performance does not meet the escrow criteria. The subscriber's funds remain in escrow. Three outcomes are possible: full refund to the subscriber, partial release to the provider (if some criteria were met), or dismissal (if the subscriber's claim is invalid). The dispute process uses the same cryptographic evidence chain that underpins the entire verification network, making disputes data-driven rather than he-said-she-said.

### Building an Evidence Package

When filing a dispute, the subscriber must present evidence. Because all signals went through the commit-reveal process, the evidence is already on-chain and verifiable. The evidence package pulls everything together.

```python
def build_evidence_package(verifier: SignalVerifier,
                            tracker: PerformanceTracker,
                            provider_id: str,
                            escrow_id: str,
                            criteria: dict) -> dict:
    """Build a complete evidence package for a dispute.

    Includes:
    - Provider audit results
    - Performance stats vs criteria
    - Specific failed criteria with evidence
    - Claim chain references for independent verification
    """
    audit = verifier.audit_provider(provider_id)
    stats = tracker.compute_stats()
    rolling = tracker.get_rolling()

    # Identify specific failures
    failures = []

    if stats["hit_rate"] < criteria.get("min_hit_rate", 0):
        failures.append({
            "criterion": "min_hit_rate",
            "required": criteria["min_hit_rate"],
            "actual": stats["hit_rate"],
            "evidence": (
                f"Verified hit rate of {stats['hit_rate']}% across "
                f"{stats['total_signals']} signals is below the "
                f"required {criteria['min_hit_rate']}%."
            ),
        })

    if stats["risk_reward_ratio"] < criteria.get("min_risk_reward", 0):
        failures.append({
            "criterion": "min_risk_reward",
            "required": criteria["min_risk_reward"],
            "actual": stats["risk_reward_ratio"],
            "evidence": (
                f"Risk/reward ratio of {stats['risk_reward_ratio']} "
                f"is below the required {criteria['min_risk_reward']}."
            ),
        })

    if stats["total_signals"] < criteria.get("min_signal_count", 0):
        failures.append({
            "criterion": "min_signal_count",
            "required": criteria["min_signal_count"],
            "actual": stats["total_signals"],
            "evidence": (
                f"Only {stats['total_signals']} verified signals issued "
                f"during the evaluation period. Minimum required: "
                f"{criteria['min_signal_count']}."
            ),
        })

    return {
        "escrow_id": escrow_id,
        "provider_id": provider_id,
        "audit": audit,
        "performance_stats": stats,
        "rolling_stats": rolling,
        "criteria": criteria,
        "failures": failures,
        "verification_rate": audit["verification_rate"],
        "claim_chain_count": audit["chain_count"],
    }
```

### Filing a Dispute

```python
def file_dispute(client, escrow_id: str, evidence: dict) -> dict:
    """File a dispute with an evidence package.

    The evidence is serialized and submitted alongside the dispute
    so the resolution process has access to verifiable data.
    """
    reason = (
        f"Performance criteria not met. "
        f"{len(evidence['failures'])} criterion/criteria failed: "
        + "; ".join(
            f"{f['criterion']}: required {f['required']}, "
            f"actual {f['actual']}"
            for f in evidence["failures"]
        )
    )

    return client._execute("open_dispute", {
        "escrow_id": escrow_id,
        "reason": reason,
    })
```

With curl:

```bash
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "open_dispute",
    "input": {
      "escrow_id": "esc_abc123",
      "reason": "Performance criteria not met. 2 criteria failed: min_hit_rate: required 55.0, actual 48.2; min_signal_count: required 15, actual 11."
    }
  }'
```

### Automated Dispute Filing

The evaluation function from Chapter 5 already identifies when criteria are not met. Combine it with the evidence builder and dispute filer for fully automated dispute resolution:

```python
def auto_evaluate_and_dispute(client, verifier, tracker,
                               escrow_id: str, provider_id: str,
                               criteria: dict) -> dict:
    """Evaluate subscription and automatically file dispute if needed."""
    result = evaluate_subscription(
        client=client,
        verifier=verifier,
        tracker=tracker,
        escrow_id=escrow_id,
        provider_id=provider_id,
        criteria=criteria,
    )

    if result["status"] == "passed":
        return {"action": "escrow_released", "result": result}

    # Build evidence and file dispute
    evidence = build_evidence_package(
        verifier=verifier,
        tracker=tracker,
        provider_id=provider_id,
        escrow_id=escrow_id,
        criteria=criteria,
    )

    dispute_result = file_dispute(
        client=client,
        escrow_id=escrow_id,
        evidence=evidence,
    )

    return {
        "action": "dispute_filed",
        "evidence": evidence,
        "dispute_result": dispute_result,
    }
```

### Resolution Outcomes

Disputes resolve into one of three outcomes:

**Full refund.** The evidence clearly shows the provider failed to meet the criteria. All escrowed funds return to the subscriber. This is the most common outcome when the evidence is backed by verified claim chains -- the data is unambiguous.

**Partial release.** The provider met some but not all criteria, or met them partially. For example, the provider delivered a 53% hit rate against a 55% minimum, but the risk/reward ratio exceeded the target significantly. A partial release acknowledges that the provider delivered meaningful value even if the strict criteria were not met. The split is typically computed proportionally to the criteria that were met.

**Dismissal.** The subscriber's claim is invalid. This happens when the evidence does not support the dispute -- for example, the subscriber computed metrics from unverified signals, or the criteria were ambiguously defined, or the provider's verified data actually meets the criteria. Escrowed funds release to the provider. Dismissals also occur when the subscriber fails to account for the difference between raw signal accuracy and execution accuracy. If the provider's verified signals met the criteria but the subscriber's actual trades did not (due to slippage, late execution, or partial fills), the dispute is against the subscriber's execution, not the provider's signals.

```python
# Provider responds to a dispute
def respond_to_dispute(client, dispute_id: str,
                        provider_evidence: dict) -> dict:
    """Provider responds to a dispute with counter-evidence."""
    return client._execute("resolve_dispute", {
        "dispute_id": dispute_id,
        "resolution": "counter_evidence",
        "evidence": provider_evidence,
    })
```

With curl:

```bash
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "resolve_dispute",
    "input": {
      "dispute_id": "dsp_xyz789",
      "resolution": "full_refund"
    }
  }'
```

### Why This Works Better Than Traditional Disputes

Traditional signal subscription disputes are subjective. "The signals were bad" is not verifiable. "I lost money" does not prove the provider was at fault -- the subscriber might have executed poorly. Cryptographic verification removes subjectivity entirely. The evidence package contains: the exact signals issued (with Ed25519 signatures and timestamp proofs), the exact outcomes (verified against market data), the exact performance statistics (computed from verified data), and the exact criteria that were agreed upon at escrow creation. There is nothing to argue about. The data either meets the criteria or it does not.

This eliminates the two most common problems with traditional dispute resolution in signal services. First, the "he said, she said" problem disappears because both parties agreed to the criteria at escrow creation and the metrics are computed from signed, immutable data. Second, the "selective evidence" problem disappears because the Merkle claim chain includes all signals, not just the ones either party wants to highlight. A provider cannot present only their winning signals, and a subscriber cannot present only the losses. The chain is complete or it is invalid.

---

## Chapter 8: Leaderboards and Discovery

### Building Signal Provider Leaderboards

The GreenHelix leaderboard ranks signal providers by composite reputation score, which is computed from verified metrics, claim chain depth, and consistency over time. Unlike self-reported leaderboards where providers game rankings by cherry-picking metrics, the GreenHelix leaderboard is computed exclusively from cryptographically verified data.

```python
def get_signal_leaderboard(client) -> list[dict]:
    """Fetch the signal provider leaderboard.

    Returns providers ranked by composite reputation score.
    Automatically filters out test/audit/stress agent IDs.
    """
    result = client._execute("get_agent_leaderboard", {})
    return result.get("agents", [])
```

With curl:

```bash
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "get_agent_leaderboard",
    "input": {}
  }'
```

### Searching by Performance Metrics

Subscribers can discover providers by searching for specific performance thresholds. This is where submitting standardized metrics (Chapter 4) pays off -- providers who submit complete metrics appear in more searches.

```python
def discover_providers(client, min_hit_rate: float = 55.0,
                        min_signals: int = 50,
                        max_drawdown: float = 15.0) -> list[dict]:
    """Discover signal providers meeting minimum criteria.

    Searches by multiple metrics and intersects results.
    """
    # Search by hit rate
    hit_rate_results = client._execute("search_agents_by_metrics", {
        "metric_name": "hit_rate",
        "min_value": min_hit_rate,
        "max_value": 100.0,
    })

    # Search by signal count
    signal_count_results = client._execute("search_agents_by_metrics", {
        "metric_name": "total_signals",
        "min_value": min_signals,
    })

    # Intersect results
    hit_rate_ids = {
        a["agent_id"] for a in hit_rate_results.get("agents", [])
    }
    signal_count_ids = {
        a["agent_id"] for a in signal_count_results.get("agents", [])
    }

    qualifying_ids = hit_rate_ids & signal_count_ids

    # Get full details for qualifying providers
    providers = []
    for agent_id in qualifying_ids:
        reputation = client._execute("get_agent_reputation", {
            "agent_id": agent_id,
        })
        claims = client._execute("get_verified_claims", {
            "agent_id": agent_id,
        })
        providers.append({
            "agent_id": agent_id,
            "trust_score": reputation.get("score"),
            "claims": claims,
        })

    # Sort by trust score descending
    providers.sort(key=lambda p: p.get("trust_score", 0), reverse=True)
    return providers
```

With curl:

```bash
# Search for providers with hit rate >= 60%
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "search_agents_by_metrics",
    "input": {
      "metric_name": "hit_rate",
      "min_value": 60.0,
      "max_value": 100.0
    }
  }'

# Search for providers with >= 100 verified signals
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "search_agents_by_metrics",
    "input": {
      "metric_name": "total_signals",
      "min_value": 100
    }
  }'
```

### Anti-Gaming Measures

Three measures prevent providers from gaming the leaderboard:

**Minimum signal count.** Providers with fewer than 30 verified signals are excluded from the leaderboard and from search results. This prevents a provider from issuing 3 lucky signals and claiming 100% accuracy. Thirty signals is the minimum sample size for statistical significance at a 95% confidence level with a 10% margin of error.

**Minimum history.** Providers must have at least 30 days of continuous verified activity (claim chains built at least weekly) to appear in search results. This prevents account churn -- creating new identities to discard bad track records. Building a 30-day verified history has a real cost in time and commitment. A provider who blows up their track record cannot simply create a new agent ID and start fresh with zero history. They must invest another 30 days of consistent, verified signal publishing before they become visible to subscribers again. This asymmetry -- destroying reputation is instant, rebuilding it takes weeks -- creates the correct incentive for providers to manage risk carefully.

**Leaderboard ID filtering.** GreenHelix automatically filters agent IDs prefixed with `test-`, `perf-`, `audit-`, or `stress-` from the leaderboard. This prevents test accounts from polluting rankings. Production signal providers should use descriptive, permanent agent IDs.

```python
def validate_provider_for_listing(client, verifier: SignalVerifier,
                                   provider_id: str) -> dict:
    """Check if a provider meets minimum listing requirements."""
    audit = verifier.audit_provider(provider_id)

    checks = {
        "min_signals": audit["total_reveals"] >= 30,
        "min_verification_rate": audit["verification_rate"] >= 90.0,
        "min_chain_count": audit["chain_count"] >= 4,
        "has_trust_score": audit["trust_score"] is not None,
        "not_test_account": not any(
            provider_id.startswith(prefix)
            for prefix in ("test-", "perf-", "audit-", "stress-")
        ),
    }

    return {
        "provider_id": provider_id,
        "eligible": all(checks.values()),
        "checks": checks,
        "audit": audit,
    }
```

### Embedding Verified Performance in Listings

When a provider lists their signal service on a marketplace, their GreenHelix verification data serves as the credential. The listing includes the agent ID, latest claim chain root hash, and a link to the verification endpoint. Any potential subscriber can independently verify every claim.

```python
def build_listing_data(client, publisher: SignalPublisher,
                        tracker: PerformanceTracker) -> dict:
    """Build a marketplace listing with verified performance data."""
    reputation = client._execute("get_agent_reputation", {
        "agent_id": publisher.agent_id,
    })
    chains = client._execute("get_claim_chains", {
        "agent_id": publisher.agent_id,
    })
    claims = client._execute("get_verified_claims", {
        "agent_id": publisher.agent_id,
    })
    stats = tracker.compute_stats()
    rolling = tracker.get_rolling()

    latest_chain = None
    if chains.get("chains"):
        latest_chain = chains["chains"][-1]

    return {
        "provider_id": publisher.agent_id,
        "trust_score": reputation.get("score"),
        "verified_performance": {
            "hit_rate": stats["hit_rate"],
            "avg_return": stats["avg_return"],
            "risk_reward_ratio": stats["risk_reward_ratio"],
            "total_signals": stats["total_signals"],
            "rolling_7d_hit_rate": rolling["7d"]["hit_rate"],
            "rolling_30d_hit_rate": rolling["30d"]["hit_rate"],
            "rolling_90d_hit_rate": rolling["90d"]["hit_rate"],
        },
        "claim_chain_root": (
            latest_chain["root_hash"] if latest_chain else None
        ),
        "claim_chain_depth": len(chains.get("chains", [])),
        "verification_endpoint": (
            f"https://sandbox.greenhelix.net/v1"
        ),
        "verification_instructions": (
            f"POST to the verification endpoint with tool "
            f"'get_verified_claims' and input "
            f'{{"agent_id": "{publisher.agent_id}"}} '
            f"to independently verify all performance claims."
        ),
    }
```

With curl, a potential subscriber verifies a listing:

```bash
# Step 1: Check the provider's trust score
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "get_agent_reputation",
    "input": {
      "agent_id": "signal-provider-momentum-x"
    }
  }'

# Step 2: Verify their claim chains
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "get_claim_chains",
    "input": {
      "agent_id": "signal-provider-momentum-x"
    }
  }'

# Step 3: Get all verified claims
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "get_verified_claims",
    "input": {
      "agent_id": "signal-provider-momentum-x"
    }
  }'

# Step 4: Search for this provider's metrics directly
curl -s -X POST https://sandbox.greenhelix.net/v1 \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "search_agents_by_metrics",
    "input": {
      "metric_name": "hit_rate",
      "min_value": 0,
      "max_value": 100
    }
  }'
```

### The Complete Discovery Workflow

A subscriber evaluating signal providers should follow this five-step process:

1. **Search by primary metric.** Use `search_agents_by_metrics` to find providers exceeding a minimum hit rate or risk/reward ratio. This narrows the candidate pool from hundreds to a manageable shortlist.

2. **Filter by sample size.** From the search results, discard any provider with fewer than 50 verified signals. Small sample sizes produce unreliable statistics. A provider with 8 signals and 87.5% accuracy has not proven anything.

3. **Audit each candidate.** Run `audit_provider()` against each remaining candidate. Check the verification rate (should be 95%+), chain count (should be 4+), and trust score (should be 70+).

4. **Check rolling trends.** Use `get_metric_averages` to compare 7-day, 30-day, and 90-day rolling performance. A provider whose 7-day metrics are significantly worse than their 90-day metrics is in decline. A provider whose 7-day metrics are improving relative to 90-day metrics is on an upswing.

5. **Subscribe with escrow.** Once a provider passes all checks, create an escrow-linked subscription with conservative criteria. Set the hit rate threshold 5-10 percentage points below the provider's verified 90-day average to account for normal variance.

```python
def full_discovery_pipeline(client, verifier, min_hit_rate=55.0,
                             min_signals=50, min_trust_score=70):
    """End-to-end provider discovery and ranking."""
    # Step 1: Search
    candidates = client._execute("search_agents_by_metrics", {
        "metric_name": "hit_rate",
        "min_value": min_hit_rate,
        "max_value": 100.0,
    })

    qualified = []
    for agent in candidates.get("agents", []):
        agent_id = agent["agent_id"]

        # Step 2: Filter by sample size
        if agent.get("value", 0) < min_signals:
            continue

        # Step 3: Audit
        audit = verifier.audit_provider(agent_id)
        if audit["verification_rate"] < 90.0:
            continue
        if audit.get("trust_score", 0) < min_trust_score:
            continue

        qualified.append({
            "agent_id": agent_id,
            "hit_rate": audit["hit_rate"],
            "total_signals": audit["total_reveals"],
            "trust_score": audit["trust_score"],
            "verification_rate": audit["verification_rate"],
        })

    # Sort by trust score
    qualified.sort(key=lambda p: p["trust_score"], reverse=True)
    return qualified
```

The listing data, combined with independent verification, creates a trust flywheel: better verified performance attracts more subscribers, more subscribers generate more revenue, more revenue funds better signal generation, better signals produce better verified performance. Providers who cannot survive this feedback loop are exactly the providers who should not be selling signals.

---

## What's Next

This guide covered the complete signal verification lifecycle: cryptographic commitment, independent verification, performance computation, escrow-linked payment, rolling monitoring, dispute resolution, and leaderboard discovery. Three companion guides extend this into adjacent domains:

- **Verified Trading Bot Reputation** -- how to build an unforgeable PnL proof for your trading bot using the same Ed25519 and Merkle chain infrastructure covered here, applied to automated trading rather than discrete signals.
- **Strategy Marketplace Playbook** -- how to list verified trading strategies as paid services with escrow-protected subscriptions, pricing tiers, and marketplace discovery.
- **Agent Dispute Resolution & Chargeback Defense** -- deep dive into the dispute resolution framework, including card network chargeback defense for agent-initiated transactions.

Full API reference and additional examples are available in the GreenHelix documentation at https://api.greenhelix.net/docs.

---

*Price: $99 | Format: Digital Guide | Updates: Lifetime access to revisions*