""" ForexBot2 — AI market outlook (advisory hourly assessment) =========================================================== Builds a broad XAU/USD market brief and asks DeepSeek whether a good setup appears to be forming. The result is written to the signal-history database and may be displayed by consumers, but it is ADVISORY-ONLY: it never changes signal scoring, gates, grades, or trade setup generation. The outlook runs outside the replayable signal path. Individual candle/news failures produce a partial brief; an unavailable LLM, malformed reply, or locked database never reaches a scan caller as an exception. The static prompt is intentional: DeepSeek prices byte-identical prompt prefixes at its cache-hit rate, so timestamps and all other changing evidence live only in the JSON user payload. """ import datetime as _dt import logging import math import os import sqlite3 import threading import time from dataclasses import dataclass from typing import Any, Dict, List, Optional from agent_panel import _normalize_candles from app_config import (get_app_config, market_outlook_enabled, signal_log_path) from config import PlatformConfig, Timeframe, XAUUSD from engine import TechnicalEngine from llm_client import get_llm_client from providers import base as _provider_base logger = logging.getLogger(__name__) DEFAULT_MAX_PER_DAY = 12 DEFAULT_CANDLE_BARS = 20 _OUTLOOK_TFS = (Timeframe.M15, Timeframe.H1, Timeframe.H4) _VALID_BIASES = ("bullish", "bearish", "neutral") _VALID_SETUPS = ("buy", "sell", "none") # STATIC prompt: repeated prompt prefixes are much cheaper with DeepSeek's # server-side cache. Variable market evidence belongs exclusively in payload. SYSTEM_PROMPT = """\ You are writing an advisory market outlook for gold (XAU/USD). You receive a JSON market brief covering recent candle shape, indicator states, price levels, news, and recent technical signals. Assess what may be forming soon; you are not a signal generator and must not provide an entry price, stop, or target. Gold doctrine: hawkish Fed / rate hikes / rising real yields / strong dollar are gold-bearish; easing / rate cuts / safe-haven flows / geopolitical stress / weak dollar are gold-bullish. Indicator "value" units: rsi 0-100 (above 70 overbought, below 30 oversold); adx 0-100 (above 25 = strong trend); macd = histogram in dollars (momentum); roc = percent; bollinger = position 0-1 (0 = lower band, 1 = upper band); keltner = dollar distance from the mid-line; ema_trend = 0-1 alignment score; divergence and candlestick = pattern counts. "candles" is per timeframe: {"ref_close", "atr", "bars"}; bars are oldest-first [open, high, low, close] in ATR units relative to the latest close. Read their shape, never quote them as prices. Hard rules: use ONLY the JSON provided. Do not invent numbers, events, or price levels. Your training data about gold, markets, and current events is stale and OFF-LIMITS: never use outside knowledge; if a fact is not in the JSON, treat it as unknown. If evidence is thin or missing, answer "neutral" with confidence below 30. Use plain everyday English for someone unfamiliar with markets: short simple sentences, no acronyms or jargon. Keep "watch" and "note" to at most 25 words each. Reply with ONLY JSON: {"bias": "bullish"|"bearish"|"neutral", "setup_forming": "buy"|"sell"|"none", "confidence": 0-100, "watch": "", "note": ""} """ @dataclass class OutlookResult: bias: str setup_forming: str confidence: float watch: str note: str created_ts: float def _finite(x: Any) -> Optional[float]: """float(x) or None for non-numeric / non-finite JSON values.""" try: value = float(x) except (TypeError, ValueError): return None return value if math.isfinite(value) else None def _round_finite(x: Any, digits: int) -> Optional[float]: value = _finite(x) return round(value, digits) if value is not None else None def _json_safe(value: Any) -> Any: """Replace any accidental NaN/inf nested in provider data with null.""" if isinstance(value, dict): return {str(key): _json_safe(item) for key, item in value.items()} if isinstance(value, (list, tuple)): return [_json_safe(item) for item in value] if isinstance(value, bool) or value is None or isinstance(value, str): return value if isinstance(value, (int, float)): return _finite(value) return value def _clamp_confidence(raw: Any) -> Optional[float]: value = _finite(raw) return max(0.0, min(100.0, value)) if value is not None else None def _word_limit(text: Any, words: int = 25) -> str: """Bound a response field by words as well as a practical char limit.""" clean = str(text or "").strip() return " ".join(clean.split()[:words])[:200] def _validate_outlook(reply: Any) -> Optional[OutlookResult]: """Validate only model replies that fully match the requested contract.""" if not isinstance(reply, dict): return None bias = reply.get("bias") setup_forming = reply.get("setup_forming") confidence = _clamp_confidence(reply.get("confidence")) if (bias not in _VALID_BIASES or setup_forming not in _VALID_SETUPS or confidence is None): return None return OutlookResult( bias=bias, setup_forming=setup_forming, confidence=confidence, watch=_word_limit(reply.get("watch")), note=_word_limit(reply.get("note")), created_ts=time.time(), ) # ── Data collection ────────────────────────────────────────────────── _dp: Any = None # Patchable singleton, mirroring agent_panel._dp. def _outlook_provider() -> Any: """Lazy provider singleton so its closed-bar cache stays warm hourly.""" global _dp if _dp is None: from providers import make_data_provider _dp = make_data_provider(PlatformConfig()) return _dp def _indicator_payload(analysis: Any) -> Dict[str, Dict[str, Any]]: """Visible scored indicators with finite, compact values only.""" out: Dict[str, Dict[str, Any]] = {} for name, indicator in (getattr(analysis, "indicators", None) or {}).items(): if name.startswith("_") or getattr(indicator, "weight", 0) == 0: continue entry: Dict[str, Any] = { "signal": getattr(getattr(indicator, "signal", None), "name", ""), "confidence": _round_finite( getattr(indicator, "confidence", None), 1), "value": _round_finite(getattr(indicator, "value", None), 2), } out[name] = entry return out def _analysis_payload(analyses: Dict[str, Any]) -> Dict[str, Any]: """Pure formatter for the technical portion of the market brief.""" indicators: Dict[str, Any] = {} market_regime: Dict[str, Optional[str]] = {} volatility_regime: Dict[str, Optional[str]] = {} key_levels: Dict[str, Dict[str, Optional[float]]] = {} for tf in _OUTLOOK_TFS: name = tf.value analysis = analyses.get(name) indicators[name] = _indicator_payload(analysis) market_regime[name] = getattr(analysis, "market_regime", None) volatility_regime[name] = getattr(analysis, "volatility_regime", None) key_levels[name] = { level: _round_finite(value, 2) for level, value in (getattr(analysis, "key_levels", None) or {}).items() } return { "indicators": indicators, "market_regime": market_regime, "volatility_regime": volatility_regime, "key_levels": key_levels, } def _empty_news_payload() -> Dict[str, Any]: return { "headlines": None, "events": None, "gold_sentiment": None, "remaining_high_impact_events": None, } def _news_payload() -> Dict[str, Any]: """Collect current news without allowing any news failure to block us.""" try: from news_engine import NewsEngine news = NewsEngine(PlatformConfig().news) context = news.get_fundamental_context() headlines = [str(getattr(h, "headline", "")) for h in (context.news_headlines or [])[:10] if getattr(h, "headline", None)] events = [ { "event": getattr(event, "event", None), "currency": getattr(event, "currency", None), "actual": getattr(event, "actual", None), "forecast": getattr(event, "forecast", None), "released": not event.is_upcoming, } for event in (context.events_today or []) if event.is_high_impact ][:8] gold_sentiment, _ = news.get_pair_sentiment(XAUUSD, context) remaining = sum( 1 for event in news.get_week_events() if event.is_high_impact and event.is_upcoming ) return { "headlines": headlines, "events": events, "gold_sentiment": _round_finite(gold_sentiment, 2), "remaining_high_impact_events": remaining, } except Exception as exc: logger.debug("market outlook: news context unavailable: %s", exc) return _empty_news_payload() def _recent_signals() -> List[Dict[str, Any]]: """Read the latest signal history through SQLite's immutable URI.""" try: path = os.path.abspath(signal_log_path()) # mode=ro (not immutable=1): immutable ignores the WAL, so the # newest rows written by the live loggers were invisible here. uri = f"file:{path}?mode=ro" con = sqlite3.connect(uri, uri=True) try: rows = con.execute( "SELECT logged_at, timeframe, signal, quality, panel_verdict " "FROM signals ORDER BY id DESC LIMIT 10").fetchall() finally: con.close() return [ { "logged_at": row[0], "timeframe": row[1], "signal": row[2], "quality": row[3], "panel_verdict": row[4], } for row in reversed(rows) ] except Exception as exc: logger.debug("market outlook: signal history unavailable: %s", exc) return [] def _build_payload( analyses: Dict[str, Any], candles: Dict[str, Any], news: Dict[str, Any], recent_signals: List[Dict[str, Any]], as_of: _dt.datetime, ) -> Dict[str, Any]: """Pure market-brief formatter. Every numeric field is finite or null.""" technical = _analysis_payload(analyses) return _json_safe({ "as_of": as_of.astimezone(_dt.timezone.utc).isoformat(), "candles": candles, **technical, "news": news, "recent_signals": recent_signals, }) def _collect_payload(settings: Dict[str, Any]) -> Dict[str, Any]: """Fetch closed bars and turn them into a structured market brief.""" analyses: Dict[str, Any] = {} candles: Dict[str, Any] = {tf.value: None for tf in _OUTLOOK_TFS} bars = int(settings.get("candles_bars", DEFAULT_CANDLE_BARS)) provider = _outlook_provider() engine = TechnicalEngine(PlatformConfig().technical) for tf in _OUTLOOK_TFS: try: df = provider.fetch_ohlcv(XAUUSD, tf) analysis = engine.analyze(df, XAUUSD, tf.value) analyses[tf.value] = analysis atr = _finite(getattr( (getattr(analysis, "indicators", None) or {}).get("atr"), "value", None)) if atr is not None and atr > 0 and df is not None and len(df) >= 2: candles[tf.value] = _normalize_candles(df, atr, bars) except Exception as exc: logger.debug("market outlook: %s data unavailable: %s", tf.value, exc) return _build_payload( analyses=analyses, candles=candles, news=_news_payload(), recent_signals=_recent_signals(), as_of=_dt.datetime.now(_dt.timezone.utc), ) # ── Orchestration, hourly dedup, and request budget ────────────────── _state_lock = threading.Lock() _last_run_hour: Optional[int] = None # UTC hour bucket of the last attempt _outlook_day: Optional[_dt.date] = None _outlook_count = 0 _warned_budget = False def _skip(reason: str) -> None: """Heartbeat for every silent early return — a dark outlook feature (zero rows for 15 days, 2026-08/09) must be visible in the log.""" logger.info("market outlook: skipped (%s)", reason) def reset_outlook_state() -> None: """Drop hourly and daily state (tests / deployment config reloads).""" global _last_run_hour, _outlook_day, _outlook_count global _warned_budget with _state_lock: _last_run_hour = None _outlook_day = None _outlook_count = 0 _warned_budget = False def _begin_attempt(cap: int, force: bool) -> bool: """Atomically apply hourly dedup, then reserve a daily attempt. Hourly dedup is by UTC hour BUCKET, not ``now - last < 3600``: the H1 loop calls this at boundary+10 s+scan time, so consecutive hours were routinely 3,590 s apart and every other hour was dropped.""" global _last_run_hour, _outlook_day, _outlook_count global _warned_budget now = time.time() hour_bucket = int(now // 3600) today = _dt.datetime.now(_dt.timezone.utc).date() with _state_lock: if not force and _last_run_hour == hour_bucket: _skip(f"hour bucket {hour_bucket} already attempted") return False if _outlook_day != today: _outlook_day = today _outlook_count = 0 _warned_budget = False if _outlook_count >= cap: if not _warned_budget: _warned_budget = True logger.warning( "market outlook: daily budget (%d) exhausted — no more " "outlooks until UTC midnight", cap) else: _skip("daily budget exhausted") return False _outlook_count += 1 _last_run_hour = hour_bucket return True def run_outlook(settings: Dict[str, Any], surface: str = "live") -> Optional[OutlookResult]: """Build, ask, validate, and persist one advisory outlook.""" try: payload = _collect_payload(settings) reply = get_llm_client().chat_json(SYSTEM_PROMPT, payload) result = _validate_outlook(reply) if result is None: logger.warning( "market outlook: no outlook produced (%s)", "LLM returned nothing — cap/timeout/disabled" if reply is None else "reply failed validation") return None try: from signal_log import log_outlook log_outlook(result, surface=surface, payload=payload) except Exception: logger.debug("market outlook: persistence unavailable", exc_info=True) return result except Exception: logger.warning("market outlook: run failed", exc_info=True) return None def maybe_run_outlook(surface: str, force: bool = False) -> Optional[OutlookResult]: """Best-effort public entry point. It NEVER raises to a scan caller. ``force`` is for explicit CLI use: it bypasses enabled, surface, and hourly gates, but not the independent outlook budget or shared LLM cap. """ try: settings = get_app_config().market_outlook_settings if not force: if not market_outlook_enabled(): _skip("disabled") return None if surface == "telegram" and not settings.get("telegram", False): _skip("telegram surface off") return None if _provider_base._is_forex_weekend(): _skip("weekend") return None cap = int(settings.get("max_per_day", DEFAULT_MAX_PER_DAY)) if cap < 1: _skip("max_per_day < 1") return None if not _begin_attempt(cap, force): return None return run_outlook(settings, surface=surface) except Exception: logger.warning("market outlook: unavailable", exc_info=True) return None