harry/Forex_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Forex Indicator Telegram Bot MVP

Browse Source 
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>
This commit is contained in:
Harwinder
2026-05-01 16:50:13 -04:00
commit b1687582ec
8 changed files with 721 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+224
View File
@@ -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 <PAIR> — latest close (e.g. /price EURUSD)\n"
"/rsi <PAIR> <TF> — RSI value\n"
"/macd <PAIR> <TF> — MACD line/signal/histogram\n"
"/bb <PAIR> <TF> — Bollinger Bands\n"
"/signal <PAIR> <TF> — 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> <TIMEFRAME>")
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 <PAIR> 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()