#!/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())