market_outlook: add hourly advisory brief
This commit is contained in:
Harwinder Sekhon
@@ -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": "<what to watch for, max 25 words>",
|
||||
"note": "<plain-English summary, max 25 words>"}
|
||||
"""
|
||||
|
||||
|
||||
@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
|
||||
Reference in New Issue
Block a user