File size: 4,975 Bytes
51f3427 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """
Gate.io Public API Integration
Fetch candle data for USDT perpetual futures
"""
import aiohttp
import pandas as pd
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
class GateIOClient:
"""Gate.io public API client for candle data"""
def __init__(self, base_url: str = "https://api.gateio.ws/api/v4"):
self.base_url = base_url
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def get_usdt_perpetuals(self) -> List[Dict]:
"""Get list of USDT perpetual futures"""
url = f"{self.base_url}/futures/usdt/contracts"
async with self.session.get(url) as response:
data = await response.json()
# Filter for USDT-settled perpetuals
perpetuals = [
{
'symbol': c['name'],
'name': c.get('name', ''),
'quanto_multiplier': c.get('quanto_multiplier', 1),
'tick_size': c.get('order_price_round', 1e-8),
'position_size': c.get('position_size', 1)
}
for c in data
if c.get('name', '').endswith('_USDT') and c.get('quanto_multiplier') is not None
]
return perpetuals
async def get_candles(
self,
symbol: str,
interval: str = '1h',
limit: int = 1000,
from_: Optional[int] = None,
to: Optional[int] = None
) -> List[Dict]:
"""Get candle data for a symbol"""
url = f"{self.base_url}/futures/usdt/candlesticks"
params = {
'contract': symbol,
'interval': interval,
'limit': limit
}
if from_:
params['from'] = from_
if to:
params['to'] = to
async with self.session.get(url, params=params) as response:
data = await response.json()
# Gate.io returns data in reverse chronological order
candles = [
{
'timestamp': datetime.fromtimestamp(int(c[0])),
'volume': float(c[1]),
'close': float(c[2]),
'high': float(c[3]),
'low': float(c[4]),
'open': float(c[5]),
'quote_volume': float(c[1]) * float(c[2]) # Approximate
}
for c in reversed(data)
]
return candles
async def get_ticker(self, symbol: str) -> Dict:
"""Get current ticker data"""
url = f"{self.base_url}/futures/usdt/ticker"
params = {'contract': symbol}
async with self.session.get(url, params=params) as response:
data = await response.json()
if isinstance(data, list) and len(data) > 0:
return {
'symbol': symbol,
'last_price': float(data[0].get('last', 0)),
'volume_24h': float(data[0].get('volume_24h', 0)),
'quote_volume_24h': float(data[0].get('volume_24h_quote', 0)),
'high_24h': float(data[0].get('high_24h', 0)),
'low_24h': float(data[0].get('low_24h', 0))
}
return {}
async def filter_symbols(
self,
min_price: float = 0.10,
min_volume_24h: float = 50000.0
) -> List[Dict]:
"""Filter symbols by price and volume criteria"""
perpetuals = await self.get_usdt_perpetuals()
filtered = []
for contract in perpetuals:
symbol = contract['symbol']
ticker = await self.get_ticker(symbol)
if ticker:
last_price = ticker.get('last_price', 0)
volume_24h = ticker.get('volume_24h', 0)
if last_price > 0 and last_price < min_price and volume_24h >= min_volume_24h:
filtered.append({
**contract,
**ticker
})
return filtered
async def backfill_candles(
self,
symbol: str,
days: int = 30,
interval: str = '1h'
) -> pd.DataFrame:
"""Backfill historical candle data"""
to_time = int(datetime.now().timestamp())
from_time = int((datetime.now() - timedelta(days=days)).timestamp())
candles = await self.get_candles(
symbol,
interval=interval,
from_=from_time,
to=to_time
)
df = pd.DataFrame(candles)
if not df.empty:
df = df.sort_values('timestamp')
return df
|