diff --git a/config.py b/config.py index 0aa2d2b..bfb16b2 100755 --- a/config.py +++ b/config.py @@ -237,6 +237,5 @@ class PlatformConfig: default_factory=lambda: [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4] ) candle_buffer_seconds: int = 10 - ws_cooldown_seconds: int = 300 log_level: str = "INFO" diff --git a/requirements.txt b/requirements.txt index 19ed6a7..7b7872d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,4 +3,3 @@ pandas>=2.0 requests>=2.28 twelvedata>=1.2.0 python-telegram-bot>=20.0 -websockets>=12.0 diff --git a/test_all.py b/test_all.py index 7c5ce8f..fe0cb1f 100644 --- a/test_all.py +++ b/test_all.py @@ -1324,8 +1324,7 @@ from testing_telegram_bot import ( _seconds_until_boundary, _check_level_breach, _format_stats_dashboard, scan_keyboard, SCAN_BUTTONS, SCAN_LOOKUP, - _update_level_cache, _last_prices, _level_cache, _level_cache_ts, - _WS_TO_PAIR, + _update_level_cache, _level_cache, _level_cache_ts, ) @@ -1428,65 +1427,50 @@ def _(): assert result <= 240 * 60 + 10 + 1 -@test("_check_level_breach detects upward cross") +@test("_check_level_breach detects price near level") def _(): _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} _level_cache_ts["TEST_PAIR"] = time.time() - _last_prices["TEST_PAIR"] = 1.0999 - result = _check_level_breach("TEST_PAIR", 1.1001) + result = _check_level_breach("TEST_PAIR", 1.10005) assert result == "swing_high_1" - # cleanup del _level_cache["TEST_PAIR"] del _level_cache_ts["TEST_PAIR"] - del _last_prices["TEST_PAIR"] -@test("_check_level_breach detects downward cross") +@test("_check_level_breach detects price at exact level") def _(): _level_cache["TEST_PAIR"] = {"swing_low_1": 1.0900} _level_cache_ts["TEST_PAIR"] = time.time() - _last_prices["TEST_PAIR"] = 1.0901 - result = _check_level_breach("TEST_PAIR", 1.0899) + result = _check_level_breach("TEST_PAIR", 1.0900) assert result == "swing_low_1" del _level_cache["TEST_PAIR"] del _level_cache_ts["TEST_PAIR"] - del _last_prices["TEST_PAIR"] -@test("_check_level_breach no breach when price stays same side") +@test("_check_level_breach no breach when price is far from level") def _(): _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} _level_cache_ts["TEST_PAIR"] = time.time() - _last_prices["TEST_PAIR"] = 1.0990 - result = _check_level_breach("TEST_PAIR", 1.0995) + result = _check_level_breach("TEST_PAIR", 1.0950) assert result is None del _level_cache["TEST_PAIR"] del _level_cache_ts["TEST_PAIR"] - del _last_prices["TEST_PAIR"] @test("_check_level_breach returns None for expired cache") def _(): _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} _level_cache_ts["TEST_PAIR"] = time.time() - 7200 # 2h old - _last_prices["TEST_PAIR"] = 1.0999 - result = _check_level_breach("TEST_PAIR", 1.1001) + result = _check_level_breach("TEST_PAIR", 1.1000) assert result is None del _level_cache["TEST_PAIR"] del _level_cache_ts["TEST_PAIR"] - del _last_prices["TEST_PAIR"] -@test("_check_level_breach returns None when no previous price") +@test("_check_level_breach returns None for empty cache") def _(): - _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} - _level_cache_ts["TEST_PAIR"] = time.time() - if "TEST_PAIR" in _last_prices: - del _last_prices["TEST_PAIR"] - result = _check_level_breach("TEST_PAIR", 1.1001) + result = _check_level_breach("NONEXISTENT_PAIR", 1.1000) assert result is None - del _level_cache["TEST_PAIR"] - del _level_cache_ts["TEST_PAIR"] @test("SCAN_BUTTONS and SCAN_LOOKUP consistency") @@ -1524,11 +1508,6 @@ def _(): _level_cache_ts.clear() -@test("_WS_TO_PAIR mapping is correct") -def _(): - assert _WS_TO_PAIR["XAU/USD"] == XAUUSD - - @test("_format_stats_dashboard returns waiting message when empty") def _(): import testing_telegram_bot diff --git a/testing_telegram_bot.py b/testing_telegram_bot.py index 85dd68b..8ff40fc 100755 --- a/testing_telegram_bot.py +++ b/testing_telegram_bot.py @@ -2,8 +2,7 @@ """ ForexBot2 — Telegram Bot ======================== -Async Telegram interface with candle-boundary auto-scanner and -real-time WebSocket level-breach monitor. +Async Telegram interface with candle-boundary auto-scanner. Manual scan: XAU/USD + EUR/USD via inline buttons. Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set. @@ -12,11 +11,6 @@ Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set. M15 fires at :00/:15/:30/:45 + 10s buffer. H1 fires at :00 + 10s buffer. H4 fires at :00 every 4 hours + 10s buffer. -WebSocket: Streams live ticks from TwelveData for XAU/USD. - 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 prevents spam. Usage: python testing_telegram_bot.py @@ -24,7 +18,6 @@ Usage: """ import asyncio -import json import logging import os import signal @@ -35,8 +28,6 @@ from datetime import datetime, timezone from zoneinfo import ZoneInfo from typing import Dict, List, Optional, Set -import websockets - from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.constants import ParseMode from telegram.ext import ( @@ -74,11 +65,6 @@ CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022") SCAN_PAIRS = [XAUUSD, EURUSD] -_WS_TO_PAIR = { - "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", @@ -315,14 +301,13 @@ def _format_stats_dashboard() -> str: display = PAIR_INFO.get(symbol, {}).get("display", symbol) pair_data = _signal_stats.get(symbol, {}) - ws_price = _last_prices.get(symbol) latest = None for tf in reversed(_DASH_TFS): if tf.value in pair_data: latest = pair_data[tf.value] break - price = ws_price or (latest.current_price if latest else 0) + price = latest.current_price if latest else 0 price_str = f"`{_fmt_price(price)}`" regime = latest.market_regime.replace("_", " ").title() if latest else "" @@ -512,17 +497,6 @@ async def _scan_timeframe( _level_cache: Dict[str, Dict[str, float]] = {} _level_cache_ts: Dict[str, float] = {} _LEVEL_CACHE_TTL = 3600 -_last_prices: Dict[str, float] = {} -_last_price_ts: 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 - -_breach_tasks: Set[asyncio.Task] = set() - def _update_level_cache(reports: List[SignalReport]) -> None: now = time.time() @@ -546,102 +520,14 @@ def _check_level_breach(symbol: str, price: float) -> Optional[str]: if cache_age > _LEVEL_CACHE_TTL: 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): + diff_pct = abs(price - level) / level if level != 0 else 0 + if diff_pct < 0.001: return name return None -def _cleanup_cooldowns() -> None: - now = time.time() - expired = [k for k, v in _scan_cooldowns.items() if now - v > _WS_COOLDOWN * 2] - for k in expired: - del _scan_cooldowns[k] - - -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} (alert disabled, cache updated)") - - try: - report = await run_signal(pair_symbol, Timeframe.M15) - _update_level_cache([report]) - _update_signal_stats([report], Timeframe.M15) - except Exception: - logger.exception(f"Breach scan failed for {display}") - - -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 - - try: - msg = json.loads(raw) - except json.JSONDecodeError: - logger.warning(f"WebSocket: malformed JSON, skipping") - continue - - if msg.get("event") != "price": - continue - - ws_symbol = msg.get("symbol") - pair_symbol = _WS_TO_PAIR.get(ws_symbol) - if not pair_symbol: - continue - - try: - price = float(msg["price"]) - except (KeyError, ValueError, TypeError) as e: - logger.warning(f"WebSocket: bad price data for {ws_symbol}: {e}") - continue - - breached = _check_level_breach(pair_symbol, price) - _last_prices[pair_symbol] = price - _last_price_ts[pair_symbol] = time.time() - - if breached: - task = asyncio.create_task( - _handle_breach(bot, ws_symbol, pair_symbol, price, breached) - ) - _breach_tasks.add(task) - task.add_done_callback(_breach_tasks.discard) - - _cleanup_cooldowns() - - 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: @@ -669,11 +555,8 @@ _AUTOSCAN_LOOPS = [ ] -_ws_task: Optional[asyncio.Task] = None - - def _start_scheduler(bot) -> None: - global _ws_task, _dashboard_task + global _dashboard_task alive = [t for t in _autoscan_tasks if not t.done()] if not alive: _autoscan_tasks.clear() @@ -683,31 +566,21 @@ def _start_scheduler(bot) -> None: ) 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") - if _dashboard_task is None or _dashboard_task.done(): _dashboard_task = asyncio.create_task(_dashboard_loop(bot)) logger.info(f"Dashboard {_DASHBOARD_REFRESH_INTERVAL}s refresh loop started") def _stop_scheduler() -> None: - global _ws_task, _dashboard_task + global _dashboard_task for t in _autoscan_tasks: t.cancel() _autoscan_tasks.clear() - if _ws_task and not _ws_task.done(): - _ws_task.cancel() - _ws_task = None if _dashboard_task and not _dashboard_task.done(): _dashboard_task.cancel() _dashboard_task = None _dashboard_msg_ids.clear() - for t in list(_breach_tasks): - t.cancel() - _breach_tasks.clear() - logger.info("Scheduler, WebSocket, and dashboard loop stopped") + logger.info("Scheduler and dashboard loop stopped") # ── Handlers ────────────────────────────────────────────────────────── @@ -720,7 +593,7 @@ async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: "Signals include a full trade setup with entry, SL and TP.\n\n" f"Auto-scan: *{scanning}* — /autoscan to toggle\n" "/scan — Quick scan buttons\n" - "/status — Bot & WebSocket status\n" + "/status — Bot status\n" "/help — All commands" ) await update.message.reply_text( @@ -736,13 +609,12 @@ async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: "/start — Welcome & scan buttons\n" "/scan — Quick scan buttons (XAU/USD + EUR/USD)\n" "/autoscan — Toggle auto-scan on/off\n" - "/status — Bot status, WS connection, live prices\n" + "/status — Bot status\n" "/cache — Show data cache age per pair/TF\n" "/clearcache — Wipe data cache, force fresh fetch\n" "/help — This message\n\n" "*Auto-scan* fires at candle close: M5/M15/H1/H4.\n" - f"Live dashboard auto-refreshes every {_DASHBOARD_REFRESH_INTERVAL}s.\n" - "*WebSocket* monitors live prices and triggers breach scans." + f"Live dashboard auto-refreshes every {_DASHBOARD_REFRESH_INTERVAL}s." ) await update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN) @@ -757,42 +629,11 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None else: lines.append("⏹ Auto-scan: *stopped*") - if _ws_task and not _ws_task.done(): - lines.append("📡 WebSocket: *connected*") - else: - lines.append("❌ WebSocket: *disconnected*") - - if _last_prices: - lines.append("\n*Live Prices (WebSocket):*") - for ws_sym, pair in _WS_TO_PAIR.items(): - price = _last_prices.get(pair) - display = PAIR_INFO.get(pair, {}).get("display", ws_sym) - age = time.time() - _last_price_ts.get(pair, 0) - if price is None: - lines.append(f" {display}: waiting...") - else: - lines.append(f" {display}: `{_fmt_price(price)}` ({age:.0f}s ago)") - else: - lines.append("\n_No live prices yet (WS not streaming)_") - cached_pairs = len(_level_cache) cached_levels = sum(len(v) for v in _level_cache.values()) if cached_levels: oldest_age = time.time() - min(_level_cache_ts.values()) if _level_cache_ts else 0 lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs (oldest {oldest_age:.0f}s)") - else: - lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan") - - active_cooldowns = sum( - 1 for v in _scan_cooldowns.values() - if time.time() - v < _WS_COOLDOWN - ) - if active_cooldowns: - lines.append(f"*Cooldowns:* {active_cooldowns} pair(s) in cooldown") - - active_breach = sum(1 for t in _breach_tasks if not t.done()) - if active_breach: - lines.append(f"*Breach tasks:* {active_breach} running") await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)