From 058ef57d8ff021be9e99e1e8b3bbee901549d4e6 Mon Sep 17 00:00:00 2001 From: harwinder Date: Mon, 11 May 2026 16:33:23 -0400 Subject: [PATCH] Pro tier upgrade: quant rearchitect + WebSocket level-breach monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Restore XAU/USD (Pro plan), drop USD/CAD, raise rate limit to 55/min - Reduce signal engine to 4 uncorrelated pillars (EMA, RSI, BB, Candlestick) with regime-adaptive weights (trending boosts EMA, ranging boosts RSI+BB) - Add structure-based S/R (swing highs/lows) for trade setup targeting - Fix confidence death spiral: all adjustments now additive, not multiplicative - Dynamic R:R trade setup: SL behind swing level, TP at next structure level, reject if R:R < 1.5 - Fix news engine for all 4 pairs (was hardcoded to USDJPY) - Add WebSocket monitor: streams live ticks, triggers M15 scan on level breach with 5-min cooldown per pair — catches breakouts between candle boundaries - Add correlation filter (EUR/USD+GBP/USD, XAU/USD+USD/JPY exposure warnings) --- config.py | 26 ++-- data_provider.py | 6 +- engine.py | 141 +++++++++---------- main.py | 21 +-- news_engine.py | 29 ++-- signal_aggregator.py | 93 ++++++++----- telegram_bot.py | 313 +++++++++++++++++++++++++++++++++++-------- ws_twelvedata.py | 2 +- 8 files changed, 435 insertions(+), 196 deletions(-) mode change 100644 => 100755 engine.py mode change 100644 => 100755 main.py mode change 100644 => 100755 news_engine.py mode change 100644 => 100755 ws_twelvedata.py diff --git a/config.py b/config.py index 96e521b..3b6d867 100755 --- a/config.py +++ b/config.py @@ -39,13 +39,13 @@ class MarketSession(Enum): USDJPY = "USDJPY=X" EURUSD = "EURUSD=X" GBPUSD = "GBPUSD=X" -USDCAD = "USDCAD=X" +XAUUSD = "XAU/USD" PAIR_INFO = { USDJPY: {"display": "USD/JPY", "pip_value": 0.01, "is_jpy_pair": True}, EURUSD: {"display": "EUR/USD", "pip_value": 0.0001, "is_jpy_pair": False}, GBPUSD: {"display": "GBP/USD", "pip_value": 0.0001, "is_jpy_pair": False}, - USDCAD: {"display": "USD/CAD", "pip_value": 0.0001, "is_jpy_pair": False}, + XAUUSD: {"display": "XAU/USD", "pip_value": 0.10, "is_jpy_pair": False}, } # Defaults (kept for backward compat) @@ -92,6 +92,7 @@ class TechnicalConfig: volume_ma_period: int = 20 pivot_lookback: int = 1 divergence_lookback: int = 14 + swing_sr_lookback: int = 50 disabled_indicators: List[str] = field(default_factory=list) @@ -102,12 +103,11 @@ class SignalConfig: min_confidence: float = 55.0 weights: Dict[str, float] = field(default_factory=lambda: { - "trend": 0.25, - "momentum": 0.20, - "volatility": 0.15, - "price_action": 0.15, - "multi_timeframe": 0.15, - "sentiment": 0.10, + "ema_trend": 1.0, + "rsi": 1.5, + "bollinger": 1.5, + "candlestick": 1.5, + "divergence": 1.0, }) mtf_agreement_threshold: int = 2 # of 3 timeframes @@ -115,13 +115,14 @@ class SignalConfig: default_risk_pct: float = 1.0 default_rr_ratio: float = 2.0 - atr_sl_multiplier: float = 1.5 # SL distance = 1.5 × ATR + atr_sl_multiplier: float = 1.5 + min_rr_ratio: float = 1.5 @dataclass class NewsConfig: """Economic calendar & sentiment.""" - tracked_currencies: List[str] = field(default_factory=lambda: ["USD", "JPY", "EUR", "GBP", "CAD"]) + tracked_currencies: List[str] = field(default_factory=lambda: ["USD", "JPY", "EUR", "GBP", "XAU"]) min_impact: str = "medium" pre_event_buffer_minutes: int = 30 post_event_buffer_minutes: int = 15 @@ -168,11 +169,12 @@ class PlatformConfig: refresh_interval: int = 300 scan_pairs: List[str] = field( - default_factory=lambda: [EURUSD, USDJPY, GBPUSD, USDCAD] + default_factory=lambda: [EURUSD, USDJPY, GBPUSD, XAUUSD] ) scan_timeframes: List[Timeframe] = field( - default_factory=lambda: [Timeframe.M15, Timeframe.H1] + 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/data_provider.py b/data_provider.py index 0840889..956284d 100755 --- a/data_provider.py +++ b/data_provider.py @@ -49,7 +49,7 @@ class ForexDataProvider: Timeframe.D1: 86400, Timeframe.W1: 604800, } - RATE_LIMIT = 8 + RATE_LIMIT = 55 RATE_WINDOW = 60.0 def __init__(self, config: PlatformConfig): @@ -67,6 +67,8 @@ class ForexDataProvider: @staticmethod def _to_td_symbol(symbol: str) -> str: + if "/" in symbol: + return symbol clean = symbol.replace("=X", "") return f"{clean[:3]}/{clean[3:]}" @@ -132,7 +134,7 @@ class ForexDataProvider: # ── Private ──────────────────────────────────────────────────── def _throttle(self, credits: int = 1) -> None: - """Free tier: 8 credits / minute. Block until budget allows.""" + """Pro tier: 55 credits / minute. Block until budget allows.""" while True: now = time.time() while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: diff --git a/engine.py b/engine.py old mode 100644 new mode 100755 index 19eb264..7253034 --- a/engine.py +++ b/engine.py @@ -87,16 +87,15 @@ class TechnicalEngine: if "ema_trend" not in disabled: self._compute_ema_signals(df, result) if "rsi" not in disabled: self._compute_rsi(df, result) - if "macd" not in disabled: self._compute_macd(df, result) if "bollinger" not in disabled: self._compute_bollinger(df, result) - if "stochastic" not in disabled: self._compute_stochastic(df, result) if "adx" not in disabled: self._compute_adx(df, result) if "atr" not in disabled: self._compute_atr(df, result) - if "ichimoku" not in disabled: self._compute_ichimoku(df, result) + + self._compute_macd_internal(df, result) + if "divergence" not in disabled: self._detect_divergences(df, result) + if "structure" not in disabled: self._compute_structure_sr(df, result) if "candlestick" not in disabled: self._detect_candlestick_patterns(df, result) - if "pivot" not in disabled: self._compute_support_resistance(df, result) - if "volume" not in disabled: self._compute_volume_analysis(df, result) # ── Classify market regime ───────────────────────────────── self._classify_regime(df, result) @@ -222,14 +221,14 @@ class TechnicalEngine: result.indicators["rsi"] = IndicatorSignal( name="RSI", value=rsi_val, signal=signal, - confidence=conf, description=desc, weight=1.8, + confidence=conf, description=desc, weight=1.5, ) # Attach RSI series for divergence detection (dynamic attribute, filtered by _prefix) result._rsi_series = rsi # type: ignore[attr-defined] - def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis): - """MACD with histogram momentum analysis.""" + def _compute_macd_internal(self, df: pd.DataFrame, result: TechnicalAnalysis): + """Compute MACD histogram internally for divergence detection (not a signal pillar).""" c = self.config close = df["Close"] @@ -239,62 +238,6 @@ class TechnicalEngine: signal_line = macd_line.ewm(span=c.macd_signal, adjust=False).mean() histogram = macd_line - signal_line - macd_val = macd_line.iloc[-1] - signal_val = signal_line.iloc[-1] - hist_val = histogram.iloc[-1] - - # Check for crossover - crossover = False - crossover_type = "" - if len(macd_line) >= 2: - prev_macd = macd_line.iloc[-2] - prev_signal = signal_line.iloc[-2] - if prev_macd <= prev_signal and macd_val > signal_val: - crossover = True - crossover_type = "bullish" - elif prev_macd >= prev_signal and macd_val < signal_val: - crossover = True - crossover_type = "bearish" - - # Histogram momentum (is it expanding or contracting?) - hist_momentum = "" - if len(histogram) >= 3: - h1, h2, h3 = histogram.iloc[-3], histogram.iloc[-2], histogram.iloc[-1] - if h3 > h2 > h1 and h3 > 0: - hist_momentum = "accelerating bullish" - elif h3 < h2 < h1 and h3 < 0: - hist_momentum = "accelerating bearish" - elif abs(h3) < abs(h2) and h3 * h2 > 0: - hist_momentum = "decelerating" - - # Signal - if crossover and crossover_type == "bullish": - signal = SignalStrength.STRONG_BUY - conf = 75 - desc = f"MACD bullish crossover | Hist: {hist_val:.6f}" - elif crossover and crossover_type == "bearish": - signal = SignalStrength.STRONG_SELL - conf = 75 - desc = f"MACD bearish crossover | Hist: {hist_val:.6f}" - elif macd_val > signal_val and hist_val > 0: - signal = SignalStrength.BUY - conf = 55 - desc = f"MACD above signal, positive histogram | {hist_momentum}" - elif macd_val < signal_val and hist_val < 0: - signal = SignalStrength.SELL - conf = 55 - desc = f"MACD below signal, negative histogram | {hist_momentum}" - else: - signal = SignalStrength.NEUTRAL - conf = 35 - desc = f"MACD mixed | Hist: {hist_val:.6f}" - - result.indicators["macd"] = IndicatorSignal( - name="MACD", value=hist_val, signal=signal, - confidence=conf, description=desc, weight=1.2, - ) - - # Attach histogram for divergence detection result._macd_histogram = histogram # type: ignore[attr-defined] def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis): @@ -672,11 +615,39 @@ class TechnicalEngine: name="Divergence", value=len(divergences), signal=signal, confidence=70, description=" | ".join(divergences), - weight=3.5, + weight=1.0, ) # Note: divergences are NOT added to pattern_signals to avoid # duplicate reason collection (the indicator entry covers them). + def _compute_structure_sr(self, df: pd.DataFrame, result: TechnicalAnalysis): + """Detect swing high/low structure levels for S/R and trade setup.""" + lookback = min(self.config.swing_sr_lookback, len(df) - 1) + if lookback < 10: + return + + price = df["Close"].iloc[-1] + high_window = df["High"].iloc[-lookback:] + low_window = df["Low"].iloc[-lookback:] + + swing_highs = self._find_swing_points(high_window, mode="high", window=3) + swing_lows = self._find_swing_points(low_window, mode="low", window=3) + + highs_above = sorted([v for _, v in swing_highs if v > price]) + highs_below = sorted([v for _, v in swing_highs if v <= price], reverse=True) + lows_below = sorted([v for _, v in swing_lows if v < price], reverse=True) + lows_above = sorted([v for _, v in swing_lows if v >= price]) + + for i, val in enumerate(highs_above[:3], 1): + result.key_levels[f"swing_high_{i}"] = float(val) + for i, val in enumerate(lows_below[:3], 1): + result.key_levels[f"swing_low_{i}"] = float(val) + + if not highs_above and highs_below: + result.key_levels["swing_high_1"] = float(highs_below[0]) + if not lows_below and lows_above: + result.key_levels["swing_low_1"] = float(lows_above[0]) + def _detect_candlestick_patterns(self, df: pd.DataFrame, result: TechnicalAnalysis): """Detect significant candlestick patterns on recent bars.""" if len(df) < 5: @@ -738,11 +709,33 @@ class TechnicalEngine: else: signal = SignalStrength.NEUTRAL + conf = 50 + atr_val = result.key_levels.get("atr", 0) + latest = c + proximity = atr_val * 0.5 if atr_val > 0 else 0 + + if proximity > 0: + swing_lows = sorted( + [v for k, v in result.key_levels.items() if k.startswith("swing_low_")] + ) + swing_highs = sorted( + [v for k, v in result.key_levels.items() if k.startswith("swing_high_")] + ) + near_support = any(abs(latest - sl) <= proximity for sl in swing_lows) + near_resistance = any(abs(latest - sh) <= proximity for sh in swing_highs) + + if bullish > bearish and near_support: + conf += 20 + patterns.append("(at support)") + elif bearish > bullish and near_resistance: + conf += 20 + patterns.append("(at resistance)") + result.indicators["candlestick"] = IndicatorSignal( name="Candlestick", value=float(bullish - bearish), - signal=signal, confidence=50, + signal=signal, confidence=conf, description=" | ".join(patterns), - weight=1.2, + weight=1.5, ) result.pattern_signals.extend(patterns) @@ -854,10 +847,18 @@ class TechnicalEngine: total_weight = 0.0 total_confidence = 0.0 + _trending = { + "ema_trend": 2.0, "rsi": 1.5, "bollinger": 0.8, + "candlestick": 1.5, "divergence": 1.0, + } regime_weights = { - "trending_up": {"divergence": 1.5, "rsi": 1.2, "stochastic": 1.1, "candlestick": 1.2, "ema_trend": 1.0, "macd": 1.0, "adx": 1.0, "ichimoku": 0.8, "bollinger": 0.7, "pivot": 0.8}, - "trending_down": {"divergence": 1.5, "rsi": 1.2, "stochastic": 1.1, "candlestick": 1.2, "ema_trend": 1.0, "macd": 1.0, "adx": 1.0, "ichimoku": 0.8, "bollinger": 0.7, "pivot": 0.8}, - "ranging": {"divergence": 1.3, "rsi": 1.4, "stochastic": 1.4, "bollinger": 1.5, "candlestick": 1.3, "pivot": 1.3, "ema_trend": 0.5, "macd": 0.6, "adx": 0.4, "ichimoku": 0.2}, + "trending_up": _trending, + "trending_down": _trending, + "trending": _trending, + "ranging": { + "ema_trend": 0.5, "rsi": 2.0, "bollinger": 2.0, + "candlestick": 1.5, "divergence": 1.5, + }, } active_regime_weights = regime_weights.get(result.market_regime, {}) diff --git a/main.py b/main.py old mode 100644 new mode 100755 index 687f274..066988c --- a/main.py +++ b/main.py @@ -76,9 +76,8 @@ def print_header(title: str = "USD/JPY SIGNAL INTELLIGENCE"): # ── Live signal renderer ─────────────────────────────────────────── _INDICATOR_PREFIXES = ( - "EMA Trend", "RSI", "MACD", "Bollinger Bands", "Stochastic", - "ADX", "Ichimoku", "Divergence", "Candlestick", "Pivot Points", - "News", "Regime", + "EMA Trend", "RSI", "Bollinger Bands", "Divergence", + "Candlestick", "ADX", "ATR", "News", "Regime", ) @@ -170,13 +169,17 @@ def print_signal_report(report: SignalReport, detail: bool = False): levels = report.primary_analysis.key_levels if levels: - shown = [lvl for lvl in ["r3", "r2", "r1", "pivot", "s1", "s2", "s3"] if lvl in levels] - if shown: - print(f"\n {Colors.BOLD}Pivot Levels{Colors.RESET}") - for lvl in shown: - val = levels[lvl] + swing_keys = sorted( + [k for k in levels if k.startswith("swing_")], + key=lambda k: levels[k], reverse=True, + ) + if swing_keys: + print(f"\n {Colors.BOLD}Structure Levels{Colors.RESET}") + for k in swing_keys: + val = levels[k] color = Colors.RED if val > report.current_price else Colors.GREEN - print(f" {lvl.upper():>6} {color}{val:.3f}{Colors.RESET}") + label = k.replace("_", " ").title() + print(f" {label:>20} {color}{val:.3f}{Colors.RESET}") def report_to_dict(report: SignalReport) -> dict: diff --git a/news_engine.py b/news_engine.py old mode 100644 new mode 100755 index 49ab66f..e9266ec --- a/news_engine.py +++ b/news_engine.py @@ -104,6 +104,7 @@ class NewsEngine: "AUD": ["aussie", "rba", "australian", "australia"], "CAD": ["loonie", "boc", "canadian", "canada"], "NZD": ["kiwi", "rbnz", "new zealand"], + "XAU": ["gold", "xau", "precious metal", "safe haven", "bullion"], } # High-impact event types (these move markets significantly) @@ -187,15 +188,21 @@ class NewsEngine: return context + @staticmethod + def _parse_pair(pair_symbol: str) -> Tuple[str, str]: + if "/" in pair_symbol: + parts = pair_symbol.split("/") + return parts[0], parts[1] + clean = pair_symbol.replace("=X", "") + return clean[:3], clean[3:] + def get_event_risk_level(self, pair_symbol: str, context: FundamentalContext) -> Tuple[str, str]: """ Get event risk level for a specific pair. Returns (risk_level, description). risk_level: "clear", "caution", "danger" """ - clean = pair_symbol.replace("=X", "") - base = clean[:3] - quote = clean[3:] + base, quote = self._parse_pair(pair_symbol) # Check upcoming high-impact events for event in context.events_upcoming_1h: @@ -222,9 +229,7 @@ class NewsEngine: Returns (score, description) where score is -1 to +1. Positive = bullish for the pair, negative = bearish. """ - clean = pair_symbol.replace("=X", "") - base = clean[:3] - quote = clean[3:] + base, quote = self._parse_pair(pair_symbol) base_sent = context.news_sentiment.get(base, 0.0) quote_sent = context.news_sentiment.get(quote, 0.0) @@ -413,7 +418,11 @@ class NewsEngine: @staticmethod def _get_affected_pairs(currency: str) -> List[str]: - """Map a currency event to USD/JPY (the only watched pair).""" - if currency in ("USD", "JPY"): - return ["USDJPY=X"] - return [] + _MAP = { + "USD": ["EURUSD=X", "USDJPY=X", "GBPUSD=X", "XAU/USD"], + "EUR": ["EURUSD=X"], + "JPY": ["USDJPY=X"], + "GBP": ["GBPUSD=X"], + "XAU": ["XAU/USD"], + } + return _MAP.get(currency, []) diff --git a/signal_aggregator.py b/signal_aggregator.py index 7305205..5f53078 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -334,54 +334,44 @@ class SignalAggregator: score = report.primary_score confidence = report.primary_confidence - # MTF if report.mtf_agreement: - confidence = min(confidence + 15, 95) + confidence += 10 all_scores = list(report.mtf_scores.values()) if all_scores: mtf_avg = sum(all_scores) / len(all_scores) score = score * 0.6 + mtf_avg * 0.4 else: - confidence *= 0.7 + confidence -= 15 if len(report.mtf_scores) >= 2: scores = list(report.mtf_scores.values()) has_bull = any(s > 10 for s in scores) has_bear = any(s < -10 for s in scores) if has_bull and has_bear: report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes") - confidence *= 0.6 - # News sentiment if abs(report.news_sentiment) > 0.2: if (report.news_sentiment > 0 and score > 0) or \ (report.news_sentiment < 0 and score < 0): - confidence = min(confidence + 5, 95) + confidence += 5 elif (report.news_sentiment > 0 and score < -10) or \ (report.news_sentiment < 0 and score > 10): report.warnings.append("News sentiment CONFLICTS with technical signal") - confidence *= 0.85 + confidence -= 10 - # Event risk if report.event_risk == "danger": report.warnings.append(f"EVENT RISK: {report.event_description}") - confidence *= 0.3 + confidence = min(confidence, 30) report.action_recommendation = f"WAIT — {report.event_description}" elif report.event_risk == "caution": report.warnings.append(f"Upcoming event: {report.event_description}") - confidence *= 0.7 + confidence -= 15 - # Volatility regime if report.volatility_regime == "extreme": report.warnings.append("EXTREME volatility — wider stops needed, reduced position size") - confidence *= 0.8 + confidence -= 10 elif report.volatility_regime == "low": report.warnings.append("Low volatility — potential for breakout, tight ranges") - # EMA200 distance gate — empirically the strongest single discriminator - # across 2024 trend / 2023 chop / 2026 hostile windows. Sweet spot is - # 0.5–1.5% from EMA200: closer = false-breakout chop, farther = chase. - # Within band keeps confidence; outside band cuts it sharply so the - # trade falls below the B-grade threshold and is rejected. primary = report.primary_analysis if primary is not None: ema200 = primary.key_levels.get("ema_trend") @@ -389,17 +379,18 @@ class SignalAggregator: ema_dist_pct = abs(report.current_price - ema200) / ema200 * 100.0 _lo = float(os.environ.get("FB2_EMA_LO", "0.5")) _hi = float(os.environ.get("FB2_EMA_HI", "1.5")) - _mult = float(os.environ.get("FB2_EMA_MULT", "0.55")) if ema_dist_pct < _lo: report.warnings.append( - f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk, signal damped" + f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk" ) - confidence *= _mult + confidence -= 20 elif ema_dist_pct > _hi: report.warnings.append( - f"FAR FROM EMA200 ({ema_dist_pct:.2f}%) — over-extended, signal damped" + f"FAR FROM EMA200 ({ema_dist_pct:.2f}%) — over-extended" ) - confidence *= _mult + confidence -= 20 + + confidence = max(0.0, min(95.0, confidence)) # Final signal if score >= 40 and confidence >= 65: @@ -452,21 +443,57 @@ class SignalAggregator: return None is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) - rr = self.config.signals.default_rr_ratio - sl_mult = self.config.signals.atr_sl_multiplier - - sl_distance = atr * sl_mult - tp1_distance = sl_distance * rr - tp2_distance = sl_distance * rr * 1.5 + levels = report.primary_analysis.key_levels + min_rr = self.config.signals.min_rr_ratio if is_buy: - entry, sl = price, price - sl_distance - tp1, tp2 = price + tp1_distance, price + tp2_distance + swing_lows = sorted( + [v for k, v in levels.items() if k.startswith("swing_low_") and v < price], + reverse=True, + ) + if swing_lows: + sl_price = swing_lows[0] - atr * 0.5 + else: + sl_price = price - atr * 1.5 + if price - sl_price < atr * 1.0: + sl_price = price - atr * 1.0 + + swing_highs = sorted( + v for k, v in levels.items() if k.startswith("swing_high_") and v > price + atr * 0.3 + ) + tp1 = swing_highs[0] if len(swing_highs) >= 1 else price + atr * 2.0 + tp2 = swing_highs[1] if len(swing_highs) >= 2 else price + atr * 3.0 + + sl_distance = price - sl_price direction = "BUY" + entry, sl = price, sl_price else: - entry, sl = price, price + sl_distance - tp1, tp2 = price - tp1_distance, price - tp2_distance + swing_highs = sorted( + v for k, v in levels.items() if k.startswith("swing_high_") and v > price + ) + if swing_highs: + sl_price = swing_highs[0] + atr * 0.5 + else: + sl_price = price + atr * 1.5 + if sl_price - price < atr * 1.0: + sl_price = price + atr * 1.0 + + swing_lows = sorted( + [v for k, v in levels.items() if k.startswith("swing_low_") and v < price - atr * 0.3], + reverse=True, + ) + tp1 = swing_lows[0] if len(swing_lows) >= 1 else price - atr * 2.0 + tp2 = swing_lows[1] if len(swing_lows) >= 2 else price - atr * 3.0 + + sl_distance = sl_price - price direction = "SELL" + entry, sl = price, sl_price + + tp1_distance = abs(tp1 - entry) + rr = tp1_distance / sl_distance if sl_distance > 0 else 0 + + if rr < min_rr: + return None risk_pips = sl_distance / pip_value reward_pips = tp1_distance / pip_value @@ -479,7 +506,7 @@ class SignalAggregator: stop_loss=sl, take_profit_1=tp1, take_profit_2=tp2, - risk_reward=rr, + risk_reward=round(rr, 2), risk_pips=risk_pips, reward_pips=reward_pips, pip_value=pip_value, diff --git a/telegram_bot.py b/telegram_bot.py index d765770..e901606 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -2,19 +2,27 @@ """ ForexBot2 — Telegram Bot ======================== -Async Telegram interface with candle-boundary auto-scanner. +Async Telegram interface with candle-boundary auto-scanner and +real-time WebSocket level-breach monitor. -Manual scan: 4 pairs × 2 timeframes via inline buttons. +Manual scan: 4 pairs × 4 timeframes via inline buttons. Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set. /autoscan toggles for additional chats or to stop. + M5 fires at :00/:05/:10/… + 10s buffer. M15 fires at :00/:15/:30/:45 + 10s buffer. - H1 fires at :00 + 70s buffer (staggered to respect 8 credits/min). + H1 fires at :00 + 10s buffer. + H4 fires at :00 every 4 hours + 10s buffer. +WebSocket: Streams live ticks from TwelveData for all 4 pairs. + When price crosses a key structure level (swing S/R, BB, + EMA200) cached from the last boundary scan, triggers an + immediate M15 scan and sends a Level Break Alert. + 5-minute cooldown per pair prevents spam. -Credit budget (free tier): - M15 boundary: 4 pairs × 1 credit = 4 credits - H1 boundary: 4 pairs × 1 credit = 4 credits (staggered to next minute) - Top of hour: 4 + 4 = 8 across two minutes → ≤ 8 / min - Daily total: 384 (M15) + 96 (H1) = 480 / day (limit 800) +Plan: TwelveData Pro (55 credits/min, 1500 WS credits). +Credit budget (worst case, 4-hour boundary): + Unique fetches: 4 pairs × ~6 TFs = ~24 calls (rest cached) → fits in 55/min. + WebSocket: 4 symbols = 4 credits (one-time on connect). + Level breach scans: ~3 REST calls each, ~10/hour worst case. Usage: python telegram_bot.py @@ -22,12 +30,16 @@ Usage: """ import asyncio +import json import logging import os import sys +import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -from typing import List, Optional +from typing import Dict, List, Optional + +import websockets from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.constants import ParseMode @@ -40,7 +52,7 @@ from telegram.ext import ( from config import ( PlatformConfig, SignalStrength, Timeframe, - EURUSD, USDJPY, GBPUSD, USDCAD, PAIR_INFO, + EURUSD, USDJPY, GBPUSD, XAUUSD, PAIR_INFO, ) from data_provider import ForexDataProvider from signal_aggregator import SignalAggregator, SignalReport @@ -63,31 +75,60 @@ CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022") # ── Pairs & buttons ─────────────────────────────────────────────────── -SCAN_PAIRS = [EURUSD, USDJPY, GBPUSD, USDCAD] +SCAN_PAIRS = [EURUSD, USDJPY, GBPUSD, XAUUSD] + +CORRELATION_GROUPS = { + "usd_negative": {"pairs": {EURUSD, GBPUSD}, "label": "USD-weakness"}, + "risk_sentiment": {"pairs": {XAUUSD, USDJPY}, "label": "Risk-sentiment"}, +} + +_WS_TO_PAIR = { + "EUR/USD": EURUSD, + "GBP/USD": GBPUSD, + "USD/JPY": USDJPY, + "XAU/USD": XAUUSD, +} +_WS_SYMBOLS = list(_WS_TO_PAIR.keys()) + +_LEVEL_KEYS = ( + "swing_high_1", "swing_high_2", "swing_high_3", + "swing_low_1", "swing_low_2", "swing_low_3", + "bb_upper", "bb_lower", "ema_trend", +) SCAN_BUTTONS = [ + {"cb": "scan_eurusd_m5", "symbol": EURUSD, "tf": Timeframe.M5, "label": "EUR/USD 5M"}, {"cb": "scan_eurusd_m15", "symbol": EURUSD, "tf": Timeframe.M15, "label": "EUR/USD 15M"}, {"cb": "scan_eurusd_h1", "symbol": EURUSD, "tf": Timeframe.H1, "label": "EUR/USD 1H"}, + {"cb": "scan_eurusd_h4", "symbol": EURUSD, "tf": Timeframe.H4, "label": "EUR/USD 4H"}, + {"cb": "scan_usdjpy_m5", "symbol": USDJPY, "tf": Timeframe.M5, "label": "USD/JPY 5M"}, {"cb": "scan_usdjpy_m15", "symbol": USDJPY, "tf": Timeframe.M15, "label": "USD/JPY 15M"}, {"cb": "scan_usdjpy_h1", "symbol": USDJPY, "tf": Timeframe.H1, "label": "USD/JPY 1H"}, + {"cb": "scan_usdjpy_h4", "symbol": USDJPY, "tf": Timeframe.H4, "label": "USD/JPY 4H"}, + {"cb": "scan_gbpusd_m5", "symbol": GBPUSD, "tf": Timeframe.M5, "label": "GBP/USD 5M"}, {"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"}, {"cb": "scan_gbpusd_h1", "symbol": GBPUSD, "tf": Timeframe.H1, "label": "GBP/USD 1H"}, - {"cb": "scan_usdcad_m15", "symbol": USDCAD, "tf": Timeframe.M15, "label": "USD/CAD 15M"}, - {"cb": "scan_usdcad_h1", "symbol": USDCAD, "tf": Timeframe.H1, "label": "USD/CAD 1H"}, + {"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"}, + {"cb": "scan_xauusd_m5", "symbol": XAUUSD, "tf": Timeframe.M5, "label": "XAU/USD 5M"}, + {"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"}, + {"cb": "scan_xauusd_h1", "symbol": XAUUSD, "tf": Timeframe.H1, "label": "XAU/USD 1H"}, + {"cb": "scan_xauusd_h4", "symbol": XAUUSD, "tf": Timeframe.H4, "label": "XAU/USD 4H"}, ] SCAN_LOOKUP = {b["cb"]: b for b in SCAN_BUTTONS} -# Manual scans use richer confirmations (may spend extra cached credits) TF_CONFIRMATIONS = { + Timeframe.M5: [Timeframe.M15, Timeframe.H1], Timeframe.M15: [Timeframe.M5, Timeframe.H1], Timeframe.H1: [Timeframe.M15, Timeframe.H4], + Timeframe.H4: [Timeframe.H1, Timeframe.D1], } -# Auto-scan uses only cross-TF (already cached, zero extra credits) AUTOSCAN_CONFIRMATIONS = { - Timeframe.M15: [Timeframe.H1], - Timeframe.H1: [Timeframe.M15], + Timeframe.M5: [Timeframe.M15, Timeframe.H1], + Timeframe.M15: [Timeframe.M5, Timeframe.H1], + Timeframe.H1: [Timeframe.M15, Timeframe.H4], + Timeframe.H4: [Timeframe.H1, Timeframe.D1], } _executor = ThreadPoolExecutor(max_workers=2) @@ -209,6 +250,33 @@ def format_signal_message(report: SignalReport) -> str: return "\n".join(lines) +# ── Correlation filter ─────────────────────────────────────────────── + +def _detect_correlated_exposure(reports: List[SignalReport]) -> List[str]: + warnings = [] + for group_info in CORRELATION_GROUPS.values(): + group_pairs = group_info["pairs"] + group_label = group_info["label"] + active = [ + r for r in reports + if r.symbol in group_pairs + and r.final_signal not in (SignalStrength.NEUTRAL,) + ] + if len(active) >= 2: + directions = set() + for r in active: + if r.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): + directions.add("BUY") + else: + directions.add("SELL") + if len(directions) == 1: + names = ", ".join(r.display_name for r in active) + warnings.append( + f"⚠️ Same exposure ({group_label}): {names} both {directions.pop()}" + ) + return warnings + + # ── Auto-scan summary format ───────────────────────────────────────── def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str: @@ -235,6 +303,10 @@ def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str: else: lines.append("No actionable signals this cycle.") + corr_warnings = _detect_correlated_exposure(reports) + for w in corr_warnings: + lines.append(w) + return "\n".join(lines) @@ -242,10 +314,10 @@ def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str: def scan_keyboard() -> InlineKeyboardMarkup: rows = [] - for i in range(0, len(SCAN_BUTTONS), 2): + for i in range(0, len(SCAN_BUTTONS), 4): rows.append([ - InlineKeyboardButton(f"🔍 {b['label']}", callback_data=b["cb"]) - for b in SCAN_BUTTONS[i:i + 2] + InlineKeyboardButton(b["label"], callback_data=b["cb"]) + for b in SCAN_BUTTONS[i:i + 4] ]) return InlineKeyboardMarkup(rows) @@ -315,6 +387,8 @@ async def _scan_timeframe( if not reports: return + _update_level_cache(reports) + summary = _format_scan_summary(reports, tf) actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] @@ -328,60 +402,175 @@ async def _scan_timeframe( logger.exception(f"Failed to send to chat {chat_id}") -async def _m15_loop(bot) -> None: - """Sleep until each M15 candle close, then scan all pairs.""" +# ── WebSocket level-breach monitor ────────────────────────────────── + +_level_cache: Dict[str, Dict[str, float]] = {} +_last_prices: Dict[str, float] = {} +_scan_cooldowns: Dict[str, float] = {} + +_cfg = PlatformConfig() +_WS_API_KEY = os.environ.get("TWELVE_DATA_API_KEY") or _cfg.twelve_data_api_key +_WS_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={_WS_API_KEY}" +_WS_COOLDOWN = _cfg.ws_cooldown_seconds + + +def _update_level_cache(reports: List[SignalReport]) -> None: + for r in reports: + if r.primary_analysis and r.primary_analysis.key_levels: + levels = { + k: v for k, v in r.primary_analysis.key_levels.items() + if k in _LEVEL_KEYS + } + if levels: + _level_cache[r.symbol] = levels + + +def _check_level_breach(symbol: str, price: float) -> Optional[str]: + levels = _level_cache.get(symbol) + if not levels: + return None + + last = _last_prices.get(symbol) + if last is None: + return None + + for name, level in levels.items(): + if (last < level <= price) or (last > level >= price): + return name + + return None + + +async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, level_name: str) -> None: + now = time.time() + if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN: + return + + _scan_cooldowns[pair_symbol] = now + display = PAIR_INFO.get(pair_symbol, {}).get("display", ws_symbol) + label = level_name.replace("_", " ").title() + logger.info(f"Level breach: {display} crossed {label} at {price}") + + try: + report = await run_signal(pair_symbol, Timeframe.M15) + _update_level_cache([report]) + text = ( + f"🔔 *Level Break — {display}*\n" + f"Price crossed *{label}* at `{price}`\n" + f"━━━━━━━━━━━━━━━━━━━━\n" + + format_signal_message(report) + ) + except Exception as exc: + logger.exception(f"Breach scan failed for {display}") + text = f"🔔 *Level Break — {display}*\nPrice crossed {label} at `{price}`\n❌ Scan failed: {exc}" + + for chat_id in list(_autoscan_chats): + try: + await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN) + except Exception: + logger.exception(f"Failed to send breach alert to {chat_id}") + + +async def _ws_monitor(bot) -> None: + backoff = 1 + while _autoscan_chats: + try: + async with websockets.connect(_WS_URL) as ws: + subscribe = { + "action": "subscribe", + "params": {"symbols": ",".join(_WS_SYMBOLS)}, + } + await ws.send(json.dumps(subscribe)) + logger.info(f"WebSocket connected, subscribed to {len(_WS_SYMBOLS)} symbols") + backoff = 1 + + async for raw in ws: + if not _autoscan_chats: + break + + msg = json.loads(raw) + if msg.get("event") != "price": + continue + + ws_symbol = msg.get("symbol") + pair_symbol = _WS_TO_PAIR.get(ws_symbol) + if not pair_symbol: + continue + + price = float(msg["price"]) + breached = _check_level_breach(pair_symbol, price) + _last_prices[pair_symbol] = price + + if breached: + asyncio.create_task( + _handle_breach(bot, ws_symbol, pair_symbol, price, breached) + ) + + except asyncio.CancelledError: + break + except Exception: + logger.exception(f"WebSocket error, reconnecting in {backoff}s") + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 60) + + logger.info("WebSocket monitor stopped") + + +async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None: + label = tf.value try: while _autoscan_chats: - wait = _seconds_until_boundary(15, buffer=10) - logger.info(f"M15 auto-scan: next in {wait:.0f}s") + wait = _seconds_until_boundary(interval_minutes, buffer=10) + logger.info(f"{label} auto-scan: next in {wait:.0f}s") await asyncio.sleep(wait) if not _autoscan_chats: break - logger.info("M15 auto-scan: firing") + logger.info(f"{label} auto-scan: firing") await _scan_timeframe( - bot, Timeframe.M15, AUTOSCAN_CONFIRMATIONS[Timeframe.M15] + bot, tf, AUTOSCAN_CONFIRMATIONS[tf] ) except asyncio.CancelledError: pass finally: - logger.info("M15 auto-scan loop stopped") + logger.info(f"{label} auto-scan loop stopped") -async def _h1_loop(bot) -> None: - """Sleep until each H1 candle close + 60s stagger, then scan all pairs.""" - try: - while _autoscan_chats: - wait = _seconds_until_boundary(60, buffer=70) - logger.info(f"H1 auto-scan: next in {wait:.0f}s") - await asyncio.sleep(wait) - if not _autoscan_chats: - break - logger.info("H1 auto-scan: firing") - await _scan_timeframe( - bot, Timeframe.H1, AUTOSCAN_CONFIRMATIONS[Timeframe.H1] - ) - except asyncio.CancelledError: - pass - finally: - logger.info("H1 auto-scan loop stopped") +_AUTOSCAN_LOOPS = [ + (Timeframe.M5, 5), + (Timeframe.M15, 15), + (Timeframe.H1, 60), + (Timeframe.H4, 240), +] + + +_ws_task: Optional[asyncio.Task] = None def _start_scheduler(bot) -> None: - """Start the M15 and H1 scanning loops if not already running.""" + global _ws_task alive = [t for t in _autoscan_tasks if not t.done()] if not alive: _autoscan_tasks.clear() - _autoscan_tasks.append(asyncio.create_task(_m15_loop(bot))) - _autoscan_tasks.append(asyncio.create_task(_h1_loop(bot))) - logger.info("Candle-boundary scheduler started") + for tf, interval in _AUTOSCAN_LOOPS: + _autoscan_tasks.append( + asyncio.create_task(_candle_loop(bot, tf, interval)) + ) + logger.info("Candle-boundary scheduler started (M5/M15/H1/H4)") + + if _ws_task is None or _ws_task.done(): + _ws_task = asyncio.create_task(_ws_monitor(bot)) + logger.info("WebSocket level-breach monitor started") def _stop_scheduler() -> None: - """Cancel all scanning loops.""" + global _ws_task for t in _autoscan_tasks: t.cancel() _autoscan_tasks.clear() - logger.info("Candle-boundary scheduler stopped") + if _ws_task and not _ws_task.done(): + _ws_task.cancel() + _ws_task = None + logger.info("Candle-boundary scheduler and WebSocket monitor stopped") # ── Handlers ────────────────────────────────────────────────────────── @@ -417,13 +606,16 @@ async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No _autoscan_chats.add(chat_id) _start_scheduler(context.bot) + next_m5 = _seconds_until_boundary(5, buffer=10) next_m15 = _seconds_until_boundary(15, buffer=10) - next_h1 = _seconds_until_boundary(60, buffer=70) + next_h1 = _seconds_until_boundary(60, buffer=10) + next_h4 = _seconds_until_boundary(240, buffer=10) await update.message.reply_text( "✅ *Auto-scan enabled*\n\n" - f"Scanning {len(SCAN_PAIRS)} pairs on M15 + H1 at candle close.\n" - f"Next M15 scan in {next_m15 / 60:.0f} min\n" - f"Next H1 scan in {next_h1 / 60:.0f} min\n\n" + f"Scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4 at candle close.\n" + f"📡 Live price monitoring active — level-break alerts enabled.\n" + f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | " + f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n" "Use /autoscan again to stop.", parse_mode=ParseMode.MARKDOWN, ) @@ -470,20 +662,23 @@ async def _post_init(app: Application) -> None: _autoscan_chats.add(chat_id) _start_scheduler(app.bot) + next_m5 = _seconds_until_boundary(5, buffer=10) next_m15 = _seconds_until_boundary(15, buffer=10) - next_h1 = _seconds_until_boundary(60, buffer=70) + next_h1 = _seconds_until_boundary(60, buffer=10) + next_h4 = _seconds_until_boundary(240, buffer=10) logger.info( f"Auto-scan started for chat {chat_id} — " - f"M15 in {next_m15:.0f}s, H1 in {next_h1:.0f}s" + f"M5 {next_m5:.0f}s, M15 {next_m15:.0f}s, H1 {next_h1:.0f}s, H4 {next_h4:.0f}s" ) try: await app.bot.send_message( chat_id, f"🤖 *ForexBot2 started*\n\n" - f"Auto-scanning {len(SCAN_PAIRS)} pairs on M15 + H1.\n" - f"Next M15 in {next_m15 / 60:.0f} min | " - f"Next H1 in {next_h1 / 60:.0f} min\n\n" + f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n" + f"📡 Live price monitoring active — level-break alerts enabled.\n" + f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | " + f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n" f"/autoscan to stop | Manual buttons via /start", parse_mode=ParseMode.MARKDOWN, ) diff --git a/ws_twelvedata.py b/ws_twelvedata.py old mode 100644 new mode 100755 index 5f10b0c..a931111 --- a/ws_twelvedata.py +++ b/ws_twelvedata.py @@ -8,7 +8,7 @@ from datetime import datetime, timezone API_KEY = "f563b22ffdfd4924955decbe1a8c8701" WS_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={API_KEY}" -SYMBOLS = ["EUR/USD", "GBP/USD", "USD/JPY"] +SYMBOLS = ["EUR/USD", "GBP/USD", "USD/JPY", "XAU/USD"] async def stream_ticks():