from langchain.tools import BaseTool
import requests
class TheCMXTool(BaseTool):
name = "crypto_universal_access"
description = "Access any crypto service with universal API"
def _run(self, query: str) -> str:
headers = {'X-Universal-Key': 'your_thecmx_key'}
response = requests.get(
'https://thecmx.com/api/universal/binance/spot_price?symbol=BTCUSDT',
headers=headers
)
return f"BTC Price: ${response.json()['price']}"
agent = initialize_agent([TheCMXTool()], llm)
result = agent.run("What's the current Bitcoin price?")
class TheCMXPlugin(AutoGPTPlugin):
def __init__(self):
self.name = "TheCMX Universal Crypto API"
self.api_key = os.getenv('THECMX_API_KEY')
def get_crypto_data(self, exchange: str, symbol: str):
"""Get crypto data from any exchange"""
url = f'https://thecmx.com/api/universal/{exchange}/spot_price'
params = {'symbol': symbol}
headers = {'X-Universal-Key': self.api_key}
response = requests.get(url, params=params, headers=headers)
return response.json()
def analyze_portfolio(self, assets: list):
"""Analyze portfolio across multiple data sources"""
portfolio_data = {}
for asset in assets:
portfolio_data[asset] = {
'binance': self.get_crypto_data('binance', asset),
'coinbase': self.get_crypto_data('coinbase', asset),
'nansen': self.get_onchain_data('nansen', asset)
}
return portfolio_data
from openclaw import skill
import requests
@skill("crypto_universal_access")
def get_crypto_data(exchange: str, symbol: str, data_type: str = "price"):
"""
Universal crypto data access across 500+ APIs
Args:
exchange: Exchange or service name (binance, coingecko, nansen, etc.)
symbol: Trading symbol or asset identifier
data_type: Type of data (price, orderbook, sentiment, etc.)
"""
api_key = os.getenv('THECMX_API_KEY')
headers = {'X-Universal-Key': api_key}
if data_type == "price":
url = f'https://thecmx.com/api/universal/{exchange}/spot_price'
params = {'symbol': symbol}
elif data_type == "sentiment":
url = f'https://thecmx.com/api/universal/{exchange}/social_sentiment'
params = {'asset': symbol}
response = requests.get(url, params=params, headers=headers)
return response.json()
from thecmx import TheCMXClient
client = TheCMXClient(api_key="your_universal_key")
btc_binance = client.get_price("binance", "BTCUSDT")
btc_coinbase = client.get_price("coinbase", "BTC-USD")
wallet_data = client.get_wallet_labels("nansen", "0x123...")
sentiment = client.get_social_sentiment("santiment", "bitcoin")
class ArbitrageBot:
def scan_opportunities(self):
exchanges = ["binance", "coinbase", "kraken", "okx"]
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
opportunities = []
for symbol in symbols:
prices = {}
for exchange in exchanges:
prices[exchange] = client.get_price(exchange, symbol)
max_price = max(prices.values())
min_price = min(prices.values())
spread = (max_price - min_price) / min_price * 100
if spread > 0.5:
opportunities.append({
'symbol': symbol,
'spread': spread,
'prices': prices
})
return opportunities