Add WebSocket dual-connection test script

Probes whether Twelve Data allows two simultaneous WS connections
for one symbol (SDK-based, 30s run, prints per-connection verdict).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

CLAUDE.md

@@ -39,10 +39,16 @@ telegram_bot.py  main.py
 **Signal Pipeline:** OHLCV → indicators → regime classification → regime-weighted composite score → MTF agreement check → news/event adjustment → diminishing penalty cap → final signal (STRONG_BUY/BUY/NEUTRAL/SELL/STRONG_SELL) → quality grade (A+/A/B/C/D/F) → trade setup if grade C+.
 
 **Regime-Adaptive Weights:** Market classified as `trending_up`, `trending_down`, `ranging`, or `transitioning` via ADX + EMA + ADX slope. Gold-specific weights per regime:
-- **Trending:** MACD (2.0), EMA (2.0), Divergence (2.0), Keltner (1.5), ROC (1.5)
+- **Trending:** MACD (2.0), EMA (2.0), Keltner (1.5), ROC (1.5), Divergence (1.0)
 - **Ranging:** Divergence (2.5), Keltner (2.0), RSI (1.5), BB (1.5), MACD (1.5)
 - **Transitioning:** Keltner (2.5), MACD (2.5), ROC (2.0), Divergence (2.0), EMA (1.5)
 
+ROC is directionally redundant with MACD/Keltner/EMA (0.8–0.94 correlation) but its weight is load-bearing: composite thresholds are calibrated with it — removing it backtested 13R worse. Don't zero it without retuning thresholds.
+
+**RSI (regime-aware, default):** In trends RSI buys pullback zones instead of fading strength (classic mode backtested worse for gold). In `transitioning` only extremes (≥70/≤30) signal.
+
+**Divergence:** 40-bar lookback; price and indicator swings must align within ±2 bars to count as one divergence observation.
+
 **Transitioning Regime:** ADX < 25 but ADX slope > 3.0 = trend emerging. Catches trend birth 5-10 bars before ADX crosses 25. Boosts quality grade for early entries.
 
 **Diminishing Penalty Cap:** Penalties sorted by magnitude, applied with exponential decay (1st at 100%, 2nd at 50%, 3rd at 25%). Hard cap at 30% total. Replaces old additive stacking that could gut 30-50% confidence.
@@ -99,6 +105,10 @@ python backtest/gpu_backtest.py --pair XAUUSD --days 365
 
 The `backtest/` directory contains `gpu_backtest.py` — a GPU-accelerated backtester (CuPy with NumPy fallback) that replays historical OHLCV through the real SignalAggregator. Pre-computes all indicator series including ROC, Keltner, ADX slope for 100x+ speedup. Uses parquet caching for data persistence.
 
+Fill realism: charges a round-trip spread per trade (`--spread`, default $0.30 for XAU), fills gap-through SL/TP at the bar open, and resolves trades against the RAW price path (the parquet cache stores raw bars; validation/cleaning runs on load — signals see cleaned data, trade resolution sees real spikes). Entry is at the next bar's open after the signal bar. Weekend/session filters evaluate bar-close (decision) time to match the live bot's wall clock.
+
+GPU note: CuPy needs the NVIDIA runtime pip wheels (`nvidia-cuda-runtime-cu12`, `nvidia-cuda-nvrtc-cu12`, `nvidia-curand-cu12`, `nvidia-nvjitlink-cu12`) or it silently falls back to NumPy — check the "Backend:" line in the run log. The bar-replay loop itself is single-threaded Python either way (~340 bars/s); GPU only accelerates indicator precompute and SL/TP scanning.
+
 ## Code Conventions
 
 - Python 3.12, type hints on public methods

backtest/gpu_backtest.py

@@ -780,11 +780,26 @@ class GPUTechnicalEngine(TechnicalEngine):
             signal = SignalStrength.SELL if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL
             direction = "bearish"
 
-        desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}"
+        adx_slope = float(cache.adx_slope[pos]) if not np.isnan(cache.adx_slope[pos]) else 0.0
+
+        slope_desc = ""
+        emerging_threshold = c.adx_slope_emerging_threshold
+        if adx_val < c.adx_strong_trend and adx_slope > emerging_threshold:
+            slope_desc = " | TREND EMERGING (ADX rising fast)"
+            conf = min(conf + 15, 85)
+        elif adx_val >= c.adx_strong_trend and adx_slope > 0:
+            slope_desc = f" | Trend strengthening (slope +{adx_slope:.1f})"
+            conf = min(conf + 5, 85)
+        elif adx_val >= c.adx_strong_trend and adx_slope < -emerging_threshold:
+            slope_desc = f" | Trend weakening (slope {adx_slope:.1f})"
+            conf = max(conf - 10, 30)
+
+        desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}{slope_desc}"
         result.indicators["adx"] = IndicatorSignal(
             name="ADX", value=adx_val, signal=signal, confidence=conf,
             description=desc, weight=0.5,
         )
+        result._adx_slope = adx_slope
 
     # ── ATR override ──────────────────────────────────────────────
 
@@ -967,11 +982,18 @@ def fetch_history(
     start: datetime,
     end: datetime,
     use_cache: bool = True,
+    clean: bool = True,
 ) -> pd.DataFrame:
     """Fetch OHLCV between start/end with parquet caching and pagination.
 
     Twelve Data caps responses at 5000 bars; we paginate by date range so
     multi-month M5/M15 backtests work.
+
+    The parquet cache stores the RAW series; `_validate_and_clean` runs on
+    load. Pass clean=False to get the raw price path — trade resolution must
+    use it, because cleaning drops >10σ spike bars (news candles) that hit
+    stops in reality. (Caches written before this change already contain
+    cleaned data; a refetch is needed to restore dropped spike bars.)
     """
     CACHE_DIR.mkdir(parents=True, exist_ok=True)
     path = _cache_path(symbol, timeframe, start, end)
@@ -980,7 +1002,7 @@ def fetch_history(
         df = pd.read_parquet(path)
         if df.index.tz is None:
             df.index = df.index.tz_localize("UTC")
-        return df
+        return provider._validate_and_clean(df, symbol, timeframe) if clean else df
 
     logger.info(f"Fetching {symbol} {timeframe.value} from {start.date()} to {end.date()}")
 
@@ -1033,18 +1055,20 @@ def fetch_history(
 
     df = pd.concat(chunks).sort_index()
     df = df[~df.index.duplicated(keep="last")]
-    df = provider._validate_and_clean(df, symbol, timeframe)
-    # Trim to requested window after cleaning. `start`/`end` are already
-    # tz-aware UTC datetimes — wrapping with pd.Timestamp(..., tz=...) would
-    # raise on already-localized inputs.
+    # Fetches ending "now" include the still-forming candle — drop it so the
+    # cache only ever contains closed bars.
+    df = provider._drop_forming_candle(df, timeframe)
+    # Trim to requested window. `start`/`end` are already tz-aware UTC
+    # datetimes — wrapping with pd.Timestamp(..., tz=...) would raise on
+    # already-localized inputs.
     end_ts = pd.Timestamp(end)
     start_ts = pd.Timestamp(start)
     df = df[(df.index >= start_ts) & (df.index <= end_ts)]
 
     if use_cache:
         df.to_parquet(path)
-        logger.info(f"Cached {len(df)} bars to {path.name}")
-    return df
+        logger.info(f"Cached {len(df)} raw bars to {path.name}")
+    return provider._validate_and_clean(df, symbol, timeframe) if clean else df
 
 
 # ═════════════════════════════════════════════════════════════════════
@@ -1141,6 +1165,7 @@ def simulate_trades(
     trades: List[Trade],
     sim_df: pd.DataFrame,
     max_bars: int = 200,
+    spread: float = 0.0,
 ) -> None:
     """Walk forward bar-by-bar after each entry and resolve SL / TP1.
 
@@ -1148,12 +1173,17 @@ def simulate_trades(
     matrices of post-entry highs/lows and finds the first hit per trade.
 
     SL/TP-same-bar tie: SL assumed first (conservative).
+    Gap-through fills: a bar *opening* beyond SL/TP fills at the open (worse
+    than SL on gaps against, better than TP on gaps in favor).
+    `spread` is a round-trip cost in price units subtracted from every
+    trade's PnL before computing R.
     Times out (no hit) → marked TIMEOUT, exit at the last simulated bar.
     """
     if not trades:
         return
 
     # Build per-trade post-entry windows (clipped to max_bars).
+    opens_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
     highs_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
     lows_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
     closes_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
@@ -1161,6 +1191,7 @@ def simulate_trades(
     exit_times: List[List[pd.Timestamp]] = []
 
     sim_index = sim_df.index
+    sim_open = sim_df["Open"].to_numpy(dtype=np.float64)
     sim_high = sim_df["High"].to_numpy(dtype=np.float64)
     sim_low = sim_df["Low"].to_numpy(dtype=np.float64)
     sim_close = sim_df["Close"].to_numpy(dtype=np.float64)
@@ -1180,6 +1211,7 @@ def simulate_trades(
         end_pos = min(entry_pos + max_bars, len(sim_index))
         n = end_pos - entry_pos
         valid_lens[i] = n
+        opens_arr[i, :n] = sim_open[entry_pos:end_pos]
         highs_arr[i, :n] = sim_high[entry_pos:end_pos]
         lows_arr[i, :n] = sim_low[entry_pos:end_pos]
         closes_arr[i, :n] = sim_close[entry_pos:end_pos]
@@ -1237,14 +1269,26 @@ def simulate_trades(
         if sl_hit_idx < n_valid and sl_hit_idx <= tp_hit_idx:
             t.exit_reason = "SL"
             t.bars_held = sl_hit_idx + 1
-            t.exit_price = t.sl
-            t.r_multiple = -1.0
+            exit_price = t.sl
+            bar_open = opens_arr[i, sl_hit_idx]
+            # Gap through the stop: the bar opened beyond SL → filled at open.
+            if not np.isnan(bar_open):
+                if (t.direction == "BUY" and bar_open < t.sl) or \
+                   (t.direction == "SELL" and bar_open > t.sl):
+                    exit_price = bar_open
+            t.exit_price = exit_price
             t.exit_time = exit_times[i][sl_hit_idx]
         elif tp_hit_idx < n_valid:
             t.exit_reason = "TP1"
             t.bars_held = tp_hit_idx + 1
-            t.exit_price = t.tp1
-            t.r_multiple = t.rr_planned
+            exit_price = t.tp1
+            bar_open = opens_arr[i, tp_hit_idx]
+            # Gap beyond the target in our favor → filled at open.
+            if not np.isnan(bar_open):
+                if (t.direction == "BUY" and bar_open > t.tp1) or \
+                   (t.direction == "SELL" and bar_open < t.tp1):
+                    exit_price = bar_open
+            t.exit_price = exit_price
             t.exit_time = exit_times[i][tp_hit_idx]
         else:
             # Time out at last simulated bar.
@@ -1252,12 +1296,15 @@ def simulate_trades(
             t.bars_held = int(n_valid)
             last_idx = int(n_valid) - 1
             t.exit_price = float(closes_arr[i, last_idx])
-            risk = abs(t.entry_price - t.sl)
-            if risk > 0:
-                pnl = (t.exit_price - t.entry_price) if t.direction == "BUY" else (t.entry_price - t.exit_price)
-                t.r_multiple = pnl / risk
             t.exit_time = exit_times[i][last_idx]
 
+        # Realized R from actual fill prices, net of round-trip spread.
+        risk = abs(t.entry_price - t.sl)
+        if risk > 0 and t.exit_price is not None:
+            pnl = ((t.exit_price - t.entry_price) if t.direction == "BUY"
+                   else (t.entry_price - t.exit_price))
+            t.r_multiple = (pnl - spread) / risk
+
         # MFE / MAE measured only over bars [0 .. bars_held).
         bars = t.bars_held
         risk = abs(t.entry_price - t.sl)
@@ -1297,13 +1344,26 @@ class BacktestConfig:
     # Restrict to specific strategy tags (XAU / EUR_PB / GBP_PB / GENERIC).
     # Empty = take all. Use to A/B test whether GENERIC trades add or destroy edge.
     strategies_only: List[str] = field(default_factory=list)
+    # Round-trip transaction cost in price units, charged on every trade
+    # (XAU/USD retail spread ≈ $0.30). 0 disables.
+    spread: float = 0.30
 
 
 def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]:
-    """Apply XAU weekend + allowed-weekday filter (live `analyze_pair` does this;
-    `analyze_from_dataframes` does not, so we replicate here)."""
+    """Apply XAU weekend + allowed-weekday filter matching live `analyze_pair`.
+
+    Forex market closed: Friday 22:00 UTC -> Sunday 22:00 UTC (Sydney open).
+    Must match `_is_forex_weekend` in signal_aggregator.py exactly.
+    """
     xau = cfg.xau_strategy
-    if xau.weekend_filter and ts.weekday() >= 5:
+    if xau.weekend_filter:
+        wd = ts.weekday()
+        hour = ts.hour
+        if wd == 4 and hour >= 22:
+            return "WEEKEND_FILTER"
+        if wd == 5:
+            return "WEEKEND_FILTER"
+        if wd == 6 and hour < 22:
             return "WEEKEND_FILTER"
     if xau.allowed_weekdays and ts.weekday() not in xau.allowed_weekdays:
         return "DAY_FILTER"
@@ -1388,12 +1448,23 @@ def run_backtest(
         decision_time = ts + primary_dur
 
         # XAU calendar filter (matches live analyze_pair behavior).
+        # Live evaluates wall-clock time ≈ bar close, so filter on
+        # decision_time, not the bar's open timestamp.
         if PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD":
-            block = _is_xau_blocked_by_calendar(ts, platform_cfg)
+            block = _is_xau_blocked_by_calendar(decision_time, platform_cfg)
             if block:
                 skipped_reasons[block] = skipped_reasons.get(block, 0) + 1
                 continue
 
+        # Session window filter (matches live _apply_session_filter_live,
+        # which checks datetime.now() ≈ bar close = decision_time).
+        bt_filter = platform_cfg.backtest
+        if bt_filter.session_filter:
+            bar_hour = decision_time.hour
+            if not (bt_filter.session_start_hour <= bar_hour < bt_filter.session_end_hour):
+                skipped_reasons["SESSION_WINDOW"] = skipped_reasons.get("SESSION_WINDOW", 0) + 1
+                continue
+
         # Cooldown.
         if pos - last_entry_pos < bt_cfg.cooldown_bars:
             continue
@@ -1443,14 +1514,13 @@ def run_backtest(
             continue
 
         ts_setup: TradeSetup = report.trade_setup
-        # Entry on the NEXT bar's open price would be ideal, but live signals fire
-        # at bar close and we use that close as entry (matches setup.entry_price).
-        # For simulation realism we step entry to next bar so bar `pos` is not
-        # used to resolve the trade outcome (bar `pos` is the signal bar).
+        # Entry at the OPEN of the next bar — live signals fire at bar close,
+        # execution happens on the next candle open (realistic fill).
         entry_pos = pos + 1
         if entry_pos >= len(primary_df):
             continue
         entry_time = primary_df.index[entry_pos]
+        entry_price = float(primary_df["Open"].iloc[entry_pos])
 
         breakdown, agree, oppose, neutral, total = _build_indicator_breakdown(
             report, ts_setup.direction
@@ -1458,17 +1528,30 @@ def run_backtest(
         reasons = (report.bullish_reasons if ts_setup.direction == "BUY"
                    else report.bearish_reasons)
 
+        # Adjust SL/TP relative to actual entry (preserve distance from setup).
+        sl_distance = abs(ts_setup.entry_price - ts_setup.stop_loss)
+        tp1_distance = abs(ts_setup.take_profit_1 - ts_setup.entry_price)
+        tp2_distance = abs(ts_setup.take_profit_2 - ts_setup.entry_price)
+        if ts_setup.direction == "BUY":
+            adj_sl = entry_price - sl_distance
+            adj_tp1 = entry_price + tp1_distance
+            adj_tp2 = entry_price + tp2_distance
+        else:
+            adj_sl = entry_price + sl_distance
+            adj_tp1 = entry_price - tp1_distance
+            adj_tp2 = entry_price - tp2_distance
+
         trades.append(Trade(
             pair=bt_cfg.pair,
             entry_time=entry_time,
-            entry_price=ts_setup.entry_price,
-            sl=ts_setup.stop_loss,
-            tp1=ts_setup.take_profit_1,
-            tp2=ts_setup.take_profit_2,
+            entry_price=entry_price,
+            sl=adj_sl,
+            tp1=adj_tp1,
+            tp2=adj_tp2,
             direction=ts_setup.direction,
             pip_value=ts_setup.pip_value,
-            risk_pips=ts_setup.risk_pips,
-            reward_pips=ts_setup.reward_pips,
+            risk_pips=sl_distance / ts_setup.pip_value,
+            reward_pips=tp1_distance / ts_setup.pip_value,
             rr_planned=ts_setup.risk_reward,
             quality=report.signal_quality,
             final_score=report.final_score,
@@ -1495,8 +1578,16 @@ def run_backtest(
         logger.info(f"Skipped bars by reason: {skipped_reasons}")
 
     # ── Resolve trade outcomes ──
+    # Resolve against the RAW price path: cleaning removes >10σ spike bars
+    # (news candles) that hit stops in reality. Cache hit → no API call.
+    sim_df = fetch_history(
+        provider, bt_cfg.pair, bt_cfg.primary_tf,
+        start=fetch_start, end=bt_cfg.end,
+        use_cache=bt_cfg.use_cache, clean=False,
+    )
     t_sim = time.perf_counter()
-    simulate_trades(trades, primary_df, max_bars=bt_cfg.max_hold_bars)
+    simulate_trades(trades, sim_df, max_bars=bt_cfg.max_hold_bars,
+                    spread=bt_cfg.spread)
     logger.info(
         f"Trade simulation took {time.perf_counter() - t_sim:.2f}s "
         f"(GPU={'yes' if _GPU_AVAILABLE else 'no'})"
@@ -1825,6 +1916,8 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
     p.add_argument("--strategies", default="",
                    help="Comma-separated strategy tags to include (XAU,EUR_PB,GBP_PB,GENERIC). Empty = all.")
     p.add_argument("--no-cache", action="store_true", help="Skip parquet cache (force API fetch)")
+    p.add_argument("--spread", type=float, default=0.30,
+                   help="Round-trip cost in price units per trade (XAU ~$0.30; 0 disables)")
     p.add_argument("--export-csv", help="Optional path to dump per-trade CSV (incl. indicator breakdown)")
     p.add_argument("--detail", action="store_true",
                    help="Print full per-trade reasoning (indicators, signals, agreement, why entered)")
@@ -1883,6 +1976,7 @@ def main(argv: Optional[List[str]] = None) -> int:
         use_cache=not args.no_cache,
         quality_filter=[q.strip() for q in args.quality.split(",") if q.strip()],
         strategies_only=[s.strip() for s in args.strategies.split(",") if s.strip()],
+        spread=args.spread,
     )
 
     logger.info(

config.py

@@ -70,13 +70,13 @@ class TechnicalConfig:
     adx_strong_trend: float = 25.0
     adx_very_strong: float = 50.0
 
-    divergence_lookback: int = 14
+    divergence_lookback: int = 40
     swing_sr_lookback: int = 50
 
     disabled_indicators: List[str] = field(default_factory=list)
 
     ema_mode: str = "4ema"          # "4ema" (9/21/50/200) or "2ema" (50/200)
-    rsi_mode: str = "classic"       # "classic" or "regime_aware"
+    rsi_mode: str = "regime_aware"  # "classic" or "regime_aware"
     macd_as_pillar: bool = True     # MACD histogram is a scored signal
 
     # ROC (Rate of Change)
@@ -147,10 +147,10 @@ class XauStrategyConfig:
     sl_atr_multiplier: float = 2.0
     min_ema200_tfs: int = 2
     allow_generic_trades: bool = True
-    blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2])
+    blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2, 10, 11, 13, 16, 21])
     weekend_filter: bool = True
     min_atr_threshold: float = 1.0
-    v2_sl_atr_multiplier: float = 2.5
+    v2_sl_atr_multiplier: float = 2.0
     buy_threshold_boost: int = 0
     m5_confirmation_bars: int = 0
     h4_rsi_gate: bool = False
@@ -158,6 +158,9 @@ class XauStrategyConfig:
     rsi_zone: tuple = (40.0, 65.0)
     allowed_weekdays: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 6])  # Mon-Fri + Sun (Sydney open)
     sell_min_quality: List[str] = field(default_factory=lambda: ["A+", "A", "B"])
+    # Volatility regimes in which SELL setups are allowed (low/normal-vol
+    # shorts lose consistently against gold's macro uptrend).
+    sell_vol_regimes: List[str] = field(default_factory=lambda: ["high", "extreme"])
 
 
 @dataclass
@@ -170,7 +173,7 @@ class FilterConfig:
     session_filter: bool = True
     session_start_hour: int = 3
     session_end_hour: int = 23
-    blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2])
+    blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2, 10, 11, 13, 16, 21])
 
     allowed_weekdays: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 6])  # Mon-Fri + Sun (Sydney open)
 

data_provider.py

@@ -121,6 +121,7 @@ class ForexDataProvider:
             min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000),
         )
         df = self._fetch_twelve_data(symbol, timeframe, outputsize)
+        df = self._drop_forming_candle(df, timeframe)
         df = self._validate_and_clean(df, symbol, timeframe)
 
         if not self._is_data_fresh(df, symbol, timeframe):
@@ -133,6 +134,21 @@ class ForexDataProvider:
         self._cache_timestamps[cache_key] = now
         return df
 
+    def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame:
+        """Drop the still-forming last candle so analysis only sees closed bars.
+
+        Twelve Data's time_series includes the current partial candle; the
+        scanner fires seconds after the boundary, so without this guard every
+        indicator reads a seconds-old stub bar and signals repaint.
+        """
+        if df is None or df.empty:
+            return df
+        candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
+        now = pd.Timestamp.now(tz="UTC")
+        if df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now:
+            df = df.iloc[:-1]
+        return df
+
     def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool:
         """Check if the last candle is within acceptable age for this timeframe."""
         if df is None or df.empty:

engine.py

@@ -478,11 +478,14 @@ class TechnicalEngine:
         tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
 
         # Directional Movement
-        plus_dm = high.diff()
-        minus_dm = -low.diff()
+        plus_dm_raw = high.diff()
+        minus_dm_raw = -low.diff()
 
-        plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0)
-        minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0)
+        # Compare raw vs raw: sequential filtering would zero +DM on a tie
+        # but keep -DM (compared against the already-zeroed +DM) — a
+        # systematic bearish nudge. Wilder zeroes both on ties.
+        plus_dm = plus_dm_raw.where((plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0), 0.0)
+        minus_dm = minus_dm_raw.where((minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0), 0.0)
 
         atr = tr.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean()
         plus_di = 100 * (plus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() / atr)
@@ -762,20 +765,34 @@ class TechnicalEngine:
             price_window = close.iloc[-lookback:]
             ind_window = indicator_series.iloc[-lookback:]
 
+            # A price swing and an indicator swing only form one divergence
+            # observation if they mark the SAME market turn — require the
+            # swing bars to line up within a small tolerance, otherwise
+            # unrelated wiggles get compared and the signal is noise.
+            def _aligned(price_pts, ind_pts, tol=2):
+                pairs = []
+                for p_idx, p_val in price_pts:
+                    best = min(ind_pts, key=lambda ip: abs(ip[0] - p_idx), default=None)
+                    if best is not None and abs(best[0] - p_idx) <= tol:
+                        pairs.append(((p_idx, p_val), best))
+                return pairs
+
             # Bullish divergence: lower price low but higher indicator low
             price_lows = self._find_swing_points(price_window, mode="low")
             ind_lows = self._find_swing_points(ind_window, mode="low")
-
-            if len(price_lows) >= 2 and len(ind_lows) >= 2:
-                if price_lows[-1][1] < price_lows[-2][1] and ind_lows[-1][1] > ind_lows[-2][1]:
+            low_pairs = _aligned(price_lows, ind_lows)
+            if len(low_pairs) >= 2:
+                (p1, i1), (p2, i2) = low_pairs[-2], low_pairs[-1]
+                if p2[1] < p1[1] and i2[1] > i1[1]:
                     divergences.append(f"Bullish {name} divergence (reversal warning)")
 
             # Bearish divergence: higher price high but lower indicator high
             price_highs = self._find_swing_points(price_window, mode="high")
             ind_highs = self._find_swing_points(ind_window, mode="high")
-
-            if len(price_highs) >= 2 and len(ind_highs) >= 2:
-                if price_highs[-1][1] > price_highs[-2][1] and ind_highs[-1][1] < ind_highs[-2][1]:
+            high_pairs = _aligned(price_highs, ind_highs)
+            if len(high_pairs) >= 2:
+                (p1, i1), (p2, i2) = high_pairs[-2], high_pairs[-1]
+                if p2[1] > p1[1] and i2[1] < i1[1]:
                     divergences.append(f"Bearish {name} divergence (reversal warning)")
 
         if divergences:
@@ -976,6 +993,22 @@ class TechnicalEngine:
                 desc = f"RSI {rsi_val:.1f} — Overbought in downtrend, possible reversal"
             else:
                 return
+        elif regime == "transitioning":
+            # Trend is being born — RSI can't tell direction yet, and
+            # range-fade rules would sell the newborn trend. Only extremes
+            # carry information here.
+            if rsi_val >= 70:
+                signal = SignalStrength.SELL
+                conf = 50
+                desc = f"RSI {rsi_val:.1f} — Overbought while trend forming"
+            elif rsi_val <= 30:
+                signal = SignalStrength.BUY
+                conf = 50
+                desc = f"RSI {rsi_val:.1f} — Oversold while trend forming"
+            else:
+                signal = SignalStrength.NEUTRAL
+                conf = 30
+                desc = f"RSI {rsi_val:.1f} — No RSI edge while trend forms"
         else:  # ranging
             if rsi_val >= 70:
                 signal = SignalStrength.SELL
@@ -1037,10 +1070,15 @@ class TechnicalEngine:
         if self.config.regime_weights_override:
             regime_weights = self.config.regime_weights_override
         else:
+            # NOTE: ROC is directionally redundant with MACD/Keltner/EMA
+            # (0.8-0.94 correlation), but zeroing its weight was backtested
+            # 2026-07 and LOST 13R — the redundant vote effectively raises
+            # the composite bar and the thresholds are calibrated with it.
+            # Don't remove without retuning score thresholds.
             _trending = {
                 "ema_trend": 2.0, "rsi": 1.0, "macd": 2.0,
                 "bollinger": 0.5, "keltner": 1.5, "roc": 1.5,
-                "candlestick": 0.8, "divergence": 2.0, "adx": 0.8,
+                "candlestick": 0.8, "divergence": 1.0, "adx": 0.8,
             }
             regime_weights = {
                 "trending_up": _trending,

signal_aggregator.py

@@ -269,10 +269,10 @@ class SignalAggregator:
             (high - close.shift()).abs(),
             (low - close.shift()).abs(),
         ], axis=1).max(axis=1)
-        plus_dm = high.diff()
-        minus_dm = -low.diff()
-        plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0)
-        minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0)
+        plus_dm_raw = high.diff()
+        minus_dm_raw = -low.diff()
+        plus_dm = plus_dm_raw.where((plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0), 0.0)
+        minus_dm = minus_dm_raw.where((minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0), 0.0)
         atr_s = tr.ewm(alpha=1/period, min_periods=period, adjust=False).mean()
         plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s)
         minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s)
@@ -411,7 +411,17 @@ class SignalAggregator:
                 report.signal_quality = "B"
                 report.action_recommendation = f"{report.final_signal.name.replace('_', ' ')} — Trend forming, watch closely"
 
-        # ── 6d. SELL quality gate ─────────────────────────────────
+        # ── 6d. Exhausted trend quality cap ───────────────────────
+        if (report.market_regime.startswith("trending")
+                and report.exhaustion_count >= 2
+                and report.signal_quality in ("C", "D")):
+            report.warnings.append(
+                "EXHAUSTED TREND FILTER: C-grade blocked in exhausted trend"
+            )
+            report.signal_quality = "D"
+            report.trade_setup = None
+
+        # ── 6e. SELL quality gate ─────────────────────────────────
         xau_is_sell_dir = (
             report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL)
             or (report.final_signal == SignalStrength.NEUTRAL and report.final_score < 0)
@@ -425,6 +435,38 @@ class SignalAggregator:
             report.signal_quality = "D"
             report.trade_setup = None
 
+        # ── 6f. MTF opposition gate ───────────────────────────────
+        # Never trade against a multi-timeframe consensus: if the
+        # threshold number of TFs point opposite to the trade direction,
+        # block the setup (counter-trend entries in gold bleed).
+        if report.signal_quality in ("A+", "A", "B", "C") and report.mtf_signals:
+            sigs = list(report.mtf_signals.values())
+            bull_n = sum(1 for s in sigs if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY))
+            bear_n = sum(1 for s in sigs if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))
+            thr = self.config.signals.mtf_agreement_threshold
+            opposed = (bear_n >= thr and report.final_score > 0) or \
+                      (bull_n >= thr and report.final_score < 0)
+            if opposed:
+                report.warnings.append(
+                    "MTF OPPOSITION: majority of timeframes point against signal — no trade"
+                )
+                report.signal_quality = "D"
+                report.trade_setup = None
+
+        # ── 6g. SELL volatility gate ──────────────────────────────
+        # Shorts in quiet markets fade against gold's macro uptrend;
+        # only take sells when volatility regime confirms real selling.
+        if (self.config.xau_strategy.enabled
+                and xau_is_sell_dir
+                and report.signal_quality in ("A+", "A", "B", "C")
+                and report.volatility_regime
+                and report.volatility_regime not in self.config.xau_strategy.sell_vol_regimes):
+            report.warnings.append(
+                f"SELL FILTER: volatility '{report.volatility_regime}' too quiet for shorts"
+            )
+            report.signal_quality = "D"
+            report.trade_setup = None
+
         # ── 7. Trade setup for actionable signals ──────────────────
         if report.trade_setup is None and report.signal_quality in ("A+", "A"):
             atr = (report.primary_analysis.key_levels.get("atr", 0)
@@ -615,6 +657,12 @@ class SignalAggregator:
         bearish = sum(1 for s in signals if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))
 
         threshold = self.config.signals.mtf_agreement_threshold
+        # Agreement must point the same way as the primary timeframe —
+        # two bullish confirmation TFs must not "confirm" a bearish primary.
+        if report.primary_score > 0:
+            return bullish >= threshold
+        if report.primary_score < 0:
+            return bearish >= threshold
         return bullish >= threshold or bearish >= threshold
 
     def _compute_final_signal(self, report: SignalReport):
@@ -776,7 +824,6 @@ class SignalAggregator:
             momentum_fading = roc_fading or macd_fading
         exhaustion_blocked = exhaustion_count >= 2 and momentum_fading
 
-        # Apply penalties with diminishing returns (cap at 30%)
         MAX_PENALTY = 30.0
         penalties.sort(key=lambda x: x[1], reverse=True)
         total_penalty = 0.0
@@ -870,11 +917,9 @@ class SignalAggregator:
         tf = self.config.signals.tp_tightening_factor
         pre_tight_min_rr = min_rr / tf if tf > 0 else min_rr
 
-        mature_move = abs(report.final_score) > 30
-        tp_mult = 1.0 if mature_move else 1.5
-        tp_swing_cap = 2.0 if mature_move else 2.5
+        tp_mult = 1.5
+        tp_swing_cap = 2.5
 
-        # Fix 3: Wider SL multiplier for XAU/USD (1.8x vs default 1.5x)
         sl_atr_floor = (self.config.xau_strategy.v2_sl_atr_multiplier
                         if report.display_name == "XAU/USD" else 1.5)
 
@@ -945,6 +990,13 @@ class SignalAggregator:
         tp1 = entry + (tp1 - entry) * tf
         tp2 = entry + (tp2 - entry) * tf
 
+        # TP2 must sit beyond TP1 — swing picks can invert when TP1 falls
+        # back to the ATR target while TP2 takes a nearer second swing.
+        if is_buy:
+            tp2 = max(tp2, tp1 + atr)
+        else:
+            tp2 = min(tp2, tp1 - atr)
+
         tp1_distance = abs(tp1 - entry)
         rr = tp1_distance / sl_distance if sl_distance > 0 else 0
 

telegram_bot.py

@@ -38,6 +38,7 @@ import websockets
 
 from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
 from telegram.constants import ParseMode
+from telegram.error import BadRequest
 from telegram.ext import (
     Application,
     CallbackQueryHandler,
@@ -177,6 +178,18 @@ def _indicator_line(name: str, ind) -> str:
     return f"[{tag}] *{_escape_md(ind.name)}*: {_escape_md(short)}"
 
 
+def _live_price_line(symbol: str) -> Optional[str]:
+    """Fresh WebSocket tick for this symbol, if we have one (<2 min old)."""
+    price = _last_prices.get(symbol)
+    ts = _last_price_ts.get(symbol, 0)
+    if price is None:
+        return None
+    age = time.time() - ts
+    if age > 120:
+        return None
+    return f"📡 Live: *{_fmt_price(price)}*  ({age:.0f}s tick)"
+
+
 def format_signal_message(report: SignalReport) -> str:
     sig = report.final_signal
     emoji = _SIGNAL_EMOJI.get(sig, "⬜")
@@ -193,7 +206,12 @@ def format_signal_message(report: SignalReport) -> str:
         f"🔍 *{report.display_name} Signal Analysis*  ({report.primary_analysis.timeframe if report.primary_analysis else ''})",
         f"🕐 {now}",
         "",
-        f"💰 Price: *{_fmt_price(report.current_price)}*",
+        f"💰 Close: *{_fmt_price(report.current_price)}*",
+    ]
+    live = _live_price_line(report.symbol)
+    if live:
+        lines.append(live)
+    lines += [
         f"📊 Regime: {regime_str} | Vol: {vol_str}",
         "",
         "━━━━━━━━━━━━━━━━━━━━",
@@ -221,14 +239,16 @@ def format_signal_message(report: SignalReport) -> str:
     if report.trade_setup and sig not in (SignalStrength.NEUTRAL,):
         ts = report.trade_setup
         dir_emoji = "📈" if ts.direction == "BUY" else "📉"
+        risk_usd = ts.risk_pips * ts.pip_value
+        reward_usd = ts.reward_pips * ts.pip_value
         lines += [
             "",
             "━━━━━━━━━━━━━━━━━━━━",
             f"🚀 *TRADE SETUP*",
             f"{dir_emoji} Direction: *{ts.direction}*",
             f"▶ Entry:  `{_fmt_price(ts.entry_price)}`",
-            f"🛑 SL:     `{_fmt_price(ts.stop_loss)}`  ({ts.risk_pips:.1f} pips)",
-            f"🎯 TP1:    `{_fmt_price(ts.take_profit_1)}`  ({ts.reward_pips:.1f} pips)",
+            f"🛑 SL:     `{_fmt_price(ts.stop_loss)}`  (${risk_usd:.2f} / {ts.risk_pips:.0f} pips)",
+            f"🎯 TP1:    `{_fmt_price(ts.take_profit_1)}`  (${reward_usd:.2f} / {ts.reward_pips:.0f} pips)",
         ]
         if ts.take_profit_2:
             lines.append(f"🎯 TP2:    `{_fmt_price(ts.take_profit_2)}`")
@@ -283,6 +303,45 @@ def scan_keyboard() -> InlineKeyboardMarkup:
     return InlineKeyboardMarkup(rows)
 
 
+def main_menu_keyboard() -> InlineKeyboardMarkup:
+    """Full interactive menu: scans + settings + status in one keyboard."""
+    tf_buttons = [b for b in SCAN_BUTTONS if b["tf"] is not None]
+    all_button = next((b for b in SCAN_BUTTONS if b["tf"] is None), None)
+    autoscan_label = "🔔 Auto-scan: ON" if _autoscan_chats else "🔕 Auto-scan: OFF"
+    quality_label = f"🎚 Alerts: {_alert_min_quality}+" if _alert_min_quality != "A" else "🎚 Alerts: A only"
+    rows = [
+        [InlineKeyboardButton(b["label"], callback_data=b["cb"]) for b in tf_buttons],
+    ]
+    if all_button:
+        rows.append([InlineKeyboardButton(all_button["label"], callback_data=all_button["cb"])])
+    rows.append([
+        InlineKeyboardButton(autoscan_label, callback_data="menu_autoscan"),
+        InlineKeyboardButton(quality_label, callback_data="menu_quality"),
+    ])
+    rows.append([
+        InlineKeyboardButton("📊 Status", callback_data="menu_status"),
+        InlineKeyboardButton("ℹ️ Help", callback_data="menu_help"),
+    ])
+    return InlineKeyboardMarkup(rows)
+
+
+async def _safe_edit(query, text: str, reply_markup=None) -> None:
+    """edit_message_text that survives 'message is not modified' and
+    Markdown parse failures (falls back to plain text)."""
+    try:
+        await query.edit_message_text(
+            text, parse_mode=ParseMode.MARKDOWN, reply_markup=reply_markup
+        )
+    except BadRequest as exc:
+        if "not modified" in str(exc).lower():
+            return
+        logger.warning(f"Markdown edit failed ({exc}); retrying as plain text")
+        try:
+            await query.edit_message_text(text, reply_markup=reply_markup)
+        except BadRequest:
+            pass
+
+
 # ── Signal runner (shared data provider for cache reuse) ──────────────
 
 def _run_signal_sync(
@@ -313,6 +372,40 @@ async def run_signal(
     )
 
 
+# ── Alert gating (quality threshold + dedup) ─────────────────────────
+
+_QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1}
+_ALERT_QUALITY_CYCLE = ["C", "B", "A"]   # C+ → B+ → A-only
+_alert_min_quality = "C"
+
+# Fingerprint of the last alert per (symbol, timeframe) so a persisting
+# signal doesn't re-send an identical message every candle.
+_last_alert_fp: Dict[str, tuple] = {}
+_last_alert_ts: Dict[str, float] = {}
+_ALERT_REARM_SECONDS = 3600   # identical signal re-alerts after 1h
+
+
+def _passes_quality(quality: str) -> bool:
+    return _QUALITY_RANK.get(quality, 0) >= _QUALITY_RANK.get(_alert_min_quality, 1)
+
+
+def _should_alert(key: str, report: SignalReport) -> bool:
+    """True when this signal is new/changed for the key, or re-armed."""
+    setup = report.trade_setup
+    fp = (
+        report.final_signal.name,
+        report.signal_quality,
+        setup.direction if setup else None,
+    )
+    now = time.time()
+    if (_last_alert_fp.get(key) == fp
+            and now - _last_alert_ts.get(key, 0) < _ALERT_REARM_SECONDS):
+        return False
+    _last_alert_fp[key] = fp
+    _last_alert_ts[key] = now
+    return True
+
+
 # ── Candle-boundary scheduler ────────────────────────────────────────
 
 _autoscan_chats: Set[int] = set()
@@ -352,7 +445,11 @@ async def _scan_timeframe(
 
     _update_level_cache(reports)
 
-    actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
+    actionable = [
+        r for r in reports
+        if _passes_quality(r.signal_quality)
+        and _should_alert(f"{r.symbol}_{tf.value}", r)
+    ]
     for chat_id in list(_autoscan_chats):
         for report in actionable:
             detail = format_signal_message(report)
@@ -427,18 +524,46 @@ async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, le
     _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)")
+    logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15")
 
     try:
         report = await run_signal(pair_symbol, Timeframe.M15)
         _update_level_cache([report])
     except Exception:
         logger.exception(f"Breach scan failed for {display}")
+        return
+
+    # Alert only when the rescan produced an actionable, changed signal —
+    # same gates as the auto-scanner, so a breach can't spam the channel.
+    if (report.final_signal == SignalStrength.NEUTRAL
+            or not _passes_quality(report.signal_quality)
+            or not _should_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)):
+        return
+
+    header = (
+        f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed "
+        f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n"
+    )
+    detail = header + format_signal_message(report)
+    for chat_id in list(_autoscan_chats):
+        try:
+            await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
+        except Exception as exc:
+            logger.warning(f"Failed to send breach alert to chat {chat_id}: {exc}")
+
+
+async def _ws_heartbeat(ws) -> None:
+    """Twelve Data drops idle WS connections; heartbeat every 10s keeps it
+    alive through quiet periods (weekends, low-liquidity hours)."""
+    while True:
+        await asyncio.sleep(10)
+        await ws.send(json.dumps({"action": "heartbeat"}))
 
 
 async def _ws_monitor(bot) -> None:
     backoff = 1
     while _autoscan_chats:
+        hb_task: Optional[asyncio.Task] = None
         try:
             async with websockets.connect(_WS_URL) as ws:
                 subscribe = {
@@ -448,6 +573,7 @@ async def _ws_monitor(bot) -> None:
                 await ws.send(json.dumps(subscribe))
                 logger.info(f"WebSocket connected, subscribed to {len(_WS_SYMBOLS)} symbols")
                 backoff = 1
+                hb_task = asyncio.create_task(_ws_heartbeat(ws))
 
                 async for raw in ws:
                     if not _autoscan_chats:
@@ -459,7 +585,13 @@ async def _ws_monitor(bot) -> None:
                         logger.warning(f"WebSocket: malformed JSON, skipping")
                         continue
 
-                    if msg.get("event") != "price":
+                    event = msg.get("event")
+                    if event == "subscribe-status":
+                        fails = msg.get("fails") or []
+                        if fails:
+                            logger.warning(f"WebSocket: failed to subscribe: {fails}")
+                        continue
+                    if event != "price":
                         continue
 
                     ws_symbol = msg.get("symbol")
@@ -492,6 +624,9 @@ async def _ws_monitor(bot) -> None:
             logger.exception(f"WebSocket error, reconnecting in {backoff}s")
             await asyncio.sleep(backoff)
             backoff = min(backoff * 2, 60)
+        finally:
+            if hb_task is not None:
+                hb_task.cancel()
 
     logger.info("WebSocket monitor stopped")
 
@@ -572,27 +707,31 @@ async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
     await update.message.reply_text(
         text,
         parse_mode=ParseMode.MARKDOWN,
-        reply_markup=scan_keyboard(),
+        reply_markup=main_menu_keyboard(),
     )
 
 
-async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
-    text = (
+_HELP_TEXT = (
     "📖 *ForexBot2 Commands*\n\n"
-        "/start — Welcome & scan buttons\n"
-        "/scan — Quick XAU/USD scan buttons\n"
+    "/start — Welcome & interactive menu\n"
+    "/scan — Quick XAU/USD scan menu\n"
     "/autoscan — Toggle auto-scan on/off\n"
     "/status — Bot status, WS connection, live prices\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"
-        "*WebSocket* monitors live prices and triggers breach scans."
+    "*Alerts* button sets the minimum quality grade that gets sent.\n"
+    "*WebSocket* streams live prices; key-level breaches trigger an "
+    "instant M15 rescan and alert (5-min cooldown)."
 )
-    await update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN)
 
 
-async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    await update.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN)
+
+
+def _status_text() -> str:
     lines = ["📊 *ForexBot2 Status*\n"]
 
     if _autoscan_chats:
@@ -639,7 +778,13 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
     if active_breach:
         lines.append(f"*Breach tasks:* {active_breach} running")
 
-    await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
+    lines.append(f"\n*Alert quality:* {_alert_min_quality}+ "
+                 f"| re-arm {_ALERT_REARM_SECONDS // 60}m")
+    return "\n".join(lines)
+
+
+async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    await update.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN)
 
 
 async def cmd_cache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
@@ -715,7 +860,7 @@ async def cmd_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
     await update.message.reply_text(
         "🔍 *XAU/USD — Pick a timeframe*",
         parse_mode=ParseMode.MARKDOWN,
-        reply_markup=scan_keyboard(),
+        reply_markup=main_menu_keyboard(),
     )
 
 
@@ -731,30 +876,60 @@ async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         await _cb_scan_all(query)
         return
 
-    await query.edit_message_text(
-        f"⏳ Scanning XAU/USD {btn['label']}…",
-        parse_mode=ParseMode.MARKDOWN,
-    )
+    await _safe_edit(query, f"⏳ Scanning XAU/USD {btn['label']}…")
 
     try:
         report = await run_signal(btn["symbol"], btn["tf"])
         text = format_signal_message(report)
     except Exception as exc:
         logger.exception("Signal scan failed")
-        text = f"❌ Scan failed: {exc}"
+        text = f"❌ Scan failed: {_escape_md(str(exc))}"
 
-    await query.edit_message_text(
-        text,
-        parse_mode=ParseMode.MARKDOWN,
-        reply_markup=scan_keyboard(),
-    )
+    await _safe_edit(query, text, reply_markup=main_menu_keyboard())
+
+
+async def cb_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Interactive menu actions: autoscan toggle, quality cycle, status, help."""
+    global _alert_min_quality
+    query = update.callback_query
+    action = query.data
+
+    if action == "menu_autoscan":
+        chat_id = query.message.chat.id
+        if chat_id in _autoscan_chats:
+            _autoscan_chats.discard(chat_id)
+            if not _autoscan_chats:
+                _stop_scheduler()
+            await query.answer("Auto-scan stopped")
+        else:
+            _autoscan_chats.add(chat_id)
+            _start_scheduler(context.bot)
+            await query.answer("Auto-scan started")
+        try:
+            await query.edit_message_reply_markup(reply_markup=main_menu_keyboard())
+        except BadRequest:
+            pass
+
+    elif action == "menu_quality":
+        i = _ALERT_QUALITY_CYCLE.index(_alert_min_quality) if _alert_min_quality in _ALERT_QUALITY_CYCLE else 0
+        _alert_min_quality = _ALERT_QUALITY_CYCLE[(i + 1) % len(_ALERT_QUALITY_CYCLE)]
+        await query.answer(f"Alerting {_alert_min_quality}+ signals")
+        try:
+            await query.edit_message_reply_markup(reply_markup=main_menu_keyboard())
+        except BadRequest:
+            pass
+
+    elif action == "menu_status":
+        await query.answer()
+        await query.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN)
+
+    elif action == "menu_help":
+        await query.answer()
+        await query.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN)
 
 
 async def _cb_scan_all(query) -> None:
-    await query.edit_message_text(
-        "⏳ Scanning XAU/USD across M5 / M15 / H1 / H4…",
-        parse_mode=ParseMode.MARKDOWN,
-    )
+    await _safe_edit(query, "⏳ Scanning XAU/USD across M5 / M15 / H1 / H4…")
 
     tfs = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]
     tasks = [run_signal(XAUUSD, tf) for tf in tfs]
@@ -803,11 +978,7 @@ async def _cb_scan_all(query) -> None:
 
     lines.append("_Use buttons below for detailed single-TF view._")
 
-    await query.edit_message_text(
-        "\n".join(lines),
-        parse_mode=ParseMode.MARKDOWN,
-        reply_markup=scan_keyboard(),
-    )
+    await _safe_edit(query, "\n".join(lines), reply_markup=main_menu_keyboard())
 
 
 # ── Boot-time auto-start ──────────────────────────────────────────────
@@ -908,6 +1079,7 @@ def main() -> None:
     app.add_handler(CommandHandler("cache", cmd_cache))
     app.add_handler(CommandHandler("clearcache", cmd_clearcache))
     app.add_handler(CallbackQueryHandler(cb_scan, pattern="^scan_"))
+    app.add_handler(CallbackQueryHandler(cb_menu, pattern="^menu_"))
 
     logger.info("Bot running — press Ctrl+C to stop")
     app.run_polling(allowed_updates=Update.ALL_TYPES)

test_all.py

@@ -155,7 +155,7 @@ def _():
     assert tc.atr_period == 14
     assert tc.adx_period == 14
     assert tc.ema_mode == "4ema"
-    assert tc.rsi_mode == "classic"
+    assert tc.rsi_mode == "regime_aware"
     assert tc.macd_as_pillar is True
     assert tc.disabled_indicators == []
     assert tc.roc_period == 10

test_ws_dual.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Test whether Twelve Data allows 2 simultaneous WebSocket connections for EUR/USD.
+
+Uses the twelvedata SDK which handles the WS protocol correctly.
+Runs connection 1, then adds connection 2 after 10s, monitors for 30s total.
+"""
+
+import threading
+import time
+
+from twelvedata import TDClient
+
+API_KEY = "f563b22ffdfd4924955decbe1a8c8701"
+SYMBOL = "EUR/USD"
+DURATION = 30
+
+results = {
+    "WS-1": {"ticks": 0, "errors": [], "alive": False},
+    "WS-2": {"ticks": 0, "errors": [], "alive": False},
+}
+
+
+def make_handlers(label):
+    def on_event(event):
+        if event.get("event") == "price":
+            results[label]["ticks"] += 1
+            price = event.get("price", "?")
+            print(f"[{label}] tick #{results[label]['ticks']:3d}  price={price}")
+
+    def on_error(error):
+        results[label]["errors"].append(str(error))
+        print(f"[{label}] ERROR: {error}")
+
+    def on_close(status):
+        results[label]["alive"] = False
+        print(f"[{label}] CLOSED (status={status})")
+
+    def on_connect():
+        results[label]["alive"] = True
+        print(f"[{label}] CONNECTED — subscribed to {SYMBOL}")
+
+    return on_event, on_error, on_close, on_connect
+
+
+def run_ws(label):
+    td = TDClient(apikey=API_KEY)
+    on_event, on_error, on_close, on_connect = make_handlers(label)
+    ws = td.websocket(
+        symbols=SYMBOL,
+        on_event=on_event,
+        on_error=on_error,
+        on_close=on_close,
+        on_connect=on_connect,
+    )
+    ws.connect()
+    return ws
+
+
+def main():
+    print(f"Testing dual WebSocket for {SYMBOL} ({DURATION}s)\n")
+
+    print("Starting WS-1...")
+    ws1 = run_ws("WS-1")
+
+    time.sleep(10)
+    print(f"\nWS-1 ticks so far: {results['WS-1']['ticks']}")
+    print("Starting WS-2...\n")
+    ws2 = run_ws("WS-2")
+
+    remaining = DURATION - 10
+    time.sleep(remaining)
+
+    print("\nStopping connections...")
+    try:
+        ws1.disconnect()
+    except Exception:
+        pass
+    try:
+        ws2.disconnect()
+    except Exception:
+        pass
+
+    time.sleep(1)
+
+    print("\n" + "=" * 50)
+    print("RESULTS")
+    print("=" * 50)
+    for label, info in sorted(results.items()):
+        status = "alive" if info["alive"] else "dead"
+        errs = info["errors"] or ["none"]
+        print(f"  {label}: {status}  ticks={info['ticks']}  errors={errs[0]}")
+
+    ws1_ok = results["WS-1"]["ticks"] > 0 and not results["WS-1"]["errors"]
+    ws2_ok = results["WS-2"]["ticks"] > 0 and not results["WS-2"]["errors"]
+    ws1_died = results["WS-1"]["alive"] is False and results["WS-1"]["ticks"] > 0
+
+    if ws1_ok and ws2_ok:
+        print("\n✓ Both connections worked — dual WS is supported")
+    elif ws2_ok and ws1_died:
+        print("\n✗ WS-2 replaced WS-1 — only 1 connection allowed at a time")
+    elif ws1_ok and not ws2_ok:
+        print("\n✗ WS-2 was rejected — only 1 connection allowed")
+    else:
+        print("\n? Inconclusive — check errors above")
+
+
+if __name__ == "__main__":
+    main()