AI-Driven Sentiment Arbitrage: Real-Time NLP Trading with Python, FinBERT, and the MT5 API

import requests
from bs4 import BeautifulSoup
import time

class NewsFeedIngestor:
    def __init__(self, rss_url):
        self.rss_url = rss_url
        self.seen_headlines = set()

    def fetch_latest_headlines(self):
        try:
            response = requests.get(self.rss_url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)
            soup = BeautifulSoup(response.content, "xml")
            items = soup.find_all("item")
            
            new_headlines = []
            for item in items:
                title = item.title.text.strip()
                link = item.link.text.strip()
                if title not in self.seen_headlines:
                    self.seen_headlines.add(title)
                    new_headlines.append({"title": title, "link": link})
            
            return new_headlines
        except Exception as e:
            print(f"Ingestion error from {self.rss_url}: {e}")
            return []

pip install torch transformers

import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification

class FinBertSentimentProcessor:
    def __init__(self):
        self.model_name = "ProsusAI/finbert"
        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(self.model_name)
        
        # Deploy on CUDA if available to achieve sub-10ms inference latency
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)
        self.model.eval()

    def analyze_sentiment(self, text):
        inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True, max_length=512)
        inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            
        # Extract logits and apply Softmax to get probabilities
        probabilities = torch.softmax(outputs.logits, dim=-1).cpu().numpy()[0]
        
        # FinBERT labels: 0 -> Positive, 1 -> Negative, 2 -> Neutral
        pos, neg, neu = probabilities[0], probabilities[1], probabilities[2]
        
        # Compute continuous Sentiment Score S_t
        sentiment_score = pos - neg
        return sentiment_score, {"positive": pos, "negative": neg, "neutral": neu}

import MetaTrader5 as mt5
from news_ingestor import NewsFeedIngestor
from finbert_processor import FinBertSentimentProcessor

# Initialize local MT5 and AI components
processor = FinBertSentimentProcessor()
ingestor = NewsFeedIngestor("https://finance.yahoo.com/rss/headlines?s=GC=F")

if not mt5.initialize():
    print("MT5 initialization failed")
    exit(1)

def dispatch_sentiment_order(symbol, sentiment_score):
    # Retrieve current bid/ask
    tick = mt5.symbol_info_tick(symbol)
    if not tick:
        return
        
    ask = tick.ask
    bid = tick.bid
    
    # Establish thresholds: Score > 0.4 triggers Buy; Score < -0.4 triggers Sell
    if sentiment_score > 0.4:
        # Construct BUY request
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": symbol,
            "volume": 0.1,  # In production, replace with calculated optimal lot size
            "type": mt5.ORDER_TYPE_BUY,
            "price": ask,
            "sl": ask - 5.0,  # $5 Stop Loss on Gold
            "tp": ask + 15.0, # $15 Take Profit
            "deviation": 20,
            "magic": 999120,
            "comment": f"FinBERT Sentiment BUY: {sentiment_score:.2f}",
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_IOC,
        }
        result = mt5.order_send(request)
        print(f"Sent BUY Order. Status: {result.comment}")
        
    elif sentiment_score < -0.4:
        # Construct SELL request
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": symbol,
            "volume": 0.1,
            "type": mt5.ORDER_TYPE_SELL,
            "price": bid,
            "sl": bid + 5.0,
            "tp": bid - 15.0,
            "deviation": 20,
            "magic": 999120,
            "comment": f"FinBERT Sentiment SELL: {sentiment_score:.2f}",
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_IOC,
        }
        result = mt5.order_send(request)
        print(f"Sent SELL Order. Status: {result.comment}")

# Main execution loop polling feeds every 30 seconds
while True:
    new_items = ingestor.fetch_latest_headlines()
    for item in new_items:
        score, details = processor.analyze_sentiment(item["title"])
        print(f"Headline: {item['title']} | Sentiment: {score:.3f}")
        
        # Dispatch order based on calculated score
        dispatch_sentiment_order("XAUUSD", score)
        
    time.sleep(30)