From b1687582ecd98fadb64601489866e4b1434122b8 Mon Sep 17 00:00:00 2001 From: Harwinder Date: Fri, 1 May 2026 16:50:13 -0400 Subject: [PATCH] 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 --- .gitignore | 5 ++ README.md | 50 +++++++++++ config.py | 19 ++++ indicators.py | 128 +++++++++++++++++++++++++++ main.py | 224 +++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 4 + signals.py | 189 +++++++++++++++++++++++++++++++++++++++ twelvedata.py | 102 +++++++++++++++++++++ 8 files changed, 721 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 config.py create mode 100644 indicators.py create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 signals.py create mode 100644 twelvedata.py 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