How to Use ME10001: Stablecoin Peg Metrics in Trading

Stablecoin peg data is foundational for every crypto trade. When USDT depegs, it ripples through BTC spot, perpetuals, MSTR stock, and Bitcoin ETFs simultaneously—but at different speeds and magnitudes. Understanding these dynamics creates alpha.

Why Madjik's ME10001 Data is Unique

Traditional data sources show you USDT price on one exchange. Madjik aggregates:

• 50+ liquidity sources (CEX, DEX, OTC desks)
• Real-time reserve composition (not quarterly attestations)
• Redemption velocity (early warning of runs)
• Cross-exchange spread matrices (arbitrage opportunities)

This 360° view lets you act before others see the signal.

Strategy 1: Cross-Asset Depeg Arbitrage (No Leverage)

The Opportunity: When USDT trades at $0.995, BTC/USDT shows ~0.5% higher BTC price than BTC/USD. This creates simultaneous arbitrage across spot, ETFs, and MSTR.

Instruments

Python Implementation

import requests
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime

@dataclass
class DepegArbitrage:
    """Cross-asset arbitrage detector using Madjik ME10001 data."""
    
    api_key: str
    base_url: str = "https://api.madjik.io/v1"
    
    def get_headers(self) -> Dict:
        return {"X-API-Key": self.api_key}
    
    def fetch_metric(self, endpoint: str, params: Dict = None) -> Dict:
        """Fetch metric from Madjik API."""
        url = f"{self.base_url}/metrics/{endpoint}"
        response = requests.get(url, headers=self.get_headers(), params=params)
        response.raise_for_status()
        return response.json()
    
    def detect_arbitrage(self, min_spread_pct: float = 0.3) -> Dict:
        """
        Detect depeg arbitrage opportunity across all instrument types.
        
        Returns actionable trade with risk assessment.
        """
        # Fetch ME10001 metrics
        peg = self.fetch_metric("ME10001/absval/now")
        spread = self.fetch_metric("ME10001/spread/now")
        reserves = self.fetch_metric("ME10001/reserves/now")
        velocity = self.fetch_metric("ME10001/velocity/now")
        
        usdt_price = peg["value"]
        btc_spread = spread["data"]["btc_usd_usdt_spread_pct"]
        cash_reserves = reserves["data"]["cash_pct"]
        redemption_velocity = velocity["value"]
        
        # Risk scoring
        systemic_risk = self._calculate_systemic_risk(
            usdt_price, cash_reserves, redemption_velocity
        )
        
        if usdt_price >= 0.998:
            return {"signal": "NO_DEPEG", "usdt_price": usdt_price}
        
        if btc_spread < min_spread_pct:
            return {"signal": "SPREAD_TOO_NARROW", "spread": btc_spread}
        
        if systemic_risk > 70:
            return {
                "signal": "ARBITRAGE_EXISTS_BUT_RISKY",
                "usdt_price": usdt_price,
                "spread": btc_spread,
                "systemic_risk": systemic_risk,
                "warning": "Reserve backing or velocity suggests potential non-repeg",
                "recommendation": "Use IBIT/MSTR instead of USDT-based instruments"
            }
        
        # Calculate profit potential
        gross_profit_pct = (1 / usdt_price - 1) * 100
        
        return {
            "signal": "DEPEG_ARBITRAGE",
            "timestamp": datetime.utcnow().isoformat(),
            "market_data": {
                "usdt_price": usdt_price,
                "btc_spread_pct": btc_spread,
                "cash_reserves_pct": cash_reserves,
                "redemption_velocity": redemption_velocity
            },
            "profit_potential": {
                "gross_pct": round(gross_profit_pct, 3),
                "after_fees_est": round(gross_profit_pct - 0.15, 3)  # ~15bps fees
            },
            "execution_plan": {
                "no_leverage": self._no_leverage_plan(usdt_price, btc_spread),
                "with_leverage": self._leveraged_plan(usdt_price, systemic_risk)
            },
            "risk_assessment": {
                "systemic_risk_score": systemic_risk,
                "counterparty_risks": self._list_counterparty_risks(),
                "execution_risks": self._list_execution_risks()
            },
            "why_madjik_critical": [
                "Real-time peg from 50+ sources vs single exchange",
                "Reserve composition shows backing quality",
                "Velocity predicts repeg probability",
                "Cross-exchange spread identifies best execution"
            ]
        }
    
    def _calculate_systemic_risk(self, peg: float, reserves: float, velocity: float) -> int:
        """Calculate systemic risk score 0-100."""
        peg_risk = max(0, (1 - peg) * 500)  # 0.99 = 5, 0.95 = 25
        reserve_risk = max(0, 100 - reserves)
        velocity_risk = min(50, velocity * 10)  # Cap at 50
        
        return int(min(100, peg_risk * 0.4 + reserve_risk * 0.4 + velocity_risk * 0.2))
    
    def _no_leverage_plan(self, usdt_price: float, spread: float) -> Dict:
        """Generate no-leverage execution plan."""
        return {
            "strategy": "Cash arbitrage across instruments",
            "legs": [
                {
                    "step": 1,
                    "action": "Acquire USDT at discount",
                    "instrument": "USDT/USD OTC or DEX",
                    "price": f"${usdt_price:.4f}",
                    "counterparty_risk": "OTC desk or smart contract"
                },
                {
                    "step": 2,
                    "action": "Buy BTC with discounted USDT",
                    "instrument": "BTC/USDT spot",
                    "venue": "Binance",
                    "counterparty_risk": "Binance exchange"
                },
                {
                    "step": 3,
                    "action": "Hedge BTC exposure",
                    "options": [
                        {
                            "instrument": "Sell BTC/USD spot",
                            "venue": "Coinbase",
                            "risk": "Execution timing"
                        },
                        {
                            "instrument": "Short IBIT ETF",
                            "venue": "IBKR",
                            "risk": "ETF premium/discount"
                        },
                        {
                            "instrument": "Short MSTR stock",
                            "venue": "IBKR",
                            "risk": "MSTR basis to BTC NAV"
                        }
                    ]
                },
                {
                    "step": 4,
                    "action": "Wait for USDT repeg",
                    "expected_duration": "Hours to days",
                    "exit_trigger": "USDT > $0.999"
                }
            ],
            "capital_required": "$100,000 minimum for meaningful profit",
            "expected_return": f"{spread:.2f}% gross"
        }
    
    def _leveraged_plan(self, usdt_price: float, risk_score: int) -> Dict:
        """Generate leveraged execution plan."""
        max_leverage = 3 if risk_score < 40 else 2 if risk_score < 60 else 1
        
        return {
            "strategy": "Leveraged basis trade",
            "max_leverage": f"{max_leverage}x",
            "legs": [
                {
                    "instrument": "BTC-USDT Perpetual",
                    "venue": "Binance",
                    "direction": "LONG",
                    "leverage": f"{max_leverage}x",
                    "margin": "USDT (at discount)",
                    "counterparty_risks": [
                        "Binance exchange solvency",
                        "USDT counterparty on margin",
                        "Auto-deleveraging if insurance depleted"
                    ]
                },
                {
                    "instrument": "BTC-USD Perpetual",
                    "venue": "Kraken or dYdX",
                    "direction": "SHORT",
                    "leverage": f"{max_leverage}x",
                    "margin": "USD",
                    "counterparty_risks": [
                        "Exchange solvency",
                        "Funding rate divergence"
                    ]
                }
            ],
            "critical_checks_before_entry": [
                "ME10012 cascade probability < 50%",
                "ME10006 exchange solvency > 70",
                "Funding rate spread < 0.05%"
            ],
            "warning": "Leverage amplifies both gains and liquidation risk"
        }
    
    def _list_counterparty_risks(self) -> List[Dict]:
        return [
            {
                "risk": "USDT Issuer (Tether)",
                "severity": "HIGH during depeg",
                "mitigation": "Check ME10001/reserves before entry"
            },
            {
                "risk": "Exchange (Binance)",
                "severity": "MEDIUM",
                "mitigation": "Check ME10006/solvency, limit exposure"
            },
            {
                "risk": "Withdrawal freeze",
                "severity": "HIGH during stress",
                "mitigation": "Check ME10006/withdrawal status"
            }
        ]
    
    def _list_execution_risks(self) -> List[Dict]:
        return [
            {
                "risk": "Repeg timing uncertainty",
                "impact": "Capital locked longer than expected",
                "mitigation": "Size for multi-day hold"
            },
            {
                "risk": "Spread compression before execution",
                "impact": "Reduced profit",
                "mitigation": "Use limit orders, execute quickly"
            },
            {
                "risk": "MSTR/ETF basis moves against you",
                "impact": "Hedge doesn't offset",
                "mitigation": "Use spot BTC/USD for tightest hedge"
            }
        ]


# Usage
arb = DepegArbitrage(api_key="your_api_key")
result = arb.detect_arbitrage(min_spread_pct=0.3)
print(result)

Strategy 2: Flight-to-Safety Rotation (With/Without Leverage)

The Setup: When stablecoin stress emerges, rotate from USDT-denominated positions to USD-settled instruments. This protects against both depeg losses AND exchange counterparty risk.

Without Leverage

class FlightToSafety:
    """
    Portfolio rotation strategy during stablecoin stress.
    No leverage - pure risk reduction.
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.madjik.io/v1"
    
    def assess_and_rotate(self, portfolio: Dict) -> Dict:
        """
        Assess stablecoin risk and generate rotation plan.
        
        portfolio = {
            "btc_usdt_perp": 50000,  # $50k notional
            "eth_usdt_perp": 30000,
            "usdt_spot": 20000,
            "btc_spot_coinbase": 100000,
            "ibit_etf": 50000,
            "mstr_stock": 25000
        }
        """
        # Fetch risk metrics
        risk_score = self._calculate_composite_risk()
        
        if risk_score < 30:
            return {
                "action": "HOLD",
                "risk_score": risk_score,
                "message": "Stablecoin risk within normal range"
            }
        
        # Calculate rotation based on risk level
        rotation_pct = min(100, risk_score)
        
        rotations = []
        
        # Rotate USDT perps to USD perps or spot
        usdt_perp_exposure = portfolio.get("btc_usdt_perp", 0) + portfolio.get("eth_usdt_perp", 0)
        if usdt_perp_exposure > 0:
            rotate_amount = usdt_perp_exposure * (rotation_pct / 100)
            rotations.append({
                "from": "USDT-margined perpetuals",
                "to": "USD-margined perpetuals (Kraken/CME) or spot",
                "amount": f"${rotate_amount:,.0f}",
                "urgency": "HIGH" if risk_score > 60 else "MEDIUM",
                "counterparty_change": "Removes USDT issuer risk"
            })
        
        # Convert USDT spot to USD or BTC
        usdt_spot = portfolio.get("usdt_spot", 0)
        if usdt_spot > 0:
            rotations.append({
                "from": "USDT spot holdings",
                "to": "USDC, USD, or BTC spot",
                "amount": f"${usdt_spot:,.0f}",
                "urgency": "CRITICAL" if risk_score > 70 else "HIGH",
                "note": "Direct exposure to depeg"
            })
        
        # Increase ETF allocation (safest)
        if risk_score > 50:
            rotations.append({
                "action": "INCREASE",
                "instrument": "IBIT/FBTC ETF",
                "reason": "USD-settled, regulated, no crypto counterparty",
                "target_allocation": f"{20 + risk_score/5:.0f}% of crypto exposure"
            })
        
        return {
            "action": "ROTATE",
            "risk_score": risk_score,
            "risk_level": "CRITICAL" if risk_score > 70 else "HIGH" if risk_score > 50 else "ELEVATED",
            "rotations": rotations,
            "post_rotation_portfolio": self._calculate_post_rotation(portfolio, rotations),
            "why_madjik_critical": [
                "Velocity metric shows redemption pressure 2-4 hours before price impact",
                "Reserve composition reveals backing quality (not just total)",
                "Exchange health scores prevent rotating INTO a failing venue"
            ]
        }
    
    def _calculate_composite_risk(self) -> int:
        """Calculate composite stablecoin risk score."""
        headers = {"X-API-Key": self.api_key}
        
        peg = requests.get(f"{self.base_url}/metrics/ME10001/absval/now", headers=headers).json()
        reserves = requests.get(f"{self.base_url}/metrics/ME10001/reserves/now", headers=headers).json()
        velocity = requests.get(f"{self.base_url}/metrics/ME10001/velocity/now", headers=headers).json()
        exchange = requests.get(f"{self.base_url}/metrics/ME10006/solvency/now", 
                               headers=headers, params={"exchange": "binance"}).json()
        
        peg_risk = max(0, (1 - peg["value"]) * 300)
        reserve_risk = max(0, 100 - reserves["data"]["cash_pct"]) * 0.5
        velocity_risk = min(30, velocity["value"] * 8)
        exchange_risk = max(0, 100 - exchange["value"]) * 0.3
        
        return int(min(100, peg_risk + reserve_risk + velocity_risk + exchange_risk))


# Usage
fts = FlightToSafety("your_api_key")
portfolio = {
    "btc_usdt_perp": 50000,
    "usdt_spot": 20000,
    "ibit_etf": 50000
}
rotation_plan = fts.assess_and_rotate(portfolio)
print(rotation_plan)

With Leverage (Hedged)

def leveraged_flight_to_safety():
    """
    Use leverage to HEDGE existing exposure during stress,
    not to amplify directional bets.
    """
    # Fetch risk metrics
    peg = fetch_metric("ME10001/absval/now")
    cascade = fetch_metric("ME10012/cascade/now")
    
    usdt_price = peg["value"]
    cascade_prob = cascade["value"]
    
    if usdt_price < 0.995 and cascade_prob < 50:
        return {
            "strategy": "LEVERAGED_HEDGE",
            "purpose": "Protect long exposure during uncertainty",
            "execution": {
                "existing_position": "Long BTC via USDT perp",
                "hedge": {
                    "instrument": "BTC-USD Perpetual",
                    "venue": "Kraken",
                    "direction": "SHORT",
                    "size": "100% of USDT perp notional",
                    "leverage": "2x",
                    "margin": "$USD (no USDT exposure)"
                }
            },
            "result": {
                "btc_exposure": "Neutral (hedged)",
                "stablecoin_exposure": "Long USDT (margin) vs Short USD (margin)",
                "profit_if_repeg": "USDT margin appreciates to $1",
                "loss_if_no_repeg": "USDT margin losses"
            },
            "critical_warning": "Only if cascade_prob < 50%. Higher = close everything."
        }
    
    elif cascade_prob > 50:
        return {
            "strategy": "DELEVERAGE_COMPLETELY",
            "reason": f"Cascade probability {cascade_prob}% too high for leverage",
            "action": "Close all leveraged positions, hold only spot"
        }

AI Agent Integration (MCP Protocol)

{
  "name": "madjik_stablecoin_agent",
  "description": "AI agent for stablecoin risk monitoring and cross-asset trading",
  "version": "1.0.0",
  "server": {
    "command": "python",
    "args": ["-m", "madjik_mcp.stablecoin_server"]
  },
  "tools": [
    {
      "name": "get_usdt_peg",
      "description": "Get current USDT peg value from 50+ sources",
      "inputSchema": {
        "type": "object",
        "properties": {
          "stablecoin": {
            "type": "string",
            "enum": ["USDT", "USDC", "DAI", "TUSD"],
            "default": "USDT"
          }
        }
      },
      "returns": {
        "value": "number (peg price)",
        "sources_count": "number",
        "confidence": "number 0-1"
      }
    },
    {
      "name": "get_reserve_health",
      "description": "Get stablecoin reserve composition and backing quality",
      "inputSchema": {
        "type": "object",
        "properties": {
          "stablecoin": {"type": "string"}
        },
        "required": ["stablecoin"]
      },
      "returns": {
        "cash_pct": "number",
        "treasuries_pct": "number",
        "commercial_paper_pct": "number",
        "backing_ratio": "number",
        "last_attestation": "date"
      }
    },
    {
      "name": "detect_arbitrage",
      "description": "Find depeg arbitrage opportunities across spot, perps, ETFs, MSTR",
      "inputSchema": {
        "type": "object",
        "properties": {
          "min_spread_pct": {"type": "number", "default": 0.3},
          "include_leveraged": {"type": "boolean", "default": false}
        }
      }
    },
    {
      "name": "assess_rotation_need",
      "description": "Determine if portfolio should rotate away from USDT",
      "inputSchema": {
        "type": "object",
        "properties": {
          "portfolio": {
            "type": "object",
            "description": "Current portfolio positions"
          }
        },
        "required": ["portfolio"]
      }
    }
  ],
  "prompts": [
    {
      "name": "depeg_response_guide",
      "description": "How to respond when USDT depegs",
      "template": "USDT is at ${peg_value}. Reserve backing is ${cash_pct}% cash. Redemption velocity is ${velocity}x normal. For a portfolio with ${usdt_exposure} USDT exposure, recommend: 1) Immediate actions, 2) Hedging strategy, 3) Instruments to use/avoid."
    }
  ],
  "resources": [
    {
      "uri": "madjik://metrics/ME10001",
      "name": "Stablecoin Metrics",
      "description": "All ME10001 endpoints for stablecoin analysis"
    }
  ]
}

MCP Server Implementation

# madjik_mcp/stablecoin_server.py
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent, Resource
import httpx

server = Server("madjik-stablecoin")
API_KEY = "your_api_key"
BASE_URL = "https://api.madjik.io/v1"

@server.tool()
async def get_usdt_peg(stablecoin: str = "USDT") -> dict:
    """Get current stablecoin peg with confidence score."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/metrics/ME10001/absval/now",
            headers={"X-API-Key": API_KEY},
            params={"stablecoin": stablecoin}
        )
        data = response.json()
    
    peg = data["value"]
    return {
        "stablecoin": stablecoin,
        "peg_value": peg,
        "deviation_pct": round((1 - peg) * 100, 3),
        "status": "STABLE" if peg > 0.998 else "WARNING" if peg > 0.995 else "CRITICAL",
        "sources_aggregated": data.get("metadata", {}).get("sources_count", 50),
        "recommendation": _get_peg_recommendation(peg)
    }

def _get_peg_recommendation(peg: float) -> str:
    if peg > 0.998:
        return "Normal operations. No action needed."
    elif peg > 0.995:
        return "Monitor closely. Consider reducing USDT exposure to 50%."
    elif peg > 0.990:
        return "Elevated risk. Rotate to USD instruments (IBIT, USD-margined perps)."
    else:
        return "CRITICAL: Exit all USDT positions. Use spot BTC or ETFs only."

@server.tool()
async def detect_arbitrage(min_spread_pct: float = 0.3, include_leveraged: bool = False) -> dict:
    """Find arbitrage opportunities across instruments."""
    async with httpx.AsyncClient() as client:
        peg = (await client.get(
            f"{BASE_URL}/metrics/ME10001/absval/now",
            headers={"X-API-Key": API_KEY}
        )).json()
        
        spread = (await client.get(
            f"{BASE_URL}/metrics/ME10001/spread/now",
            headers={"X-API-Key": API_KEY}
        )).json()
        
        reserves = (await client.get(
            f"{BASE_URL}/metrics/ME10001/reserves/now",
            headers={"X-API-Key": API_KEY}
        )).json()
    
    usdt_price = peg["value"]
    btc_spread = spread["data"]["btc_usd_usdt_spread_pct"]
    cash_reserves = reserves["data"]["cash_pct"]
    
    if usdt_price >= 0.998 or btc_spread < min_spread_pct:
        return {"opportunity": False, "reason": "No significant depeg or spread"}
    
    result = {
        "opportunity": True,
        "usdt_price": usdt_price,
        "spread_pct": btc_spread,
        "gross_profit_pct": round((1 / usdt_price - 1) * 100, 3),
        "reserve_backing": f"{cash_reserves}% cash",
        "confidence": "HIGH" if cash_reserves > 70 else "MEDIUM" if cash_reserves > 50 else "LOW",
        "execution": {
            "no_leverage": {
                "buy": "BTC with USDT on Binance",
                "sell": "BTC/USD on Coinbase OR short IBIT ETF",
                "risk": "Exchange counterparty, repeg timing"
            }
        }
    }
    
    if include_leveraged:
        result["execution"]["with_leverage"] = {
            "long": "BTC-USDT perp (Binance) at 2x",
            "short": "BTC-USD perp (Kraken) at 2x",
            "risk": "Liquidation, funding divergence, USDT margin risk"
        }
    
    return result

@server.resource("madjik://metrics/ME10001")
async def get_stablecoin_resource() -> Resource:
    """Resource listing all ME10001 endpoints."""
    return Resource(
        uri="madjik://metrics/ME10001",
        name="ME10001 Stablecoin Peg Metrics",
        description="Endpoints: absval, relcha, spread, reserves, velocity, backing_ratio",
        mimeType="application/json"
    )

if __name__ == "__main__":
    asyncio.run(server.run())

A2A (Agent-to-Agent) Integration

import asyncio
from dataclasses import dataclass, field
from typing import List, Dict, Callable
from datetime import datetime
import json

@dataclass
class A2AMessage:
    """Standard A2A message format."""
    sender_id: str
    message_type: str
    payload: Dict
    priority: int  # 1-5, 5 = critical
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    requires_ack: bool = False

class StablecoinSentinelAgent:
    """
    Dedicated agent monitoring stablecoin health 24/7.
    Broadcasts alerts to portfolio and execution agents.
    """
    
    def __init__(self, agent_id: str, api_key: str):
        self.agent_id = agent_id
        self.api_key = api_key
        self.subscribers: List[Callable] = []
        self.alert_thresholds = {
            "peg_warning": 0.998,
            "peg_critical": 0.995,
            "reserve_warning": 60,
            "velocity_warning": 2.0
        }
    
    def subscribe(self, callback: Callable):
        """Subscribe to alerts from this agent."""
        self.subscribers.append(callback)
    
    async def monitor_loop(self, interval_seconds: int = 10):
        """Continuous monitoring with A2A broadcasts."""
        while True:
            try:
                status = await self._get_stablecoin_status()
                alerts = self._evaluate_alerts(status)
                
                for alert in alerts:
                    await self._broadcast(alert)
                
            except Exception as e:
                print(f"Monitor error: {e}")
            
            await asyncio.sleep(interval_seconds)
    
    async def _get_stablecoin_status(self) -> Dict:
        """Fetch current stablecoin status from Madjik."""
        import httpx
        headers = {"X-API-Key": self.api_key}
        
        async with httpx.AsyncClient() as client:
            peg = (await client.get(
                "https://api.madjik.io/v1/metrics/ME10001/absval/now",
                headers=headers
            )).json()
            
            reserves = (await client.get(
                "https://api.madjik.io/v1/metrics/ME10001/reserves/now",
                headers=headers
            )).json()
            
            velocity = (await client.get(
                "https://api.madjik.io/v1/metrics/ME10001/velocity/now",
                headers=headers
            )).json()
        
        return {
            "peg": peg["value"],
            "cash_reserves": reserves["data"]["cash_pct"],
            "velocity": velocity["value"]
        }
    
    def _evaluate_alerts(self, status: Dict) -> List[A2AMessage]:
        """Evaluate status and generate alerts."""
        alerts = []
        
        if status["peg"] < self.alert_thresholds["peg_critical"]:
            alerts.append(A2AMessage(
                sender_id=self.agent_id,
                message_type="STABLECOIN_CRITICAL",
                payload={
                    "peg": status["peg"],
                    "recommended_action": "EXIT_ALL_USDT_IMMEDIATELY",
                    "safe_instruments": ["IBIT", "FBTC", "BTC/USD spot", "MSTR"]
                },
                priority=5,
                requires_ack=True
            ))
        
        elif status["peg"] < self.alert_thresholds["peg_warning"]:
            alerts.append(A2AMessage(
                sender_id=self.agent_id,
                message_type="STABLECOIN_WARNING",
                payload={
                    "peg": status["peg"],
                    "reserves": status["cash_reserves"],
                    "recommended_action": "REDUCE_USDT_EXPOSURE",
                    "rotation_target": "50% to USD instruments"
                },
                priority=4
            ))
        
        if status["velocity"] > self.alert_thresholds["velocity_warning"]:
            alerts.append(A2AMessage(
                sender_id=self.agent_id,
                message_type="REDEMPTION_VELOCITY_HIGH",
                payload={
                    "velocity": status["velocity"],
                    "interpretation": "Redemption pressure elevated - potential run forming",
                    "lead_time": "2-4 hours before price impact typically"
                },
                priority=4
            ))
        
        return alerts
    
    async def _broadcast(self, message: A2AMessage):
        """Broadcast message to all subscribers."""
        for callback in self.subscribers:
            try:
                if asyncio.iscoroutinefunction(callback):
                    await callback(message)
                else:
                    callback(message)
            except Exception as e:
                print(f"Broadcast error to subscriber: {e}")


class PortfolioManagerAgent:
    """
    Agent that receives alerts and manages portfolio risk.
    """
    
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.positions = {}
        self.pending_rotations = []
    
    async def handle_alert(self, message: A2AMessage):
        """Process incoming A2A alert."""
        print(f"[{self.agent_id}] Received: {message.message_type} (priority {message.priority})")
        
        if message.message_type == "STABLECOIN_CRITICAL":
            await self._emergency_rotation(message.payload)
        
        elif message.message_type == "STABLECOIN_WARNING":
            await self._gradual_rotation(message.payload)
        
        elif message.message_type == "REDEMPTION_VELOCITY_HIGH":
            await self._prepare_rotation(message.payload)
    
    async def _emergency_rotation(self, payload: Dict):
        """Execute emergency rotation away from USDT."""
        print(f"[{self.agent_id}] EMERGENCY ROTATION TRIGGERED")
        
        rotations = [
            {"action": "CLOSE", "instrument": "BTC-USDT-PERP", "venue": "all"},
            {"action": "CONVERT", "from": "USDT", "to": "BTC", "venue": "Binance"},
            {"action": "OPEN", "instrument": "IBIT", "amount": "equivalent_exposure", "venue": "IBKR"}
        ]
        
        for rotation in rotations:
            print(f"  Executing: {rotation}")
            # Execute via exchange APIs
    
    async def _gradual_rotation(self, payload: Dict):
        """Execute gradual rotation based on risk level."""
        target_pct = int(payload.get("rotation_target", "50").split("%")[0])
        print(f"[{self.agent_id}] Gradual rotation: {target_pct}% to USD instruments")


# Run the multi-agent system
async def main():
    sentinel = StablecoinSentinelAgent("sentinel-001", "your_api_key")
    portfolio_mgr = PortfolioManagerAgent("portfolio-001")
    
    # Subscribe portfolio manager to sentinel alerts
    sentinel.subscribe(portfolio_mgr.handle_alert)
    
    # Run monitoring
    await sentinel.monitor_loop(interval_seconds=10)

if __name__ == "__main__":
    asyncio.run(main())

Risk Matrix

Recommended Endpoints