Files
Harwinder b1687582ec Initial commit: Forex Indicator Telegram Bot MVP
Multi-indicator signal engine with RSI, MACD, Bollinger Bands,
EMA trend filter, ADX strength gate, Stochastic, ATR, and
higher-timeframe trend confirmation. Twelve Data API client
with in-memory cache and rate limiting.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-01 16:50:13 -04:00

103 lines
3.0 KiB
Python

"""Twelve Data API client with in-memory cache and simple rate limiting."""
from __future__ import annotations
import asyncio
import logging
import time
from typing import Any
import httpx
import pandas as pd
from config import SUPPORTED_PAIRS, SUPPORTED_TIMEFRAMES, TWELVE_DATA_API_KEY
logger = logging.getLogger(__name__)
_API_URL = "https://api.twelvedata.com/time_series"
_TTL_SECONDS: dict[str, int] = {
"15min": 800,
"1h": 3500,
"4h": 14000,
"1day": 43000,
}
_cache: dict[tuple[str, str], tuple[float, pd.DataFrame]] = {}
_call_times: list[float] = []
_lock = asyncio.Lock()
_MAX_CALLS_PER_MIN = 8
_WINDOW = 60.0
def _to_api_symbol(pair: str) -> str:
"""Convert 'EURUSD' to 'EUR/USD'."""
return f"{pair[:3]}/{pair[3:]}"
async def _respect_rate_limit() -> None:
now = time.monotonic()
cutoff = now - _WINDOW
while _call_times and _call_times[0] < cutoff:
_call_times.pop(0)
if len(_call_times) >= _MAX_CALLS_PER_MIN:
sleep_for = _WINDOW - (now - _call_times[0]) + 0.1
logger.info("Rate limit hit, sleeping %.2fs", sleep_for)
await asyncio.sleep(sleep_for)
_call_times.append(time.monotonic())
async def fetch_ohlc(pair: str, timeframe: str, size: int = 200) -> pd.DataFrame:
"""Fetch OHLC candles for a forex pair from Twelve Data.
Returns a DataFrame indexed by datetime (ascending) with columns
[open, high, low, close]. Cached per (pair, timeframe) with timeframe-aware TTL.
"""
pair = pair.upper()
if pair not in SUPPORTED_PAIRS:
raise ValueError(f"Unsupported pair: {pair}")
if timeframe not in SUPPORTED_TIMEFRAMES:
raise ValueError(f"Unsupported timeframe: {timeframe}")
key = (pair, timeframe)
ttl = _TTL_SECONDS[timeframe]
now = time.time()
async with _lock:
cached = _cache.get(key)
if cached and (now - cached[0]) < ttl:
return cached[1]
await _respect_rate_limit()
params: dict[str, Any] = {
"symbol": _to_api_symbol(pair),
"interval": timeframe,
"outputsize": size,
"apikey": TWELVE_DATA_API_KEY,
"format": "JSON",
}
try:
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.get(_API_URL, params=params)
resp.raise_for_status()
data = resp.json()
except httpx.HTTPError as e:
logger.error("Twelve Data HTTP error: %s", e)
raise
if data.get("status") == "error" or "values" not in data:
msg = data.get("message", "Unknown API error")
logger.error("Twelve Data API error: %s", msg)
raise ValueError(f"API error: {msg}")
df = pd.DataFrame(data["values"])
df["datetime"] = pd.to_datetime(df["datetime"])
for col in ("open", "high", "low", "close"):
df[col] = df[col].astype(float)
df = df.set_index("datetime")[["open", "high", "low", "close"]].sort_index()
_cache[key] = (now, df)
return df