1#!/usr/bin/env python3
2"""
3CoinGecko Market Discovery Tools
4 
5Tools for discovering trending coins, top movers, and newly listed coins.
6"""
7 
8import os
9from dotenv import load_dotenv
10import json
11import argparse
12from typing import Dict, Any
13 
14from core.http_client import proxied_get
15 
16# Load environment variables
17load_dotenv()
18 
19 
20# MCP Tool Schemas
21MCP_TRENDING_SCHEMA = {
22    "name": "cg_trending",
23    "title": "CoinGecko Trending Coins",
24    "description": "Get trending coins in the last 24 hours based on user search data.",
25    "inputSchema": {
26        "type": "object",
27        "properties": {},
28        "additionalProperties": False
29    }
30}
31 
32MCP_TOP_GAINERS_LOSERS_SCHEMA = {
33    "name": "cg_top_gainers_losers",
34    "title": "CoinGecko Top Gainers/Losers",
35    "description": "Get top 30 gainers and losers by price change percentage.",
36    "inputSchema": {
37        "type": "object",
38        "properties": {
39            "vs_currency": {
40                "type": "string",
41                "description": "Target currency for price data",
42                "default": "usd"
43            },
44            "duration": {
45                "type": "string",
46                "description": "Time duration: 1h, 24h, 7d, 14d, 30d, 60d, 1y",
47                "default": "24h",
48                "enum": ["1h", "24h", "7d", "14d", "30d", "60d", "1y"]
49            }
50        },
51        "additionalProperties": False
52    }
53}
54 
55MCP_NEW_COINS_SCHEMA = {
56    "name": "cg_new_coins",
57    "title": "CoinGecko Recently Added Coins",
58    "description": "Get recently added coins to CoinGecko.",
59    "inputSchema": {
60        "type": "object",
61        "properties": {},
62        "additionalProperties": False
63    }
64}
65 
66 
67def get_api_key() -> str:
68    """Get CoinGecko API key from environment."""
69    api_key = os.getenv("COINGECKO_API_KEY")
70    if not api_key:
71        raise ValueError("COINGECKO_API_KEY environment variable is required")
72    return api_key
73 
74 
75def get_trending() -> Dict[str, Any]:
76    """
77    Get trending coins in the last 24 hours.
78 
79    Based on user search data on CoinGecko.
80 
81    Returns:
82        Dictionary with trending coins, nfts, and categories
83    """
84    api_key = get_api_key()
85 
86    url = "https://pro-api.coingecko.com/api/v3/search/trending"
87    headers = {"x-cg-pro-api-key": api_key}
88 
89    response = proxied_get(url, headers=headers, timeout=15)
90    response.raise_for_status()
91    data = response.json()
92 
93    # Format coins
94    trending_coins = []
95    for item in data.get("coins", []):
96        coin = item.get("item", {})
97        trending_coins.append({
98            "rank": coin.get("score", 0) + 1,
99            "id": coin.get("id", ""),
100            "symbol": coin.get("symbol", "").upper(),
101            "name": coin.get("name", ""),
102            "market_cap_rank": coin.get("market_cap_rank"),
103            "price_btc": coin.get("price_btc"),
104            "price_change_24h": coin.get("data", {}).get("price_change_percentage_24h", {}).get("usd"),
105            "market_cap": coin.get("data", {}).get("market_cap"),
106            "total_volume": coin.get("data", {}).get("total_volume"),
107            "sparkline": coin.get("data", {}).get("sparkline")
108        })
109 
110    # Format NFTs
111    trending_nfts = []
112    for nft in data.get("nfts", []):
113        trending_nfts.append({
114            "id": nft.get("id", ""),
115            "name": nft.get("name", ""),
116            "symbol": nft.get("symbol", ""),
117            "floor_price_24h_change": nft.get("floor_price_24h_percentage_change")
118        })
119 
120    # Format categories
121    trending_categories = []
122    for cat in data.get("categories", []):
123        trending_categories.append({
124            "id": cat.get("id"),
125            "name": cat.get("name", ""),
126            "market_cap_change_24h": cat.get("market_cap_1h_change")
127        })
128 
129    return {
130        "coins": trending_coins,
131        "nfts": trending_nfts,
132        "categories": trending_categories
133    }
134 
135 
136def get_top_gainers_losers(
137    vs_currency: str = "usd",
138    duration: str = "24h"
139) -> Dict[str, Any]:
140    """
141    Get top 30 gainers and losers by price change percentage.
142 
143    Args:
144        vs_currency: Target currency (usd, eur, btc)
145        duration: Time duration (1h, 24h, 7d, 14d, 30d, 60d, 1y)
146 
147    Returns:
148        Dictionary with top_gainers and top_losers lists
149    """
150    api_key = get_api_key()
151 
152    url = "https://pro-api.coingecko.com/api/v3/coins/top_gainers_losers"
153    headers = {"x-cg-pro-api-key": api_key}
154    params = {
155        "vs_currency": vs_currency,
156        "duration": duration
157    }
158 
159    response = proxied_get(url, headers=headers, params=params, timeout=15)
160    response.raise_for_status()
161    data = response.json()
162 
163    # Format gainers
164    gainers = []
165    for coin in data.get("top_gainers", []):
166        gainers.append({
167            "id": coin.get("id", ""),
168            "symbol": coin.get("symbol", "").upper(),
169            "name": coin.get("name", ""),
170            "image": coin.get("image", ""),
171            "price": coin.get(vs_currency),
172            "price_change_percentage": coin.get(f"{vs_currency}_24h_change")
173        })
174 
175    # Format losers
176    losers = []
177    for coin in data.get("top_losers", []):
178        losers.append({
179            "id": coin.get("id", ""),
180            "symbol": coin.get("symbol", "").upper(),
181            "name": coin.get("name", ""),
182            "image": coin.get("image", ""),
183            "price": coin.get(vs_currency),
184            "price_change_percentage": coin.get(f"{vs_currency}_24h_change")
185        })
186 
187    return {
188        "vs_currency": vs_currency,
189        "duration": duration,
190        "top_gainers": gainers,
191        "top_losers": losers
192    }
193 
194 
195def get_new_coins() -> Dict[str, Any]:
196    """
197    Get recently added coins to CoinGecko.
198 
199    Returns:
200        Dictionary with list of newly added coins
201    """
202    api_key = get_api_key()
203 
204    url = "https://pro-api.coingecko.com/api/v3/coins/list/new"
205    headers = {"x-cg-pro-api-key": api_key}
206 
207    response = proxied_get(url, headers=headers, timeout=15)
208    response.raise_for_status()
209    data = response.json()
210 
211    coins = []
212    for coin in data:
213        coins.append({
214            "id": coin.get("id", ""),
215            "symbol": coin.get("symbol", "").upper(),
216            "name": coin.get("name", ""),
217            "activated_at": coin.get("activated_at")
218        })
219 
220    return {
221        "new_coins": coins,
222        "count": len(coins)
223    }
224 
225 
226def main():
227    """CLI interface for market discovery tools."""
228    parser = argparse.ArgumentParser(
229        description="CoinGecko Market Discovery Tools",
230        formatter_class=argparse.RawDescriptionHelpFormatter
231    )
232 
233    subparsers = parser.add_subparsers(dest="command")
234 
235    # Trending command
236    subparsers.add_parser("trending", help="Get trending coins")
237 
238    # Gainers/losers command
239    gl_parser = subparsers.add_parser("gainers-losers", help="Get top gainers and losers")
240    gl_parser.add_argument("--currency", default="usd")
241    gl_parser.add_argument(
242        "--duration", default="24h",
243        choices=["1h", "24h", "7d", "14d", "30d", "60d", "1y"]
244    )
245 
246    # New coins command
247    subparsers.add_parser("new", help="Get newly added coins")
248 
249    parser.add_argument("--schema", action="store_true", help="Output MCP schema")
250 
251    args = parser.parse_args()
252 
253    if args.schema:
254        schemas = {
255            "trending": MCP_TRENDING_SCHEMA,
256            "top_gainers_losers": MCP_TOP_GAINERS_LOSERS_SCHEMA,
257            "new_coins": MCP_NEW_COINS_SCHEMA
258        }
259        print(json.dumps(schemas, indent=2))
260        return 0
261 
262    try:
263        if args.command == "trending":
264            result = get_trending()
265        elif args.command == "gainers-losers":
266            result = get_top_gainers_losers(args.currency, args.duration)
267        elif args.command == "new":
268            result = get_new_coins()
269        else:
270            parser.print_help()
271            return 0
272 
273        print(json.dumps(result, indent=2, ensure_ascii=False))
274        return 0
275 
276    except Exception as e:
277        print(json.dumps({"error": str(e)}, indent=2))
278        return 1
279 
280 
281if __name__ == "__main__":
282    exit(main())
283