diff --git a/data_provider.py b/data_provider.py index ab0b4ab..dac5cca 100644 --- a/data_provider.py +++ b/data_provider.py @@ -73,6 +73,7 @@ class ForexDataProvider: self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} self._request_times: deque = deque() # timestamps of recent API credits used + self._no_api_data: set = set() # td_symbols with no values from API (use synthetic) self._session = requests.Session() self._session.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)" @@ -281,7 +282,20 @@ class ForexDataProvider: values = sym_data.get("values") if not values: - logger.warning(f"No values in batch response for {td_sym}") + if td_sym not in self._no_api_data: + logger.warning( + f"No API data for {td_sym} — using synthetic data" + ) + self._no_api_data.add(td_sym) + # Cache synthetic so fetch_ohlcv won't make a redundant API call + try: + syn = self._generate_synthetic_data(symbol, timeframe, lookback_days) + syn = self._validate_and_clean(syn, symbol, timeframe) + cache_key = f"{symbol}_{timeframe.value}" + self._cache[cache_key] = syn + self._cache_timestamps[cache_key] = now + except Exception: + pass continue records = [ diff --git a/requirements.txt b/requirements.txt index 2379543..615386f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ numpy>=1.24 pandas>=2.0 requests>=2.28 +python-telegram-bot[job-queue]>=20.0 diff --git a/signal_aggregator.py b/signal_aggregator.py index 4010e29..38122bb 100644 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -202,7 +202,11 @@ class SignalAggregator: report.news_description = sent_desc # ── 5. Spread Estimate ───────────────────────────────────── - report.spread_estimate = self.data_provider.get_spread_estimate(symbol) + # Derive spread from already-fetched primary_df to avoid a separate M5 API call + if primary_df is not None and not primary_df.empty: + report.spread_estimate = float((primary_df["High"] - primary_df["Low"]).median()) + else: + report.spread_estimate = self.data_provider.get_spread_estimate(symbol) # ── 6. Compute Final Signal ──────────────────────────────── self._compute_final_signal(report) diff --git a/telegram_bot.py b/telegram_bot.py new file mode 100644 index 0000000..f22aa34 --- /dev/null +++ b/telegram_bot.py @@ -0,0 +1,588 @@ +#!/usr/bin/env python3 +""" +Forex Signal Intelligence Platform - Telegram Bot +=================================================== +Exposes all platform functionality through a Telegram bot interface. + +Usage: + export TELEGRAM_BOT_TOKEN="your_bot_token_here" + python telegram_bot.py + +Commands: + /scan [TF] — Scan all major pairs + /signals [TF] — Only actionable signals (grade B+) + /pair EURUSD [TF] — Detailed analysis of one pair + /all [TF] — Scan major + cross pairs + /tf H1 — Set default timeframe + /watch [minutes] — Enable periodic scan alerts + /unwatch — Disable alerts + /status — Current settings + /help — Full command reference +""" + +import asyncio +import logging +import os +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from typing import Dict, Optional + +from telegram import Update +from telegram.constants import ParseMode +from telegram.ext import Application, CallbackContext, CommandHandler, ContextTypes + +from config import ( + CROSS_PAIRS, MAJOR_PAIRS, SYMBOL_DISPLAY, + PlatformConfig, SignalStrength, Timeframe, +) +from signal_aggregator import SignalAggregator, SignalReport + +logger = logging.getLogger(__name__) + +# ── Constants ────────────────────────────────────────────────────── + +TF_MAP = { + "M5": Timeframe.M5, + "M15": Timeframe.M15, + "M30": Timeframe.M30, + "H1": Timeframe.H1, + "H4": Timeframe.H4, + "D1": Timeframe.D1, +} + +TF_CONFIRMATIONS = { + "M5": [Timeframe.M1, Timeframe.M15], + "M15": [Timeframe.M5, Timeframe.H1], + "M30": [Timeframe.M15, Timeframe.H1], + "H1": [Timeframe.M15, Timeframe.H4], + "H4": [Timeframe.H1, Timeframe.D1], + "D1": [Timeframe.H4, Timeframe.W1], +} + +SIGNAL_EMOJI = { + SignalStrength.STRONG_BUY: "🟢⬆⬆", + SignalStrength.BUY: "🟢⬆", + SignalStrength.NEUTRAL: "⬜↔", + SignalStrength.SELL: "🔴⬇", + SignalStrength.STRONG_SELL: "🔴⬇⬇", +} + +SIGNAL_NAME = { + SignalStrength.STRONG_BUY: "STRONG BUY", + SignalStrength.BUY: "BUY", + SignalStrength.NEUTRAL: "NEUTRAL", + SignalStrength.SELL: "SELL", + SignalStrength.STRONG_SELL: "STRONG SELL", +} + +GRADE_EMOJI = { + "A+": "⭐⭐", + "A": "⭐", + "B": "🔵", + "C": "🟡", + "D": "🟠", + "F": "⚫", +} + +WATCH_MIN_INTERVAL = 5 # minimum watch interval in minutes +WATCH_MAX_INTERVAL = 120 # maximum watch interval in minutes +WATCH_DEFAULT = 5 + + +# ── Per-chat state ───────────────────────────────────────────────── + +class ChatState: + def __init__(self): + self.timeframe: str = "H1" + self.pairs: list = list(MAJOR_PAIRS) + self.watch_enabled: bool = False + + +# ── HTML helpers ─────────────────────────────────────────────────── + +def _esc(text: str) -> str: + """Escape HTML special characters for Telegram HTML parse mode.""" + return text.replace("&", "&").replace("<", "<").replace(">", ">") + + +def _split_messages(lines: list[str], limit: int = 4000) -> list[str]: + """Join lines into messages that stay under Telegram's 4096-char limit.""" + messages, current, current_len = [], [], 0 + for line in lines: + line_len = len(line) + 1 # +1 for newline + if current and current_len + line_len > limit: + messages.append("\n".join(current)) + current, current_len = [], 0 + current.append(line) + current_len += line_len + if current: + messages.append("\n".join(current)) + return messages + + +# ── Message formatters ───────────────────────────────────────────── + +def _format_scan(reports: list[SignalReport], timeframe: str, title: str = "FOREX SIGNAL SCAN") -> str: + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + lines = [ + f"📊 {title} — {timeframe}", + "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━", + f"🕐 {now}\n", + ] + + for r in reports: + sig_em = SIGNAL_EMOJI.get(r.final_signal, "❔") + grade_em = GRADE_EMOJI.get(r.signal_quality, "") + sig_name = SIGNAL_NAME.get(r.final_signal, "?") + event_flag = " ⚠️" if r.event_risk == "danger" else (" ⚡" if r.event_risk == "caution" else "") + + lines.append( + f"{sig_em} {_esc(r.display_name)} " + f"{grade_em}{r.signal_quality} " + f"{_esc(sig_name):<11} " + f"{r.final_confidence:.0f}%{event_flag}" + ) + + lines += ["", "Use /pair EURUSD for detailed analysis"] + return "\n".join(lines) + + +def _format_pair_detail(r: SignalReport) -> list[str]: + """ + Return a list of message strings for a single pair. + Split across messages so each stays under Telegram's 4096-char limit. + """ + sig_em = SIGNAL_EMOJI.get(r.final_signal, "❔") + grade_em = GRADE_EMOJI.get(r.signal_quality, "") + sig_name = SIGNAL_NAME.get(r.final_signal, "?") + + # ── Message 1: Header, MTF, Trade Setup, Key Levels ─────────── + m1 = [ + f"{sig_em} {_esc(r.display_name)} — {_esc(sig_name)}", + "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━", + f"Price: {r.current_price:.5f}", + f"Regime: {_esc(r.market_regime)} Vol: {_esc(r.volatility_regime)}", + "", + f"Score: {r.final_score:+.1f} " + f"Grade: {grade_em}{r.signal_quality} " + f"Conf: {r.final_confidence:.0f}%", + f"Action: {_esc(r.action_recommendation)}", + ] + + if r.event_risk != "clear": + icon = "⚠️" if r.event_risk == "danger" else "⚡" + m1.append(f"{icon} Event: {_esc(r.event_description)}") + + if r.news_description and r.news_description != "Neutral news sentiment": + m1.append(f"📰 News: {_esc(r.news_description)}") + + # Multi-timeframe + if r.mtf_signals: + agree = "✅ ALIGNED" if r.mtf_agreement else "⚡ DIVERGENT" + m1 += ["", f"📊 Multi-Timeframe: {agree}"] + for tf, sig in r.mtf_signals.items(): + score = r.mtf_scores.get(tf, 0) + m1.append( + f" {tf:>5} {SIGNAL_EMOJI.get(sig, '?')} " + f"{score:+.1f}" + ) + + # Trade setup + if r.trade_setup: + ts = r.trade_setup + dir_em = "🟢" if ts.direction == "BUY" else "🔴" + m1 += [ + "", + "💼 Trade Setup", + f" Direction: {dir_em} {ts.direction}", + f" Entry: {ts.entry_price:.5f}", + f" Stop Loss: {ts.stop_loss:.5f} ({ts.risk_pips:.1f} pips)", + f" Take Profit 1: {ts.take_profit_1:.5f} ({ts.reward_pips:.1f} pips)", + f" Take Profit 2: {ts.take_profit_2:.5f}", + f" Risk:Reward: 1:{ts.risk_reward:.1f}", + ] + + # Key levels + if r.primary_analysis and r.primary_analysis.key_levels: + lvls = r.primary_analysis.key_levels + shown = [(k, lvls[k]) for k in ("r3", "r2", "r1", "pivot", "s1", "s2", "s3") if k in lvls] + if shown: + m1 += ["", "📍 Key Levels:"] + for k, v in shown: + arrow = "🔴" if v > r.current_price else "🟢" + m1.append(f" {arrow} {k.upper():>6}: {v:.5f}") + + # ── Message 2: Indicators + Reasons ─────────────────────────── + m2 = [] + + if r.primary_analysis: + ind_lines = [] + for name, ind in r.primary_analysis.indicators.items(): + if name.startswith("_") or ind.weight == 0: + continue + em = SIGNAL_EMOJI.get(ind.signal, "↔") + # Trim long descriptions to keep message readable + desc = _esc(ind.description[:90]) + ind_lines.append( + f" {em} {_esc(ind.name)} {ind.confidence:.0f}%" + f"\n {desc}" + ) + if ind_lines: + m2 += ["📉 Indicators:"] + ind_lines + + if r.bullish_reasons: + if m2: + m2.append("") + m2.append("🟢 Bullish Factors:") + for reason in r.bullish_reasons[:5]: + m2.append(f" ▲ {_esc(reason)}") + + if r.bearish_reasons: + if m2: + m2.append("") + m2.append("🔴 Bearish Factors:") + for reason in r.bearish_reasons[:5]: + m2.append(f" ▼ {_esc(reason)}") + + if r.warnings: + if m2: + m2.append("") + m2.append("⚠️ Warnings:") + for w in r.warnings: + m2.append(f" ! {_esc(w)}") + + result = _split_messages(m1) + if m2: + result += _split_messages(m2) + return result + + +# ══════════════════════════════════════════════════════════════════ +# BOT CLASS +# ══════════════════════════════════════════════════════════════════ + +class ForexTelegramBot: + def __init__(self): + self._chat_states: Dict[int, ChatState] = {} + self._executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="forex") + + def _state(self, chat_id: int) -> ChatState: + if chat_id not in self._chat_states: + self._chat_states[chat_id] = ChatState() + return self._chat_states[chat_id] + + def _build_config(self, state: ChatState) -> PlatformConfig: + config = PlatformConfig() + config.primary_timeframe = TF_MAP[state.timeframe] + config.confirmation_timeframes = TF_CONFIRMATIONS.get( + state.timeframe, [Timeframe.M15, Timeframe.H4] + ) + config.watch_pairs = list(state.pairs) + return config + + # ── Blocking analysis (run in thread pool) ───────────────────── + + def _run_scan(self, state: ChatState) -> list[SignalReport]: + config = self._build_config(state) + return SignalAggregator(config).analyze_all_pairs() + + def _run_pair(self, symbol: str, state: ChatState) -> Optional[SignalReport]: + config = self._build_config(state) + config.watch_pairs = [symbol] + reports = SignalAggregator(config).analyze_all_pairs() + return reports[0] if reports else None + + async def _analyze(self, func, *args): + """Run a blocking analysis function in the thread pool.""" + loop = asyncio.get_running_loop() + return await loop.run_in_executor(self._executor, func, *args) + + # ── Parse helpers ────────────────────────────────────────────── + + @staticmethod + def _parse_tf(args: list[str]) -> Optional[str]: + """Return the first arg that is a valid timeframe key, or None.""" + for a in args: + if a.upper() in TF_MAP: + return a.upper() + return None + + # ── Commands ─────────────────────────────────────────────────── + + async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + text = ( + "🌐 Forex Signal Intelligence Platform\n" + "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\n" + "Multi-layer technical + fundamental signal analysis across major Forex pairs.\n\n" + "Commands:\n" + " /scan — Scan all major pairs\n" + " /signals — Actionable signals only (grade B+)\n" + " /pair EURUSD — Detailed single-pair analysis\n" + " /all — Include cross pairs in scan\n" + " /tf H1 — Set default timeframe\n" + " /watch — Periodic scan alerts (every 5 min)\n" + " /unwatch — Disable alerts\n" + " /status — Current settings\n" + " /help — Full command reference\n\n" + "⚠️ Educational/research use only. Not financial advice." + ) + await update.message.reply_text(text, parse_mode=ParseMode.HTML) + + async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + text = ( + "📖 Command Reference\n\n" + "/scan — Scan all major pairs (uses default TF)\n" + "/scan H4 — Scan with a specific timeframe\n\n" + "/signals — Grade B+ signals only\n" + "/signals H4 — With a specific timeframe\n\n" + "/pair EURUSD — Detailed analysis (uses default TF)\n" + "/pair EURUSD H4 — With a specific timeframe\n\n" + "/all — Scan major + cross pairs\n" + "/all H4 — With a specific timeframe\n\n" + "/tf H1 — Set your default timeframe\n" + " Options: M5 M15 M30 H1 H4 D1\n\n" + "/watch — Alert every 5 min (actionable signals only)\n" + "/watch 15 — Alert every 15 min (5–120 min range)\n" + "/unwatch — Stop alerts\n\n" + "/status — View current settings\n\n" + "Signal key:\n" + " 🟢⬆⬆ Strong Buy 🟢⬆ Buy ⬜↔ Neutral\n" + " 🔴⬇ Sell 🔴⬇⬇ Strong Sell\n\n" + "Grade key:\n" + " A+ ⭐⭐ A ⭐ B 🔵 C 🟡 D 🟠 F ⚫" + ) + await update.message.reply_text(text, parse_mode=ParseMode.HTML) + + async def cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + state = self._state(update.effective_chat.id) + pairs_display = ", ".join(SYMBOL_DISPLAY.get(p, p) for p in state.pairs) + watch_str = "✅ enabled" if state.watch_enabled else "❌ disabled" + text = ( + "⚙️ Current Settings\n\n" + f"Timeframe: {state.timeframe}\n" + f"Pairs: {_esc(pairs_display)}\n" + f"Watch alerts: {watch_str}" + ) + await update.message.reply_text(text, parse_mode=ParseMode.HTML) + + async def cmd_tf(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + tf = self._parse_tf(context.args or []) + if not tf: + opts = " ".join(TF_MAP.keys()) + await update.message.reply_text( + f"Usage: /tf <timeframe>\nOptions: {opts}", + parse_mode=ParseMode.HTML, + ) + return + self._state(update.effective_chat.id).timeframe = tf + await update.message.reply_text( + f"✅ Default timeframe set to {tf}", parse_mode=ParseMode.HTML + ) + + async def cmd_scan(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + chat_id = update.effective_chat.id + state = self._state(chat_id) + + # Allow inline TF override: /scan H4 + scan_state = ChatState() + scan_state.timeframe = self._parse_tf(context.args or []) or state.timeframe + scan_state.pairs = list(state.pairs) + + msg = await update.message.reply_text("🔄 Scanning markets, please wait…") + try: + reports = await self._analyze(self._run_scan, scan_state) + text = _format_scan(reports, scan_state.timeframe) + await msg.edit_text(text, parse_mode=ParseMode.HTML) + except Exception as e: + logger.error(f"Scan error: {e}") + await msg.edit_text(f"❌ Scan failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) + + async def cmd_signals(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + chat_id = update.effective_chat.id + state = self._state(chat_id) + + scan_state = ChatState() + scan_state.timeframe = self._parse_tf(context.args or []) or state.timeframe + scan_state.pairs = list(state.pairs) + + msg = await update.message.reply_text("🔄 Scanning for signals…") + try: + reports = await self._analyze(self._run_scan, scan_state) + actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] + if actionable: + text = _format_scan(actionable, scan_state.timeframe, title="🎯 ACTIONABLE SIGNALS") + else: + text = ( + f"🎯 ACTIONABLE SIGNALS — {scan_state.timeframe}\n\n" + "No grade B+ signals right now.\n" + "Use /scan to see all pairs." + ) + await msg.edit_text(text, parse_mode=ParseMode.HTML) + except Exception as e: + logger.error(f"Signals error: {e}") + await msg.edit_text(f"❌ Scan failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) + + async def cmd_pair(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + chat_id = update.effective_chat.id + state = self._state(chat_id) + args = context.args or [] + + if not args: + await update.message.reply_text( + "Usage: /pair <symbol> [timeframe]\n" + "Example: /pair EURUSD H4", + parse_mode=ParseMode.HTML, + ) + return + + raw = args[0].upper().replace("/", "").replace("_", "") + if not raw.endswith("=X"): + raw += "=X" + + pair_state = ChatState() + pair_state.timeframe = self._parse_tf(args[1:]) or state.timeframe + + display = SYMBOL_DISPLAY.get(raw, raw) + msg = await update.message.reply_text(f"🔄 Analyzing {_esc(display)}…") + try: + report = await self._analyze(self._run_pair, raw, pair_state) + if report is None: + await msg.edit_text( + f"❌ No data returned for {_esc(raw)}.", + parse_mode=ParseMode.HTML, + ) + return + messages = _format_pair_detail(report) + await msg.edit_text(messages[0], parse_mode=ParseMode.HTML) + for m in messages[1:]: + await update.message.reply_text(m, parse_mode=ParseMode.HTML) + except Exception as e: + logger.error(f"Pair analysis error for {raw}: {e}") + await msg.edit_text(f"❌ Analysis failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) + + async def cmd_all(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + chat_id = update.effective_chat.id + state = self._state(chat_id) + + all_state = ChatState() + all_state.timeframe = self._parse_tf(context.args or []) or state.timeframe + all_state.pairs = list(MAJOR_PAIRS) + list(CROSS_PAIRS) + + msg = await update.message.reply_text("🔄 Scanning all pairs (major + cross)…") + try: + reports = await self._analyze(self._run_scan, all_state) + text = _format_scan(reports, all_state.timeframe, title="FULL PAIR SCAN") + await msg.edit_text(text, parse_mode=ParseMode.HTML) + except Exception as e: + logger.error(f"All-pairs scan error: {e}") + await msg.edit_text(f"❌ Scan failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) + + async def cmd_watch(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + chat_id = update.effective_chat.id + state = self._state(chat_id) + + interval_minutes = WATCH_DEFAULT + if context.args: + try: + interval_minutes = int(context.args[0]) + interval_minutes = max(WATCH_MIN_INTERVAL, min(WATCH_MAX_INTERVAL, interval_minutes)) + except ValueError: + pass + + # Remove any existing watch job for this chat + for job in context.job_queue.get_jobs_by_name(f"watch_{chat_id}"): + job.schedule_removal() + + state.watch_enabled = True + context.job_queue.run_repeating( + self._watch_job, + interval=interval_minutes * 60, + first=5, + chat_id=chat_id, + name=f"watch_{chat_id}", + ) + await update.message.reply_text( + f"✅ Watch alerts enabled — scanning every {interval_minutes} min.\n" + "Alerts fire when grade B+ signals appear. Use /unwatch to stop.", + parse_mode=ParseMode.HTML, + ) + + async def cmd_unwatch(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + chat_id = update.effective_chat.id + self._state(chat_id).watch_enabled = False + for job in context.job_queue.get_jobs_by_name(f"watch_{chat_id}"): + job.schedule_removal() + await update.message.reply_text("❌ Watch alerts disabled.") + + # ── Watch job ────────────────────────────────────────────────── + + async def _watch_job(self, context: CallbackContext): + chat_id = context.job.chat_id + state = self._state(chat_id) + if not state.watch_enabled: + return + + try: + reports = await self._analyze(self._run_scan, state) + actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] + if actionable: + text = ( + "🔔 SIGNAL ALERT\n\n" + + _format_scan(actionable, state.timeframe, title="ACTIONABLE SIGNALS") + ) + else: + now = datetime.now(timezone.utc).strftime("%H:%M UTC") + text = f"📊 Watch scan {now} — No grade B+ signals detected." + await context.bot.send_message(chat_id, text, parse_mode=ParseMode.HTML) + except Exception as e: + logger.error(f"Watch job error for chat {chat_id}: {e}") + + # ── Application builder ──────────────────────────────────────── + + def build_application(self) -> Application: + token = os.environ.get("TELEGRAM_BOT_TOKEN") + if not token: + raise ValueError( + "TELEGRAM_BOT_TOKEN environment variable is not set.\n" + "Get a token from @BotFather on Telegram, then:\n" + " export TELEGRAM_BOT_TOKEN='your_token_here'" + ) + + app = ( + Application.builder() + .token(token) + .build() + ) + + app.add_handler(CommandHandler("start", self.cmd_start)) + app.add_handler(CommandHandler("help", self.cmd_help)) + app.add_handler(CommandHandler("status", self.cmd_status)) + app.add_handler(CommandHandler("tf", self.cmd_tf)) + app.add_handler(CommandHandler("scan", self.cmd_scan)) + app.add_handler(CommandHandler("signals", self.cmd_signals)) + app.add_handler(CommandHandler("pair", self.cmd_pair)) + app.add_handler(CommandHandler("all", self.cmd_all)) + app.add_handler(CommandHandler("watch", self.cmd_watch)) + app.add_handler(CommandHandler("unwatch", self.cmd_unwatch)) + + return app + + +# ══════════════════════════════════════════════════════════════════ + +def main(): + logging.basicConfig( + level=logging.WARNING, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + + bot = ForexTelegramBot() + app = bot.build_application() + + print("🤖 Forex Signal Bot started. Press Ctrl+C to stop.") + app.run_polling(drop_pending_updates=True) + + +if __name__ == "__main__": + main()