-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathevaluate_stock_response.py
More file actions
237 lines (189 loc) · 7.58 KB
/
evaluate_stock_response.py
File metadata and controls
237 lines (189 loc) · 7.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
import re
import textstat
from finance_agent import run_finance_agent
from functions import score_based_on_financials,analyze_financials
import os
from groq import Groq
from huggingface_hub import InferenceClient
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from scrapeNews import extract_news
from functions import get_Full_ticker
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
def evaluate_stock_analysis(response):
results = {}
# Check presence of required sections
sections = ["Stock Price", "Sentiment", "Financials", "Final Verdict"]
results["section_coverage"] = sum(1 for s in sections if s.lower() in response.lower())
results["structure_score (0-1)"] = round(results["section_coverage"] / len(sections), 2)
# Count relevant emojis
emojis = ["📈", "💰", "⚠️", "✅", "📉", "⭐", "📎", "🟢", "🔴", "🟡"]
results["emoji_count"] = sum(response.count(e) for e in emojis)
# Basic text stats
results["word_count"] = len(response.split())
# Readability metrics
results["readability"] = {
"flesch_reading_ease": textstat.flesch_reading_ease(response),
"flesch_kincaid_grade": textstat.flesch_kincaid_grade(response),
"gunning_fog": textstat.gunning_fog(response),
"automated_readability_index": textstat.automated_readability_index(response)
}
# Try to classify sentiment from the Verdict section
results["verdict_sentiment"] = "Missing"
verdict_match = re.search(r"\*\*⭐ Final Verdict\*\*(.*?)\*\*", response, re.DOTALL | re.IGNORECASE)
if verdict_match:
verdict_text = verdict_match.group(1).strip().lower()
if "buy" in verdict_text or "positive" in verdict_text:
results["verdict_sentiment"] = "Positive"
elif "sell" in verdict_text or "negative" in verdict_text:
results["verdict_sentiment"] = "Negative"
elif "neutral" in verdict_text or "caution" in verdict_text:
results["verdict_sentiment"] = "Neutral"
else:
results["verdict_sentiment"] = "Unclear"
if results["verdict_sentiment"] == "Missing":
lower_text = response.lower()
if "bullish" in lower_text or "buy" in lower_text or "positive" in lower_text:
results["verdict_sentiment"] = "Positive"
elif "bearish" in lower_text or "sell" in lower_text or "negative" in lower_text:
results["verdict_sentiment"] = "Negative"
elif "neutral" in lower_text or "caution" in lower_text:
results["verdict_sentiment"] = "Neutral"
else:
results["verdict_sentiment"] = "Unclear"
evaluation = ""
for key, value in results.items():
evaluation += f"{key}: {value}\n"
return evaluation
def evaluateScore(symbol):
check = analyze_financials(symbol)
cash_summary = check['cash_summary']
valuation_summary=check['valuation_summary']
total_summary = {**cash_summary,**valuation_summary}
score = score_based_on_financials(total_summary)
print(score)
def finbert(sentiment):
client = InferenceClient(
provider="hf-inference",
api_key=os.environ.get("HUGGIN")
)
result = client.text_classification(
sentiment,
model="ProsusAI/finbert",
)
score=0
for entry in result:
if entry.label.lower()=='positive':
score+=100*entry.score
elif entry.label.lower()=='neutral':
score+=50*entry.score
score = round(score,2)
return score
def vader(sentiment):
analyzer = SentimentIntensityAnalyzer()
score = analyzer.polarity_scores(sentiment)['compound']
vader_score = (score + 1) / 2 * 100
return vader_score
def llama_sentiment_score(text):
prompt = f"""
Classify the sentiment of the following financial news headline or summary as a number between 0 and 100:
- 0 = extremely negative
- 50 = neutral
- 100 = extremely positive
Only return the number. No explanation.
Text:
"{text}"
"""
chat_completion = client.chat.completions.create(
model="llama3-70b-8192",
messages=[{"role": "user", "content": prompt}],
max_tokens=10
)
try:
return float(chat_completion.choices[0].message.content.strip())
except:
return None
def evaluate_sentiment(sentiment):
llama_score = llama_sentiment_score(sentiment)
finbert_score = finbert(sentiment)
vader_score = vader(sentiment)
result={
"llama":llama_score,
"finbert":finbert_score,
"vader":vader_score
}
# print(result)
return result
def news_sentiment_score(symbol):
news = extract_news(symbol)
news_list = news.split('\n')
size = len(news_list)
result={
"llama":0,
"finbert":0,
"vader":0
}
for article in news_list:
print(article)
evalu = evaluate_sentiment(article)
result['finbert']+=evalu['finbert']
result['llama']+=evalu['llama']
result['vader']+=evalu['vader']
for key in result:
result[key] = round(result[key] / size, 2) # average to 2 decimal places
print(result)
return result
# 🧪 Example Usage
# if __name__ == "__main__":
# try:
# with open("tatamotors_sample.txt", "r") as f:
# response_text = f.read()
# except FileNotFoundError:
# print("❌ File 'tatamotors_sample.txt' not found. Please place it in the same folder.")
# exit()
# print("📊 Evaluation Report:")
# evaluation = evaluate_stock_analysis(response_text)
# for key, value in evaluation.items():
# print(f"{key}: {value}")
test_prompts = {
"What’s the outlook for TCS this quarter?": "TCS.NS",
"Can you evaluate Apple’s stock performance?": "AAPL",
"Tell me if I should invest in coca cola right now.": "KO",
"Is Nvidia still a good long-term bet?": "NVDA",
"What's the financial standing of Reliance?": "RELIANCE.NS",
"Give me a breakdown on Microsoft stock.": "MSFT",
"I’m considering HDFC Bank — what are the risks?": "HDFCBANK.NS",
"What’s going on with Tesla these days?": "TSLA",
"Any updates on the earnings for yesBank?": "YESBANK.NS",
"Evaluate Alphabet’s current market position.": "GOOGL"
}
def extract_company(prompt):
chat_completion = client.chat.completions.create(
messages=[
{"role": "system", "content": "You extract official company names from finance-related user queries."},
{"role": "user", "content": f"""
You are a financial assistant. Extract the **stock ticker** from the following user query:
\"\"\"{prompt}\"\"\"
- If no company name is found, return 'None'.
- Do **not** include any other text.
"""}
],
model="llama3-70b-8192",
max_tokens=15
)
company_name = chat_completion.choices[0].message.content.strip()
if company_name.lower() != "none":
if company_name and '.' in company_name:
company_name = company_name.split('.')[0]
return get_Full_ticker(company_name)
return None
# correct = 0
# print("\n🔍 Ticker Extraction Test Results:\n")
# for prompt, expected in test_prompts.items():
# extracted = extract_company(prompt)
# status = "✅" if extracted == expected else "❌"
# print(f"{status} Prompt: {prompt}\nExpected: {expected}, Extracted: {extracted}\n")
# if extracted == expected:
# correct += 1
# accuracy = (correct / len(test_prompts)) * 100
# print(f"🎯 Final Accuracy: {correct}/{len(test_prompts)} = {accuracy:.2f}%\n")
news_sentiment_score('ABNB')