Turn Earnings Call Text into Gold Price Signals in 45 Minutes

# Personal note: Learned this after corrupting 200+ files with encoding errors
import pandas as pd
from pathlib import Path
import re
from datetime import datetime

def clean_earnings_text(raw_text):
    """Remove artifacts that break NLP models"""
    # Remove operator instructions
    text = re.sub(r'Operator:.*?\n', '', raw_text)
    # Fix common OCR errors in PDFs
    text = text.replace('”"', '--').replace('’', "'")
    # Remove multiple spaces
    text = re.sub(r'\s+', ' ', text)
    
    # Watch out: Don't strip all newlines or you lose speaker changes
    return text.strip()

def load_earnings_corpus(data_dir):
    """Load and index all transcripts"""
    transcripts = []
    
    for file_path in Path(data_dir).glob("*.txt"):
        # Extract metadata from filename: TICKER_YYYY-MM-DD.txt
        parts = file_path.stem.split('_')
        ticker = parts[0]
        date = datetime.strptime(parts[1], '%Y-%m-%d')
        
        with open(file_path, 'r', encoding='utf-8') as f:
            raw_text = f.read()
        
        transcripts.append({
            'ticker': ticker,
            'date': date,
            'text': clean_earnings_text(raw_text),
            'word_count': len(raw_text.split())
        })
    
    return pd.DataFrame(transcripts)

# Load corpus
corpus_df = load_earnings_corpus('./earnings_data')
print(f"Loaded {len(corpus_df)} transcripts")
print(f"Date range: {corpus_df['date'].min()} to {corpus_df['date'].max()}")

import numpy as np
from collections import Counter

class EarningsFeatureExtractor:
    """Extract gold-mining specific signals from text"""
    
    def __init__(self):
        # Personal note: Built this list from 6 months of manual reading
        self.production_terms = [
            'ounces produced', 'production guidance', 'grade decline',
            'mill throughput', 'recovery rate', 'reserve replacement'
        ]
        
        self.cost_terms = [
            'all-in sustaining cost', 'aisc', 'cost pressure',
            'labor inflation', 'energy costs', 'royalty'
        ]
        
        self.risk_terms = [
            'permitting delay', 'community opposition', 'water shortage',
            'geotechnical', 'strike', 'political risk', 'expropriation'
        ]
        
        self.expansion_terms = [
            'expansion project', 'brownfield', 'greenfield',
            'acquisition target', 'merger', 'capex increase'
        ]
    
    def extract_features(self, text):
        """Generate feature vector from transcript"""
        text_lower = text.lower()
        words = text_lower.split()
        
        features = {}
        
        # 1. Term frequency features (normalized)
        doc_length = len(words)
        features['production_intensity'] = sum(
            text_lower.count(term) for term in self.production_terms
        ) / doc_length * 1000  # Per 1000 words
        
        features['cost_intensity'] = sum(
            text_lower.count(term) for term in self.cost_terms
        ) / doc_length * 1000
        
        features['risk_intensity'] = sum(
            text_lower.count(term) for term in self.risk_terms
        ) / doc_length * 1000
        
        features['expansion_intensity'] = sum(
            text_lower.count(term) for term in self.expansion_terms
        ) / doc_length * 1000
        
        # 2. Sentiment proxies
        # Watch out: Don't use generic sentiment—mining is industry-specific
        positive_words = ['beat expectations', 'record production', 
                         'ahead of schedule', 'outperform', 'upside']
        negative_words = ['below guidance', 'delay', 'shortfall', 
                         'impairment', 'suspend', 'downgrade']
        
        features['positive_ratio'] = sum(
            text_lower.count(w) for w in positive_words
        ) / (doc_length / 1000)
        
        features['negative_ratio'] = sum(
            text_lower.count(w) for w in negative_words
        ) / (doc_length / 1000)
        
        # 3. Forward-looking statements
        future_terms = ['guidance', 'forecast', 'expect', 'project', 'outlook']
        features['forward_looking_density'] = sum(
            text_lower.count(term) for term in future_terms
        ) / doc_length * 1000
        
        # 4. Uncertainty markers
        uncertainty = ['uncertain', 'unclear', 'difficult to predict',
                      'may', 'might', 'could', 'possibly']
        features['uncertainty_score'] = sum(
            text_lower.count(term) for term in uncertainty
        ) / doc_length * 1000
        
        # 5. Numerical density (proxy for quantitative discussion)
        numbers = re.findall(r'\b\d+(?:\.\d+)?\b', text)
        features['numerical_density'] = len(numbers) / doc_length * 1000
        
        # 6. Executive Q&A engagement
        qa_section = re.search(r'question-and-answer.*', text_lower, re.DOTALL)
        if qa_section:
            qa_text = qa_section.group()
            features['qa_length_ratio'] = len(qa_text.split()) / doc_length
            features['question_count'] = qa_text.count('question:')
        else:
            features['qa_length_ratio'] = 0
            features['question_count'] = 0
        
        # 7. Hedging language (executives protecting themselves)
        hedges = ['approximately', 'roughly', 'around', 'about', 'estimated']
        features['hedging_intensity'] = sum(
            text_lower.count(h) for h in hedges
        ) / doc_length * 1000
        
        return features

# Extract features for entire corpus
extractor = EarningsFeatureExtractor()
feature_list = []

for idx, row in corpus_df.iterrows():
    features = extractor.extract_features(row['text'])
    features['ticker'] = row['ticker']
    features['date'] = row['date']
    feature_list.append(features)
    
    if idx % 100 == 0:
        print(f"Processed {idx}/{len(corpus_df)} transcripts")

features_df = pd.DataFrame(feature_list)
print(f"\nExtracted {len(features_df.columns)-2} features")
print(features_df.describe())

# Personal note: This step requires 4GB GPU or takes 3x longer on CPU
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
from torch.nn.functional import softmax

class FinBERTFeatureExtractor:
    """Use pre-trained financial BERT for deep features"""
    
    def __init__(self):
        self.tokenizer = AutoTokenizer.from_pretrained(
            "ProsusAI/finbert"
        )
        self.model = AutoModelForSequenceClassification.from_pretrained(
            "ProsusAI/finbert"
        )
        self.model.eval()
        
        # Watch out: Move to GPU if available
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.model.to(self.device)
    
    def extract_sentence_sentiment(self, text, max_length=512):
        """Get sentiment for text chunks"""
        # Split into sentences (finbert has 512 token limit)
        sentences = re.split(r'[.!?]+', text)
        
        sentiments = []
        for sentence in sentences[:50]:  # Limit to first 50 sentences
            if len(sentence.strip()) < 10:
                continue
            
            inputs = self.tokenizer(
                sentence, 
                return_tensors="pt", 
                truncation=True, 
                max_length=max_length,
                padding=True
            ).to(self.device)
            
            with torch.no_grad():
                outputs = self.model(**inputs)
                probs = softmax(outputs.logits, dim=1)
            
            # Labels: negative, neutral, positive
            sentiments.append(probs[0].cpu().numpy())
        
        # Aggregate sentiment across document
        if sentiments:
            avg_sentiment = np.mean(sentiments, axis=0)
            return {
                'finbert_negative': float(avg_sentiment[0]),
                'finbert_neutral': float(avg_sentiment[1]),
                'finbert_positive': float(avg_sentiment[2])
            }
        return {'finbert_negative': 0, 'finbert_neutral': 1, 'finbert_positive': 0}

# Add FinBERT features
finbert = FinBERTFeatureExtractor()

finbert_features = []
for idx, row in corpus_df.iterrows():
    sentiment = finbert.extract_sentence_sentiment(row['text'])
    sentiment['ticker'] = row['ticker']
    sentiment['date'] = row['date']
    finbert_features.append(sentiment)
    
    if idx % 50 == 0:
        print(f"FinBERT processed {idx}/{len(corpus_df)}")

finbert_df = pd.DataFrame(finbert_features)

# Merge with existing features
features_df = features_df.merge(
    finbert_df, 
    on=['ticker', 'date'], 
    how='left'
)

print(f"Total features: {len(features_df.columns)-2}")

import yfinance as yf

def align_features_with_target(features_df, target_ticker='GC=F', lag_days=5):
    """Merge text features with gold price movements"""
    
    # Download gold futures data
    gold = yf.download(
        target_ticker,
        start=features_df['date'].min(),
        end=features_df['date'].max() + pd.Timedelta(days=30),
        progress=False
    )
    
    # Calculate forward returns (what we're predicting)
    gold['return_5d'] = gold['Close'].pct_change(5).shift(-5)
    gold['return_10d'] = gold['Close'].pct_change(10).shift(-10)
    gold['return_20d'] = gold['Close'].pct_change(20).shift(-20)
    
    gold_features = gold[['return_5d', 'return_10d', 'return_20d']].reset_index()
    gold_features.columns = ['date'] + list(gold_features.columns[1:])
    
    # Merge on date (broadcasts earnings features to daily prices)
    # Watch out: This creates multiple rows per earnings report
    merged = features_df.merge(
        gold_features,
        left_on='date',
        right_on='date',
        how='left'
    )
    
    # Forward-fill earnings features to next earnings date
    merged = merged.sort_values(['ticker', 'date'])
    merged = merged.groupby('ticker').apply(
        lambda x: x.ffill(limit=90)  # Max 90 days between earnings
    ).reset_index(drop=True)
    
    # Remove NaN targets (insufficient future data)
    merged = merged.dropna(subset=['return_5d', 'return_10d', 'return_20d'])
    
    print(f"Training samples: {len(merged)}")
    print(f"Date range: {merged['date'].min()} to {merged['date'].max()}")
    print(f"\nTarget distribution:")
    print(merged[['return_5d', 'return_10d', 'return_20d']].describe())
    
    return merged

# Create modeling dataset
modeling_df = align_features_with_target(features_df)

# Save for modeling
modeling_df.to_parquet('gold_earnings_features.parquet', index=False)
print("\nSaved features to gold_earnings_features.parquet")

def validate_features(df):
    """Check for common feature engineering errors"""
    
    issues = []
    
    # 1. Check for data leakage (future info in features)
    date_cols = df.select_dtypes(include=['datetime64']).columns
    if len(date_cols) > 1:
        issues.append(f"⚠️  Multiple date columns: {list(date_cols)}")
    
    # 2. Check feature variance
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    low_variance = []
    for col in numeric_cols:
        if df[col].std() < 0.01:
            low_variance.append(col)
    
    if low_variance:
        issues.append(f"⚠️  Low variance features: {low_variance}")
    
    # 3. Check for inf/-inf values
    inf_cols = []
    for col in numeric_cols:
        if np.isinf(df[col]).any():
            inf_cols.append(col)
    
    if inf_cols:
        issues.append(f"❌ Infinite values in: {inf_cols}")
    
    # 4. Check missing value patterns
    missing = df.isnull().sum()
    missing = missing[missing > 0]
    if len(missing) > 0:
        issues.append(f"⚠️  Missing values:\n{missing}")
    
    # 5. Check feature correlations (detect redundant features)
    corr_matrix = df[numeric_cols].corr().abs()
    upper_triangle = corr_matrix.where(
        np.triu(np.ones(corr_matrix.shape), k=1).astype(bool)
    )
    high_corr = [
        column for column in upper_triangle.columns 
        if any(upper_triangle[column] > 0.95)
    ]
    
    if high_corr:
        issues.append(f"⚠️  Highly correlated features (>0.95): {high_corr}")
    
    # Report
    if issues:
        print("🔍 Feature Validation Issues:\n")
        for issue in issues:
            print(issue)
    else:
        print("✅ All features passed validation")
    
    # Summary stats
    print(f"\n📊 Feature Summary:")
    print(f"   Total samples: {len(df):,}")
    print(f"   Features: {len(numeric_cols)}")
    print(f"   Date range: {df['date'].min()} to {df['date'].max()}")
    print(f"   Tickers: {df['ticker'].nunique()}")

# Validate
validate_features(modeling_df)