用 Claude 和 CoinGecko 构建实时加密货币分析 Agent

在波动剧烈的加密货币世界里，实时数据可能决定一笔交易究竟能否盈利，也可能让你错失机会。对于开发者、交易者和爱好者来说，构建能够获取、分析市场数据并据此采取行动的自动化工具，已经变得不可或缺。在这篇实战指南中，我们将结合 CoinGecko API 与 Anthropic 的 Claude AI，逐步构建一个实时加密货币情报 Agent。

为什么选择 CoinGecko + Claude？

CoinGecko API 是使用最广泛的免费加密货币数据 API 之一，覆盖数千种代币，提供实时价格、市值、交易量、历史数据等信息。另一方面，Claude 擅长推理、总结和工具调用。将两者结合起来，你就能构建一个不只是展示数字，还能解读数据、解释情况并给出建议的 Agent。

Anthropic API key（console.anthropic.com）

CoinGecko API key（可选；免费套餐：每分钟 50 次调用）

requests、anthropic、python-dotenv

第 1 步：从 CoinGecko 获取实时数据

CoinGecko 的 /simple/price endpoint 是最常用的核心接口。下面是一个简洁的封装：

import requests

def get_crypto_price(crypto_id='bitcoin', vs_currency='usd'):
    url = "https://api.coingecko.com/api/v3/simple/price"
    params = {
        'ids': crypto_id,
        'vs_currencies': vs_currency,
        'include_market_cap': 'true',
        'include_24hr_vol': 'true',
        'include_24hr_change': 'true'
    }
    headers = {'accept': 'application/json'}
    if COINGECKO_API_KEY:
        headers['x-cg-demo-api-key'] = COINGECKO_API_KEY

    resp = requests.get(url, params=params, headers=headers)
    return resp.json() if resp.status_code == 200 else None

# Usage
btc = get_crypto_price('bitcoin')
print(f"BTC: ${btc['bitcoin']['usd']}")

{
  "bitcoin": {
    "usd": 67300,
    "usd_market_cap": 1320000000000,
    "usd_24h_vol": 28000000000,
    "usd_24h_change": 2.35
  }
}

第 2 步：让 Claude 获得获取价格的能力（Tool Use）

Claude 的 tool use（function calling）功能允许它自行决定何时调用外部 API。我们定义一个工具 schema，再让 Claude 负责调度调用：

import anthropic

client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

price_tool = {
    "name": "get_crypto_price",
    "description": "Get live price, market cap, 24h volume and change for any crypto",
    "input_schema": {
        "type": "object",
        "properties": {
            "crypto_id": {
                "type": "string",
                "description": "CoinGecko ID, e.g. 'bitcoin', 'ethereum', 'solana'"
            },
            "currency": {
                "type": "string",
                "description": "Fiat currency, default 'usd'"
            }
        },
        "required": ["crypto_id"]
    }
}

def chat_with_crypto_agent(user_message):
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=messages,
            tools=[price_tool]
        )

        if response.stop_reason == "tool_use":
            # Claude wants to call our tool
            block = next(b for b in response.content if b.type == "tool_use")
            if block.name == "get_crypto_price":
                data = get_crypto_price(
                    block.input["crypto_id"],
                    block.input.get("currency", "usd")
                )
                messages.append({
                    "role": "user",
                    "content": [{
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(data)
                    }]
                })
        else:
            return response.content[0].text

print(chat_with_crypto_agent(
    "What's Ethereum at right now, and how's it performing vs yesterday?"
))

Claude 会识别用户意图，调用 get_crypto_price('ethereum')，接收返回的数据，然后生成类似下面这样的自然语言回答：

“Ethereum 当前价格为 3,450 美元，过去 24 小时上涨了 4.1%。其市值为 4,140 亿美元，24 小时交易量为 220 亿美元。这是一个表现强劲的上涨日——4.1% 的涨幅表明市场动能积极，可能是由……”

第 3 步：构建实时价格缓存

每次请求都调用 API 会白白消耗 rate limit。下面构建一个后台价格缓存，每 30 秒刷新一次：

import threading, time

price_cache = {}
cache_lock = threading.Lock()

def fetch_prices_for_coins(coin_ids=['bitcoin','ethereum','solana']):
    url = "https://api.coingecko.com/api/v3/simple/price"
    params = {
        'ids': ','.join(coin_ids),
        'vs_currencies': 'usd',
        'include_market_cap': 'true',
        'include_24hr_vol': 'true',
        'include_24hr_change': 'true'
    }
    headers = {'accept': 'application/json'}
    if COINGECKO_API_KEY:
        headers['x-cg-demo-api-key'] = COINGECKO_API_KEY

    try:
        resp = requests.get(url, params=params, headers=headers)
        if resp.status_code == 200:
            with cache_lock:
                price_cache.clear()
                price_cache.update(resp.json())
    except Exception as e:
        print(f"Cache update error: {e}")

def start_price_refresh(interval=30):
    def loop():
        while True:
            fetch_prices_for_coins()
            time.sleep(interval)
    t = threading.Thread(target=loop, daemon=True)
    t.start()

现在，你的 Agent 可以在几毫秒内作出响应，无须等待 API 返回结果。

第 4 步：添加热门代币与市场广度信息

CoinGecko 的 /search/trending endpoint 可以揭示当前最热门的代币：

def get_trending():
    url = "https://api.coingecko.com/api/v3/search/trending"
    resp = requests.get(url, headers={'accept': 'application/json'})
    coins = resp.json()['coins']
    return [{
        'name': c['item']['name'],
        'symbol': c['item']['symbol'],
        'market_cap_rank': c['item']['market_cap_rank'],
        'score': c['item']['score']
    } for c in coins[:7]]

把它作为第二个工具添加到 Claude Agent 中，你就可以提出这样的问题：

“今天最热门的代币有哪些？快速告诉我当前的市场脉搏。”

第 5 步：整合所有功能——一个可用于生产环境的 Agent

下面是完整的架构：

┌─────────────┐     ┌─────────────────┐     ┌──────────────┐
│   User      │────▶│  Claude Agent   │────▶│ CoinGecko    │
│  Question    │     │  (Tool Router)  │     │ API + Cache  │
└─────────────┘     └─────────────────┘     └──────────────┘
                            │
                    ┌───────▼───────┐
                    │  Natural      │
                    │  Language     │
                    │  Response     │
                    └───────────────┘

完整代码大约 150 行。这个 Agent 可以：

• 获取任意代币的实时价格
• 报告 24 小时涨跌幅、交易量和市值
• 展示热门代币
• 回答有关市场的自然语言问题
• 使用后台缓存，实现亚秒级响应

超越基础功能：扩展思路

将 CoinGecko 丰富的市场数据与 Claude 的推理能力结合起来，只需不到 200 行 Python 代码，就能构建一个真正实用的加密货币情报 Agent。这两项服务的免费套餐都相当慷慨，足以满足个人项目和原型开发的需求。

真正的威力来自 Claude 的 tool use：它知道何时获取数据，也知道如何解读数据，因此你不需要构建复杂的规则引擎。这种模式（LLM + 实时数据 API）的用途远不止加密货币：股票市场、天气、新闻、体育比分——任何需要将实时数据与自然语言结合起来的场景，都可以使用它。

Anthropic Tool Use 指南

你会用这种模式构建什么？欢迎在下方留言！🚀

如需采取进一步措施，你可以考虑屏蔽此人和/或举报滥用行为。