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

Pro tier upgrade: quant rearchitect + WebSocket level-breach monitor

Browse Source 
- Restore XAU/USD (Pro plan), drop USD/CAD, raise rate limit to 55/min
- Reduce signal engine to 4 uncorrelated pillars (EMA, RSI, BB, Candlestick)
  with regime-adaptive weights (trending boosts EMA, ranging boosts RSI+BB)
- Add structure-based S/R (swing highs/lows) for trade setup targeting
- Fix confidence death spiral: all adjustments now additive, not multiplicative
- Dynamic R:R trade setup: SL behind swing level, TP at next structure level,
  reject if R:R < 1.5
- Fix news engine for all 4 pairs (was hardcoded to USDJPY)
- Add WebSocket monitor: streams live ticks, triggers M15 scan on level breach
  with 5-min cooldown per pair — catches breakouts between candle boundaries
- Add correlation filter (EUR/USD+GBP/USD, XAU/USD+USD/JPY exposure warnings)
This commit is contained in:
harwinder
2026-05-11 16:33:23 -04:00
parent cc66e65f2d
commit 058ef57d8f
8 changed files with 435 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_bot.py
+254 -59
View File
@@ -2,19 +2,27 @@
"""
ForexBot2 — Telegram Bot
========================
Async Telegram interface with candle-boundary auto-scanner.
Async Telegram interface with candle-boundary auto-scanner and
real-time WebSocket level-breach monitor.
Manual scan: 4 pairs × 2 timeframes via inline buttons.
Manual scan: 4 pairs × 4 timeframes via inline buttons.
Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set.
/autoscan toggles for additional chats or to stop.
M5 fires at :00/:05/:10/… + 10s buffer.
M15 fires at :00/:15/:30/:45 + 10s buffer.
H1 fires at :00 + 70s buffer (staggered to respect 8 credits/min).
H1 fires at :00 + 10s buffer.
H4 fires at :00 every 4 hours + 10s buffer.
WebSocket: Streams live ticks from TwelveData for all 4 pairs.
When price crosses a key structure level (swing S/R, BB,
EMA200) cached from the last boundary scan, triggers an
immediate M15 scan and sends a Level Break Alert.
5-minute cooldown per pair prevents spam.
Credit budget (free tier):
M15 boundary: 4 pairs × 1 credit = 4 credits
H1 boundary: 4 pairs × 1 credit = 4 credits (staggered to next minute)
Top of hour: 4 + 4 = 8 across two minutes → ≤ 8 / min
Daily total: 384 (M15) + 96 (H1) = 480 / day (limit 800)
Plan: TwelveData Pro (55 credits/min, 1500 WS credits).
Credit budget (worst case, 4-hour boundary):
Unique fetches: 4 pairs × ~6 TFs = ~24 calls (rest cached) → fits in 55/min.
WebSocket: 4 symbols = 4 credits (one-time on connect).
Level breach scans: ~3 REST calls each, ~10/hour worst case.
Usage:
python telegram_bot.py
@@ -22,12 +30,16 @@ Usage:
"""
import asyncio
import json
import logging
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import List, Optional
from typing import Dict, List, Optional
import websockets
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ParseMode
@@ -40,7 +52,7 @@ from telegram.ext import (
from config import (
PlatformConfig, SignalStrength, Timeframe,
EURUSD, USDJPY, GBPUSD, USDCAD, PAIR_INFO,
EURUSD, USDJPY, GBPUSD, XAUUSD, PAIR_INFO,
)
from data_provider import ForexDataProvider
from signal_aggregator import SignalAggregator, SignalReport
@@ -63,31 +75,60 @@ CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022")
# ── Pairs & buttons ───────────────────────────────────────────────────
SCAN_PAIRS = [EURUSD, USDJPY, GBPUSD, USDCAD]
SCAN_PAIRS = [EURUSD, USDJPY, GBPUSD, XAUUSD]
CORRELATION_GROUPS = {
"usd_negative": {"pairs": {EURUSD, GBPUSD}, "label": "USD-weakness"},
"risk_sentiment": {"pairs": {XAUUSD, USDJPY}, "label": "Risk-sentiment"},
}
_WS_TO_PAIR = {
"EUR/USD": EURUSD,
"GBP/USD": GBPUSD,
"USD/JPY": USDJPY,
"XAU/USD": XAUUSD,
}
_WS_SYMBOLS = list(_WS_TO_PAIR.keys())
_LEVEL_KEYS = (
"swing_high_1", "swing_high_2", "swing_high_3",
"swing_low_1", "swing_low_2", "swing_low_3",
"bb_upper", "bb_lower", "ema_trend",
)
SCAN_BUTTONS = [
{"cb": "scan_eurusd_m5", "symbol": EURUSD, "tf": Timeframe.M5, "label": "EUR/USD 5M"},
{"cb": "scan_eurusd_m15", "symbol": EURUSD, "tf": Timeframe.M15, "label": "EUR/USD 15M"},
{"cb": "scan_eurusd_h1", "symbol": EURUSD, "tf": Timeframe.H1, "label": "EUR/USD 1H"},
{"cb": "scan_eurusd_h4", "symbol": EURUSD, "tf": Timeframe.H4, "label": "EUR/USD 4H"},
{"cb": "scan_usdjpy_m5", "symbol": USDJPY, "tf": Timeframe.M5, "label": "USD/JPY 5M"},
{"cb": "scan_usdjpy_m15", "symbol": USDJPY, "tf": Timeframe.M15, "label": "USD/JPY 15M"},
{"cb": "scan_usdjpy_h1", "symbol": USDJPY, "tf": Timeframe.H1, "label": "USD/JPY 1H"},
{"cb": "scan_usdjpy_h4", "symbol": USDJPY, "tf": Timeframe.H4, "label": "USD/JPY 4H"},
{"cb": "scan_gbpusd_m5", "symbol": GBPUSD, "tf": Timeframe.M5, "label": "GBP/USD 5M"},
{"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"},
{"cb": "scan_gbpusd_h1", "symbol": GBPUSD, "tf": Timeframe.H1, "label": "GBP/USD 1H"},
{"cb": "scan_usdcad_m15", "symbol": USDCAD, "tf": Timeframe.M15, "label": "USD/CAD 15M"},
{"cb": "scan_usdcad_h1", "symbol": USDCAD, "tf": Timeframe.H1, "label": "USD/CAD 1H"},
{"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"},
{"cb": "scan_xauusd_m5", "symbol": XAUUSD, "tf": Timeframe.M5, "label": "XAU/USD 5M"},
{"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"},
{"cb": "scan_xauusd_h1", "symbol": XAUUSD, "tf": Timeframe.H1, "label": "XAU/USD 1H"},
{"cb": "scan_xauusd_h4", "symbol": XAUUSD, "tf": Timeframe.H4, "label": "XAU/USD 4H"},
]
SCAN_LOOKUP = {b["cb"]: b for b in SCAN_BUTTONS}
# Manual scans use richer confirmations (may spend extra cached credits)
TF_CONFIRMATIONS = {
Timeframe.M5: [Timeframe.M15, Timeframe.H1],
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
}
# Auto-scan uses only cross-TF (already cached, zero extra credits)
AUTOSCAN_CONFIRMATIONS = {
Timeframe.M15: [Timeframe.H1],
Timeframe.H1: [Timeframe.M15],
Timeframe.M5: [Timeframe.M15, Timeframe.H1],
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
}
_executor = ThreadPoolExecutor(max_workers=2)
@@ -209,6 +250,33 @@ def format_signal_message(report: SignalReport) -> str:
return "\n".join(lines)
# ── Correlation filter ───────────────────────────────────────────────
def _detect_correlated_exposure(reports: List[SignalReport]) -> List[str]:
warnings = []
for group_info in CORRELATION_GROUPS.values():
group_pairs = group_info["pairs"]
group_label = group_info["label"]
active = [
r for r in reports
if r.symbol in group_pairs
and r.final_signal not in (SignalStrength.NEUTRAL,)
]
if len(active) >= 2:
directions = set()
for r in active:
if r.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
directions.add("BUY")
else:
directions.add("SELL")
if len(directions) == 1:
names = ", ".join(r.display_name for r in active)
warnings.append(
f"⚠️ Same exposure ({group_label}): {names} both {directions.pop()}"
)
return warnings
# ── Auto-scan summary format ─────────────────────────────────────────
def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
@@ -235,6 +303,10 @@ def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
else:
lines.append("No actionable signals this cycle.")
corr_warnings = _detect_correlated_exposure(reports)
for w in corr_warnings:
lines.append(w)
return "\n".join(lines)
@@ -242,10 +314,10 @@ def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
def scan_keyboard() -> InlineKeyboardMarkup:
rows = []
for i in range(0, len(SCAN_BUTTONS), 2):
for i in range(0, len(SCAN_BUTTONS), 4):
rows.append([
InlineKeyboardButton(f"🔍 {b['label']}", callback_data=b["cb"])
for b in SCAN_BUTTONS[i:i + 2]
InlineKeyboardButton(b["label"], callback_data=b["cb"])
for b in SCAN_BUTTONS[i:i + 4]
])
return InlineKeyboardMarkup(rows)
@@ -315,6 +387,8 @@ async def _scan_timeframe(
if not reports:
return
_update_level_cache(reports)
summary = _format_scan_summary(reports, tf)
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")]
@@ -328,60 +402,175 @@ async def _scan_timeframe(
logger.exception(f"Failed to send to chat {chat_id}")
async def _m15_loop(bot) -> None:
"""Sleep until each M15 candle close, then scan all pairs."""
# ── WebSocket level-breach monitor ──────────────────────────────────
_level_cache: Dict[str, Dict[str, float]] = {}
_last_prices: Dict[str, float] = {}
_scan_cooldowns: Dict[str, float] = {}
_cfg = PlatformConfig()
_WS_API_KEY = os.environ.get("TWELVE_DATA_API_KEY") or _cfg.twelve_data_api_key
_WS_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={_WS_API_KEY}"
_WS_COOLDOWN = _cfg.ws_cooldown_seconds
def _update_level_cache(reports: List[SignalReport]) -> None:
for r in reports:
if r.primary_analysis and r.primary_analysis.key_levels:
levels = {
k: v for k, v in r.primary_analysis.key_levels.items()
if k in _LEVEL_KEYS
}
if levels:
_level_cache[r.symbol] = levels
def _check_level_breach(symbol: str, price: float) -> Optional[str]:
levels = _level_cache.get(symbol)
if not levels:
return None
last = _last_prices.get(symbol)
if last is None:
return None
for name, level in levels.items():
if (last < level <= price) or (last > level >= price):
return name
return None
async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, level_name: str) -> None:
now = time.time()
if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN:
return
_scan_cooldowns[pair_symbol] = now
display = PAIR_INFO.get(pair_symbol, {}).get("display", ws_symbol)
label = level_name.replace("_", " ").title()
logger.info(f"Level breach: {display} crossed {label} at {price}")
try:
report = await run_signal(pair_symbol, Timeframe.M15)
_update_level_cache([report])
text = (
f"🔔 *Level Break — {display}*\n"
f"Price crossed *{label}* at `{price}`\n"
f"━━━━━━━━━━━━━━━━━━━━\n"
+ format_signal_message(report)
)
except Exception as exc:
logger.exception(f"Breach scan failed for {display}")
text = f"🔔 *Level Break — {display}*\nPrice crossed {label} at `{price}`\n❌ Scan failed: {exc}"
for chat_id in list(_autoscan_chats):
try:
await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
except Exception:
logger.exception(f"Failed to send breach alert to {chat_id}")
async def _ws_monitor(bot) -> None:
backoff = 1
while _autoscan_chats:
try:
async with websockets.connect(_WS_URL) as ws:
subscribe = {
"action": "subscribe",
"params": {"symbols": ",".join(_WS_SYMBOLS)},
}
await ws.send(json.dumps(subscribe))
logger.info(f"WebSocket connected, subscribed to {len(_WS_SYMBOLS)} symbols")
backoff = 1
async for raw in ws:
if not _autoscan_chats:
break
msg = json.loads(raw)
if msg.get("event") != "price":
continue
ws_symbol = msg.get("symbol")
pair_symbol = _WS_TO_PAIR.get(ws_symbol)
if not pair_symbol:
continue
price = float(msg["price"])
breached = _check_level_breach(pair_symbol, price)
_last_prices[pair_symbol] = price
if breached:
asyncio.create_task(
_handle_breach(bot, ws_symbol, pair_symbol, price, breached)
)
except asyncio.CancelledError:
break
except Exception:
logger.exception(f"WebSocket error, reconnecting in {backoff}s")
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 60)
logger.info("WebSocket monitor stopped")
async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
label = tf.value
try:
while _autoscan_chats:
wait = _seconds_until_boundary(15, buffer=10)
logger.info(f"M15 auto-scan: next in {wait:.0f}s")
wait = _seconds_until_boundary(interval_minutes, buffer=10)
logger.info(f"{label} auto-scan: next in {wait:.0f}s")
await asyncio.sleep(wait)
if not _autoscan_chats:
break
logger.info("M15 auto-scan: firing")
logger.info(f"{label} auto-scan: firing")
await _scan_timeframe(
bot, Timeframe.M15, AUTOSCAN_CONFIRMATIONS[Timeframe.M15]
bot, tf, AUTOSCAN_CONFIRMATIONS[tf]
)
except asyncio.CancelledError:
pass
finally:
logger.info("M15 auto-scan loop stopped")
logger.info(f"{label} auto-scan loop stopped")
async def _h1_loop(bot) -> None:
"""Sleep until each H1 candle close + 60s stagger, then scan all pairs."""
try:
while _autoscan_chats:
wait = _seconds_until_boundary(60, buffer=70)
logger.info(f"H1 auto-scan: next in {wait:.0f}s")
await asyncio.sleep(wait)
if not _autoscan_chats:
break
logger.info("H1 auto-scan: firing")
await _scan_timeframe(
bot, Timeframe.H1, AUTOSCAN_CONFIRMATIONS[Timeframe.H1]
)
except asyncio.CancelledError:
pass
finally:
logger.info("H1 auto-scan loop stopped")
_AUTOSCAN_LOOPS = [
(Timeframe.M5, 5),
(Timeframe.M15, 15),
(Timeframe.H1, 60),
(Timeframe.H4, 240),
]
_ws_task: Optional[asyncio.Task] = None
def _start_scheduler(bot) -> None:
"""Start the M15 and H1 scanning loops if not already running."""
global _ws_task
alive = [t for t in _autoscan_tasks if not t.done()]
if not alive:
_autoscan_tasks.clear()
_autoscan_tasks.append(asyncio.create_task(_m15_loop(bot)))
_autoscan_tasks.append(asyncio.create_task(_h1_loop(bot)))
logger.info("Candle-boundary scheduler started")
for tf, interval in _AUTOSCAN_LOOPS:
_autoscan_tasks.append(
asyncio.create_task(_candle_loop(bot, tf, interval))
)
logger.info("Candle-boundary scheduler started (M5/M15/H1/H4)")
if _ws_task is None or _ws_task.done():
_ws_task = asyncio.create_task(_ws_monitor(bot))
logger.info("WebSocket level-breach monitor started")
def _stop_scheduler() -> None:
"""Cancel all scanning loops."""
global _ws_task
for t in _autoscan_tasks:
t.cancel()
_autoscan_tasks.clear()
logger.info("Candle-boundary scheduler stopped")
if _ws_task and not _ws_task.done():
_ws_task.cancel()
_ws_task = None
logger.info("Candle-boundary scheduler and WebSocket monitor stopped")
# ── Handlers ──────────────────────────────────────────────────────────
@@ -417,13 +606,16 @@ async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
_autoscan_chats.add(chat_id)
_start_scheduler(context.bot)
next_m5 = _seconds_until_boundary(5, buffer=10)
next_m15 = _seconds_until_boundary(15, buffer=10)
next_h1 = _seconds_until_boundary(60, buffer=70)
next_h1 = _seconds_until_boundary(60, buffer=10)
next_h4 = _seconds_until_boundary(240, buffer=10)
await update.message.reply_text(
"✅ *Auto-scan enabled*\n\n"
f"Scanning {len(SCAN_PAIRS)} pairs on M15 + H1 at candle close.\n"
f"Next M15 scan in {next_m15 / 60:.0f} min\n"
f"Next H1 scan in {next_h1 / 60:.0f} min\n\n"
f"Scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4 at candle close.\n"
f"📡 Live price monitoring active — level-break alerts enabled.\n"
f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | "
f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n"
"Use /autoscan again to stop.",
parse_mode=ParseMode.MARKDOWN,
)
@@ -470,20 +662,23 @@ async def _post_init(app: Application) -> None:
_autoscan_chats.add(chat_id)
_start_scheduler(app.bot)
next_m5 = _seconds_until_boundary(5, buffer=10)
next_m15 = _seconds_until_boundary(15, buffer=10)
next_h1 = _seconds_until_boundary(60, buffer=70)
next_h1 = _seconds_until_boundary(60, buffer=10)
next_h4 = _seconds_until_boundary(240, buffer=10)
logger.info(
f"Auto-scan started for chat {chat_id} — "
f"M15 in {next_m15:.0f}s, H1 in {next_h1:.0f}s"
f"M5 {next_m5:.0f}s, M15 {next_m15:.0f}s, H1 {next_h1:.0f}s, H4 {next_h4:.0f}s"
)
try:
await app.bot.send_message(
chat_id,
f"🤖 *ForexBot2 started*\n\n"
f"Auto-scanning {len(SCAN_PAIRS)} pairs on M15 + H1.\n"
f"Next M15 in {next_m15 / 60:.0f} min | "
f"Next H1 in {next_h1 / 60:.0f} min\n\n"
f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n"
f"📡 Live price monitoring active — level-break alerts enabled.\n"
f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | "
f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n"
f"/autoscan to stop | Manual buttons via /start",
parse_mode=ParseMode.MARKDOWN,
)