ForexBot2/news_model.py

"""Local XAU headline scorer: the owner's ML-NEWS transformer on CPU.

Vendored from ML-NEWS news_transformer.py at commit afa26dc; bundle
trained_at 2026-09-27T23:15:30.403442+00:00. Keep the scoring functions
identical to that file (the one difference: ``load_model`` has no default
directory here, the path always comes from config.json); ML-NEWS trains
and exports, ForexBot2 only scores.

A fine-tuned DeBERTa-v3-small encoder with two heads, exported to ONNX
with an int8 embedding table. Scoring needs only onnxruntime, tokenizers
and numpy. For each headline:

- ``p_relevant``: probability the news moves the gold price (trained on
  DeepSeek |xau| >= 0.25 vs <= 0.10)
- ``p_bullish``: probability that, if it does, gold goes up
- ``relevant``: ``p_relevant >= relevance_threshold`` (from meta.json)
- ``label``: ``bullish`` / ``bearish``, or ``neutral`` when not relevant
- ``score``: ``p_relevant * (2 * p_bullish - 1)`` in [-1, 1], the same
  scale as DeepSeek's gold score the archive stores in ``xau_score``

The model has a GOLD head only. news_engine writes its score into
``xau_score`` and the XAU per-currency entry, never into USD or any other
currency (get_pair_sentiment subtracts USD from XAU).

Import contract: ``onnxruntime`` and ``tokenizers`` are imported INSIDE
``load_model`` so ``import news_model`` (and therefore ``import
news_engine``) succeeds on a box without them; ``get_news_model`` then
returns None once, with one warning, and the DeepSeek path stays.

Configuration comes from app_config (``news_model`` section, env
``NEWS_MODEL_MODE``): ``news_model_path()``, ``news_model_threads()``.
"""

from __future__ import annotations

import html
import json
import logging
import os
import threading
from pathlib import Path
from typing import Any, Iterable, Optional

import numpy as np

logger = logging.getLogger(__name__)

MAX_LENGTH = 64
#: DeepSeek |xau| at or above this is "moves gold"; at or below
#: RELEVANT_MAX_IRRELEVANT it is "does not"; the band between carries no
#: relevance label in training.
RELEVANT_MIN = 0.25
RELEVANT_MAX_IRRELEVANT = 0.10


def clean_headline(headline: str) -> str:
    """The exact text normalisation used in training and scoring."""
    text = " ".join(html.unescape(str(headline or "")).split())
    if not text:
        raise ValueError("headline is empty")
    return text


def _sigmoid(x: np.ndarray) -> np.ndarray:
    return 1.0 / (1.0 + np.exp(-x))


def load_model(model_dir: str | Path,
               threads: int | None = None) -> dict[str, Any]:
    """Load the ONNX model, tokenizer and metadata. Only load trusted files."""
    import onnxruntime as ort
    from tokenizers import Tokenizer

    model_dir = Path(model_dir)
    meta_path = model_dir / "meta.json"
    if not meta_path.is_file():
        raise FileNotFoundError(f"model not found: {meta_path}")
    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    tokenizer = Tokenizer.from_file(str(model_dir / "tokenizer.json"))
    tokenizer.enable_truncation(max_length=int(meta.get("max_length", MAX_LENGTH)))
    tokenizer.enable_padding(pad_id=int(meta["pad_token_id"]), pad_token=meta["pad_token"])
    options = ort.SessionOptions()
    if threads:
        options.intra_op_num_threads = max(1, min(int(threads), os.cpu_count() or 1))
        options.inter_op_num_threads = 1
    # ORT intra-op threads spin-wait after each run by default; in a mostly
    # idle bot process that is 4 cores at 100% between cycles.
    options.add_session_config_entry("session.intra_op.allow_spinning", "0")
    session = ort.InferenceSession(str(model_dir / meta.get("onnx_file", "model.onnx")),
                                   options, providers=["CPUExecutionProvider"])
    return {"meta": meta, "tokenizer": tokenizer, "session": session}


def score_batch(bundle: dict[str, Any], headlines: Iterable[str]) -> list[dict[str, Any]]:
    texts = [clean_headline(h) for h in headlines]
    if not texts:
        return []
    encodings = bundle["tokenizer"].encode_batch(texts)
    ids = np.array([e.ids for e in encodings], dtype=np.int64)
    mask = np.array([e.attention_mask for e in encodings], dtype=np.int64)
    rel_logit, dir_logit = bundle["session"].run(
        ["relevance_logit", "direction_logit"],
        {"input_ids": ids, "attention_mask": mask})
    p_rel = _sigmoid(rel_logit.astype(np.float64))
    p_bull = _sigmoid(dir_logit.astype(np.float64))
    threshold = float(bundle["meta"].get("relevance_threshold", 0.5))
    out = []
    for text, pr, pb in zip(texts, p_rel, p_bull):
        relevant = bool(pr >= threshold)
        direction = "bullish" if pb >= 0.5 else "bearish"
        label = direction if relevant else "neutral"
        confidence = pr * max(pb, 1 - pb) if relevant else 1 - pr
        out.append({
            "target": "XAUUSD",
            "text": text,
            "relevant": relevant,
            "p_relevant": round(float(pr), 4),
            "direction": direction,
            "p_bullish": round(float(pb), 4),
            "label": label,
            "score": round(float(pr * (2 * pb - 1)), 4),
            "confidence": round(float(confidence), 4),
        })
    return out


def score_news(bundle: dict[str, Any], headline: str) -> dict[str, Any]:
    return score_batch(bundle, [headline])[0]


# ── Process-wide singleton (ForexBot2 addition, not in ML-NEWS) ──────────
#
# One ONNX session per process (~0.6 GB RSS, 3–5 s to create). A failed
# load is remembered: the bot logs one warning and keeps the DeepSeek /
# keyword path for the rest of its life instead of retrying every cycle.

_lock = threading.Lock()
_bundle: Optional[dict] = None
_failed: bool = False          # tried and failed; don't retry, don't re-log
_warm_started: bool = False


def get_news_model() -> Optional[dict]:
    """The loaded bundle, or None. Loads once per process. Never raises."""
    global _bundle, _failed
    if _bundle is not None or _failed:
        return _bundle
    with _lock:
        if _bundle is not None or _failed:
            return _bundle
        try:
            import app_config
            path = app_config.news_model_path()
            if not path:
                raise FileNotFoundError("news_model.path is not configured")
            threads = app_config.news_model_threads()
            b = load_model(path, threads=threads)
            logger.info(
                "news model loaded from %s (trained_at %s, threshold %s, "
                "%d threads)", path, b["meta"].get("trained_at"),
                b["meta"].get("relevance_threshold"), threads)
            _bundle = b
        except Exception as e:  # ImportError, FileNotFoundError, bad JSON, ORT
            logger.warning(
                "news model unavailable, DeepSeek/keyword path stays: %s: %s",
                type(e).__name__, e)
            _failed = True
        return _bundle


def teacher_tag() -> Optional[str]:
    """Provenance tag for archive rows scored by the loaded model, e.g.
    ``xau_transformer@2026-09-27T23:15:30…``. Reads the already-loaded
    bundle only; never triggers a load. None when nothing is loaded."""
    b = _bundle
    if b is None:
        return None
    return f"xau_transformer@{b['meta'].get('trained_at', 'unknown')}"


def warm_up_in_background() -> None:
    """Create the ONNX session on a daemon thread, once per process, so
    the first news cycle does not pay the 3–5 s load under
    NewsEngine._cache_lock. Idempotent; a later get_news_model() call
    simply waits on the lock if the thread is still loading."""
    global _warm_started
    if _warm_started or _bundle is not None or _failed:
        return
    with _lock:
        if _warm_started:
            return
        _warm_started = True
    threading.Thread(target=get_news_model, name="news-model-warmup",
                     daemon=True).start()


def reset_for_tests() -> None:
    """Forget any loaded or failed state (test_all.py only)."""
    global _bundle, _failed, _warm_started
    with _lock:
        _bundle = None
        _failed = False
        _warm_started = False