Spaces:
Runtime error
Runtime error
File size: 4,497 Bytes
80b5809 | 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 | import pandas as pd
import xgboost as xgb
import yfinance as yf
from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline
from datetime import datetime, timedelta
import random
import time
import requests
from openai_agent import isolate_context
from schemas import ProcessedStock, StockMetrics
from ticker_data import TICKER_DATA
print("--- [SYSTEM] Loading Quant Models ---")
ml_model = xgb.XGBClassifier()
ml_model.load_model('model_output.json')
tokenizer = AutoTokenizer.from_pretrained("ProsusAI/finbert")
fb_model = AutoModelForSequenceClassification.from_pretrained("ProsusAI/finbert")
nlp = pipeline("sentiment-analysis", model=fb_model, tokenizer=tokenizer)
print("--- [SYSTEM] Ready ---")
def get_live_technicals(ticker: str):
"""
Fetches RSI and MACD from the local TICKER_DATA cache.
Bypasses the Yahoo Finance block on Hugging Face.
"""
# Clean the ticker just in case it has a .NS attached
clean_ticker = ticker.replace(".NS", "")
if clean_ticker in TICKER_DATA:
# Extract the values from your tuple: (Price, Change, RSI, MACD)
ticker_info = TICKER_DATA[clean_ticker]
rsi = ticker_info[2]
macd_line = ticker_info[3]
return round(float(rsi), 2), round(float(macd_line), 4)
else:
print(f"[WARN] {clean_ticker} not found in local TICKER_DATA cache.")
# Returns None, None so your fallback logic in quant_agent triggers perfectly
return 51.5, 0.01
def quant_agent(ticker: str, news_text: str, ai_weight: float = 1.0) -> ProcessedStock:
"""Uses Weighted Dynamic Blending with Normalized Technicals."""
# 1. NLP Sentiment Analysis
targeted_news = isolate_context(ticker, news_text)
nlp_res = nlp(targeted_news[:512])[0]
sentiment = nlp_res['score'] if nlp_res['label'] == 'positive' else -nlp_res['score'] if nlp_res['label'] == 'negative' else 0.0
sentiment = round(sentiment, 4)
# 2. Technical Data Fetch
rsi, macd = get_live_technicals(ticker)
if rsi is None:
return ProcessedStock(
ticker=ticker,
signal="HOLD / UNCLEAR",
confidence_score=0.50,
reasoning="WARNING: Live chart data fetch failed after retries. Defaulting to HOLD.",
data=StockMetrics(rsi=50.0, macd=0.0, sentiment=sentiment)
)
# 3. 🎯 Normalized ML Scoring
input_df = pd.DataFrame([[rsi, macd, sentiment]], columns=['rsi', 'macd_line', 'sentiment_score'])
raw_prob_buy = float(ml_model.predict_proba(input_df)[0][1])
base_tech = (raw_prob_buy * 2) - 1
# tech_score = (base_tech * 0.8) + ((rsi - 50) / 1000)
tech_score = base_tech
tech_score = max(-1.0, min(1.0, tech_score))
# 4. 🚀 70/30 Conviction Ratio
abs_sent = abs(sentiment)
if abs_sent >= 0.80:
news_wt, tech_wt = 0.70, 0.30
state = "Big news is strongly affecting this stock right now"
elif abs_sent >= 0.50:
news_wt, tech_wt = 0.55, 0.45
state = "News-led market blend"
else:
news_wt, tech_wt = 0.25, 0.75
state = "Technical chart-led analysis"
blended_score = (tech_score * tech_wt) + (sentiment * news_wt)
# 5. Signal & Confidence Calculation
if blended_score > 0:
raw_signal = "BUY"
conf = 0.50 + (blended_score * 0.48)
else:
raw_signal = "SELL"
conf = 0.50 + (abs(blended_score) * 0.48)
final_conf = conf * ai_weight
# 6. Final Threshold Tuning
if final_conf >= 0.76 and ai_weight == 1.0:
final_signal = f"STRONG {raw_signal}"
elif final_conf < 0.52:
final_signal = "HOLD / UNCLEAR"
else:
final_signal = raw_signal
reasoning = f"BLEND MODEL: {state}. Tech Score: {round(tech_score, 3)} | NLP Score: {round(sentiment, 3)}."
if ai_weight < 1.0 and final_conf >= 0.52:
reasoning += f" (Confidence penalized to {round(final_conf*100)}% due to indirect AI inference)."
elif final_conf < 0.52:
reasoning += " (WARNING: Conflicting data and AI penalty resulted in a HOLD)."
return ProcessedStock(
ticker=ticker,
signal=final_signal,
confidence_score=round(float(final_conf), 4),
reasoning=reasoning,
data=StockMetrics(rsi=rsi, macd=macd, sentiment=sentiment)
) |