244 lines
8.3 KiB
Python
244 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
DEFAULT_TIMEOUT = 20
|
|
|
|
|
|
def fetch_json(url, headers=None, timeout=DEFAULT_TIMEOUT):
|
|
merged_headers = {
|
|
"Accept": "application/json",
|
|
"User-Agent": "Mozilla/5.0 (compatible; OpenClaw Coin Hunter/1.0)",
|
|
}
|
|
if headers:
|
|
merged_headers.update(headers)
|
|
req = urllib.request.Request(url, headers=merged_headers)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
data = resp.read()
|
|
return json.loads(data.decode("utf-8"))
|
|
|
|
|
|
def print_json(data):
|
|
print(json.dumps(data, ensure_ascii=False, indent=2))
|
|
|
|
|
|
def bybit_ticker(symbol: str):
|
|
url = (
|
|
"https://api.bybit.com/v5/market/tickers?category=spot&symbol="
|
|
+ urllib.parse.quote(symbol.upper())
|
|
)
|
|
payload = fetch_json(url)
|
|
items = payload.get("result", {}).get("list", [])
|
|
if not items:
|
|
raise SystemExit(f"No Bybit spot ticker found for {symbol}")
|
|
item = items[0]
|
|
out = {
|
|
"provider": "bybit",
|
|
"symbol": symbol.upper(),
|
|
"lastPrice": item.get("lastPrice"),
|
|
"price24hPcnt": item.get("price24hPcnt"),
|
|
"highPrice24h": item.get("highPrice24h"),
|
|
"lowPrice24h": item.get("lowPrice24h"),
|
|
"turnover24h": item.get("turnover24h"),
|
|
"volume24h": item.get("volume24h"),
|
|
"bid1Price": item.get("bid1Price"),
|
|
"ask1Price": item.get("ask1Price"),
|
|
}
|
|
print_json(out)
|
|
|
|
|
|
def bybit_klines(symbol: str, interval: str, limit: int):
|
|
params = urllib.parse.urlencode({
|
|
"category": "spot",
|
|
"symbol": symbol.upper(),
|
|
"interval": interval,
|
|
"limit": str(limit),
|
|
})
|
|
url = f"https://api.bybit.com/v5/market/kline?{params}"
|
|
payload = fetch_json(url)
|
|
rows = payload.get("result", {}).get("list", [])
|
|
out = {
|
|
"provider": "bybit",
|
|
"symbol": symbol.upper(),
|
|
"interval": interval,
|
|
"candles": [
|
|
{
|
|
"startTime": r[0],
|
|
"open": r[1],
|
|
"high": r[2],
|
|
"low": r[3],
|
|
"close": r[4],
|
|
"volume": r[5],
|
|
"turnover": r[6],
|
|
}
|
|
for r in rows
|
|
],
|
|
}
|
|
print_json(out)
|
|
|
|
|
|
def dexscreener_search(query: str):
|
|
url = "https://api.dexscreener.com/latest/dex/search/?q=" + urllib.parse.quote(query)
|
|
payload = fetch_json(url)
|
|
pairs = payload.get("pairs") or []
|
|
out = []
|
|
for p in pairs[:10]:
|
|
out.append({
|
|
"chainId": p.get("chainId"),
|
|
"dexId": p.get("dexId"),
|
|
"pairAddress": p.get("pairAddress"),
|
|
"url": p.get("url"),
|
|
"baseToken": p.get("baseToken"),
|
|
"quoteToken": p.get("quoteToken"),
|
|
"priceUsd": p.get("priceUsd"),
|
|
"liquidityUsd": (p.get("liquidity") or {}).get("usd"),
|
|
"fdv": p.get("fdv"),
|
|
"marketCap": p.get("marketCap"),
|
|
"volume24h": (p.get("volume") or {}).get("h24"),
|
|
"buys24h": ((p.get("txns") or {}).get("h24") or {}).get("buys"),
|
|
"sells24h": ((p.get("txns") or {}).get("h24") or {}).get("sells"),
|
|
})
|
|
print_json({"provider": "dexscreener", "query": query, "pairs": out})
|
|
|
|
|
|
def dexscreener_token(chain: str, address: str):
|
|
url = f"https://api.dexscreener.com/tokens/v1/{urllib.parse.quote(chain)}/{urllib.parse.quote(address)}"
|
|
payload = fetch_json(url)
|
|
pairs = payload if isinstance(payload, list) else payload.get("pairs") or []
|
|
out = []
|
|
for p in pairs[:10]:
|
|
out.append({
|
|
"chainId": p.get("chainId"),
|
|
"dexId": p.get("dexId"),
|
|
"pairAddress": p.get("pairAddress"),
|
|
"baseToken": p.get("baseToken"),
|
|
"quoteToken": p.get("quoteToken"),
|
|
"priceUsd": p.get("priceUsd"),
|
|
"liquidityUsd": (p.get("liquidity") or {}).get("usd"),
|
|
"fdv": p.get("fdv"),
|
|
"marketCap": p.get("marketCap"),
|
|
"volume24h": (p.get("volume") or {}).get("h24"),
|
|
})
|
|
print_json({"provider": "dexscreener", "chain": chain, "address": address, "pairs": out})
|
|
|
|
|
|
def coingecko_search(query: str):
|
|
url = "https://api.coingecko.com/api/v3/search?query=" + urllib.parse.quote(query)
|
|
payload = fetch_json(url)
|
|
coins = payload.get("coins") or []
|
|
out = []
|
|
for c in coins[:10]:
|
|
out.append({
|
|
"id": c.get("id"),
|
|
"name": c.get("name"),
|
|
"symbol": c.get("symbol"),
|
|
"marketCapRank": c.get("market_cap_rank"),
|
|
"thumb": c.get("thumb"),
|
|
})
|
|
print_json({"provider": "coingecko", "query": query, "coins": out})
|
|
|
|
|
|
def coingecko_coin(coin_id: str):
|
|
params = urllib.parse.urlencode({
|
|
"localization": "false",
|
|
"tickers": "false",
|
|
"market_data": "true",
|
|
"community_data": "false",
|
|
"developer_data": "false",
|
|
"sparkline": "false",
|
|
})
|
|
url = f"https://api.coingecko.com/api/v3/coins/{urllib.parse.quote(coin_id)}?{params}"
|
|
payload = fetch_json(url)
|
|
md = payload.get("market_data") or {}
|
|
out = {
|
|
"provider": "coingecko",
|
|
"id": payload.get("id"),
|
|
"symbol": payload.get("symbol"),
|
|
"name": payload.get("name"),
|
|
"marketCapRank": payload.get("market_cap_rank"),
|
|
"currentPriceUsd": (md.get("current_price") or {}).get("usd"),
|
|
"marketCapUsd": (md.get("market_cap") or {}).get("usd"),
|
|
"fullyDilutedValuationUsd": (md.get("fully_diluted_valuation") or {}).get("usd"),
|
|
"totalVolumeUsd": (md.get("total_volume") or {}).get("usd"),
|
|
"priceChangePercentage24h": md.get("price_change_percentage_24h"),
|
|
"priceChangePercentage7d": md.get("price_change_percentage_7d"),
|
|
"priceChangePercentage30d": md.get("price_change_percentage_30d"),
|
|
"circulatingSupply": md.get("circulating_supply"),
|
|
"totalSupply": md.get("total_supply"),
|
|
"maxSupply": md.get("max_supply"),
|
|
"homepage": (payload.get("links") or {}).get("homepage", [None])[0],
|
|
}
|
|
print_json(out)
|
|
|
|
|
|
def birdeye_token(address: str):
|
|
api_key = os.getenv("BIRDEYE_API_KEY") or os.getenv("BIRDEYE_APIKEY")
|
|
if not api_key:
|
|
raise SystemExit("Birdeye requires BIRDEYE_API_KEY in the environment")
|
|
url = "https://public-api.birdeye.so/defi/token_overview?address=" + urllib.parse.quote(address)
|
|
payload = fetch_json(url, headers={
|
|
"x-api-key": api_key,
|
|
"x-chain": "solana",
|
|
})
|
|
print_json({"provider": "birdeye", "address": address, "data": payload.get("data")})
|
|
|
|
|
|
def build_parser():
|
|
parser = argparse.ArgumentParser(description="Coin Hunter market data probe")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
p = sub.add_parser("bybit-ticker", help="Fetch Bybit spot ticker")
|
|
p.add_argument("symbol")
|
|
|
|
p = sub.add_parser("bybit-klines", help="Fetch Bybit spot klines")
|
|
p.add_argument("symbol")
|
|
p.add_argument("--interval", default="60", help="Bybit interval, e.g. 1, 5, 15, 60, 240, D")
|
|
p.add_argument("--limit", type=int, default=10)
|
|
|
|
p = sub.add_parser("dex-search", help="Search DexScreener by query")
|
|
p.add_argument("query")
|
|
|
|
p = sub.add_parser("dex-token", help="Fetch DexScreener token pairs by chain/address")
|
|
p.add_argument("chain")
|
|
p.add_argument("address")
|
|
|
|
p = sub.add_parser("gecko-search", help="Search CoinGecko")
|
|
p.add_argument("query")
|
|
|
|
p = sub.add_parser("gecko-coin", help="Fetch CoinGecko coin by id")
|
|
p.add_argument("coin_id")
|
|
|
|
p = sub.add_parser("birdeye-token", help="Fetch Birdeye token overview (Solana)")
|
|
p.add_argument("address")
|
|
|
|
return parser
|
|
|
|
|
|
def main():
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
if args.command == "bybit-ticker":
|
|
bybit_ticker(args.symbol)
|
|
elif args.command == "bybit-klines":
|
|
bybit_klines(args.symbol, args.interval, args.limit)
|
|
elif args.command == "dex-search":
|
|
dexscreener_search(args.query)
|
|
elif args.command == "dex-token":
|
|
dexscreener_token(args.chain, args.address)
|
|
elif args.command == "gecko-search":
|
|
coingecko_search(args.query)
|
|
elif args.command == "gecko-coin":
|
|
coingecko_coin(args.coin_id)
|
|
elif args.command == "birdeye-token":
|
|
birdeye_token(args.address)
|
|
else:
|
|
parser.error("Unknown command")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|