38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
from typing import List, Literal, Dict, Any
|
|
|
|
BINANCE_REST = "https://api.binance.com/api/v3/klines"
|
|
|
|
Interval = Literal["1m", "3m", "5m", "15m", "30m", "1h", "2h", "4h", "6h", "8h", "12h", "1d"]
|
|
|
|
|
|
async def fetch_klines(symbol: str, interval: Interval, limit: int = 500) -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetch OHLCV klines from Binance REST. Returns list of dicts with fields:
|
|
time, open, high, low, close, volume
|
|
"""
|
|
params = {"symbol": symbol.upper().replace("/", ""), "interval": interval, "limit": min(max(limit, 1), 1000)}
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
r = await client.get(BINANCE_REST, params=params)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
out: List[Dict[str, Any]] = []
|
|
for row in data:
|
|
# Binance format
|
|
# [ openTime, open, high, low, close, volume, closeTime, ... ]
|
|
out.append(
|
|
{
|
|
"time": int(row[0] // 1000),
|
|
"open": float(row[1]),
|
|
"high": float(row[2]),
|
|
"low": float(row[3]),
|
|
"close": float(row[4]),
|
|
"volume": float(row[5]),
|
|
}
|
|
)
|
|
# Ensure ascending by time
|
|
out.sort(key=lambda x: x["time"])
|
|
return out
|