From 2a2ccb13dd08f01e2b79ab56213d45c459d51d23 Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Tue, 18 Aug 2026 13:52:18 -0400 Subject: [PATCH] market_outlook: add hourly advisory brief --- CLAUDE.md | 3 +- app_config.py | 38 +++- config.json.example | 6 + main.py | 42 ++++- market_outlook.py | 410 ++++++++++++++++++++++++++++++++++++++++++++ signal_log.py | 53 ++++++ telegram_bot.py | 23 +++ test_all.py | 276 +++++++++++++++++++++++++++++ 8 files changed, 848 insertions(+), 3 deletions(-) create mode 100644 market_outlook.py diff --git a/CLAUDE.md b/CLAUDE.md index e55eaab..982a9fa 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -35,7 +35,7 @@ Python 3.12; a `venv/` exists in-repo. Dependencies in `requirements.txt` (numpy ## Configuration & credentials -Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `data_providers.streaming` (optional tick-stream provider for the level-breach monitor), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_ACCOUNT_ID`/`OANDA_API_TOKEN`/`OANDA_BASE_URL`, `DEEPSEEK_API_KEY` (news LLM; the `deepseek` config.json block holds model/enabled/timeout/daily cap). The `agent_panel` config.json block (all-defaults = feature dark) controls the advisory LLM agent panel: `enabled`, per-surface `telegram`/`dashboard` toggles, `min_quality`, `max_panels_per_day`, `reuse_ttl_seconds`, plus the technical-payload enrichment knobs `candles_enabled`/`candle_bars_primary`/`candle_bars_confirm`. +Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `data_providers.streaming` (optional tick-stream provider for the level-breach monitor), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_ACCOUNT_ID`/`OANDA_API_TOKEN`/`OANDA_BASE_URL`, `DEEPSEEK_API_KEY` (news LLM; the `deepseek` config.json block holds model/enabled/timeout/daily cap). The `agent_panel` config.json block (all-defaults = feature dark) controls the advisory LLM agent panel: `enabled`, per-surface `telegram`/`dashboard` toggles, `min_quality`, `max_panels_per_day`, `reuse_ttl_seconds`, plus the technical-payload enrichment knobs `candles_enabled`/`candle_bars_primary`/`candle_bars_confirm`. The dark-by-default `market_outlook` block controls the hourly advisory market brief: `enabled`, `telegram`, `max_per_day`, `candles_bars`; `MARKET_OUTLOOK_ENABLED` overrides its enabled setting. ## Architecture @@ -62,6 +62,7 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - **`signal_log.py`** — SQLite signal-history logger (stdlib only). Hooked into `SignalAggregator.analyze_pair` only (the live entry point — the backtester never writes here, preserving parity), so every live scan from every surface (Telegram `telegram`, CLI `cli`, Oanda daemon `oanda_daemon`, dashboard `dashboard`) lands one row in `logs/signals.db`: signal/grade/price, MTF + trade-setup + reasons JSON, and the news at scan time (event risk + calendar events + scored headlines JSON, re-fetched via the NewsEngine class-level caches). Config.json `signal_log` block (`enabled`, `path`; env `SIGNAL_LOG_ENABLED`/`SIGNAL_LOG_PATH`). **Fail-safe by contract** — a logging failure can never break a scan. Panel fields arrive post-hoc: `log_signal` stashes the rowid on the report and `agent_panel.attach_panel` writes the verdict back onto the same row via `log_panel_update` (no rowid — backtest replay or logging disabled — is a silent no-op). - **`news_engine.py`** — Forex Factory calendar (MyFXBook RSS fallback) + RSS keyword sentiment with gold-inverse keywords. Class-level caches (fresh engine per scan). Optional DeepSeek enrichment via `llm_client.py` (requests-only, no SDK): batched headline sentiment overriding keyword scores, gold-direction on release surprises, and a market-summary blurb (`SignalReport.news_summary`) — **display/warning-only, keyword fallback on any failure, inert without a key**; prompts are static module constants (DeepSeek prefix-cache pricing) and requests send `thinking: disabled` (V4-Flash reasoning otherwise blows the timeout). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`) — the LLM does not change that. - **`agent_panel.py`** — ai-hedge-fund-style advisory panel: 4 DeepSeek persona calls (macro/technical/risk/contrarian, concurrent on a panel-private pool) + 1 synthesizer producing an **ADVISORY-ONLY** consensus (`AGREE/NEUTRAL/OPPOSE` + confidence + rationale) written onto `SignalReport.panel_*` fields. Attached **post-hoc** by consumers (`telegram_bot` alert path, `main --panel`, `dashboard_publisher`) — never runs inside `analyze_from_dataframes` (backtest parity, test-locked) and no gate reads its fields. Static prompts (prefix caching); spend bounded by an exact per-bar cache, a reuse TTL keyed (symbol, TF, direction) so persisting signals don't re-spend every bar, and a `max_panels_per_day` UTC budget (default 20 ≈ ≤100 requests) on top of the shared 500/day client cap. `<2` valid persona takes skips the synthesizer; any failure degrades to no panel, never a changed signal. The technical persona's payload is enriched (2026-08-15): raw indicator values plus ATR-normalized recent candles per TF (`candles_enabled`/`candle_bars_primary` 20/`candle_bars_confirm` 10 config keys; code does the arithmetic — raw prices are the model's weakest input, per `backtest/research/llm_news_eval_2026-08-06.md`). Prompts forbid training-data/outside knowledge and cap notes at 15 words / rationales at 30 to bound output tokens. +- **`market_outlook.py`** — hourly DeepSeek market brief for XAU/USD, using ATR-normalized M15/H1/H4 candles, visible indicator states, levels, news, and immutable recent signal history. It returns a written **ADVISORY-ONLY** bias/setup-forming assessment, persists it independently in `outlooks` within `logs/signals.db`, and may be sent post-scan by Telegram or requested with `main.py --outlook`; it never imports or runs inside the signal path. A lock enforces hourly dedup plus its own UTC-day cap (default 12), and provider/news failures leave a safe partial brief while LLM/database failures degrade to no outlook. - **`telegram_bot.py`** — owner-locked bot + subscriber channel; candle-boundary auto-scan loops (M5/M15/H1/H4, boundary+10 s), alert quality threshold + fingerprint dedup (committed only after a successful send), 📰 LLM market-context line on alerts when available, WebSocket level-breach monitor fed by the streaming provider (`data_providers.streaming`, falling back to the active provider; auto-disabled when the resolved provider can't stream or has no API key — Oanda does not stream, Tiingo does). Rotating log: `logs/telegram.log`, 3-h rotation × 240 backups. - **`oanda_daemon.py` / `oanda_executor.py`** — practice-account auto-execution. Book-parity rules (see daemon docstring): no broker-side TP (trailing model — TP1/TP2 advisory), 5-bar per-TF cooldown (wall-clock approximation, so it expires over weekends where the backtester would still count bars), trail ratchets at bar close once per *observed* bar (a missed/failed cycle skips those bars entirely — the intra-gap high/low never enters the peak). Accepted divergence: one position total; orders send `positionFill=OPEN_ONLY` so a state/broker divergence cancels the order instead of silently netting. State/broker re-sync runs at boot and after any suspicious fill (`_reconcile_positions`). - **`dashboard_publisher.py` / `static/dashboard.html`** — web dashboard as a **static publisher** (live LAMP server): standalone daemon scans at candle closes via `produce_signal` and atomically writes `snapshot.json`/`news.json`/`dashboard.html` to `$DASHBOARD_OUT_DIR` (Apache serves them; page is public by owner decision). **Hard independence rule: never imports, and is never imported by, `telegram_bot`/`oanda_daemon`** — confirmation map and boundary math are deliberate local copies, so a web-side failure can only mean a stale page. Due-TF classification uses the RAW wake time (flooring wake−buffer skips every H1/H4 scan — 2026-08-02 bug). News panel reads `NewsEngine.get_fundamental_context()` plus the additive `get_week_events()` accessor (display-only; no signal path calls it); as a separate process it has its own news caches and DeepSeek daily cap. News/calendar rows are clickable (detail dialog); headlines carry `id` = sha1(headline) in news.json, and exactly one on-demand endpoint exists: `static/news_summary.php` (installed by `install_static`), which serves a DeepSeek gold-impact summary per headline — whitelist-gated against news.json ids, cache-forever in `ai/`, own fail-closed daily cap (150/day, file-based, separate from the publisher's), key at `/etc/gold-dashboard/deepseek.key`, never reads repo config. Deployment checklist: `deploy/README-dashboard.md`. diff --git a/app_config.py b/app_config.py index b188441..a1ff92e 100644 --- a/app_config.py +++ b/app_config.py @@ -86,7 +86,8 @@ _VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") _SCANNABLE_TFS = ("M5", "M15", "H1", "H4") _KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level", - "oanda", "deepseek", "agent_panel", "signal_log"} + "oanda", "deepseek", "agent_panel", "market_outlook", + "signal_log"} _KNOWN_OANDA = { "account_id", "api_token", "base_url", "lot_size", "quality_gate", "max_hold_bars", @@ -99,11 +100,19 @@ _KNOWN_AGENT_PANEL = { "max_panels_per_day", "reuse_ttl_seconds", "candles_enabled", "candle_bars_primary", "candle_bars_confirm", } +_KNOWN_MARKET_OUTLOOK = {"enabled", "telegram", "max_per_day", "candles_bars"} _KNOWN_SIGNAL_LOG = {"enabled", "path"} _KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"} _KNOWN_SCANNING = {"pairs", "timeframes"} _KNOWN_PAIR_KEYS = {"blocked_hours_utc"} +_MARKET_OUTLOOK_DEFAULTS = { + "enabled": False, + "telegram": False, + "max_per_day": 12, + "candles_bars": 20, +} + class AppConfigError(Exception): """config.json is present but invalid — fail fast at boot.""" @@ -125,6 +134,8 @@ class AppConfig: oanda_settings: Dict[str, Any] = field(default_factory=dict) deepseek_settings: Dict[str, Any] = field(default_factory=dict) agent_panel_settings: Dict[str, Any] = field(default_factory=dict) + market_outlook_settings: Dict[str, Any] = field( + default_factory=lambda: dict(_MARKET_OUTLOOK_DEFAULTS)) signal_log_settings: Dict[str, Any] = field(default_factory=dict) log_level: str = "INFO" @@ -321,6 +332,20 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig: "between 1 and 100") app.agent_panel_settings = dict(panel) + outlook = raw.get("market_outlook", {}) + _check_keys(outlook, _KNOWN_MARKET_OUTLOOK, '"market_outlook"') + for flag in ("enabled", "telegram"): + if flag in outlook and not isinstance(outlook[flag], bool): + raise AppConfigError( + f"config.json: market_outlook.{flag} must be true or false") + for key in ("max_per_day", "candles_bars"): + if key in outlook: + value = outlook[key] + if isinstance(value, bool) or not isinstance(value, int) or value < 1: + raise AppConfigError( + f"config.json: market_outlook.{key} must be a positive integer") + app.market_outlook_settings.update(outlook) + slog = raw.get("signal_log", {}) _check_keys(slog, _KNOWN_SIGNAL_LOG, '"signal_log"') if "enabled" in slog and not isinstance(slog["enabled"], bool): @@ -431,6 +456,17 @@ def deepseek_api_key(app: Optional[AppConfig] = None) -> str: app.deepseek_settings.get("api_key", "")) +# ── AI market-outlook resolution helper ────────────────────────────── + +def market_outlook_enabled(app: Optional[AppConfig] = None) -> bool: + """Whether the advisory outlook feature is enabled (env wins).""" + env = os.environ.get("MARKET_OUTLOOK_ENABLED") + if env: + return env.strip().lower() in ("1", "true", "yes", "on") + app = app or get_app_config() + return bool(app.market_outlook_settings.get("enabled", False)) + + # ── Signal-log resolution helpers (env → config.json → default) ────── def signal_log_enabled(app: Optional[AppConfig] = None) -> bool: diff --git a/config.json.example b/config.json.example index 8964b23..fadbd7f 100644 --- a/config.json.example +++ b/config.json.example @@ -51,6 +51,12 @@ "candle_bars_primary": 20, "candle_bars_confirm": 10 }, + "market_outlook": { + "enabled": false, + "telegram": false, + "max_per_day": 12, + "candles_bars": 20 + }, "signal_log": { "enabled": true, "path": "logs/signals.db" diff --git a/main.py b/main.py index e53f862..8f07287 100755 --- a/main.py +++ b/main.py @@ -22,6 +22,7 @@ if hasattr(sys.stderr, "reconfigure"): from agent_panel import attach_panel from config import PlatformConfig, Timeframe, SignalStrength, XAUUSD +from market_outlook import OutlookResult, maybe_run_outlook from pair_signals import PAIR_PRODUCERS from signal_aggregator import SignalAggregator, SignalReport @@ -197,6 +198,24 @@ def print_signal_report(report: SignalReport, detail: bool = False): print(f" {label:>20} {color}{val:.3f}{Colors.RESET}") +def print_market_outlook(outlook: OutlookResult | None) -> None: + """Render the independent advisory outlook for explicit CLI use.""" + print(f"\n {Colors.BOLD}AI Market Outlook{Colors.RESET}") + if outlook is None: + print(f" {Colors.GRAY}Unavailable (disabled, market closed, or no LLM response).{Colors.RESET}") + return + forming = { + "buy": "possible buy forming", + "sell": "possible sell forming", + "none": "no clear setup forming", + }[outlook.setup_forming] + print(f" {outlook.bias.title()} · {forming} · {outlook.confidence:.0f}%") + if outlook.watch: + print(f" Watch: {outlook.watch}") + if outlook.note: + print(f" {Colors.GRAY}{outlook.note}{Colors.RESET}") + + def report_to_dict(report: SignalReport) -> dict: d = { "symbol": report.symbol, @@ -283,6 +302,9 @@ Examples: parser.add_argument("--panel", action="store_true", help="Run the LLM agent panel on the result " "(advisory only; needs a DeepSeek key)") + parser.add_argument("--outlook", action="store_true", + help="Run the hourly AI market outlook now " + "(advisory only; needs a DeepSeek key)") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--verbose", "-v", action="store_true", @@ -343,12 +365,30 @@ def run_live(args, config: PlatformConfig): if getattr(args, "panel", False): attach_panel(report, surface="cli", force=True) + outlook = None + if getattr(args, "outlook", False): + outlook = maybe_run_outlook(surface="cli", force=True) + if args.json: - print(json.dumps(report_to_dict(report), indent=2)) + payload = report_to_dict(report) + if getattr(args, "outlook", False): + payload["market_outlook"] = ( + None if outlook is None else { + "bias": outlook.bias, + "setup_forming": outlook.setup_forming, + "confidence": outlook.confidence, + "watch": outlook.watch, + "note": outlook.note, + "created_ts": outlook.created_ts, + } + ) + print(json.dumps(payload, indent=2)) return print_header(f"{config.display_name} SIGNAL INTELLIGENCE") print_signal_report(report, detail=args.detail) + if getattr(args, "outlook", False): + print_market_outlook(outlook) print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") print(f"{Colors.GRAY} Disclaimer: For educational/research purposes only.{Colors.RESET}") print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") diff --git a/market_outlook.py b/market_outlook.py new file mode 100644 index 0000000..89ec89e --- /dev/null +++ b/market_outlook.py @@ -0,0 +1,410 @@ +""" +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()) + uri = f"file:{path}?immutable=1" + 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_ts: Optional[float] = None +_outlook_day: Optional[_dt.date] = None +_outlook_count = 0 +_warned_budget = False + + +def reset_outlook_state() -> None: + """Drop hourly and daily state (tests / deployment config reloads).""" + global _last_run_ts, _outlook_day, _outlook_count, _warned_budget + with _state_lock: + _last_run_ts = 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.""" + global _last_run_ts, _outlook_day, _outlook_count, _warned_budget + now = time.time() + today = _dt.datetime.now(_dt.timezone.utc).date() + with _state_lock: + if not force and _last_run_ts is not None and now - _last_run_ts < 3600: + 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) + return False + _outlook_count += 1 + _last_run_ts = now + 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) + result = _validate_outlook(get_llm_client().chat_json(SYSTEM_PROMPT, payload)) + if result is None: + return None + try: + from signal_log import log_outlook + log_outlook(result, surface=surface) + 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(): + return None + if surface == "telegram" and not settings.get("telegram", False): + return None + if _provider_base._is_forex_weekend(): + return None + cap = int(settings.get("max_per_day", DEFAULT_MAX_PER_DAY)) + if cap < 1 or 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 diff --git a/signal_log.py b/signal_log.py index f50410f..b52fda0 100644 --- a/signal_log.py +++ b/signal_log.py @@ -75,6 +75,16 @@ CREATE TABLE IF NOT EXISTS signals ( ); CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts ON signals (symbol, timeframe, logged_at); +CREATE TABLE IF NOT EXISTS outlooks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + logged_at TEXT NOT NULL, + surface TEXT NOT NULL, + bias TEXT, + setup_forming TEXT, + confidence REAL, + watch TEXT, + note TEXT +); """ # Additive migration: older DBs created before the panel columns existed @@ -97,6 +107,12 @@ INSERT INTO signals ( ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ +_OUTLOOK_INSERT = """ +INSERT INTO outlooks ( + logged_at, surface, bias, setup_forming, confidence, watch, note +) VALUES (?, ?, ?, ?, ?, ?, ?) +""" + def _finite(x: Any) -> Optional[float]: """float(x) or None for non-numeric / non-finite values (NaN/inf — @@ -323,6 +339,22 @@ class SignalLogger: (verdict, confidence, rationale, takes_json, rowid)) self._conn.commit() + def log_outlook(self, outlook: Any, surface: str) -> int: + """Insert one advisory market outlook; returns its rowid.""" + row = ( + datetime.now(timezone.utc).isoformat(), + surface, + getattr(outlook, "bias", None), + getattr(outlook, "setup_forming", None), + _finite(getattr(outlook, "confidence", None)), + getattr(outlook, "watch", None), + getattr(outlook, "note", None), + ) + with self._lock: + cur = self._conn.execute(_OUTLOOK_INSERT, row) + self._conn.commit() + return int(cur.lastrowid) + def close(self) -> None: with self._lock: self._conn.close() @@ -392,6 +424,27 @@ def log_panel_update(report: Any) -> None: exc_info=True) +def log_outlook(outlook: Any, surface: str = "live") -> None: + """Persist one advisory outlook when signal logging is enabled. + + Outlook persistence is deliberately independent of a signal row: the + feature runs on its own hourly cadence and must never affect a scan. + """ + try: + if not app_config.signal_log_enabled(): + return + path = app_config.signal_log_path() + with _loggers_lock: + lg = _loggers.get(path) + if lg is None: + lg = SignalLogger(path) + _loggers[path] = lg + lg.log_outlook(outlook, surface) + except Exception: + logger.warning("signal_log: failed to persist market outlook", + exc_info=True) + + def reset_loggers_for_tests() -> None: """Close and drop all cached loggers (test isolation).""" with _loggers_lock: diff --git a/telegram_bot.py b/telegram_bot.py index 4c3d77f..964cccd 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -73,6 +73,7 @@ from app_config import ( from providers import BaseDataProvider, make_data_provider from agent_panel import attach_panel, panel_health from llm_client import get_llm_client +from market_outlook import maybe_run_outlook from news_engine import FundamentalContext, NewsEngine from pair_signals import PAIR_PRODUCERS from signal_aggregator import SignalAggregator, SignalReport @@ -749,6 +750,28 @@ async def _scan_timeframe( if delivered: _commit_alert(key, report) + # This independent advisory call runs only after the H1 scan completes. + # Offloading the LLM request keeps the event loop responsive. + if tf == Timeframe.H1: + try: + outlook = await asyncio.get_running_loop().run_in_executor( + _executor, maybe_run_outlook, "telegram") + settings = get_app_config().market_outlook_settings + if (outlook is not None and settings.get("telegram", False) + and OWNER_ID is not None): + forming = { + "buy": "possible buy forming", + "sell": "possible sell forming", + "none": "no clear setup forming", + }[outlook.setup_forming] + message = ( + f"🔭 AI outlook: {outlook.bias}, {forming} " + f"({outlook.confidence:.0f}%) — {_escape_md(outlook.note)}" + ) + await _send_alert(bot, OWNER_ID, message) + except Exception: + logger.exception("Hourly AI market outlook failed") + # ── WebSocket level-breach monitor ────────────────────────────────── diff --git a/test_all.py b/test_all.py index d7aad57..16c4c81 100644 --- a/test_all.py +++ b/test_all.py @@ -7835,6 +7835,282 @@ class _BoomNews: raise RuntimeError("news engine down") +# ══════════════════════════════════════════════════════════════════════ +# MODULE 26: market_outlook.py (advisory hourly market brief) +# ══════════════════════════════════════════════════════════════════════ + +print("\n═══ Module 26: AI market outlook ═══") + +import market_outlook as mo + + +class _OutlookStub: + available = True + + def __init__(self, reply=None): + self.reply = reply or { + "bias": "bullish", "setup_forming": "buy", "confidence": 68, + "watch": "Look for the price to hold above its recent low.", + "note": "The short-term chart is improving while the bigger trend remains supportive.", + } + self.calls = [] + + def chat_json(self, system_prompt, user_payload): + self.calls.append((system_prompt, user_payload)) + if isinstance(self.reply, Exception): + raise self.reply + return self.reply + + +class _OutlookNews: + def __init__(self, config): + pass + + def get_fundamental_context(self): + now = datetime.now(timezone.utc) + return FundamentalContext( + news_headlines=[NewsSentiment( + headline="Gold firms as the dollar weakens", source="test", + timestamp=now, currencies_mentioned=["XAU", "USD"], + sentiment_score=0.3, keywords=[])], + events_today=[EconomicEvent( + time=now - timedelta(minutes=5), currency="USD", impact="high", + event="Inflation", actual="0.2%", forecast="0.3%")], + ) + + def get_pair_sentiment(self, symbol, context): + return 0.25, "slightly bullish" + + def get_week_events(self): + return [] + + +_OUTLOOK_ON = { + "enabled": True, + "telegram": True, + "max_per_day": 12, + "candles_bars": 20, +} + + +@test("app_config: market_outlook defaults, env override, and validation") +def _(): + defaults = load_app_config(_write_json({})).market_outlook_settings + assert defaults == { + "enabled": False, "telegram": False, + "max_per_day": 12, "candles_bars": 20, + } + app = load_app_config(_write_json({"market_outlook": _OUTLOOK_ON})) + assert app.market_outlook_settings == _OUTLOOK_ON + old = os.environ.get("MARKET_OUTLOOK_ENABLED") + try: + os.environ["MARKET_OUTLOOK_ENABLED"] = "false" + assert app_config.market_outlook_enabled(app) is False + os.environ["MARKET_OUTLOOK_ENABLED"] = "true" + assert app_config.market_outlook_enabled(app) is True + finally: + if old is None: + os.environ.pop("MARKET_OUTLOOK_ENABLED", None) + else: + os.environ["MARKET_OUTLOOK_ENABLED"] = old + for bad in ( + {"market_outlook": {"enabld": True}}, + {"market_outlook": {"enabled": "yes"}}, + {"market_outlook": {"telegram": 1}}, + {"market_outlook": {"max_per_day": 0}}, + {"market_outlook": {"max_per_day": True}}, + {"market_outlook": {"candles_bars": 2.5}}, + ): + try: + load_app_config(_write_json(bad)) + assert False, f"should have rejected {bad}" + except AppConfigError: + pass + + +@test("signal_log: market outlook table roundtrip and old-db migration") +def _(): + import sqlite3 + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + con = sqlite3.connect(path) + # Legacy pre-panel schema: it has a signals table but no outlooks + # table. Keep the index columns from the original schema present. + con.execute( + "CREATE TABLE signals (id INTEGER PRIMARY KEY, logged_at TEXT," + " symbol TEXT, timeframe TEXT)") + con.commit() + con.close() + logger = slog.SignalLogger(path) + result = mo.OutlookResult( + bias="bearish", setup_forming="sell", confidence=71.5, + watch="Wait for price to fail at resistance.", + note="Momentum is fading after a sharp rise.", created_ts=time.time()) + rowid = logger.log_outlook(result, surface="test") + logger.close() + con = sqlite3.connect(path) + row = con.execute( + "SELECT surface, bias, setup_forming, confidence, watch, note " + "FROM outlooks WHERE id=?", (rowid,)).fetchone() + con.close() + assert row == ("test", "bearish", "sell", 71.5, + "Wait for price to fail at resistance.", + "Momentum is fading after a sharp rise.") + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("market_outlook: payload is finite, normalized, and news-safe") +def _(): + try: + mo._dp = _FakeDP() + with patch("news_engine.NewsEngine", _OutlookNews): + payload = mo._collect_payload(_OUTLOOK_ON) + assert mo._normalize_candles is ap._normalize_candles + assert set(payload["candles"]) == {"15min", "1h", "4h"} + assert all(len(block["bars"]) == 20 + for block in payload["candles"].values()) + assert payload["indicators"]["1h"] + assert payload["news"]["headlines"] == ["Gold firms as the dollar weakens"] + json.dumps(payload, sort_keys=True, allow_nan=False) + + with patch("news_engine.NewsEngine", side_effect=RuntimeError("down")): + no_news = mo._collect_payload(_OUTLOOK_ON)["news"] + assert no_news == mo._empty_news_payload() + finally: + mo._dp = None + + +@test("market_outlook: reply validation clamps and rejects malformed JSON") +def _(): + validate = mo._validate_outlook + assert validate(None) is None + assert validate({"bias": "up", "setup_forming": "buy", "confidence": 50}) is None + assert validate({"bias": "bullish", "setup_forming": "wait", "confidence": 50}) is None + assert validate({"bias": "bullish", "setup_forming": "buy"}) is None + assert validate({"bias": "bullish", "setup_forming": "buy", "confidence": "hot"}) is None + result = validate({ + "bias": "neutral", "setup_forming": "none", "confidence": 150, + "watch": "word " * 30, "note": "note " * 30, + }) + assert result.confidence == 100.0 + assert len(result.watch.split()) == 25 and len(result.note.split()) == 25 + + +@test("market_outlook: disabled default spends no LLM request") +def _(): + original_app = app_config.get_app_config() + old = os.environ.pop("MARKET_OUTLOOK_ENABLED", None) + try: + app_config.set_app_config(AppConfig()) + stub = _OutlookStub() + lc._client = stub + mo.reset_outlook_state() + assert mo.maybe_run_outlook("telegram") is None + assert stub.calls == [] + finally: + if old is not None: + os.environ["MARKET_OUTLOOK_ENABLED"] = old + lc._client = None + mo.reset_outlook_state() + app_config.set_app_config(original_app) + + +@test("market_outlook: hourly dedup prevents a second request") +def _(): + original_app = app_config.get_app_config() + old = os.environ.pop("MARKET_OUTLOOK_ENABLED", None) + try: + app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON))) + stub = _OutlookStub() + lc._client = stub + mo.reset_outlook_state() + with patch("providers.base._is_forex_weekend", return_value=False), \ + patch.object(mo, "_collect_payload", return_value={}), \ + patch("signal_log.log_outlook"): + assert mo.maybe_run_outlook("telegram") is not None + assert mo.maybe_run_outlook("telegram") is None + assert len(stub.calls) == 1 + finally: + if old is not None: + os.environ["MARKET_OUTLOOK_ENABLED"] = old + lc._client = None + mo.reset_outlook_state() + app_config.set_app_config(original_app) + + +@test("market_outlook: daily budget blocks force calls after the cap") +def _(): + original_app = app_config.get_app_config() + try: + app_config.set_app_config(AppConfig(market_outlook_settings={ + **_OUTLOOK_ON, "max_per_day": 1})) + stub = _OutlookStub() + lc._client = stub + mo.reset_outlook_state() + with patch("providers.base._is_forex_weekend", return_value=False), \ + patch.object(mo, "_collect_payload", return_value={}), \ + patch("signal_log.log_outlook"): + assert mo.maybe_run_outlook("cli", force=True) is not None + assert mo.maybe_run_outlook("cli", force=True) is None + assert len(stub.calls) == 1 + finally: + lc._client = None + mo.reset_outlook_state() + app_config.set_app_config(original_app) + + +@test("market_outlook: LLM and persistence failures remain fail-safe") +def _(): + original_app = app_config.get_app_config() + try: + app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON))) + lc._client = _OutlookStub(RuntimeError("LLM down")) + mo.reset_outlook_state() + with patch("providers.base._is_forex_weekend", return_value=False), \ + patch.object(mo, "_collect_payload", return_value={}): + assert mo.maybe_run_outlook("telegram") is None + + lc._client = _OutlookStub() + mo.reset_outlook_state() + with patch("providers.base._is_forex_weekend", return_value=False), \ + patch.object(mo, "_collect_payload", return_value={}), \ + patch("signal_log.log_outlook", side_effect=RuntimeError("db locked")): + assert mo.maybe_run_outlook("telegram") is not None + finally: + lc._client = None + mo.reset_outlook_state() + app_config.set_app_config(original_app) + + +@test("market_outlook: weekend gate spends nothing") +def _(): + original_app = app_config.get_app_config() + try: + app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON))) + stub = _OutlookStub() + lc._client = stub + mo.reset_outlook_state() + with patch("providers.base._is_forex_weekend", return_value=True): + assert mo.maybe_run_outlook("telegram") is None + assert stub.calls == [] + finally: + lc._client = None + mo.reset_outlook_state() + app_config.set_app_config(original_app) + + +@test("market_outlook isolation: signal path never imports the advisory module") +def _(): + root = os.path.dirname(__file__) + for name in ("engine.py", "signal_aggregator.py"): + with open(os.path.join(root, name), encoding="utf-8") as fh: + assert "market_outlook" not in fh.read(), name + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════