Files
News-ML-Decoder/news_sentiment.py
T
Harwinder Sekhon 1d72e4006a Initial import: XAUUSD news sentiment models, evals, and data tooling
TF-IDF baseline (news_sentiment.py, models/xauusd_sentiment.joblib) and the
two-head DeBERTa-v3-small transformer (transformer/, news_transformer.py)
with its Google News crawler and DeepSeek labeller (data_collect/), the
golden-set and archive evals (eval/), and the 2026-09-22 and 2026-09-27
results and progress notes.

Transformer weights (model.pt, model.onnx) and the crawl database are not
tracked; rebuild with transformer/train.py and transformer/export_onnx.py.
The DeepSeek label cache in data/ is tracked.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-27 19:22:30 -04:00

690 lines
25 KiB
Python

#!/usr/bin/env python3
"""Train and use a scikit-learn XAUUSD news sentiment model."""
from __future__ import annotations
import argparse
import html
import json
import math
import re
import sqlite3
import sys
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from statistics import median
from typing import Any, Iterable
import joblib
import sklearn
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
accuracy_score,
balanced_accuracy_score,
classification_report,
confusion_matrix,
f1_score,
mean_absolute_error,
)
from sklearn.model_selection import train_test_split
from sklearn.pipeline import FeatureUnion, Pipeline
FORMAT_VERSION = 3
LABELS = ("bearish", "neutral", "bullish")
TARGET_SYMBOL = "XAUUSD"
TARGET_SCORE_KEY = "XAU"
DEFAULT_MODEL_PATH = Path("models/xauusd_sentiment.joblib")
SCOPES = ("all", "gold")
# The ForexBot2 test suite once wrote fixture headlines into the live archive.
TEST_FEED_MARKERS = ("feed.example", "good.example")
XAU_SUBJECT_PATTERN = re.compile(r"\b(?:gold|xau(?:/usd|usd)?|bullion)\b")
XAU_UP_PATTERN = re.compile(
r"\b(?:rall(?:y|ies|ied)|gain(?:s|ed)?|ris(?:e|es|ing|en)|"
r"rebound(?:s|ed)?|jump(?:s|ed)?|surg(?:e|es|ed|ing)|"
r"climb(?:s|ed)?|advance(?:s|d)?|recover(?:s|ed|y)?|"
r"soar(?:s|ed)?|break(?:s|ing)?[ -]?out)\b"
)
XAU_DOWN_PATTERN = re.compile(
r"\b(?:fall(?:s|ing|en)?|fell|slid(?:e|es|ing)?|"
r"slump(?:s|ed|ing)?|drop(?:s|ped|ping)?|tumble(?:s|d)?|"
r"plunge(?:s|d)?|declin(?:e|es|ed|ing)|sell[ -]?off|"
r"weaken(?:s|ed|ing)?)\b"
)
XAU_FLAT_PATTERN = re.compile(
r"\b(?:steady|stable|flat|unchanged|range[ -]?bound|"
r"consolidat(?:e|es|ed|ing))\b"
)
CLAUSE_SPLIT_PATTERN = re.compile(
r"\b(?:as|after|amid|while|despite|ahead of|on)\b|[,;:\u2013\u2014]"
)
XAU_MACRO_SUPPORT_PATTERN = re.compile(
r"(?:dollar|usd).{0,40}(?:weaken|fall|drop|declin|slide)|"
r"(?:weaker|falling|lower).{0,20}(?:dollar|usd)|"
r"(?:yield).{0,30}(?:fall|drop|declin|ease|lower)|"
r"(?:falling|lower|easing).{0,20}(?:yield)|"
r"(?:rate[ -]?cut|dovish)"
)
XAU_MACRO_PRESSURE_PATTERN = re.compile(
r"(?:dollar|usd).{0,40}(?:strengthen|rise|gain|climb)|"
r"(?:stronger|rising|higher).{0,20}(?:dollar|usd)|"
r"(?:yield).{0,30}(?:rise|climb|surge|higher)|"
r"(?:rising|higher).{0,20}(?:yield)|"
r"(?:rate[ -]?hike|hawkish)"
)
def _xau_signal_markers(headline: str) -> str:
"""Create domain features without replacing the learned text features."""
clauses = CLAUSE_SPLIT_PATTERN.split(headline.casefold())
xau_clauses = [
clause for clause in clauses if XAU_SUBJECT_PATTERN.search(clause)
]
markers: list[str] = []
if any(XAU_UP_PATTERN.search(clause) for clause in xau_clauses):
markers.extend(("xau_move_up", "xau_move_up"))
if any(XAU_DOWN_PATTERN.search(clause) for clause in xau_clauses):
markers.extend(("xau_move_down", "xau_move_down"))
if any(XAU_FLAT_PATTERN.search(clause) for clause in xau_clauses):
markers.extend(("xau_move_flat", "xau_move_flat"))
if XAU_MACRO_SUPPORT_PATTERN.search(headline.casefold()):
markers.append("xau_macro_support")
if XAU_MACRO_PRESSURE_PATTERN.search(headline.casefold()):
markers.append("xau_macro_pressure")
return " ".join(markers)
def article_text(headline: str, description: str | None = None) -> str:
"""Combine and normalize the fields used by the model."""
clean_headline = " ".join(html.unescape(str(headline or "")).split())
clean_description = " ".join(html.unescape(str(description or "")).split())
if not clean_headline and not clean_description:
raise ValueError("an article must contain a headline or description")
if not clean_description or clean_description.casefold() == clean_headline.casefold():
text = clean_headline
elif not clean_headline:
text = clean_description
else:
text = f"{clean_headline}\n{clean_description}"
markers = _xau_signal_markers(clean_headline)
return f"{text}\n{markers}" if markers else text
def score_to_label(score: float, neutral_threshold: float) -> str:
"""Convert an existing numeric label into a training class."""
if score > neutral_threshold:
return "bullish"
if score < -neutral_threshold:
return "bearish"
return "neutral"
def build_pipeline(random_state: int, min_df: int, max_features: int) -> Pipeline:
"""Build the text classification pipeline."""
return Pipeline(
steps=[
(
"features",
FeatureUnion(
transformer_list=[
(
"word_tfidf",
TfidfVectorizer(
lowercase=True,
strip_accents="unicode",
ngram_range=(1, 2),
min_df=min_df,
max_df=0.98,
max_features=max_features,
sublinear_tf=True,
),
),
(
"character_tfidf",
TfidfVectorizer(
lowercase=True,
strip_accents="unicode",
analyzer="char_wb",
ngram_range=(3, 5),
min_df=min_df,
max_features=max_features,
sublinear_tf=True,
),
),
],
transformer_weights={"word_tfidf": 1.0, "character_tfidf": 0.6},
),
),
(
"classifier",
LogisticRegression(
C=4.0,
class_weight="balanced",
max_iter=2_000,
random_state=random_state,
),
),
]
)
def load_extra_labels(labels_path: Path | None) -> dict[str, float]:
"""Read ``{"hash": ..., "xau": ...}`` JSONL rows, e.g. backfilled teacher labels."""
if labels_path is None:
return {}
if not labels_path.is_file():
raise FileNotFoundError(f"extra labels not found: {labels_path}")
extra: dict[str, float] = {}
with labels_path.open(encoding="utf-8") as handle:
for line in handle:
if not line.strip():
continue
record = json.loads(line)
if record.get("hash") and record.get("xau") is not None:
extra[str(record["hash"])] = float(record["xau"])
return extra
def load_training_examples(
database_path: Path,
score_source: str,
neutral_threshold: float,
scope: str = "all",
extra_labels_path: Path | None = None,
) -> tuple[list[str], list[str], list[float], list[str], dict[str, Any]]:
"""Read and deduplicate rows with an XAU sentiment label.
Label priority per row: the ``xau_score`` column (stored for every
teacher-scored headline since 2026-09-18), then the ``XAU`` key of
``score_by_currency_json``, then ``extra_labels_path``. ``scope="gold"``
keeps only rows whose JSON carries the ``XAU`` key, the pre-2026-09-22
training set.
"""
if not database_path.is_file():
raise FileNotFoundError(f"database not found: {database_path}")
if neutral_threshold < 0:
raise ValueError("neutral threshold cannot be negative")
if scope not in SCOPES:
raise ValueError(f"scope must be one of: {', '.join(SCOPES)}")
extra_labels = load_extra_labels(extra_labels_path)
database_uri = f"{database_path.resolve().as_uri()}?mode=ro"
with sqlite3.connect(database_uri, uri=True) as connection:
columns = {
row[1] for row in connection.execute("PRAGMA table_info(headlines)")
}
required_columns = {
"hash",
"headline",
"description",
"published_at",
"source",
"score_by_currency_json",
"score_source",
}
missing_columns = required_columns - columns
if missing_columns:
missing = ", ".join(sorted(missing_columns))
raise ValueError(f"headlines table is missing columns: {missing}")
xau_column = "xau_score" if "xau_score" in columns else "NULL"
query = f"""
SELECT hash, headline, description, published_at, source,
score_by_currency_json, {xau_column}
FROM headlines
WHERE 1 = 1
"""
parameters: tuple[str, ...] = ()
if score_source != "all":
query += " AND score_source = ?"
parameters = (score_source,)
query += " ORDER BY published_at, hash"
rows = connection.execute(query, parameters).fetchall()
grouped_examples: dict[str, dict[str, Any]] = {}
skipped_rows = 0
target_labeled_rows = 0
test_feed_rows = 0
label_sources: Counter[str] = Counter()
for (
row_hash,
headline,
description,
published_at,
source,
raw_scores,
xau_score,
) in rows:
if any(marker in str(source or "") for marker in TEST_FEED_MARKERS):
test_feed_rows += 1
continue
try:
scores_by_currency = json.loads(raw_scores) if raw_scores else {}
if not isinstance(scores_by_currency, dict):
raise ValueError("currency scores must be a JSON object")
has_json_label = TARGET_SCORE_KEY in scores_by_currency
if scope == "gold" and not has_json_label:
continue
if xau_score is not None:
label_source, score = "xau_score", float(xau_score)
elif has_json_label:
label_source = "json"
score = float(scores_by_currency[TARGET_SCORE_KEY])
elif row_hash in extra_labels:
label_source, score = "extra_labels", extra_labels[row_hash]
else:
continue
text = article_text(headline, description)
except (json.JSONDecodeError, TypeError, ValueError):
skipped_rows += 1
continue
if not math.isfinite(score):
skipped_rows += 1
continue
target_labeled_rows += 1
label_sources[label_source] += 1
key = text.casefold()
if key not in grouped_examples:
grouped_examples[key] = {"text": text, "scores": [], "dates": []}
grouped_examples[key]["scores"].append(score)
grouped_examples[key]["dates"].append(str(published_at or ""))
texts: list[str] = []
labels: list[str] = []
target_scores: list[float] = []
published_dates: list[str] = []
for example in grouped_examples.values():
target_score = float(median(example["scores"]))
texts.append(example["text"])
target_scores.append(target_score)
labels.append(score_to_label(target_score, neutral_threshold))
published_dates.append(max(example["dates"]))
label_distribution = dict(sorted(Counter(labels).items()))
if len(label_distribution) < 2:
raise ValueError("training data must contain at least two sentiment classes")
if min(label_distribution.values()) < 2:
raise ValueError("each sentiment class must contain at least two examples")
stats = {
"database": str(database_path),
"target": TARGET_SYMBOL,
"target_score_key": TARGET_SCORE_KEY,
"label_priority": ["xau_score", "json", "extra_labels"],
"label_sources": dict(sorted(label_sources.items())),
"extra_labels": str(extra_labels_path) if extra_labels_path else None,
"scope": scope,
"score_source": score_source,
"test_feed_rows": test_feed_rows,
"published_range": [min(published_dates), max(published_dates)],
"source_rows": len(rows),
"target_labeled_rows": target_labeled_rows,
"examples": len(texts),
"duplicates_collapsed": (
sum(len(example["scores"]) for example in grouped_examples.values())
- len(texts)
),
"rows_without_target_label": len(rows) - target_labeled_rows,
"skipped_rows": skipped_rows,
"neutral_threshold": neutral_threshold,
"label_distribution": label_distribution,
}
return texts, labels, target_scores, published_dates, stats
def _directional_scores(model: Pipeline, texts: Iterable[str]) -> list[float]:
probabilities = model.predict_proba(list(texts))
classes = list(model.classes_)
bullish_index = classes.index("bullish")
bearish_index = classes.index("bearish")
return [
float(row[bullish_index] - row[bearish_index]) for row in probabilities
]
def evaluate_model(
model: Pipeline,
test_texts: list[str],
test_labels: list[str],
test_target_scores: list[float],
) -> dict[str, Any]:
"""Evaluate class predictions and the derived continuous score."""
predictions = model.predict(test_texts)
directional_scores = _directional_scores(model, test_texts)
return {
"test_examples": len(test_texts),
"accuracy": float(accuracy_score(test_labels, predictions)),
"balanced_accuracy": float(
balanced_accuracy_score(test_labels, predictions)
),
"macro_f1": float(
f1_score(test_labels, predictions, labels=LABELS, average="macro")
),
"score_mae": float(mean_absolute_error(test_target_scores, directional_scores)),
"classification_report": classification_report(
test_labels,
predictions,
labels=LABELS,
output_dict=True,
zero_division=0,
),
"confusion_matrix": {
"labels": list(LABELS),
"matrix": confusion_matrix(
test_labels, predictions, labels=LABELS
).tolist(),
},
}
def train_model(
database_path: Path,
model_path: Path,
report_path: Path,
score_source: str = "llm",
neutral_threshold: float = 0.05,
test_size: float = 0.2,
random_state: int = 42,
min_df: int = 2,
max_features: int = 50_000,
scope: str = "all",
extra_labels_path: Path | None = None,
) -> dict[str, Any]:
"""Train, evaluate, and persist a model bundle."""
if not 0 < test_size < 1:
raise ValueError("test size must be between 0 and 1")
if min_df < 1:
raise ValueError("min-df must be at least 1")
if max_features < 1:
raise ValueError("max-features must be at least 1")
texts, labels, target_scores, published_dates, data_stats = (
load_training_examples(
database_path,
score_source,
neutral_threshold,
scope=scope,
extra_labels_path=extra_labels_path,
)
)
(
train_texts,
test_texts,
train_labels,
test_labels,
_,
test_target_scores,
_,
_,
) = train_test_split(
texts,
labels,
target_scores,
published_dates,
test_size=test_size,
random_state=random_state,
stratify=labels,
)
stratified_model = build_pipeline(random_state, min_df, max_features)
stratified_model.fit(train_texts, train_labels)
stratified_evaluation = evaluate_model(
stratified_model, test_texts, test_labels, test_target_scores
)
ordered_indices = sorted(
range(len(texts)), key=lambda index: (published_dates[index], index)
)
temporal_test_count = len(test_texts)
temporal_train_indices = ordered_indices[:-temporal_test_count]
temporal_test_indices = ordered_indices[-temporal_test_count:]
temporal_model = build_pipeline(random_state, min_df, max_features)
temporal_model.fit(
[texts[index] for index in temporal_train_indices],
[labels[index] for index in temporal_train_indices],
)
temporal_evaluation = evaluate_model(
temporal_model,
[texts[index] for index in temporal_test_indices],
[labels[index] for index in temporal_test_indices],
[target_scores[index] for index in temporal_test_indices],
)
temporal_evaluation.update(
{
"train_through": published_dates[temporal_train_indices[-1]],
"test_from": published_dates[temporal_test_indices[0]],
}
)
evaluation = {
"stratified_holdout": stratified_evaluation,
"temporal_holdout": temporal_evaluation,
}
final_model = build_pipeline(random_state, min_df, max_features)
final_model.fit(texts, labels)
trained_at = datetime.now(timezone.utc).isoformat()
training = {
**data_stats,
"test_size": test_size,
"random_state": random_state,
"min_df": min_df,
"max_features": max_features,
"domain_features": [
"explicit_xau_direction",
"usd_direction",
"treasury_yield_direction",
"hawkish_or_dovish_policy",
],
}
bundle = {
"format_version": FORMAT_VERSION,
"target": TARGET_SYMBOL,
"trained_at": trained_at,
"sklearn_version": sklearn.__version__,
"model_type": "domain_augmented_tfidf_logistic_regression",
"score_definition": "P(XAUUSD bullish) - P(XAUUSD bearish)",
"training": training,
"evaluation": evaluation,
"model": final_model,
}
model_path.parent.mkdir(parents=True, exist_ok=True)
report_path.parent.mkdir(parents=True, exist_ok=True)
joblib.dump(bundle, model_path)
report = {
"model_path": str(model_path),
"format_version": FORMAT_VERSION,
"target": TARGET_SYMBOL,
"trained_at": trained_at,
"sklearn_version": sklearn.__version__,
"model_type": bundle["model_type"],
"score_definition": bundle["score_definition"],
"training": training,
"evaluation": evaluation,
}
report_path.write_text(
json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
return {**report, "report_path": str(report_path)}
def load_model(model_path: str | Path = DEFAULT_MODEL_PATH) -> dict[str, Any]:
"""Load and validate a model bundle created by this module."""
path = Path(model_path)
if not path.is_file():
raise FileNotFoundError(f"model not found: {path}")
bundle = joblib.load(path)
if not isinstance(bundle, dict) or "model" not in bundle:
raise ValueError(f"invalid model bundle: {path}")
if bundle.get("format_version") != FORMAT_VERSION:
raise ValueError(
f"unsupported model format: {bundle.get('format_version')!r}"
)
if bundle.get("target") != TARGET_SYMBOL:
raise ValueError(f"model target must be {TARGET_SYMBOL}: {path}")
return bundle
def score_news(
model_bundle: dict[str, Any],
headline: str,
description: str | None = None,
) -> dict[str, Any]:
"""Score one news article for its expected XAUUSD sentiment."""
text = article_text(headline, description)
model = model_bundle["model"]
probabilities = model.predict_proba([text])[0]
classes = [str(label) for label in model.classes_]
probability_by_label = {
label: float(probabilities[index]) for index, label in enumerate(classes)
}
for label in LABELS:
probability_by_label.setdefault(label, 0.0)
label = classes[int(probabilities.argmax())]
score = probability_by_label["bullish"] - probability_by_label["bearish"]
return {
"target": TARGET_SYMBOL,
"label": label,
"score": round(score, 6),
"confidence": round(probability_by_label[label], 6),
"probabilities": {
name: round(probability_by_label[name], 6) for name in LABELS
},
}
def _score_jsonl(model_bundle: dict[str, Any], input_path: str) -> None:
handle = (
sys.stdin if input_path == "-" else Path(input_path).open(encoding="utf-8")
)
try:
for line_number, line in enumerate(handle, start=1):
if not line.strip():
continue
try:
article = json.loads(line)
except json.JSONDecodeError as error:
raise ValueError(
f"invalid JSON on input line {line_number}: {error.msg}"
) from error
if not isinstance(article, dict):
raise ValueError(f"input line {line_number} must be a JSON object")
headline = article.get(
"headline", article.get("title", article.get("text"))
)
description = article.get("description")
if headline is None and description is None:
raise ValueError(
f"input line {line_number} needs headline, title, text, or description"
)
output = dict(article)
output["sentiment"] = score_news(
model_bundle, str(headline or ""), description
)
print(json.dumps(output, ensure_ascii=False))
finally:
if handle is not sys.stdin:
handle.close()
def build_argument_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Train and run a TF-IDF XAUUSD news sentiment model."
)
subparsers = parser.add_subparsers(dest="command", required=True)
train_parser = subparsers.add_parser(
"train", help="train from XAU-labeled rows in the headlines table"
)
train_parser.add_argument("--db", type=Path, default=Path("newsdata.db"))
train_parser.add_argument("--model", type=Path, default=DEFAULT_MODEL_PATH)
train_parser.add_argument(
"--report",
type=Path,
help="metrics JSON path (default: MODEL with .metrics.json suffix)",
)
train_parser.add_argument(
"--score-source",
choices=("llm", "keyword", "all"),
default="llm",
help="which existing scores to use as labels (default: llm)",
)
train_parser.add_argument(
"--scope",
choices=SCOPES,
default="all",
help="all headlines, or only those with a gold XAU key (default: all)",
)
train_parser.add_argument(
"--extra-labels",
type=Path,
metavar="JSONL",
help="fallback labels keyed by headline hash, field xau",
)
train_parser.add_argument("--neutral-threshold", type=float, default=0.05)
train_parser.add_argument("--test-size", type=float, default=0.2)
train_parser.add_argument("--random-state", type=int, default=42)
train_parser.add_argument("--min-df", type=int, default=2)
train_parser.add_argument("--max-features", type=int, default=50_000)
score_parser = subparsers.add_parser(
"score", help="score news sentiment for XAUUSD"
)
score_parser.add_argument("--model", type=Path, default=DEFAULT_MODEL_PATH)
input_group = score_parser.add_mutually_exclusive_group(required=True)
input_group.add_argument("--headline", help="headline or short news text")
input_group.add_argument(
"--input",
metavar="JSONL",
help="JSONL path, or - to read JSONL from standard input",
)
score_parser.add_argument("--description", help="optional article description")
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_argument_parser()
args = parser.parse_args(argv)
try:
if args.command == "train":
report_path = args.report or args.model.with_suffix(".metrics.json")
report = train_model(
database_path=args.db,
model_path=args.model,
report_path=report_path,
score_source=args.score_source,
neutral_threshold=args.neutral_threshold,
test_size=args.test_size,
random_state=args.random_state,
min_df=args.min_df,
max_features=args.max_features,
scope=args.scope,
extra_labels_path=args.extra_labels,
)
print(json.dumps(report, indent=2, sort_keys=True))
elif args.input:
_score_jsonl(load_model(args.model), args.input)
else:
prediction = score_news(
load_model(args.model), args.headline, args.description
)
print(json.dumps(prediction, indent=2))
except (FileNotFoundError, OSError, sqlite3.Error, ValueError) as error:
print(f"error: {error}", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())