my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""CoinGecko API Client for Real-Time Cryptocurrency Data"""
#2
#3	import httpx
#4	from typing import Optional, List, Dict, Any
#5
#6
#7	class CoinGeckoClient:
#8	"""Client for CoinGecko Pro API"""
#9
#10	def __init__(self, api_key: str, use_pro: bool = True):
#11	"""
#12	Initialize CoinGecko client.
#13
#14	Args:
#15	api_key: CoinGecko Pro API key
#16	use_pro: Use Pro API endpoint (default: True)
#17	"""
#18	self.api_key = api_key
#19	self.base_url = "https://pro-api.coingecko.com/api/v3" if use_pro else "https://api.coingecko.com/api/v3"
#20
#21	self._client = httpx.AsyncClient(
#22	base_url=self.base_url,
#23	headers={
#24	"x-cg-pro-api-key": api_key,
#25	"Accept": "application/json",
#26	},
#27	timeout=30.0,
#28	)
#29
#30	async def get_price(
#31	self,
#32	coin_ids: List[str],
#33	vs_currencies: List[str] = ["usd"],
#34	include_market_cap: bool = True,
#35	include_24hr_vol: bool = True,
#36	include_24hr_change: bool = True,
#37	include_last_updated_at: bool = True,
#38	) -> Dict[str, Any]:
#39	"""
#40	Get current prices for coins by their IDs.
#41
#42	Args:
#43	coin_ids: List of coin IDs (e.g., ["bitcoin", "ethereum"])
#44	vs_currencies: Target currencies (default: ["usd"])
#45	include_market_cap: Include market cap data
#46	include_24hr_vol: Include 24hr volume
#47	include_24hr_change: Include 24hr price change
#48	include_last_updated_at: Include last updated timestamp
#49
#50	Returns:
#51	Dict with price data for each coin
#52	"""
#53	params = {
#54	"ids": ",".join(coin_ids),
#55	"vs_currencies": ",".join(vs_currencies),
#56	"include_market_cap": str(include_market_cap).lower(),
#57	"include_24hr_vol": str(include_24hr_vol).lower(),
#58	"include_24hr_change": str(include_24hr_change).lower(),
#59	"include_last_updated_at": str(include_last_updated_at).lower(),
#60	}
#61
#62	response = await self._client.get("/simple/price", params=params)
#63	response.raise_for_status()
#64	return response.json()
#65
#66	async def get_coin_markets(
#67	self,
#68	vs_currency: str = "usd",
#69	ids: Optional[List[str]] = None,
#70	category: Optional[str] = None,
#71	order: str = "market_cap_desc",
#72	per_page: int = 100,
#73	page: int = 1,
#74	sparkline: bool = False,
#75	price_change_percentage: Optional[str] = None,
#76	) -> List[Dict[str, Any]]:
#77	"""
#78	Get market data for coins (price, market cap, volume, etc.).
#79
#80	Args:
#81	vs_currency: Target currency (default: "usd")
#82	ids: Filter by coin IDs
#83	category: Filter by category
#84	order: Sort order (market_cap_desc, volume_desc, etc.)
#85	per_page: Results per page (max 250)
#86	page: Page number
#87	sparkline: Include 7-day sparkline
#88	price_change_percentage: Include price change for periods (e.g., "1h,24h,7d")
#89
#90	Returns:
#91	List of coin market data
#92	"""
#93	params = {
#94	"vs_currency": vs_currency,
#95	"order": order,
#96	"per_page": per_page,
#97	"page": page,
#98	"sparkline": str(sparkline).lower(),
#99	}
#100
#101	if ids:
#102	params["ids"] = ",".join(ids)
#103	if category:
#104	params["category"] = category
#105	if price_change_percentage:
#106	params["price_change_percentage"] = price_change_percentage
#107
#108	response = await self._client.get("/coins/markets", params=params)
#109	response.raise_for_status()
#110	return response.json()
#111
#112	async def get_coin_data(
#113	self,
#114	coin_id: str,
#115	localization: bool = False,
#116	tickers: bool = False,
#117	market_data: bool = True,
#118	community_data: bool = False,
#119	developer_data: bool = False,
#120	sparkline: bool = False,
#121	) -> Dict[str, Any]:
#122	"""
#123	Get detailed data for a specific coin.
#124
#125	Args:
#126	coin_id: Coin ID (e.g., "bitcoin")
#127	localization: Include localized data
#128	tickers: Include ticker data
#129	market_data: Include market data
#130	community_data: Include community data
#131	developer_data: Include developer data
#132	sparkline: Include sparkline
#133
#134	Returns:
#135	Detailed coin data
#136	"""
#137	params = {
#138	"localization": str(localization).lower(),
#139	"tickers": str(tickers).lower(),
#140	"market_data": str(market_data).lower(),
#141	"community_data": str(community_data).lower(),
#142	"developer_data": str(developer_data).lower(),
#143	"sparkline": str(sparkline).lower(),
#144	}
#145
#146	response = await self._client.get(f"/coins/{coin_id}", params=params)
#147	response.raise_for_status()
#148	return response.json()
#149
#150	async def get_market_chart(
#151	self,
#152	coin_id: str,
#153	vs_currency: str = "usd",
#154	days: int = 30,
#155	interval: Optional[str] = None,
#156	) -> Dict[str, Any]:
#157	"""
#158	Get historical market data (price, market cap, volume).
#159
#160	Args:
#161	coin_id: Coin ID (e.g., "bitcoin")
#162	vs_currency: Target currency (default: "usd")
#163	days: Number of days (1/7/14/30/90/180/365/max)
#164	interval: Data interval (daily for 90+ days, leave None for auto)
#165
#166	Returns:
#167	Historical chart data with prices, market caps, volumes
#168	"""
#169	params = {
#170	"vs_currency": vs_currency,
#171	"days": days,
#172	}
#173
#174	if interval:
#175	params["interval"] = interval
#176
#177	response = await self._client.get(f"/coins/{coin_id}/market_chart", params=params)
#178	response.raise_for_status()
#179	return response.json()
#180
#181	async def get_trending(self) -> Dict[str, Any]:
#182	"""
#183	Get trending search coins in the last 24 hours.
#184
#185	Returns:
#186	Trending coins data
#187	"""
#188	response = await self._client.get("/search/trending")
#189	response.raise_for_status()
#190	return response.json()
#191
#192	async def get_global_data(self) -> Dict[str, Any]:
#193	"""
#194	Get global cryptocurrency market data.
#195
#196	Returns:
#197	Global market data (total market cap, volume, BTC dominance, etc.)
#198	"""
#199	response = await self._client.get("/global")
#200	response.raise_for_status()
#201	return response.json()
#202
#203	async def search_coins(self, query: str) -> Dict[str, Any]:
#204	"""
#205	Search for coins, categories, and markets.
#206
#207	Args:
#208	query: Search query
#209
#210	Returns:
#211	Search results
#212	"""
#213	params = {"query": query}
#214	response = await self._client.get("/search", params=params)
#215	response.raise_for_status()
#216	return response.json()
#217
#218	async def get_coin_ohlc(
#219	self,
#220	coin_id: str,
#221	vs_currency: str = "usd",
#222	days: int = 7,
#223	) -> List[List[float]]:
#224	"""
#225	Get OHLC (Open, High, Low, Close) chart data.
#226
#227	Args:
#228	coin_id: Coin ID (e.g., "bitcoin")
#229	vs_currency: Target currency (default: "usd")
#230	days: Number of days (1/7/14/30/90/180/365)
#231
#232	Returns:
#233	OHLC data [[timestamp, open, high, low, close], ...]
#234	"""
#235	params = {
#236	"vs_currency": vs_currency,
#237	"days": days,
#238	}
#239
#240	response = await self._client.get(f"/coins/{coin_id}/ohlc", params=params)
#241	response.raise_for_status()
#242	return response.json()
#243
#244	async def get_top_gainers_losers(self) -> Dict[str, Any]:
#245	"""
#246	Get top 30 coins with largest price gain and loss (requires Pro API).
#247
#248	Returns:
#249	Top gainers and losers data
#250	"""
#251	response = await self._client.get("/coins/top_gainers_losers")
#252	response.raise_for_status()
#253	return response.json()
#254
#255	async def get_categories(self) -> List[Dict[str, Any]]:
#256	"""
#257	Get all coin categories with market data.
#258
#259	Returns:
#260	List of categories with market data
#261	"""
#262	response = await self._client.get("/coins/categories")
#263	response.raise_for_status()
#264	return response.json()
#265
#266	async def close(self):
#267	"""Close the HTTP client"""
#268	await self._client.aclose()
#269
#270	async def __aenter__(self):
#271	return self
#272
#273	async def __aexit__(self, exc_type, exc_val, exc_tb):
#274	await self.close()
#275
#276
#277	# Factory function for easy initialization
#278	def create_coingecko_client(api_key: Optional[str] = None) -> Optional[CoinGeckoClient]:
#279	"""
#280	Create a CoinGecko client from environment variables or parameters.
#281
#282	Args:
#283	api_key: CoinGecko API key (or COINGECKO_API_KEY env var)
#284
#285	Returns:
#286	CoinGeckoClient instance or None if API key is missing
#287	"""
#288	import os
#289
#290	api_key = api_key or os.getenv("COINGECKO_API_KEY")
#291
#292	if not api_key:
#293	return None
#294
#295	return CoinGeckoClient(api_key=api_key)
#296

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public