Harden Telegram bot, add EUR/GBP strategies, parallelize scans

Telegram reliability: rate-limited sends (1 msg/s), dashboard lock,
WebSocket JSON/price error handling, markdown escaping, level cache
TTL, cooldown cleanup, breach task tracking, graceful shutdown.

Telegram speed: parallelize 3-pair scans with asyncio.gather(),
increase thread pool 2→4, parallelize boot-up initial scan.

Strategy: add EUR/USD and GBP/USD pullback-buy strategies with
per-pair configs, optimize XAU TP to 0.9×ATR, block hours 8/9
for XAU and hour 17 for GBP.

Backtest: add --max-hours flag to evaluate.py for time-exit support.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
harwinder
2026-05-13 21:07:49 -04:00
parent 1f3f69c21d
commit 235241f95a
4 changed files with 494 additions and 66 deletions
+7 -2
View File
@@ -269,6 +269,7 @@ def evaluate_outcomes(
trades: List[Trade], trades: List[Trade],
m5_data: Dict[str, pd.DataFrame], m5_data: Dict[str, pd.DataFrame],
max_bars: int = 500, max_bars: int = 500,
max_hours: int = 0,
) -> List[Trade]: ) -> List[Trade]:
""" """
For each trade, scan forward in M5 data to check if TP1 or SL hit first. For each trade, scan forward in M5 data to check if TP1 or SL hit first.
@@ -305,7 +306,9 @@ def evaluate_outcomes(
continue continue
start = int(start_indices[0]) start = int(start_indices[0])
end = min(start + max_bars, len(highs_gpu)) time_limit = max_hours * 12 if max_hours > 0 else max_bars
effective_max = min(time_limit, max_bars)
end = min(start + effective_max, len(highs_gpu))
if start >= end: if start >= end:
continue continue
@@ -532,6 +535,8 @@ def main():
help="Max M5 bars to scan for SL/TP hit (default: 1000 = ~83h)") help="Max M5 bars to scan for SL/TP hit (default: 1000 = ~83h)")
parser.add_argument("--pair", default=None, parser.add_argument("--pair", default=None,
help="Single pair to evaluate (EURUSD, USDJPY, GBPUSD, XAUUSD)") help="Single pair to evaluate (EURUSD, USDJPY, GBPUSD, XAUUSD)")
parser.add_argument("--max-hours", type=int, default=0,
help="Max hours to hold a trade before time-exit (0 = disabled)")
args = parser.parse_args() args = parser.parse_args()
eval_tfs = EVAL_TIMEFRAMES eval_tfs = EVAL_TIMEFRAMES
@@ -599,7 +604,7 @@ def main():
m5_data = {k: v for k, v in m5_data.items() if v is not None} m5_data = {k: v for k, v in m5_data.items() if v is not None}
t2 = time.time() t2 = time.time()
all_trades = evaluate_outcomes(all_trades, m5_data, max_bars=args.max_outcome_bars) all_trades = evaluate_outcomes(all_trades, m5_data, max_bars=args.max_outcome_bars, max_hours=args.max_hours)
print(f" Outcomes evaluated in {time.time() - t2:.3f}s") print(f" Outcomes evaluated in {time.time() - t2:.3f}s")
# ── 4. Report ──────────────────────────────────────────────── # ── 4. Report ────────────────────────────────────────────────
+44 -5
View File
@@ -131,15 +131,52 @@ class NewsConfig:
@dataclass @dataclass
class XauStrategyConfig: class XauStrategyConfig:
"""XAU/USD oversold-bounce strategy (BUY only). """XAU/USD oversold-bounce strategy (BUY only).
Backtested: 79.7% WR, PF 2.06 over 10 months (192 trades). 1-year optimized: 89% WR, PF 4.58, +5041p (118 trades).
Entry: M15 RSI<30 + price above EMA200 on M15, H1, H4. Entry: M15 RSI<38 + price above EMA200 on M15, H1, H4.
Exit: TP=0.7×ATR, SL=2.0×ATR, max hold 8 bars (2h). Exit: TP=0.9×ATR, SL=2.0×ATR.
""" """
enabled: bool = True enabled: bool = True
rsi_entry_threshold: float = 38.0 # Wilder RSI scale; equivalent to standard RSI ~30 rsi_entry_threshold: float = 38.0 # Wilder RSI scale; equivalent to standard RSI ~30
tp_atr_multiplier: float = 0.7 tp_atr_multiplier: float = 0.9
sl_atr_multiplier: float = 2.0 sl_atr_multiplier: float = 2.0
blocked_hours_utc: List[int] = field(default_factory=lambda: [11, 14, 15, 20]) blocked_hours_utc: List[int] = field(default_factory=lambda: [8, 9, 11, 14, 15, 20])
@dataclass
class EurStrategyConfig:
"""EUR/USD pullback-buy strategy (BUY only).
Fixes: no entries at BB resistance, no overbought RSI, no chasing.
Entry: M15 RSI 30–55 (pullback zone) + BB position < 0.75
+ price > EMA200 on M15 & H1 + near EMA21 (not chasing).
Exit: TP=0.7×ATR, SL=1.5×ATR.
"""
enabled: bool = True
rsi_lower: float = 30.0
rsi_upper: float = 55.0
bb_max_position: float = 0.75
chase_max_atr_from_ema21: float = 1.5
tp_atr_multiplier: float = 0.7
sl_atr_multiplier: float = 1.0
blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5, 6, 16, 17, 18, 19, 21, 22, 23])
@dataclass
class GbpStrategyConfig:
"""GBP/USD pullback-buy strategy (BUY only).
1-year optimized: hour 17 blocked.
Entry: M15 RSI 30–55 + BB < 0.75 + price > EMA200 on M15 & H1
+ within 1.5×ATR of EMA21.
Exit: TP=0.7×ATR, SL=1.0×ATR.
"""
enabled: bool = True
rsi_lower: float = 30.0
rsi_upper: float = 55.0
bb_min_position: float = 0.0
bb_max_position: float = 0.75
chase_max_atr_from_ema21: float = 1.5
tp_atr_multiplier: float = 0.7
sl_atr_multiplier: float = 1.0
blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5, 6, 14, 15, 16, 17, 19, 21, 22, 23])
@dataclass @dataclass
@@ -166,6 +203,8 @@ class PlatformConfig:
news: NewsConfig = field(default_factory=NewsConfig) news: NewsConfig = field(default_factory=NewsConfig)
backtest: FilterConfig = field(default_factory=FilterConfig) backtest: FilterConfig = field(default_factory=FilterConfig)
xau_strategy: XauStrategyConfig = field(default_factory=XauStrategyConfig) xau_strategy: XauStrategyConfig = field(default_factory=XauStrategyConfig)
eur_strategy: EurStrategyConfig = field(default_factory=EurStrategyConfig)
gbp_strategy: GbpStrategyConfig = field(default_factory=GbpStrategyConfig)
symbol: str = USDJPY symbol: str = USDJPY
display_name: str = DISPLAY display_name: str = DISPLAY
+278 -1
View File
@@ -144,7 +144,17 @@ class SignalAggregator:
and self.config.xau_strategy.enabled and self.config.xau_strategy.enabled
and any("XAU STRATEGY" in r for r in report.bullish_reasons) and any("XAU STRATEGY" in r for r in report.bullish_reasons)
) )
if not xau_active: eur_active = (
report.display_name == "EUR/USD"
and self.config.eur_strategy.enabled
and any("EUR STRATEGY" in r for r in report.bullish_reasons)
)
gbp_active = (
report.display_name == "GBP/USD"
and self.config.gbp_strategy.enabled
and any("GBP STRATEGY" in r for r in report.bullish_reasons)
)
if not xau_active and not eur_active and not gbp_active:
self._apply_session_filter_live(report) self._apply_session_filter_live(report)
return report return report
@@ -354,6 +364,40 @@ class SignalAggregator:
) )
report.trade_setup = self._generate_xau_trade_setup(report, dfs) report.trade_setup = self._generate_xau_trade_setup(report, dfs)
# ── 6d. EUR/USD pullback-buy strategy ─────────────────────
if report.display_name == "EUR/USD" and self.config.eur_strategy.enabled:
if self._check_eur_pullback_buy(report, dfs):
report.final_signal = SignalStrength.BUY
report.final_score = 40.0
report.final_confidence = 75.0
report.signal_quality = "A"
report.action_recommendation = (
"BUY — EUR pullback entry "
"(RSI pullback + not at resistance + M15/H1 above EMA200)"
)
report.bullish_reasons.insert(
0,
"EUR STRATEGY: M15 RSI pullback + BB mid-band + dual-TF EMA200 confirmation",
)
report.trade_setup = self._generate_eur_trade_setup(report, dfs)
# ── 6e. GBP/USD pullback-buy strategy ─────────────────────
if report.display_name == "GBP/USD" and self.config.gbp_strategy.enabled:
if self._check_gbp_pullback_buy(report, dfs):
report.final_signal = SignalStrength.BUY
report.final_score = 40.0
report.final_confidence = 75.0
report.signal_quality = "A"
report.action_recommendation = (
"BUY — GBP pullback entry "
"(RSI pullback + not at resistance + M15/H1 above EMA200)"
)
report.bullish_reasons.insert(
0,
"GBP STRATEGY: M15 RSI pullback + BB mid-band + dual-TF EMA200 confirmation",
)
report.trade_setup = self._generate_gbp_trade_setup(report, dfs)
# ── 7. Trade setup for actionable signals ────────────────── # ── 7. Trade setup for actionable signals ──────────────────
if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"): if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"):
is_sell_direction = ( is_sell_direction = (
@@ -486,6 +530,239 @@ class SignalAggregator:
pip_value=pip_value, pip_value=pip_value,
) )
def _check_eur_pullback_buy(
self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame]
) -> bool:
"""Return True when EUR/USD pullback-buy conditions are met.
Conditions (all must be true):
1. Current bar hour not in blocked list
2. M15 RSI(14) in 30–55 (pullback, not overbought)
3. M15 BB position < 0.75 (not at resistance)
4. M15 price > EMA200 (uptrend)
5. H1 price > EMA200 (higher-TF confirmation)
6. Price within 1.5×ATR of EMA21 (not chasing)
"""
cfg = self.config.eur_strategy
try:
bar_hour = pd.Timestamp(report.timestamp).hour
if bar_hour in cfg.blocked_hours_utc:
return False
except Exception:
return False
m15_analysis = None
h1_analysis = None
if report.primary_analysis:
tf = report.primary_analysis.timeframe
if tf == Timeframe.M15.value:
m15_analysis = report.primary_analysis
elif tf == Timeframe.H1.value:
h1_analysis = report.primary_analysis
for ca in report.confirmation_analyses:
if ca.timeframe == Timeframe.M15.value:
m15_analysis = ca
elif ca.timeframe == Timeframe.H1.value:
h1_analysis = ca
if not m15_analysis or not h1_analysis:
return False
m15_rsi = m15_analysis.indicators.get("rsi")
if not m15_rsi or m15_rsi.value < cfg.rsi_lower or m15_rsi.value > cfg.rsi_upper:
return False
m15_bb = m15_analysis.indicators.get("bollinger")
if m15_bb and m15_bb.value > cfg.bb_max_position:
return False
m15_ema200 = m15_analysis.key_levels.get("ema_trend")
m15_df = dfs.get(Timeframe.M15)
if m15_df is None or m15_ema200 is None:
return False
m15_price = float(m15_df["Close"].iloc[-1])
if m15_price <= m15_ema200:
return False
h1_ema200 = h1_analysis.key_levels.get("ema_trend")
h1_df = dfs.get(Timeframe.H1)
if h1_ema200 is None:
return False
h1_price = float(h1_df["Close"].iloc[-1]) if h1_df is not None else report.current_price
if h1_price <= h1_ema200:
return False
m15_atr = m15_analysis.key_levels.get("atr", 0)
m15_ema21 = m15_analysis.key_levels.get("ema_medium")
if m15_atr > 0 and m15_ema21:
dist = abs(m15_price - m15_ema21) / m15_atr
if dist > cfg.chase_max_atr_from_ema21:
return False
return True
def _generate_eur_trade_setup(
self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame]
) -> Optional[TradeSetup]:
"""Generate EUR/USD-specific trade setup with tight TP / wide SL."""
cfg = self.config.eur_strategy
m15_analysis = None
if report.primary_analysis and report.primary_analysis.timeframe == Timeframe.M15.value:
m15_analysis = report.primary_analysis
else:
for ca in report.confirmation_analyses:
if ca.timeframe == Timeframe.M15.value:
m15_analysis = ca
break
if not m15_analysis:
return None
atr = m15_analysis.key_levels.get("atr", 0)
if atr <= 0:
return None
price = report.current_price
pip_value = 0.0001
tp1 = price + cfg.tp_atr_multiplier * atr
sl = price - cfg.sl_atr_multiplier * atr
tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr
sl_distance = price - sl
tp_distance = tp1 - price
rr = tp_distance / sl_distance if sl_distance > 0 else 0
return TradeSetup(
pair=report.symbol,
display_name=report.display_name,
direction="BUY",
entry_price=price,
stop_loss=round(sl, 5),
take_profit_1=round(tp1, 5),
take_profit_2=round(tp2 , 5),
risk_reward=round(rr, 2),
risk_pips=round(sl_distance / pip_value, 1),
reward_pips=round(tp_distance / pip_value, 1),
pip_value=pip_value,
)
def _check_gbp_pullback_buy(
self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame]
) -> bool:
"""GBP/USD pullback-buy — same logic as EUR, different config."""
cfg = self.config.gbp_strategy
try:
bar_hour = pd.Timestamp(report.timestamp).hour
if bar_hour in cfg.blocked_hours_utc:
return False
except Exception:
return False
m15_analysis = None
h1_analysis = None
if report.primary_analysis:
tf = report.primary_analysis.timeframe
if tf == Timeframe.M15.value:
m15_analysis = report.primary_analysis
elif tf == Timeframe.H1.value:
h1_analysis = report.primary_analysis
for ca in report.confirmation_analyses:
if ca.timeframe == Timeframe.M15.value:
m15_analysis = ca
elif ca.timeframe == Timeframe.H1.value:
h1_analysis = ca
if not m15_analysis or not h1_analysis:
return False
m15_rsi = m15_analysis.indicators.get("rsi")
if not m15_rsi or m15_rsi.value < cfg.rsi_lower or m15_rsi.value > cfg.rsi_upper:
return False
m15_bb = m15_analysis.indicators.get("bollinger")
if m15_bb and (m15_bb.value < cfg.bb_min_position or m15_bb.value > cfg.bb_max_position):
return False
m15_ema200 = m15_analysis.key_levels.get("ema_trend")
m15_df = dfs.get(Timeframe.M15)
if m15_df is None or m15_ema200 is None:
return False
m15_price = float(m15_df["Close"].iloc[-1])
if m15_price <= m15_ema200:
return False
h1_ema200 = h1_analysis.key_levels.get("ema_trend")
h1_df = dfs.get(Timeframe.H1)
if h1_ema200 is None:
return False
h1_price = float(h1_df["Close"].iloc[-1]) if h1_df is not None else report.current_price
if h1_price <= h1_ema200:
return False
m15_atr = m15_analysis.key_levels.get("atr", 0)
m15_ema21 = m15_analysis.key_levels.get("ema_medium")
if m15_atr > 0 and m15_ema21:
dist = abs(m15_price - m15_ema21) / m15_atr
if dist > cfg.chase_max_atr_from_ema21:
return False
return True
def _generate_gbp_trade_setup(
self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame]
) -> Optional[TradeSetup]:
"""Generate GBP/USD trade setup with tight TP / balanced SL."""
cfg = self.config.gbp_strategy
m15_analysis = None
if report.primary_analysis and report.primary_analysis.timeframe == Timeframe.M15.value:
m15_analysis = report.primary_analysis
else:
for ca in report.confirmation_analyses:
if ca.timeframe == Timeframe.M15.value:
m15_analysis = ca
break
if not m15_analysis:
return None
atr = m15_analysis.key_levels.get("atr", 0)
if atr <= 0:
return None
price = report.current_price
pip_value = 0.0001
tp1 = price + cfg.tp_atr_multiplier * atr
sl = price - cfg.sl_atr_multiplier * atr
tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr
sl_distance = price - sl
tp_distance = tp1 - price
rr = tp_distance / sl_distance if sl_distance > 0 else 0
return TradeSetup(
pair=report.symbol,
display_name=report.display_name,
direction="BUY",
entry_price=price,
stop_loss=round(sl, 5),
take_profit_1=round(tp1, 5),
take_profit_2=round(tp2, 5),
risk_reward=round(rr, 2),
risk_pips=round(sl_distance / pip_value, 1),
reward_pips=round(tp_distance / pip_value, 1),
pip_value=pip_value,
)
def _check_mtf_agreement(self, report: SignalReport) -> bool: def _check_mtf_agreement(self, report: SignalReport) -> bool:
signals = list(report.mtf_signals.values()) signals = list(report.mtf_signals.values())
if len(signals) < 2: if len(signals) < 2:
+165 -58
View File
@@ -35,11 +35,12 @@ import asyncio
import json import json
import logging import logging
import os import os
import signal
import sys import sys
import time import time
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Dict, List, Optional from typing import Dict, List, Optional, Set
import websockets import websockets
@@ -104,7 +105,7 @@ SCAN_BUTTONS = [
{"cb": "scan_eurusd_m15", "symbol": EURUSD, "tf": Timeframe.M15, "label": "EUR/USD 15M"}, {"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_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_eurusd_h4", "symbol": EURUSD, "tf": Timeframe.H4, "label": "EUR/USD 4H"},
{"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"}, {"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_gbpusd_h1", "symbol": GBPUSD, "tf": Timeframe.H1, "label": "GBP/USD 1H"},
{"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"}, {"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"},
{"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"}, {"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"},
@@ -128,7 +129,60 @@ AUTOSCAN_CONFIRMATIONS = {
Timeframe.H4: [Timeframe.H1, Timeframe.D1], Timeframe.H4: [Timeframe.H1, Timeframe.D1],
} }
_executor = ThreadPoolExecutor(max_workers=2) _executor = ThreadPoolExecutor(max_workers=4)
# ── Telegram rate limiter (1 msg/sec per chat) ────────────────────────
_send_lock = asyncio.Lock()
_last_send_per_chat: Dict[int, float] = {}
_TELEGRAM_MIN_INTERVAL = 1.0
_TELEGRAM_SEND_TIMEOUT = 30
async def _rate_limited_send(bot, chat_id: int, text: str, **kwargs) -> Optional[object]:
async with _send_lock:
now = time.time()
last = _last_send_per_chat.get(chat_id, 0)
wait = _TELEGRAM_MIN_INTERVAL - (now - last)
if wait > 0:
await asyncio.sleep(wait)
try:
result = await asyncio.wait_for(
bot.send_message(chat_id, text, **kwargs),
timeout=_TELEGRAM_SEND_TIMEOUT,
)
_last_send_per_chat[chat_id] = time.time()
return result
except asyncio.TimeoutError:
logger.error(f"Telegram send_message timed out for chat {chat_id}")
return None
except Exception:
logger.exception(f"Telegram send_message failed for chat {chat_id}")
return None
async def _rate_limited_edit(bot, text: str, chat_id: int, message_id: int, **kwargs) -> bool:
try:
await asyncio.wait_for(
bot.edit_message_text(text, chat_id=chat_id, message_id=message_id, **kwargs),
timeout=_TELEGRAM_SEND_TIMEOUT,
)
return True
except asyncio.TimeoutError:
logger.error(f"Telegram edit_message timed out for chat {chat_id}")
return False
except Exception as exc:
if "message is not modified" not in str(exc).lower():
logger.warning(f"Telegram edit_message failed for chat {chat_id}: {exc}")
return False
# ── Markdown escaping ─────────────────────────────────────────────────
def _escape_md(text: str) -> str:
for ch in ("*", "_", "`", "[", "]"):
text = text.replace(ch, "\\" + ch)
return text
# ── Shared data provider (single cache across all scans) ───────────── # ── Shared data provider (single cache across all scans) ─────────────
@@ -183,7 +237,7 @@ def _mtf_line(report: SignalReport) -> str:
def _indicator_line(name: str, ind) -> str: def _indicator_line(name: str, ind) -> str:
tag = _SIGNAL_TAG.get(ind.signal, "Neutral") tag = _SIGNAL_TAG.get(ind.signal, "Neutral")
short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description
return f"[{tag}] *{ind.name}*: {short}" return f"[{tag}] *{_escape_md(ind.name)}*: {_escape_md(short)}"
def format_signal_message(report: SignalReport) -> str: def format_signal_message(report: SignalReport) -> str:
@@ -196,16 +250,18 @@ def format_signal_message(report: SignalReport) -> str:
lines: list[str] = [] lines: list[str] = []
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
regime_str = _escape_md(report.market_regime.replace("_", " ").title())
vol_str = _escape_md(str(report.volatility_regime))
lines += [ lines += [
f"🔍 *{report.display_name} Signal Analysis* ({report.primary_analysis.timeframe if report.primary_analysis else ''})", f"🔍 *{report.display_name} Signal Analysis* ({report.primary_analysis.timeframe if report.primary_analysis else ''})",
f"🕐 {now}", f"🕐 {now}",
"", "",
f"💰 Price: *{report.current_price:.5f}*" if report.current_price < 10 else f"💰 Price: *{report.current_price:.3f}*", f"💰 Price: *{report.current_price:.5f}*" if report.current_price < 10 else f"💰 Price: *{report.current_price:.3f}*",
f"📊 Regime: {report.market_regime.replace('_', ' ').title()} | Vol: {report.volatility_regime}", f"📊 Regime: {regime_str} | Vol: {vol_str}",
"", "",
"━━━━━━━━━━━━━━━━━━━━", "━━━━━━━━━━━━━━━━━━━━",
f"{emoji} *{label}* {qe}{report.signal_quality} | Score: {report.final_score:+.1f} | {report.final_confidence:.0f}% conf", f"{emoji} *{label}* {qe}{report.signal_quality} | Score: {report.final_score:+.1f} | {report.final_confidence:.0f}% conf",
f"_{report.action_recommendation}_", f"_{_escape_md(report.action_recommendation)}_",
f"{mtf_tag}", f"{mtf_tag}",
] ]
@@ -223,7 +279,7 @@ def format_signal_message(report: SignalReport) -> str:
if report.warnings: if report.warnings:
lines.append("") lines.append("")
for w in report.warnings: for w in report.warnings:
lines.append(f"⚠️ {w}") lines.append(f"⚠️ {_escape_md(w)}")
if report.trade_setup and sig not in (SignalStrength.NEUTRAL,): if report.trade_setup and sig not in (SignalStrength.NEUTRAL,):
ts = report.trade_setup ts = report.trade_setup
@@ -366,7 +422,7 @@ def _format_stats_dashboard() -> str:
sig = _SHORT_SIGNAL.get(r.final_signal, "Neutral") sig = _SHORT_SIGNAL.get(r.final_signal, "Neutral")
tf_parts.append(f"{label}: {emoji}{sig}") tf_parts.append(f"{label}: {emoji}{sig}")
lines.append(f"*{display}* {price_str} _{regime}_") lines.append(f"*{display}* {price_str} _{_escape_md(regime)}_")
lines.append(" | ".join(tf_parts)) lines.append(" | ".join(tf_parts))
if latest and latest.trade_setup and latest.final_signal not in (SignalStrength.NEUTRAL,): if latest and latest.trade_setup and latest.final_signal not in (SignalStrength.NEUTRAL,):
@@ -390,40 +446,35 @@ _dashboard_msg_ids: Dict[int, int] = {}
_dashboard_task: Optional[asyncio.Task] = None _dashboard_task: Optional[asyncio.Task] = None
_last_dashboard_refresh: float = 0 _last_dashboard_refresh: float = 0
_DASHBOARD_THROTTLE = 5 _DASHBOARD_THROTTLE = 5
_dashboard_lock = asyncio.Lock()
async def _send_dashboard(bot, chat_id: int) -> None: async def _send_dashboard(bot, chat_id: int) -> None:
text = _format_stats_dashboard() text = _format_stats_dashboard()
try: msg = await _rate_limited_send(bot, chat_id, text, parse_mode=ParseMode.MARKDOWN)
msg = await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN) if msg:
_dashboard_msg_ids[chat_id] = msg.message_id _dashboard_msg_ids[chat_id] = msg.message_id
except Exception:
logger.exception(f"Failed to send dashboard to {chat_id}")
async def _refresh_dashboard(bot, force: bool = False) -> None: async def _refresh_dashboard(bot, force: bool = False) -> None:
global _last_dashboard_refresh global _last_dashboard_refresh
now = time.time() async with _dashboard_lock:
if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE: now = time.time()
return if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE:
_last_dashboard_refresh = now return
_last_dashboard_refresh = now
text = _format_stats_dashboard() text = _format_stats_dashboard()
for chat_id in list(_autoscan_chats): for chat_id in list(_autoscan_chats):
msg_id = _dashboard_msg_ids.get(chat_id) msg_id = _dashboard_msg_ids.get(chat_id)
if not msg_id: if not msg_id:
await _send_dashboard(bot, chat_id) await _send_dashboard(bot, chat_id)
continue continue
try: ok = await _rate_limited_edit(
await bot.edit_message_text( bot, text, chat_id=chat_id, message_id=msg_id,
text, chat_id=chat_id, message_id=msg_id,
parse_mode=ParseMode.MARKDOWN, parse_mode=ParseMode.MARKDOWN,
) )
except Exception as exc: if not ok:
if "message is not modified" in str(exc).lower():
pass
else:
logger.warning(f"Dashboard edit failed, sending new: {exc}")
await _send_dashboard(bot, chat_id) await _send_dashboard(bot, chat_id)
@@ -484,7 +535,7 @@ async def run_signal(
# ── Candle-boundary scheduler ──────────────────────────────────────── # ── Candle-boundary scheduler ────────────────────────────────────────
_autoscan_chats: set[int] = set() _autoscan_chats: Set[int] = set()
_autoscan_tasks: list[asyncio.Task] = [] _autoscan_tasks: list[asyncio.Task] = []
@@ -507,13 +558,14 @@ async def _scan_timeframe(
confirmation_tfs: List[Timeframe], confirmation_tfs: List[Timeframe],
) -> None: ) -> None:
"""Scan all pairs for one timeframe and broadcast results.""" """Scan all pairs for one timeframe and broadcast results."""
tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS]
results = await asyncio.gather(*tasks, return_exceptions=True)
reports: List[SignalReport] = [] reports: List[SignalReport] = []
for symbol in SCAN_PAIRS: for i, result in enumerate(results):
try: if isinstance(result, Exception):
report = await run_signal(symbol, tf, confirmation_tfs) logger.exception(f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}")
reports.append(report) else:
except Exception: reports.append(result)
logger.exception(f"Auto-scan failed for {symbol} {tf.value}")
if not reports: if not reports:
return return
@@ -530,17 +582,17 @@ async def _scan_timeframe(
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")] actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
for chat_id in list(_autoscan_chats): for chat_id in list(_autoscan_chats):
for report in actionable: for report in actionable:
try: detail = format_signal_message(report)
detail = format_signal_message(report) await _rate_limited_send(bot, chat_id, detail, parse_mode=ParseMode.MARKDOWN)
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
except Exception:
logger.exception(f"Failed to send to chat {chat_id}")
# ── WebSocket level-breach monitor ────────────────────────────────── # ── WebSocket level-breach monitor ──────────────────────────────────
_level_cache: Dict[str, Dict[str, float]] = {} _level_cache: Dict[str, Dict[str, float]] = {}
_level_cache_ts: Dict[str, float] = {}
_LEVEL_CACHE_TTL = 3600
_last_prices: Dict[str, float] = {} _last_prices: Dict[str, float] = {}
_last_price_ts: Dict[str, float] = {}
_scan_cooldowns: Dict[str, float] = {} _scan_cooldowns: Dict[str, float] = {}
_cfg = PlatformConfig() _cfg = PlatformConfig()
@@ -548,8 +600,11 @@ _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_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={_WS_API_KEY}"
_WS_COOLDOWN = _cfg.ws_cooldown_seconds _WS_COOLDOWN = _cfg.ws_cooldown_seconds
_breach_tasks: Set[asyncio.Task] = set()
def _update_level_cache(reports: List[SignalReport]) -> None: def _update_level_cache(reports: List[SignalReport]) -> None:
now = time.time()
for r in reports: for r in reports:
if r.primary_analysis and r.primary_analysis.key_levels: if r.primary_analysis and r.primary_analysis.key_levels:
levels = { levels = {
@@ -558,6 +613,7 @@ def _update_level_cache(reports: List[SignalReport]) -> None:
} }
if levels: if levels:
_level_cache[r.symbol] = levels _level_cache[r.symbol] = levels
_level_cache_ts[r.symbol] = now
def _check_level_breach(symbol: str, price: float) -> Optional[str]: def _check_level_breach(symbol: str, price: float) -> Optional[str]:
@@ -565,6 +621,10 @@ def _check_level_breach(symbol: str, price: float) -> Optional[str]:
if not levels: if not levels:
return None return None
cache_age = time.time() - _level_cache_ts.get(symbol, 0)
if cache_age > _LEVEL_CACHE_TTL:
return None
last = _last_prices.get(symbol) last = _last_prices.get(symbol)
if last is None: if last is None:
return None return None
@@ -576,6 +636,13 @@ def _check_level_breach(symbol: str, price: float) -> Optional[str]:
return None 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: async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, level_name: str) -> None:
now = time.time() now = time.time()
if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN: if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN:
@@ -611,7 +678,12 @@ async def _ws_monitor(bot) -> None:
if not _autoscan_chats: if not _autoscan_chats:
break break
msg = json.loads(raw) try:
msg = json.loads(raw)
except json.JSONDecodeError:
logger.warning(f"WebSocket: malformed JSON, skipping")
continue
if msg.get("event") != "price": if msg.get("event") != "price":
continue continue
@@ -620,14 +692,24 @@ async def _ws_monitor(bot) -> None:
if not pair_symbol: if not pair_symbol:
continue continue
price = float(msg["price"]) 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) breached = _check_level_breach(pair_symbol, price)
_last_prices[pair_symbol] = price _last_prices[pair_symbol] = price
_last_price_ts[pair_symbol] = time.time()
if breached: if breached:
asyncio.create_task( task = asyncio.create_task(
_handle_breach(bot, ws_symbol, pair_symbol, price, breached) _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: except asyncio.CancelledError:
break break
@@ -701,6 +783,9 @@ def _stop_scheduler() -> None:
_dashboard_task.cancel() _dashboard_task.cancel()
_dashboard_task = None _dashboard_task = None
_dashboard_msg_ids.clear() _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, WebSocket, and dashboard loop stopped")
@@ -729,7 +814,7 @@ async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"/start — Show scan buttons\n" "/start — Show scan buttons\n"
"/autoscan — Toggle auto-scan on/off\n" "/autoscan — Toggle auto-scan on/off\n"
"/status — Bot status, WS connection, live prices\n" "/status — Bot status, WS connection, live prices\n"
"/cache — Show when each pair/TF was last fetched\n" "/cache — Show data cache age per pair/TF\n"
"/clearcache — Wipe data cache, force fresh fetch\n" "/clearcache — Wipe data cache, force fresh fetch\n"
"/help — This message\n\n" "/help — This message\n\n"
"*Auto-scan* fires at candle close: M5/M15/H1/H4.\n" "*Auto-scan* fires at candle close: M5/M15/H1/H4.\n"
@@ -759,19 +844,21 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
for ws_sym, pair in _WS_TO_PAIR.items(): for ws_sym, pair in _WS_TO_PAIR.items():
price = _last_prices.get(pair) price = _last_prices.get(pair)
display = PAIR_INFO.get(pair, {}).get("display", ws_sym) display = PAIR_INFO.get(pair, {}).get("display", ws_sym)
age = time.time() - _last_price_ts.get(pair, 0)
if price is None: if price is None:
lines.append(f" {display}: waiting...") lines.append(f" {display}: waiting...")
elif price < 10: elif price < 10:
lines.append(f" {display}: `{price:.5f}`") lines.append(f" {display}: `{price:.5f}` ({age:.0f}s ago)")
else: else:
lines.append(f" {display}: `{price:.3f}`") lines.append(f" {display}: `{price:.3f}` ({age:.0f}s ago)")
else: else:
lines.append("\n_No live prices yet (WS not streaming)_") lines.append("\n_No live prices yet (WS not streaming)_")
cached_pairs = len(_level_cache) cached_pairs = len(_level_cache)
cached_levels = sum(len(v) for v in _level_cache.values()) cached_levels = sum(len(v) for v in _level_cache.values())
if cached_levels: if cached_levels:
lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs") 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: else:
lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan") lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan")
@@ -782,6 +869,10 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
if active_cooldowns: if active_cooldowns:
lines.append(f"*Cooldowns:* {active_cooldowns} pair(s) in cooldown") 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) await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
@@ -906,13 +997,19 @@ async def _post_init(app: Application) -> None:
logger.info("Running initial scan to populate level cache and signal stats...") logger.info("Running initial scan to populate level cache and signal stats...")
try: try:
init_reports = [] tasks = [
for symbol in SCAN_PAIRS: asyncio.wait_for(run_signal(sym, Timeframe.M15), timeout=30)
report = await run_signal(symbol, Timeframe.M15) for sym in SCAN_PAIRS
init_reports.append(report) ]
_update_level_cache(init_reports) results = await asyncio.gather(*tasks, return_exceptions=True)
_update_signal_stats(init_reports, Timeframe.M15) init_reports = [r for r in results if isinstance(r, SignalReport)]
logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels") for i, r in enumerate(results):
if isinstance(r, Exception):
logger.error(f"Initial scan failed for {SCAN_PAIRS[i]}: {r}")
if init_reports:
_update_level_cache(init_reports)
_update_signal_stats(init_reports, Timeframe.M15)
logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels")
except Exception: except Exception:
logger.exception("Initial scan failed — WS will wait for first candle scan") logger.exception("Initial scan failed — WS will wait for first candle scan")
@@ -928,8 +1025,8 @@ async def _post_init(app: Application) -> None:
) )
try: try:
await app.bot.send_message( await _rate_limited_send(
chat_id, app.bot, chat_id,
f"🤖 *ForexBot2 started*\n\n" f"🤖 *ForexBot2 started*\n\n"
f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n" f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n"
f"📊 Live dashboard refreshes every 60s.\n" f"📊 Live dashboard refreshes every 60s.\n"
@@ -944,6 +1041,15 @@ async def _post_init(app: Application) -> None:
logger.exception("Failed to send boot message") logger.exception("Failed to send boot message")
# ── Graceful shutdown ─────────────────────────────────────────────────
async def _post_shutdown(app: Application) -> None:
logger.info("Shutting down ForexBot2...")
_stop_scheduler()
_executor.shutdown(wait=False)
logger.info("Shutdown complete")
# ── Entry point ─────────────────────────────────────────────────────── # ── Entry point ───────────────────────────────────────────────────────
def main() -> None: def main() -> None:
@@ -956,6 +1062,7 @@ def main() -> None:
Application.builder() Application.builder()
.token(TOKEN) .token(TOKEN)
.post_init(_post_init) .post_init(_post_init)
.post_shutdown(_post_shutdown)
.build() .build()
) )