Machine Learning in Trading Strategy Design › Sentiment Analysis for Trading Signals

import re


def preprocess_financial_text(text):
    """
    Preprocess financial text for sentiment analysis.
    """
    # Convert to lowercase
    text = text.lower()

    # Preserve ticker symbols (typically $AAPL or just AAPL)
    ticker_pattern = r"\$?[A-Z]{1,5}(?=\s|$|,|\.)"
    tickers = re.findall(ticker_pattern, text, re.IGNORECASE)

    # Handle common financial abbreviations
    abbreviations = {
        "q1": "first quarter",
        "q2": "second quarter",
        "q3": "third quarter",
        "q4": "fourth quarter",
        "yoy": "year over year",
        "qoq": "quarter over quarter",
        "eps": "earnings per share",
        "p/e": "price to earnings",
        "roi": "return on investment",
        "ebitda": "ebitda",
    }
    for abbr, expansion in abbreviations.items():
        text = re.sub(r"\b" + abbr + r"\b", expansion, text)

    # Remove URLs
    text = re.sub(r"http\S+|www\.\S+", "", text)

    # Remove special characters but keep sentence structure
    text = re.sub(r"[^\w\s\.\,\!\?]", "", text)

    # Normalize whitespace
    text = " ".join(text.split())

    return text, tickers


# Example financial headlines
headlines = [
    "Apple beats Q3 earnings estimates, stock surges 5% in after-hours trading",
    "Fed signals rate cuts ahead, markets rally on dovish commentary",
    "Tesla misses delivery targets, shares plunge amid production concerns",
    "Goldman Sachs upgrades MSFT to buy, cites cloud growth momentum",
    "Oil prices crash as OPEC+ fails to reach production agreement",
]

processed_headlines = [preprocess_financial_text(h) for h in headlines]

# Financial sentiment lexicon (simplified Loughran-McDonald style)
positive_words = {
    "beat",
    "beats",
    "exceeded",
    "surged",
    "surge",
    "rally",
    "rallied",
    "upgrade",
    "upgrades",
    "growth",
    "profit",
    "profitable",
    "gain",
    "gains",
    "positive",
    "strong",
    "strength",
    "outperform",
    "buy",
    "bullish",
    "momentum",
    "opportunity",
    "success",
    "successful",
    "improve",
    "improved",
}

negative_words = {
    "miss",
    "misses",
    "missed",
    "plunge",
    "plunged",
    "crash",
    "crashed",
    "downgrade",
    "downgrades",
    "loss",
    "losses",
    "decline",
    "declined",
    "negative",
    "weak",
    "weakness",
    "underperform",
    "sell",
    "bearish",
    "concern",
    "concerns",
    "risk",
    "risks",
    "fail",
    "failed",
    "warning",
}


def dictionary_sentiment(text, positive_dict, negative_dict):
    """
    Calculate sentiment score using dictionary approach.
    Returns score between -1 (very negative) and +1 (very positive).
    """
    words = text.lower().split()

    pos_count = sum(1 for w in words if w in positive_dict)
    neg_count = sum(1 for w in words if w in negative_dict)

    total = pos_count + neg_count
    if total == 0:
        return 0.0

    # Normalized sentiment score
    score = (pos_count - neg_count) / total

    return score, pos_count, neg_count


# Analyze headlines
sentiment_results = []
for original, (processed, tickers) in zip(headlines, processed_headlines):
    score, pos, neg = dictionary_sentiment(
        processed, positive_words, negative_words
    )
    sentiment_results.append(
        {
            "headline": original[:40] + "...",
            "sentiment": score,
            "positive_words": pos,
            "negative_words": neg,
        }
    )

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline

# Create synthetic training data for demonstration
# In practice, you would use labeled datasets like Financial PhraseBank
training_texts = [
    "Company reports record profits and raises dividend",
    "Strong quarterly results exceed analyst expectations",
    "Revenue growth accelerates amid market expansion",
    "Management announces stock buyback program",
    "Earnings beat consensus estimates significantly",
    "Company misses revenue targets by wide margin",
    "Profit warning issued ahead of quarterly report",
    "Sales decline accelerates as competition intensifies",
    "Management cuts full year guidance substantially",
    "Restructuring charges lead to quarterly loss",
    "Market conditions remain challenging for growth",
    "Regulatory concerns weigh on stock performance",
    "Supply chain disruptions impact production capacity",
    "Debt levels raise concerns among analysts",
    "Customer losses accelerate amid service issues",
]
training_labels = [
    1,
    1,
    1,
    1,
    1,
    0,
    0,
    0,
    0,
    0,
    0,
    0,
    0,
    0,
    0,
]  # 1=positive, 0=negative

# Build sentiment classification pipeline
sentiment_pipeline = Pipeline(
    [
        ("tfidf", TfidfVectorizer(ngram_range=(1, 2), max_features=500)),
        ("classifier", MultinomialNB(alpha=0.1)),
    ]
)

sentiment_pipeline.fit(training_texts, training_labels)

# Analyze new headlines
ml_predictions = []
for original, (processed, tickers) in zip(headlines, processed_headlines):
    prob = sentiment_pipeline.predict_proba([processed])[0]
    ml_predictions.append(
        {
            "headline": original[:40] + "...",
            "negative_prob": prob[0],
            "positive_prob": prob[1],
            "prediction": "Positive" if prob[1] > 0.5 else "Negative",
        }
    )

def aggregate_sentiment_signals(sentiment_scores, timestamps, decay_halflife=3):
    """
    Aggregate multiple sentiment observations into a single signal.
    Uses exponential decay weighting: more recent observations matter more.

    Parameters:

    - sentiment_scores: Array of individual sentiment scores
    - timestamps: Array of observation times (in days relative to now)
    - decay_halflife: Half-life for exponential decay in days
    """
    # Calculate decay weights
    decay_rate = np.log(2) / decay_halflife
    weights = np.exp(-decay_rate * timestamps)
    weights = weights / weights.sum()  # Normalize

    # Weighted average sentiment
    weighted_sentiment = np.sum(sentiment_scores * weights)

    # Sentiment dispersion (disagreement)
    sentiment_std = np.sqrt(
        np.sum(weights * (sentiment_scores - weighted_sentiment) ** 2)
    )

    # Sentiment momentum: compare recent vs older
    recent_mask = timestamps <= decay_halflife
    if recent_mask.sum() > 0 and (~recent_mask).sum() > 0:
        recent_sentiment = sentiment_scores[recent_mask].mean()
        older_sentiment = sentiment_scores[~recent_mask].mean()
        sentiment_momentum = recent_sentiment - older_sentiment
    else:
        sentiment_momentum = 0.0

    return {
        "aggregate_sentiment": weighted_sentiment,
        "sentiment_dispersion": sentiment_std,
        "sentiment_momentum": sentiment_momentum,
    }


# Example: aggregate sentiment for a stock over past week
np.random.seed(42)
n_observations = 20
sentiment_scores = np.random.normal(
    0.1, 0.3, n_observations
)  # Slightly positive on average
timestamps = np.random.uniform(0, 7, n_observations)  # Days ago

aggregated = aggregate_sentiment_signals(sentiment_scores, timestamps)