Coverage for api \ blockchain_endpoints.py: 0.00%

1""" 

2Blockchain Explorer API Endpoints 

3Provides endpoints for Ethereum, BSC, and Tron blockchain data 

4""" 

5 

6import os 

7import logging 

8import httpx 

9from datetime import datetime, timedelta 

10from typing import Optional, List, Dict, Any 

11from fastapi import APIRouter, HTTPException, Query 

12from pydantic import BaseModel 

13 

14from config import EXTERNAL_PROVIDERS, PROVIDER_FALLBACK_STRATEGY 

15 

16logger = logging.getLogger(__name__) 

17 

18router = APIRouter(prefix="/api/blockchain", tags=["blockchain"]) 

19 

20 

21# ============================================================================ 

22# Models 

23# ============================================================================ 

24 

25class BlockchainTransaction(BaseModel): 

26 """Blockchain transaction model""" 

27 tx_hash: str 

28 from_address: str 

29 to_address: str 

30 value: float 

31 value_usd: Optional[float] = None 

32 timestamp: datetime 

33 block_number: Optional[int] = None 

34 blockchain: str 

35 gas_used: Optional[int] = None 

36 gas_price: Optional[float] = None 

37 

38 

39# ============================================================================ 

40# Helper Functions 

41# ============================================================================ 

42 

43async def call_etherscan_api( 

44 action: str, 

45 params: Dict[str, Any] = None, 

46 timeout: float = 10.0 

47) -> Dict[str, Any]: 

48 """Call Etherscan API""" 

49 provider = EXTERNAL_PROVIDERS.get("etherscan") 

50 if not provider or not provider.get("api_key"): 

51 raise ValueError("Etherscan API key not configured") 

52 

53 base_url = provider["base_url"] 

54 api_key = provider["api_key"] 

55 

56 params = params or {} 

57 params.update({ 

58 "module": "proxy", 

59 "action": action, 

60 "apikey": api_key 

61 }) 

62 

63 async with httpx.AsyncClient(timeout=timeout) as client: 

64 response = await client.get(base_url, params=params) 

65 response.raise_for_status() 

66 data = response.json() 

67 

68 if data.get("status") == "0" and "rate limit" in data.get("message", "").lower(): 

69 raise Exception("Rate limit exceeded") 

70 

71 return data 

72 

73 

74async def call_bscscan_api( 

75 action: str, 

76 params: Dict[str, Any] = None, 

77 timeout: float = 10.0 

78) -> Dict[str, Any]: 

79 """Call BSCScan API""" 

80 provider = EXTERNAL_PROVIDERS.get("bscscan") 

81 if not provider or not provider.get("api_key"): 

82 raise ValueError("BSCScan API key not configured") 

83 

84 base_url = provider["base_url"] 

85 api_key = provider["api_key"] 

86 

87 params = params or {} 

88 params.update({ 

89 "module": "proxy", 

90 "action": action, 

91 "apikey": api_key 

92 }) 

93 

94 async with httpx.AsyncClient(timeout=timeout) as client: 

95 response = await client.get(base_url, params=params) 

96 response.raise_for_status() 

97 data = response.json() 

98 

99 if data.get("status") == "0" and "rate limit" in data.get("message", "").lower(): 

100 raise Exception("Rate limit exceeded") 

101 

102 return data 

103 

104 

105async def call_tronscan_api( 

106 endpoint: str, 

107 params: Dict[str, Any] = None, 

108 timeout: float = 10.0 

109) -> Dict[str, Any]: 

110 """Call TronScan API""" 

111 provider = EXTERNAL_PROVIDERS.get("tronscan") 

112 if not provider or not provider.get("api_key"): 

113 raise ValueError("TronScan API key not configured") 

114 

115 base_url = provider["base_url"] 

116 api_key = provider["api_key"] 

117 

118 url = f"{base_url}/{endpoint}" 

119 headers = { 

120 "TRON-PRO-API-KEY": api_key 

121 } 

122 

123 async with httpx.AsyncClient(timeout=timeout, headers=headers) as client: 

124 response = await client.get(url, params=params or {}) 

125 response.raise_for_status() 

126 return response.json() 

127 

128 

129# ============================================================================ 

130# Ethereum Endpoints 

131# ============================================================================ 

132 

133@router.get("/ethereum/recent-transactions") 

134async def get_ethereum_transactions( 

135 limit: int = Query(default=20, ge=1, le=100), 

136 min_value_usd: Optional[float] = Query(default=100000, description="Minimum transaction value in USD") 

137): 

138 """ 

139 Get recent large Ethereum transactions 

140  

141 Args: 

142 limit: Maximum number of transactions to return 

143 min_value_usd: Minimum transaction value in USD 

144 """ 

145 try: 

146 fallback_config = PROVIDER_FALLBACK_STRATEGY.get("etherscan", {}) 

147 

148 # Get latest block 

149 try: 

150 block_data = await call_etherscan_api("eth_blockNumber") 

151 latest_block = int(block_data.get("result", "0x0"), 16) 

152 

153 # Get recent blocks (approximate - we'll get transactions from last 100 blocks) 

154 transactions = [] 

155 start_block = max(0, latest_block - 100) 

156 

157 for block_num in range(latest_block, start_block, -1): 

158 if len(transactions) >= limit: 

159 break 

160 

161 try: 

162 block_data = await call_etherscan_api( 

163 "eth_getBlockByNumber", 

164 params={ 

165 "tag": hex(block_num), 

166 "boolean": "true" 

167 } 

168 ) 

169 

170 block_result = block_data.get("result", {}) 

171 txs = block_result.get("transactions", []) 

172 

173 for tx in txs[:10]: # Limit per block 

174 if len(transactions) >= limit: 

175 break 

176 

177 value_wei = int(tx.get("value", "0x0"), 16) 

178 value_eth = value_wei / 1e18 

179 

180 # Rough USD estimate (ETH price ~$3000) 

181 value_usd = value_eth * 3000 

182 

183 if value_usd >= min_value_usd: 

184 transactions.append({ 

185 "tx_hash": tx.get("hash", ""), 

186 "from_address": tx.get("from", ""), 

187 "to_address": tx.get("to", ""), 

188 "value": value_eth, 

189 "value_usd": value_usd, 

190 "block_number": block_num, 

191 "timestamp": datetime.utcnow(), # Approximate 

192 "blockchain": "ethereum" 

193 }) 

194 except Exception as e: 

195 logger.warning(f"Error fetching block {block_num}: {e}") 

196 continue 

197 

198 return { 

199 "transactions": transactions[:limit], 

200 "count": len(transactions), 

201 "source": "etherscan" 

202 } 

203 

204 except ValueError as e: 

205 return { 

206 "error": str(e), 

207 "transactions": [], 

208 "count": 0 

209 } 

210 except Exception as e: 

211 logger.error(f"Etherscan API error: {e}") 

212 return { 

213 "error": f"Failed to fetch transactions: {str(e)}", 

214 "transactions": [], 

215 "count": 0 

216 } 

217 

218 except Exception as e: 

219 logger.error(f"Ethereum transactions error: {e}", exc_info=True) 

220 raise HTTPException(status_code=500, detail=str(e)) 

221 

222 

223@router.get("/ethereum/whale-addresses") 

224async def get_ethereum_whales( 

225 min_balance_eth: float = Query(default=10000, description="Minimum balance in ETH") 

226): 

227 """ 

228 Get Ethereum addresses with large balances (whales) 

229 """ 

230 try: 

231 # Note: This is a simplified version 

232 # Real implementation would require tracking addresses over time 

233 return { 

234 "message": "Whale address tracking requires historical data", 

235 "whales": [], 

236 "count": 0 

237 } 

238 except Exception as e: 

239 logger.error(f"Ethereum whales error: {e}") 

240 raise HTTPException(status_code=500, detail=str(e)) 

241 

242 

243# ============================================================================ 

244# BSC Endpoints 

245# ============================================================================ 

246 

247@router.get("/bsc/whale-movements") 

248async def get_bsc_whale_movements( 

249 limit: int = Query(default=20, ge=1, le=100), 

250 min_value_usd: Optional[float] = Query(default=1000000, description="Minimum transaction value in USD") 

251): 

252 """ 

253 Get large BSC transactions (whale movements) 

254 """ 

255 try: 

256 fallback_config = PROVIDER_FALLBACK_STRATEGY.get("bscscan", {}) 

257 

258 try: 

259 # Get latest block 

260 block_data = await call_bscscan_api("eth_blockNumber") 

261 latest_block = int(block_data.get("result", "0x0"), 16) 

262 

263 transactions = [] 

264 start_block = max(0, latest_block - 200) # Check more blocks for BSC 

265 

266 for block_num in range(latest_block, start_block, -1): 

267 if len(transactions) >= limit: 

268 break 

269 

270 try: 

271 block_data = await call_bscscan_api( 

272 "eth_getBlockByNumber", 

273 params={ 

274 "tag": hex(block_num), 

275 "boolean": "true" 

276 } 

277 ) 

278 

279 block_result = block_data.get("result", {}) 

280 txs = block_result.get("transactions", []) 

281 

282 for tx in txs[:10]: 

283 if len(transactions) >= limit: 

284 break 

285 

286 value_wei = int(tx.get("value", "0x0"), 16) 

287 value_bnb = value_wei / 1e18 

288 

289 # Rough USD estimate (BNB price ~$600) 

290 value_usd = value_bnb * 600 

291 

292 if value_usd >= min_value_usd: 

293 transactions.append({ 

294 "tx_hash": tx.get("hash", ""), 

295 "from_address": tx.get("from", ""), 

296 "to_address": tx.get("to", ""), 

297 "value": value_bnb, 

298 "value_usd": value_usd, 

299 "block_number": block_num, 

300 "timestamp": datetime.utcnow(), 

301 "blockchain": "bsc" 

302 }) 

303 except Exception as e: 

304 logger.warning(f"Error fetching BSC block {block_num}: {e}") 

305 continue 

306 

307 return { 

308 "transactions": transactions[:limit], 

309 "count": len(transactions), 

310 "source": "bscscan" 

311 } 

312 

313 except ValueError as e: 

314 return { 

315 "error": str(e), 

316 "transactions": [], 

317 "count": 0 

318 } 

319 except Exception as e: 

320 logger.error(f"BSCScan API error: {e}") 

321 return { 

322 "error": f"Failed to fetch transactions: {str(e)}", 

323 "transactions": [], 

324 "count": 0 

325 } 

326 

327 except Exception as e: 

328 logger.error(f"BSC whale movements error: {e}", exc_info=True) 

329 raise HTTPException(status_code=500, detail=str(e)) 

330 

331 

332# ============================================================================ 

333# Tron Endpoints 

334# ============================================================================ 

335 

336@router.get("/tron/large-transfers") 

337async def get_tron_large_transfers( 

338 limit: int = Query(default=20, ge=1, le=100), 

339 min_value_usd: Optional[float] = Query(default=100000, description="Minimum transfer value in USD") 

340): 

341 """ 

342 Get large Tron transfers 

343 """ 

344 try: 

345 fallback_config = PROVIDER_FALLBACK_STRATEGY.get("tronscan", {}) 

346 

347 try: 

348 # TronScan API endpoint for large transfers 

349 # Note: This is a simplified implementation 

350 # Real TronScan API may have different endpoints 

351 

352 # For now, return a placeholder structure 

353 return { 

354 "message": "TronScan API integration in progress", 

355 "transactions": [], 

356 "count": 0, 

357 "source": "tronscan" 

358 } 

359 

360 except ValueError as e: 

361 return { 

362 "error": str(e), 

363 "transactions": [], 

364 "count": 0 

365 } 

366 except Exception as e: 

367 logger.error(f"TronScan API error: {e}") 

368 return { 

369 "error": f"Failed to fetch transfers: {str(e)}", 

370 "transactions": [], 

371 "count": 0 

372 } 

373 

374 except Exception as e: 

375 logger.error(f"Tron large transfers error: {e}", exc_info=True) 

376 raise HTTPException(status_code=500, detail=str(e)) 

377 

378 

379# ============================================================================ 

380# Combined Whale Activity 

381# ============================================================================ 

382 

383@router.get("/whales/activity") 

384async def get_whale_activity( 

385 limit: int = Query(default=50, ge=1, le=200), 

386 min_amount_usd: float = Query(default=1000000, description="Minimum transaction amount in USD"), 

387 hours: int = Query(default=24, ge=1, le=168, description="Time period in hours") 

388): 

389 """ 

390 Get combined whale activity from all blockchains 

391 """ 

392 try: 

393 all_transactions = [] 

394 

395 # Get Ethereum transactions 

396 try: 

397 eth_data = await get_ethereum_transactions(limit=limit, min_value_usd=min_amount_usd) 

398 if eth_data.get("transactions"): 

399 all_transactions.extend(eth_data["transactions"]) 

400 except Exception as e: 

401 logger.warning(f"Failed to get Ethereum transactions: {e}") 

402 

403 # Get BSC transactions 

404 try: 

405 bsc_data = await get_bsc_whale_movements(limit=limit, min_value_usd=min_amount_usd) 

406 if bsc_data.get("transactions"): 

407 all_transactions.extend(bsc_data["transactions"]) 

408 except Exception as e: 

409 logger.warning(f"Failed to get BSC transactions: {e}") 

410 

411 # Get Tron transfers 

412 try: 

413 tron_data = await get_tron_large_transfers(limit=limit, min_value_usd=min_amount_usd) 

414 if tron_data.get("transactions"): 

415 all_transactions.extend(tron_data["transactions"]) 

416 except Exception as e: 

417 logger.warning(f"Failed to get Tron transfers: {e}") 

418 

419 # Sort by value_usd descending 

420 all_transactions.sort(key=lambda x: x.get("value_usd", 0), reverse=True) 

421 

422 return { 

423 "transactions": all_transactions[:limit], 

424 "count": len(all_transactions), 

425 "total_volume_usd": sum(tx.get("value_usd", 0) for tx in all_transactions), 

426 "by_blockchain": { 

427 "ethereum": len([t for t in all_transactions if t.get("blockchain") == "ethereum"]), 

428 "bsc": len([t for t in all_transactions if t.get("blockchain") == "bsc"]), 

429 "tron": len([t for t in all_transactions if t.get("blockchain") == "tron"]) 

430 }, 

431 "timestamp": datetime.utcnow().isoformat() 

432 } 

433 

434 except Exception as e: 

435 logger.error(f"Whale activity error: {e}", exc_info=True) 

436 raise HTTPException(status_code=500, detail=str(e)) 

437