commit b1687582ecd98fadb64601489866e4b1434122b8 Author: Harwinder Date: Fri May 1 16:50:13 2026 -0400 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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e08762e --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.env +.venv/ +__pycache__/ +*.pyc +*.pyo diff --git a/README.md b/README.md new file mode 100644 index 0000000..0d6b487 --- /dev/null +++ b/README.md @@ -0,0 +1,50 @@ +# Forex Indicator Telegram Bot + +Minimal Telegram bot that serves forex prices and technical indicators (RSI, MACD, Bollinger Bands) on demand, powered by [Twelve Data](https://twelvedata.com/). + +## Get tokens + +1. **Telegram bot token** — message [@BotFather](https://t.me/BotFather), `/newbot`, copy the token. +2. **Twelve Data API key** — sign up at [twelvedata.com](https://twelvedata.com/), free tier gives 800 req/day, 8 req/min. + +## Install + +```bash +cd forex_bot +python -m venv .venv +# Windows: .venv\Scripts\activate macOS/Linux: source .venv/bin/activate +pip install -r requirements.txt +cp .env.example .env # then edit .env and paste your tokens +``` + +> `pandas-ta` requires NumPy 1.x; the pinned `numpy==1.26.4` handles that. + +## Run + +```bash +python main.py +``` + +Then open Telegram and message your bot. + +## Commands + +| Command | Example | Description | +|---|---|---| +| `/start`, `/help` | `/help` | Welcome + command list | +| `/pairs` | `/pairs` | List supported pairs and timeframes | +| `/price ` | `/price EURUSD` | Latest close price | +| `/rsi ` | `/rsi EURUSD 1h` | RSI(14) | +| `/macd ` | `/macd GBPUSD 4h` | MACD(12,26,9) | +| `/bb ` | `/bb USDJPY 1day` | Bollinger Bands(20,2) + price | + +**Supported pairs:** EURUSD, GBPUSD, USDJPY, AUDUSD, USDCAD +**Supported timeframes:** 15min, 1h, 4h, 1day + +## Notes + +- In-memory cache per (pair, timeframe) with TTL based on timeframe — keeps you well under the free-tier rate limit. +- Simple in-process rate limiter caps to 8 calls/min. +- No database, no scheduler, no charts — on-demand text only. + +_Educational only. Not financial advice._ diff --git a/config.py b/config.py new file mode 100644 index 0000000..58c34ad --- /dev/null +++ b/config.py @@ -0,0 +1,19 @@ +"""Configuration loaded from environment variables.""" +from __future__ import annotations + +import os +from dotenv import load_dotenv + +load_dotenv() + +TELEGRAM_TOKEN: str = os.getenv("TELEGRAM_TOKEN", "") +TWELVE_DATA_API_KEY: str = os.getenv("TWELVE_DATA_API_KEY", "") + +SUPPORTED_PAIRS: list[str] = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD", "USDCAD", "XAUUSD"] +SUPPORTED_TIMEFRAMES: list[str] = ["15min", "1h", "4h", "1day"] + +if not TELEGRAM_TOKEN or not TWELVE_DATA_API_KEY: + raise RuntimeError( + "Missing TELEGRAM_TOKEN or TWELVE_DATA_API_KEY. " + "Copy .env.example to .env and fill in values." + ) diff --git a/indicators.py b/indicators.py new file mode 100644 index 0000000..1a7df9a --- /dev/null +++ b/indicators.py @@ -0,0 +1,128 @@ +"""Technical indicator functions. + +Pure pandas implementations — no third-party TA library. Each public function +takes a DataFrame with columns [open, high, low, close] indexed by datetime +ascending, and returns either a float or a dict of latest values. +""" +from __future__ import annotations + +import pandas as pd + + +def _require(df: pd.DataFrame, n: int) -> None: + if len(df) < n: + raise ValueError(f"Not enough data: need at least {n} rows, got {len(df)}") + + +def _ema(series: pd.Series, period: int) -> pd.Series: + return series.ewm(span=period, adjust=False).mean() + + +def _wilder(series: pd.Series, period: int) -> pd.Series: + """Wilder's smoothing (RMA) — same as EMA with alpha = 1/period.""" + return series.ewm(alpha=1.0 / period, adjust=False).mean() + + +def rsi(df: pd.DataFrame, period: int = 14) -> float: + """Latest RSI using Wilder's smoothing.""" + _require(df, period + 1) + delta = df["close"].diff() + gain = delta.clip(lower=0.0) + loss = -delta.clip(upper=0.0) + avg_gain = _wilder(gain, period) + avg_loss = _wilder(loss, period) + rs = avg_gain / avg_loss.replace(0.0, pd.NA) + rsi_series = (100.0 - (100.0 / (1.0 + rs))).fillna(100.0) + return float(rsi_series.iloc[-1]) + + +def macd(df: pd.DataFrame) -> dict[str, float]: + """Latest MACD line, signal, histogram (12/26/9), plus prior histogram for direction.""" + _require(df, 35) + fast = _ema(df["close"], 12) + slow = _ema(df["close"], 26) + macd_line = fast - slow + signal_line = _ema(macd_line, 9) + hist = macd_line - signal_line + return { + "macd": float(macd_line.iloc[-1]), + "signal": float(signal_line.iloc[-1]), + "hist": float(hist.iloc[-1]), + "hist_prev": float(hist.iloc[-2]), + } + + +def bollinger(df: pd.DataFrame, period: int = 20) -> dict[str, float]: + """Latest Bollinger Bands (upper/mid/lower, 2 stdev) and current price.""" + _require(df, period + 1) + close = df["close"] + mid = close.rolling(period).mean() + std = close.rolling(period).std(ddof=0) + return { + "upper": float((mid + 2.0 * std).iloc[-1]), + "mid": float(mid.iloc[-1]), + "lower": float((mid - 2.0 * std).iloc[-1]), + "price": float(close.iloc[-1]), + } + + +def ema_pair(df: pd.DataFrame, fast: int = 20, slow: int = 50) -> dict[str, float]: + """Latest fast/slow EMA values + current close.""" + _require(df, slow + 1) + return { + "fast": float(_ema(df["close"], fast).iloc[-1]), + "slow": float(_ema(df["close"], slow).iloc[-1]), + "price": float(df["close"].iloc[-1]), + } + + +def atr(df: pd.DataFrame, period: int = 14) -> float: + """Latest Average True Range (Wilder).""" + _require(df, period + 1) + high, low, close = df["high"], df["low"], df["close"] + prev_close = close.shift(1) + tr = pd.concat( + [(high - low), (high - prev_close).abs(), (low - prev_close).abs()], + axis=1, + ).max(axis=1) + return float(_wilder(tr, period).iloc[-1]) + + +def adx(df: pd.DataFrame, period: int = 14) -> float: + """Latest ADX(14) — trend strength, range 0-100.""" + _require(df, period * 2 + 1) + high, low, close = df["high"], df["low"], df["close"] + up = high.diff() + down = -low.diff() + plus_dm = pd.Series(((up > down) & (up > 0)) * up, index=df.index).clip(lower=0.0) + minus_dm = pd.Series(((down > up) & (down > 0)) * down, index=df.index).clip(lower=0.0) + + prev_close = close.shift(1) + tr = pd.concat( + [(high - low), (high - prev_close).abs(), (low - prev_close).abs()], + axis=1, + ).max(axis=1) + + atr_s = _wilder(tr, period) + plus_di = 100.0 * _wilder(plus_dm, period) / atr_s.replace(0.0, pd.NA) + minus_di = 100.0 * _wilder(minus_dm, period) / atr_s.replace(0.0, pd.NA) + dx = 100.0 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0.0, pd.NA) + adx_series = _wilder(dx.fillna(0.0), period) + return float(adx_series.iloc[-1]) + + +def stochastic(df: pd.DataFrame, k_period: int = 14, d_period: int = 3) -> dict[str, float]: + """Latest Stochastic %K (smoothed) and %D, plus prior values for cross detection.""" + _require(df, k_period + d_period + 1) + high, low, close = df["high"], df["low"], df["close"] + lowest = low.rolling(k_period).min() + highest = high.rolling(k_period).max() + raw_k = 100.0 * (close - lowest) / (highest - lowest).replace(0.0, pd.NA) + k = raw_k.rolling(d_period).mean() + d = k.rolling(d_period).mean() + return { + "k": float(k.iloc[-1]), + "d": float(d.iloc[-1]), + "k_prev": float(k.iloc[-2]), + "d_prev": float(d.iloc[-2]), + } diff --git a/main.py b/main.py new file mode 100644 index 0000000..e23a0a8 --- /dev/null +++ b/main.py @@ -0,0 +1,224 @@ +"""Telegram bot entry point and command handlers.""" +from __future__ import annotations + +import logging + +from telegram import BotCommand, Update +from telegram.ext import ( + Application, + CommandHandler, + ContextTypes, +) + +from config import SUPPORTED_PAIRS, SUPPORTED_TIMEFRAMES, TELEGRAM_TOKEN +from indicators import bollinger, macd, rsi +from signals import HTF_OF, analyze, emoji +from twelvedata import fetch_ohlc + +logging.basicConfig( + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + level=logging.INFO, +) +logger = logging.getLogger(__name__) + +DISCLAIMER = "\n\n_Educational only. Not financial advice._" + +WELCOME = ( + "*Forex Indicator Bot*\n\n" + "Commands:\n" + "/pairs — list supported pairs/timeframes\n" + "/price — latest close (e.g. /price EURUSD)\n" + "/rsi — RSI value\n" + "/macd — MACD line/signal/histogram\n" + "/bb — Bollinger Bands\n" + "/signal — multi-indicator weighted verdict with HTF trend filter\n" + "/help — this message" +) + DISCLAIMER + + +def _parse_pair_tf(args: list[str]) -> tuple[str, str]: + if len(args) != 2: + raise ValueError("Usage: ") + pair, tf = args[0].upper(), args[1].lower() + if tf == "1d": + tf = "1day" + if pair not in SUPPORTED_PAIRS: + raise ValueError(f"Unknown pair '{pair}'. Try /pairs.") + if tf not in SUPPORTED_TIMEFRAMES: + raise ValueError(f"Unknown timeframe '{tf}'. Try /pairs.") + return pair, tf + + +async def _reply(update: Update, text: str) -> None: + await update.message.reply_text(text + DISCLAIMER, parse_mode="Markdown") + + +async def start(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None: + await update.message.reply_text(WELCOME, parse_mode="Markdown") + + +async def pairs_cmd(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None: + text = ( + "*Pairs:* " + ", ".join(SUPPORTED_PAIRS) + + "\n*Timeframes:* " + ", ".join(SUPPORTED_TIMEFRAMES) + ) + await _reply(update, text) + + +async def price_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: + try: + if len(ctx.args) != 1: + raise ValueError("Usage: /price e.g. /price EURUSD") + pair = ctx.args[0].upper() + if pair not in SUPPORTED_PAIRS: + raise ValueError(f"Unknown pair '{pair}'. Try /pairs.") + df = await fetch_ohlc(pair, "1h", size=50) + last = df["close"].iloc[-1] + ts = df.index[-1].strftime("%Y-%m-%d %H:%M UTC") + await _reply(update, f"*{pair}* close: `{last:.5f}`\n_{ts}_") + except Exception as e: + logger.exception("price_cmd failed") + await _reply(update, f"Error: {e}") + + +async def rsi_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: + try: + pair, tf = _parse_pair_tf(ctx.args) + df = await fetch_ohlc(pair, tf) + await _reply(update, f"*RSI(14)* {pair} {tf}: `{rsi(df):.2f}`") + except Exception as e: + logger.exception("rsi_cmd failed") + await _reply(update, f"Error: {e}") + + +async def macd_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: + try: + pair, tf = _parse_pair_tf(ctx.args) + df = await fetch_ohlc(pair, tf) + m = macd(df) + text = ( + f"*MACD* {pair} {tf}\n" + f"MACD: `{m['macd']:.5f}`\n" + f"Signal: `{m['signal']:.5f}`\n" + f"Hist: `{m['hist']:.5f}`" + ) + await _reply(update, text) + except Exception as e: + logger.exception("macd_cmd failed") + await _reply(update, f"Error: {e}") + + +async def bb_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: + try: + pair, tf = _parse_pair_tf(ctx.args) + df = await fetch_ohlc(pair, tf) + b = bollinger(df) + text = ( + f"*Bollinger(20,2)* {pair} {tf}\n" + f"Upper: `{b['upper']:.5f}`\n" + f"Mid: `{b['mid']:.5f}`\n" + f"Lower: `{b['lower']:.5f}`\n" + f"Price: `{b['price']:.5f}`" + ) + await _reply(update, text) + except Exception as e: + logger.exception("bb_cmd failed") + await _reply(update, f"Error: {e}") + + +def _format_reading(r) -> str: + sign = "+" if r.contribution > 0 else "" + return f"• {r.name}: {r.value} ({sign}{r.contribution})" + + +async def signal_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: + try: + pair, tf = _parse_pair_tf(ctx.args) + df = await fetch_ohlc(pair, tf) + + htf = HTF_OF.get(tf) + htf_df = await fetch_ohlc(pair, htf) if htf and htf != tf else None + + report = analyze(df, htf_df) + + sl_dist = 1.5 * report["atr"] + tp_dist = 2.5 * report["atr"] + price = report["price"] + + lines = [ + f"*Signal* {pair} {tf} (HTF: {htf or 'n/a'})", + f"Price: `{price:.5f}` ATR(14): `{report['atr']:.5f}`", + "", + "*Indicators:*", + *[_format_reading(r) for r in report["readings"]], + "", + f"Score: `{report['score']:+d} / ±{report['max_score']}`", + f"*Verdict:* {emoji(report['verdict'])} *{report['verdict']}*" + f" (confidence {report['confidence']}%)", + ] + + if report["verdict"] in ("BUY", "STRONG BUY"): + lines += [ + "", + f"_Suggested zones (ATR-based):_", + f"Entry near `{price:.5f}`", + f"Stop ≈ `{price - sl_dist:.5f}` ({sl_dist:.5f} below)", + f"Target ≈ `{price + tp_dist:.5f}` ({tp_dist:.5f} above)", + ] + elif report["verdict"] in ("SELL", "STRONG SELL"): + lines += [ + "", + f"_Suggested zones (ATR-based):_", + f"Entry near `{price:.5f}`", + f"Stop ≈ `{price + sl_dist:.5f}` ({sl_dist:.5f} above)", + f"Target ≈ `{price - tp_dist:.5f}` ({tp_dist:.5f} below)", + ] + + if report["notes"]: + lines += ["", *[f"_{n}_" for n in report["notes"]]] + + await _reply(update, "\n".join(lines)) + except Exception as e: + logger.exception("signal_cmd failed") + await _reply(update, f"Error: {e}") + + +BOT_COMMANDS: list[BotCommand] = [ + BotCommand("start", "Show welcome and command list"), + BotCommand("help", "Show welcome and command list"), + BotCommand("pairs", "List supported pairs and timeframes"), + BotCommand("price", "Latest close price — /price EURUSD"), + BotCommand("rsi", "RSI(14) — /rsi EURUSD 1h"), + BotCommand("macd", "MACD(12,26,9) — /macd EURUSD 1h"), + BotCommand("bb", "Bollinger Bands(20,2) — /bb EURUSD 1h"), + BotCommand("signal", "Multi-indicator verdict — /signal EURUSD 1h"), +] + + +async def _post_init(app: Application) -> None: + await app.bot.set_my_commands(BOT_COMMANDS) + logger.info("Registered %d bot commands with Telegram", len(BOT_COMMANDS)) + + +def main() -> None: + """Run the bot.""" + app = ( + Application.builder() + .token(TELEGRAM_TOKEN) + .post_init(_post_init) + .build() + ) + app.add_handler(CommandHandler("start", start)) + app.add_handler(CommandHandler("help", start)) + app.add_handler(CommandHandler("pairs", pairs_cmd)) + app.add_handler(CommandHandler("price", price_cmd)) + app.add_handler(CommandHandler("rsi", rsi_cmd)) + app.add_handler(CommandHandler("macd", macd_cmd)) + app.add_handler(CommandHandler("bb", bb_cmd)) + app.add_handler(CommandHandler("signal", signal_cmd)) + logger.info("Bot starting...") + app.run_polling() + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9ed6406 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +python-telegram-bot>=21.6 +httpx>=0.27.2 +pandas>=2.2.3 +python-dotenv>=1.0.1 diff --git a/signals.py b/signals.py new file mode 100644 index 0000000..a028cf4 --- /dev/null +++ b/signals.py @@ -0,0 +1,189 @@ +"""Signal scoring engine. + +Each indicator contributes a score in [-2..+2] (sell..buy) with a weight. +Trend scores are gated by ADX (weak trend ⇒ halved). The weighted sum is +mapped to a verdict and confidence percentage. +""" +from __future__ import annotations + +from dataclasses import dataclass + +import pandas as pd + +from indicators import adx, atr, bollinger, ema_pair, macd, rsi, stochastic + +# Weights — tuned so HTF trend dominates and momentum confirms. +W_HTF_TREND = 3 +W_LTF_TREND = 2 +W_MACD = 2 +W_RSI = 1 +W_STOCH = 1 +W_BB = 1 +MAX_SCORE = 2 * (W_HTF_TREND + W_LTF_TREND + W_MACD + W_RSI + W_STOCH + W_BB) # 20 + +# Higher-timeframe mapping for the trend filter. +HTF_OF: dict[str, str] = { + "15min": "1h", + "1h": "4h", + "4h": "1day", + "1day": "1day", +} + + +@dataclass +class Reading: + name: str + value: str # display string for the user + raw: int # -2..+2 + weight: int + + @property + def contribution(self) -> int: + return self.raw * self.weight + + +def _trend_score(fast: float, slow: float) -> tuple[int, str]: + """Score EMA fast vs slow. ±2 if separation >0.1%, ±1 if just above/below.""" + if slow == 0: + return 0, "flat" + sep = (fast - slow) / slow + if sep >= 0.001: + return 2, "strong bullish" + if sep > 0: + return 1, "bullish" + if sep <= -0.001: + return -2, "strong bearish" + if sep < 0: + return -1, "bearish" + return 0, "flat" + + +def _macd_score(m: dict[str, float]) -> tuple[int, str]: + above = m["macd"] > m["signal"] + rising = m["hist"] > m["hist_prev"] + if above and m["hist"] > 0: + return (2, "bullish + rising") if rising else (1, "bullish") + if not above and m["hist"] < 0: + return (-2, "bearish + falling") if not rising else (-1, "bearish") + return 0, "neutral" + + +def _rsi_score(value: float) -> tuple[int, str]: + if value >= 80: + return -2, "extreme overbought" + if value >= 70: + return -1, "overbought" + if value <= 20: + return 2, "extreme oversold" + if value <= 30: + return 1, "oversold" + return 0, "neutral" + + +def _stoch_score(s: dict[str, float]) -> tuple[int, str]: + k, d, kp, dp = s["k"], s["d"], s["k_prev"], s["d_prev"] + cross_down = kp >= dp and k < d + cross_up = kp <= dp and k > d + if k >= 80 and cross_down: + return -2, "overbought + bearish cross" + if k >= 80: + return -1, "overbought" + if k <= 20 and cross_up: + return 2, "oversold + bullish cross" + if k <= 20: + return 1, "oversold" + return 0, "neutral" + + +def _bb_score(b: dict[str, float]) -> tuple[int, str]: + if b["price"] >= b["upper"]: + return -1, "above upper band" + if b["price"] <= b["lower"]: + return 1, "below lower band" + return 0, "inside bands" + + +def _verdict(score: int) -> str: + if score >= 12: + return "STRONG BUY" + if score >= 6: + return "BUY" + if score <= -12: + return "STRONG SELL" + if score <= -6: + return "SELL" + return "NEUTRAL" + + +def emoji(verdict: str) -> str: + return { + "STRONG BUY": "🟢🟢", + "BUY": "🟢", + "NEUTRAL": "⚪", + "SELL": "🔴", + "STRONG SELL": "🔴🔴", + }.get(verdict, "⚪") + + +def analyze(df: pd.DataFrame, htf_df: pd.DataFrame | None = None) -> dict: + """Run the full signal engine over a primary DataFrame and optional higher-TF DataFrame. + + Returns a dict with: readings (list[Reading]), score, max_score, confidence, + verdict, adx, atr, price, and a notes list (e.g., 'weak trend, halved'). + """ + notes: list[str] = [] + + e = ema_pair(df, 20, 50) + ltf_raw, ltf_label = _trend_score(e["fast"], e["slow"]) + + if htf_df is not None: + eh = ema_pair(htf_df, 20, 50) + htf_raw, htf_label = _trend_score(eh["fast"], eh["slow"]) + else: + htf_raw, htf_label = 0, "n/a (no higher TF)" + + adx_value = adx(df) + if adx_value < 20: + notes.append(f"ADX={adx_value:.1f} <20 → weak trend, trend scores halved") + ltf_raw = ltf_raw // 2 # towards zero + htf_raw = htf_raw // 2 + elif adx_value >= 25: + notes.append(f"ADX={adx_value:.1f} >=25 → strong trend") + + m = macd(df) + macd_raw, macd_label = _macd_score(m) + + rsi_value = rsi(df) + rsi_raw, rsi_label = _rsi_score(rsi_value) + + s = stochastic(df) + stoch_raw, stoch_label = _stoch_score(s) + + b = bollinger(df) + bb_raw, bb_label = _bb_score(b) + + readings = [ + Reading("HTF Trend (EMA20/50)", htf_label, htf_raw, W_HTF_TREND), + Reading("LTF Trend (EMA20/50)", ltf_label, ltf_raw, W_LTF_TREND), + Reading("MACD", macd_label, macd_raw, W_MACD), + Reading(f"RSI={rsi_value:.1f}", rsi_label, rsi_raw, W_RSI), + Reading(f"Stoch %K={s['k']:.1f}", stoch_label, stoch_raw, W_STOCH), + Reading("Bollinger", bb_label, bb_raw, W_BB), + ] + + score = sum(r.contribution for r in readings) + confidence = round(abs(score) / MAX_SCORE * 100) + verdict = _verdict(score) + atr_value = atr(df) + + return { + "readings": readings, + "score": score, + "max_score": MAX_SCORE, + "confidence": confidence, + "verdict": verdict, + "adx": adx_value, + "atr": atr_value, + "price": b["price"], + "notes": notes, + } diff --git a/twelvedata.py b/twelvedata.py new file mode 100644 index 0000000..18d0b72 --- /dev/null +++ b/twelvedata.py @@ -0,0 +1,102 @@ +"""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