""" Web-dashboard publisher — pushes ticks and signal reports to the LAMP webapp's ingest endpoint (webapp/ingest.php), which stores them in MySQL. Disabled unless BOTH env vars are set: WEBAPP_INGEST_URL e.g. https://example.com/webapp/ingest.php WEBAPP_INGEST_TOKEN shared secret (must match webapp/config.php) All publishing is fire-and-forget through a bounded queue drained by a daemon thread: the bot's WebSocket loop and scan executor are never blocked by a slow or dead web server. Failures are dropped with a rate-limited warning. CONTRACT: the public functions (publish_tick, publish_report) NEVER raise — any internal error is swallowed and logged. The Telegram bot calls them from its WS tick loop and scan paths without guards; a dashboard/DB/web outage must never be able to break the bot. `build_payload()` is pure (no I/O) — it converts a SignalReport into the JSON the dashboard renders, including the plain-English translation of every indicator, warning, and the overall verdict. """ import json import logging import os import queue import threading import time from typing import Any, Dict, Optional import requests from config import SignalStrength from signal_aggregator import SignalReport logger = logging.getLogger(__name__) INGEST_URL = os.environ.get("WEBAPP_INGEST_URL", "") INGEST_TOKEN = os.environ.get("WEBAPP_INGEST_TOKEN", "") TICK_MIN_INTERVAL = 1.0 # seconds between published ticks per symbol _QUEUE_MAX = 200 _WARN_EVERY = 60.0 # rate-limit failure warnings _queue: "queue.Queue[dict]" = queue.Queue(maxsize=_QUEUE_MAX) _worker_started = False _worker_lock = threading.Lock() _last_tick_pub: Dict[str, float] = {} _last_warn_ts = 0.0 def enabled() -> bool: return bool(INGEST_URL and INGEST_TOKEN) # ── Transport ──────────────────────────────────────────────────────── def _ensure_worker() -> None: global _worker_started if _worker_started: return with _worker_lock: if _worker_started: return t = threading.Thread(target=_drain, name="web-publisher", daemon=True) t.start() _worker_started = True def _warn_limited(msg: str) -> None: """Rate-limited warning (max one per _WARN_EVERY seconds).""" global _last_warn_ts now = time.time() if now - _last_warn_ts > _WARN_EVERY: _last_warn_ts = now logger.warning(msg) def _drain() -> None: """Worker loop. Hardened so the thread itself can never die: every step is guarded, and task_done() always runs so queue.join() can't hang.""" while True: try: item = _queue.get() except Exception: time.sleep(1.0) continue try: resp = requests.post(INGEST_URL, json=item, timeout=5) if resp.status_code != 200: raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:200]}") except Exception as exc: _warn_limited(f"Web publish failed ({item.get('type')}): {exc}") finally: try: _queue.task_done() except Exception: pass def _enqueue(item: dict) -> None: _ensure_worker() try: _queue.put_nowait(item) except queue.Full: pass # dashboard data is best-effort; never back-pressure the bot # ── Public API ─────────────────────────────────────────────────────── def publish_tick(symbol: str, price: float, prev: Optional[float] = None) -> None: """Publish a live tick (throttled to 1/s per symbol). No-op unless configured. Never raises (see module contract).""" if not enabled(): return try: now = time.time() if now - _last_tick_pub.get(symbol, 0.0) < TICK_MIN_INTERVAL: return _last_tick_pub[symbol] = now _enqueue({ "token": INGEST_TOKEN, "type": "tick", "symbol": symbol, "price": float(price), "prev": float(prev) if prev is not None else None, "tick_at": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(now)), }) except Exception: _warn_limited("publish_tick failed — tick dropped (bot unaffected)") def publish_report(report: SignalReport) -> None: """Publish a full signal report. No-op unless configured. Never raises (see module contract).""" if not enabled(): return try: payload = build_payload(report) _enqueue({ "token": INGEST_TOKEN, "type": "report", # display_name is the canonical "XAU/USD" form in both the # live path and backtest-style replays; report.symbol is not. "symbol": report.display_name or report.symbol, "timeframe": payload["timeframe"], "bar_ts": payload["bar_ts"], "signal": payload["signal"], "quality": payload["quality"] or "F", "score": payload["score"], "confidence": payload["confidence"], "payload": payload, }) except Exception: _warn_limited("publish_report failed — report dropped (bot unaffected)") # ── Payload builder (pure) ─────────────────────────────────────────── def build_payload(report: SignalReport) -> Dict[str, Any]: """SignalReport → dashboard JSON, with plain-English throughout.""" pa = report.primary_analysis tf = pa.timeframe if pa else "?" indicators: Dict[str, Any] = {} if pa: for name, ind in pa.indicators.items(): if name.startswith("_"): continue indicators[name] = { "value": _num(ind.value), "signal": ind.signal.name, "confidence": round(float(ind.confidence), 1), "description": ind.description, "plain": _plain_indicator(name, ind, report), } d: Dict[str, Any] = { "v": 1, "symbol": report.symbol, "display_name": report.display_name, "timeframe": tf, "bar_ts": _bar_ts(report.timestamp), "price": _num(report.current_price), "signal": report.final_signal.name, "score": round(float(report.final_score), 2), "confidence": round(float(report.final_confidence), 2), "quality": report.signal_quality, "market_regime": report.market_regime, "volatility_regime": report.volatility_regime, "regime_plain": _plain_regime(report.market_regime, report.volatility_regime), "mtf_agreement": bool(report.mtf_agreement), "mtf_signals": {k: v.name for k, v in report.mtf_signals.items()}, "mtf_plain": _plain_mtf(report), "summary": _summary(report), "warnings": list(report.warnings), "warnings_plain": [_plain_warning(w) for w in report.warnings], "indicators": indicators, "generated_at": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()), } if report.trade_setup: ts = report.trade_setup d["trade_setup"] = { "direction": ts.direction, "strategy": ts.strategy, "entry": _num(ts.entry_price), "stop_loss": _num(ts.stop_loss), "take_profit_1": _num(ts.take_profit_1), "take_profit_2": _num(ts.take_profit_2), "risk_reward": _num(ts.risk_reward), "risk_usd": _num(abs(ts.entry_price - ts.stop_loss)), } else: d["trade_setup"] = None return d def _num(v) -> Optional[float]: try: f = float(v) except (TypeError, ValueError): return None return round(f, 5) def _bar_ts(ts: str) -> str: """Pandas index label ('2026-07-17 14:45:00+00:00') → MySQL DATETIME.""" s = str(ts).replace("T", " ") s = s.split("+")[0].split(".")[0].strip() return s[:19] # ── Plain-English layer ────────────────────────────────────────────── _BUYISH = (SignalStrength.BUY, SignalStrength.STRONG_BUY) _SELLISH = (SignalStrength.SELL, SignalStrength.STRONG_SELL) def _plain_indicator(name: str, ind, report: SignalReport) -> str: sig = ind.signal desc = ind.description or "" low = desc.lower() v = None try: v = float(ind.value) except (TypeError, ValueError): pass if name == "ema_trend": if sig in _BUYISH: txt = "Price is trading above its stacked moving averages — the short-term trend points up." elif sig in _SELLISH: txt = "Price is trading below its stacked moving averages — the short-term trend points down." else: txt = "The moving averages are tangled together — no clear trend direction right now." if "compressed" in low: txt += " The averages are squeezed tight — often the calm before a bigger move." return txt if name == "rsi": r = v if v is not None else 50.0 base = f"RSI is {r:.0f}. " if r >= 70: zone = "Price has climbed unusually fast — overbought territory, where pullbacks are common." elif r >= 60: zone = "Buyers currently have the momentum." elif r > 40: zone = "Momentum is balanced — neither buyers nor sellers dominate." elif r > 30: zone = "Sellers currently have the momentum." else: zone = "Price has dropped unusually fast — oversold territory, where bounces are common." if "pullback zone" in low: zone += " In this trend, that dip is exactly what the engine looks to buy into." \ if sig in _BUYISH else \ " In this downtrend, that bounce is exactly what the engine looks to sell into." return base + zone if name == "macd": if sig == SignalStrength.STRONG_BUY: return "Upward momentum is building and accelerating." if sig == SignalStrength.BUY: return "Momentum has just turned upward." if sig == SignalStrength.SELL: return "Momentum has just turned downward." if sig == SignalStrength.STRONG_SELL: return "Downward momentum is building and accelerating." if "fading" in low: return "The most recent push is losing steam." return "Momentum is flat — no real push in either direction." if name == "bollinger": p = v if v is not None else 0.5 if p < 0: txt = "Price has broken below its typical range — unusually stretched to the downside." elif p < 0.2: txt = "Price is near the bottom of its recent range." elif p > 1: txt = "Price has broken above its typical range — unusually stretched to the upside." elif p > 0.8: txt = "Price is near the top of its recent range." else: txt = "Price is inside its normal range — nothing unusual." if "squeeze" in low: txt += " The range is tightening — bigger moves often follow." return txt if name == "adx": a = v if v is not None else 0.0 if a < 20: txt = "No meaningful trend — the market is drifting." elif a < 25: txt = "A trend may be forming, but it is not confirmed yet." elif a < 50: txt = "A confirmed trend is in force." else: txt = "A very strong trend is in force." if sig in _BUYISH: txt += " Buyers are the stronger side." elif sig in _SELLISH: txt += " Sellers are the stronger side." if "emerging" in low: txt += " Trend strength is rising quickly — a new trend may be igniting." return txt if name == "roc": r = v if v is not None else 0.0 txt = f"Price is {r:+.2f}% versus 10 bars ago" if sig in _BUYISH: return txt + " — momentum points up." if sig in _SELLISH: return txt + " — momentum points down." return txt + " — momentum is flat." if name == "keltner": if "squeeze" in low and ("released" in low or "fired" in low): side = "upward" if sig in _BUYISH else "downward" if sig in _SELLISH else "one" return f"A volatility squeeze just released with {side} force — these bursts often start real moves." if "squeeze" in low: return "Volatility is coiled tight — a bigger move often follows a squeeze like this." if sig in _BUYISH: return "Price is riding above its volatility channel — a sign the up-move wants to continue." if sig in _SELLISH: return "Price is pressing below its volatility channel — a sign the down-move wants to continue." return "Price is inside its normal volatility channel — nothing unusual." if name == "divergence": if sig in _BUYISH: return ("Price made a lower low but momentum did not — selling pressure " "may be quietly fading (bullish divergence).") if sig in _SELLISH: return ("Price made a higher high but momentum did not — buying pressure " "may be quietly fading (bearish divergence).") n = int(v) if v is not None else 0 if n > 0: return "Both bullish and bearish divergences are present — a mixed message." return "Price and momentum agree — no hidden warning signs." if name == "candlestick": if sig == SignalStrength.NEUTRAL and ("no " in low or not desc): return "No notable candle pattern on the last bar." hint = ("a bullish hint" if sig in _BUYISH else "a bearish hint" if sig in _SELLISH else "an indecision signal") return f"{desc} — {hint} from the last candle's shape." if name == "atr": a = v if v is not None else 0.0 vol = { "low": "Very quiet market.", "normal": "Normal volatility.", "high": "Livelier than usual — swings are wider.", "extreme": "Unusually wild — moves are far larger than normal.", }.get(report.volatility_regime, "") return f"Bars are moving about ${a:.2f} on average. {vol}".strip() return desc or "—" _WARNING_PLAIN = [ ("TREND EXHAUSTION", "Price has already travelled a long way without a rest — chasing here is risky."), ("EXHAUSTED TREND", "The trend looks exhausted — late entries are blocked."), ("EXTREME VOLATILITY", "The market is unusually violent right now — stops must be much wider."), ("EXTREME", "The market is unusually volatile right now."), ("LOW VOLATILITY", "Very quiet market — a breakout may be brewing."), ("SESSION FILTER", "Outside the strategy's approved trading hours — no trades are taken."), ("DAY FILTER", "Trading is disabled on this weekday."), ("CONFLICT", "The timeframes disagree with each other — a mixed picture."), ("SELL FILTER", "A safety rule is blocking short trades here."), ("SELL SUPPRESSED", "A safety rule is blocking short trades here."), ("M5 FILTER", "An M5-specific noise filter blocked this signal."), ("NO PULLBACK", "Price is far from its average — better to wait for a pullback than chase."), ("WEAK PULLBACK", "The entry is slightly extended past the ideal pullback."), ("NEAR EMA200", "Price is sitting at a major long-term average, where moves often stall or reverse."), ("FAR FROM EMA200", "Price is unusually far above/below its long-term average."), ("COUNTER-TREND", "This signal points against the bigger trend."), ("EVENT RISK", "A high-impact news event is near — expect sudden spikes."), ("STALE", "Market data is stale — the signal is suppressed."), ("RSI", "Momentum is already stretched in this direction."), ("BB ", "Price is pressed against the edge of its range — a poor entry point."), ] def _plain_warning(w: str) -> str: up = w.upper() for key, plain in _WARNING_PLAIN: if key in up: return plain return w _REGIME_PLAIN = { "trending_up": "a clear uptrend is in force", "trending_down": "a clear downtrend is in force", "trending": "there is strong momentum but its direction is unclear (mixed)", "transitioning": "a new trend may be starting", "ranging": "the market is moving sideways with no trend", "unknown": "the market state is unclear", } _VOL_PLAIN = { "low": "very quiet", "normal": "normal", "high": "livelier than usual", "extreme": "unusually wild", } def _plain_regime(regime: str, vol: str) -> str: r = _REGIME_PLAIN.get(regime, regime) v = _VOL_PLAIN.get(vol, vol) return f"Right now {r}, and volatility is {v}." def _plain_mtf(report: SignalReport) -> str: if not report.mtf_signals: return "" parts = [f"{tf} says {sig.name.replace('_', ' ')}" for tf, sig in report.mtf_signals.items()] txt = "Other timeframes: " + ", ".join(parts) + "." if report.final_signal != SignalStrength.NEUTRAL: txt += (" They back the main signal." if report.mtf_agreement else " They do not line up with the main signal.") return txt _SIG_PHRASE = { "STRONG_BUY": "Strong BUY signal", "BUY": "BUY signal", "NEUTRAL": "No trade", "SELL": "SELL signal", "STRONG_SELL": "Strong SELL signal", } def _summary(report: SignalReport) -> str: sig = report.final_signal.name phrase = _SIG_PHRASE.get(sig, sig) score = float(report.final_score) conf = float(report.final_confidence) q = report.signal_quality or "F" if report.trade_setup: ts = report.trade_setup risk = abs(ts.entry_price - ts.stop_loss) head = (f"{phrase} — grade {q}. {ts.direction} near {ts.entry_price:.2f}, " f"stop {ts.stop_loss:.2f} (${risk:.2f} risk), targets " f"{ts.take_profit_1:.2f} and {ts.take_profit_2:.2f}.") elif sig == "NEUTRAL": lean = ("leans bullish" if score >= 10 else "leans bearish" if score <= -10 else "is flat") head = (f"No trade. The overall read {lean} " f"(score {score:+.0f} on a ±100 scale, confidence {conf:.0f}/100) — " f"the engine needs at least ±25 with confidence 50 before acting. Grade {q}.") else: head = (f"{phrase} (grade {q}, score {score:+.0f}, confidence {conf:.0f}), " f"but no trade setup passed the risk checks.") return head