| |
| |
| |
| |
|
|
| import os |
| import re |
| from datetime import datetime, timedelta |
| from functools import lru_cache |
|
|
| import numpy as np |
| import pandas as pd |
| import yfinance as yf |
|
|
| |
| |
| |
| MODEL_NAME = os.getenv("FINANCIAL_MODEL", "Ani-404/finbert-model") |
|
|
|
|
| @lru_cache(maxsize=1) |
| def get_classifier(): |
| """Load and cache the financial sentiment classifier pipeline.""" |
| from transformers import pipeline |
|
|
| return pipeline( |
| "text-classification", |
| model=MODEL_NAME, |
| tokenizer=MODEL_NAME, |
| ) |
|
|
|
|
| def ingest_transcripts(project_root: str) -> pd.DataFrame: |
| """Load the sample earnings-call transcripts.""" |
| file_path = os.path.join(project_root, "finance", "sample_transcripts_advanced.csv") |
| try: |
| return pd.read_csv(file_path) |
| except FileNotFoundError: |
| raise FileNotFoundError(f"Could not find transcript data at '{file_path}'.") |
|
|
|
|
| def preprocess_and_split(text): |
| """Clean text and split it into sentences.""" |
| if not isinstance(text, str): |
| return [] |
| text = re.sub(r"\s+", " ", text).strip() |
| sentences = re.split(r"(?<=[.!?])\s+", text) |
| return [s.strip() for s in sentences if s.strip()] |
|
|
|
|
| def _label_to_score(label: str) -> int: |
| """Map a sentiment label to a simplified -1 / 0 / +1 score.""" |
| l = label.lower() |
| if "pos" in l: |
| return 1 |
| if "neg" in l: |
| return -1 |
| return 0 |
|
|
|
|
| def score_sentences(sentences, classifier=None): |
| """Run the classifier on sentences and return a -1/0/+1 score each.""" |
| if not sentences: |
| return [] |
| classifier = classifier or get_classifier() |
| results = classifier(sentences, truncation=True, max_length=128) |
| return [_label_to_score(r["label"]) for r in results] |
|
|
|
|
| def aggregate_sentiment_features(scores): |
| """Aggregate sentence scores into summary features.""" |
| if not scores: |
| return {"mean_score": 0.0, "positive_ratio": 0.0, "negative_ratio": 0.0} |
|
|
| total_sentences = len(scores) |
| positive_sentences = sum(1 for s in scores if s > 0) |
| negative_sentences = sum(1 for s in scores if s < 0) |
|
|
| return { |
| "mean_score": float(np.mean(scores)), |
| "positive_ratio": positive_sentences / total_sentences, |
| "negative_ratio": negative_sentences / total_sentences, |
| } |
|
|
|
|
| def get_stock_returns(ticker, earnings_date_str): |
| """Fetch 1-day and 5-day returns following an earnings date.""" |
| try: |
| earnings_date = datetime.strptime(earnings_date_str, "%Y-%m-%d") |
| start_date = earnings_date - timedelta(days=1) |
| end_date = earnings_date + timedelta(days=10) |
|
|
| stock_data = yf.download( |
| ticker, start=start_date, end=end_date, auto_adjust=True, progress=False |
| ) |
| if stock_data.empty: |
| return None |
|
|
| |
| if isinstance(stock_data.columns, pd.MultiIndex): |
| stock_data.columns = stock_data.columns.get_level_values(0) |
|
|
| price_on_date_series = stock_data.loc[stock_data.index >= earnings_date] |
| if price_on_date_series.empty: |
| return None |
| price_on_date = float(price_on_date_series.iloc[0]["Close"]) |
|
|
| post_earnings_prices = stock_data.loc[stock_data.index > earnings_date] |
| if len(post_earnings_prices) < 5: |
| return None |
|
|
| price_1d_after = float(post_earnings_prices.iloc[0]["Close"]) |
| price_5d_after = float(post_earnings_prices.iloc[4]["Close"]) |
|
|
| return { |
| "return_1d": (price_1d_after - price_on_date) / price_on_date, |
| "return_5d": (price_5d_after - price_on_date) / price_on_date, |
| } |
| except Exception: |
| return None |
|
|
