Harwinder
301 lines
10 KiB
Python
301 lines
10 KiB
Python
"""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 charts import generate as generate_chart
|
||
from indicators import bollinger, macd, rsi
|
||
from signals import HTF_OF, analyze, ascii_indicator, to_pips
|
||
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 chart\n"
|
||
"/chart <PAIR> <TF> - candlestick chart with EMA/BB/RSI\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 _fmt_reading(r) -> str:
|
||
"""Format one indicator row.
|
||
|
||
Shows: signal (raw) × weight → contribution
|
||
e.g. +1 × 3 → +3 HTF EMA 20/50: strong bullish
|
||
This makes it immediately clear that the number after → is
|
||
(raw signal strength) × (importance weight), not a raw count.
|
||
"""
|
||
sign = "+" if r.raw > 0 else ""
|
||
raw_str = f"{sign}{r.raw}"
|
||
csign = "+" if r.contribution > 0 else ""
|
||
contrib = f"{csign}{r.contribution}"
|
||
return (
|
||
f" `{raw_str:>3} × {r.weight} → {contrib:>3}`"
|
||
f" {r.name}: {r.value}"
|
||
)
|
||
|
||
|
||
def _score_bar(score: int, max_score: int, width: int = 10) -> str:
|
||
"""Return a filled ASCII bar showing directional strength.
|
||
|
||
e.g. score=+8, max=20 -> '####------'
|
||
We use a single-sided bar: filled squares show absolute strength.
|
||
"""
|
||
filled = round(abs(score) / max_score * width)
|
||
bar = "#" * filled + "-" * (width - filled)
|
||
return bar
|
||
|
||
|
||
def _score_label(score: int, max_score: int) -> str:
|
||
"""Translate numeric score into a plain-English strength phrase."""
|
||
pct = abs(score) / max_score * 100
|
||
if pct >= 80:
|
||
return "very strong"
|
||
if pct >= 50:
|
||
return "strong"
|
||
if pct >= 25:
|
||
return "moderate"
|
||
return "weak"
|
||
|
||
|
||
def _fmt_group(readings, group: str, header: str) -> list[str]:
|
||
rows = [r for r in readings if r.group == group]
|
||
if not rows:
|
||
return []
|
||
return [header] + [_fmt_reading(r) for r in rows]
|
||
|
||
|
||
def _fmt_price(pair: str, value: float) -> str:
|
||
"""5 decimal places for FX, 2 for XAU."""
|
||
return f"{value:.2f}" if pair == "XAUUSD" else f"{value:.5f}"
|
||
|
||
|
||
async def chart_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
|
||
try:
|
||
pair, tf = _parse_pair_tf(ctx.args)
|
||
df = await fetch_ohlc(pair, tf)
|
||
img = generate_chart(df, pair, tf)
|
||
caption = f"*{pair}* {tf} - candlestick | EMA20/50 | BB(20,2) | RSI(14)" + DISCLAIMER
|
||
await update.message.reply_photo(img, caption=caption, parse_mode="Markdown")
|
||
except Exception as e:
|
||
logger.exception("chart_cmd failed")
|
||
await _reply(update, f"Error: {e}")
|
||
|
||
|
||
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)
|
||
price = report["price"]
|
||
atr_val = report["atr"]
|
||
verdict = report["verdict"]
|
||
ts = report["timestamp"].strftime("%Y-%m-%d %H:%M UTC")
|
||
pips = to_pips(pair, atr_val)
|
||
|
||
lines: list[str] = [
|
||
f"[SIGNAL] *{pair} {tf}* | HTF: {htf or 'n/a'}",
|
||
f"`{_fmt_price(pair, price)}` | "
|
||
f"ATR `{_fmt_price(pair, atr_val)}` ({pips} pips) | _{ts}_",
|
||
"",
|
||
*_fmt_group(report["readings"], "trend",
|
||
f"*-- TREND* (ADX {report['adx_label']}"
|
||
+ (" (!) scores halved" if report["adx_weak"] else "") + ")"),
|
||
"",
|
||
*_fmt_group(report["readings"], "momentum", "*-- MOMENTUM*"),
|
||
"",
|
||
*_fmt_group(report["readings"], "volatility", "*-- VOLATILITY*"),
|
||
"",
|
||
# Score line — bar + plain-English label so the number has context
|
||
f"`[{_score_bar(report['score'], report['max_score'])}]` "
|
||
f"{ascii_indicator(verdict)} *{verdict}*",
|
||
f"Score `{report['score']:+d}` of +/-`{report['max_score']}` "
|
||
f"({_score_label(report['score'], report['max_score'])} signal, "
|
||
f"{report['confidence']}% of max)",
|
||
]
|
||
|
||
if report["context"]:
|
||
lines += ["", f"> _{report['context']}_"]
|
||
|
||
sl_dist = 1.5 * atr_val
|
||
tp_dist = 2.5 * atr_val
|
||
if verdict in ("BUY", "STRONG BUY"):
|
||
sl = _fmt_price(pair, price - sl_dist)
|
||
tp = _fmt_price(pair, price + tp_dist)
|
||
lines += [
|
||
"",
|
||
f"_ATR-based zones:_ SL `{sl}` | TP `{tp}`",
|
||
]
|
||
elif verdict in ("SELL", "STRONG SELL"):
|
||
sl = _fmt_price(pair, price + sl_dist)
|
||
tp = _fmt_price(pair, price - tp_dist)
|
||
lines += [
|
||
"",
|
||
f"_ATR-based zones:_ SL `{sl}` | TP `{tp}`",
|
||
]
|
||
|
||
caption = "\n".join(lines) + DISCLAIMER
|
||
img = generate_chart(df, pair, tf, verdict=verdict, confidence=report["confidence"])
|
||
await update.message.reply_photo(img, caption=caption, parse_mode="Markdown")
|
||
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 + chart - /signal EURUSD 1h"),
|
||
BotCommand("chart", "Candlestick chart with EMA/BB/RSI - /chart 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))
|
||
app.add_handler(CommandHandler("chart", chart_cmd))
|
||
logger.info("Bot starting...")
|
||
app.run_polling()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|