import os from flask import Flask, request, jsonify, render_template from flask_cors import CORS import logging import yfinance as yf import pandas as pd from services.sdg_llm import classify_news # ✅ NEW from services.statistical_model import get_advanced_metrics from services.scoring import get_price_trend_score, get_valuation_score, combine_scores, map_to_recommendation import requests import json app = Flask(__name__) CORS(app) # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) @app.route("/stocks", methods=["GET"]) def get_stocks(): from services.finnhub_service import search_stocks, get_company_logo, finnhub_client if not finnhub_client: error_message = "Finnhub client not initialized. Please check if the FINNHUB_API_KEY is set correctly in the backend environment." logging.error(error_message) return jsonify({"error": error_message}), 500 query = request.args.get("q", "").strip() results = [] processed_symbols = set() try: stocks_to_fetch = [] if ',' in query and query: stocks_to_fetch = [s.strip() for s in query.split(',')] elif query: search_results = search_stocks(query) for item in search_results: symbol = item.get("symbol") if not symbol or symbol in processed_symbols: continue if item.get("type") != "Common Stock": continue if '.' in symbol: continue stocks_to_fetch.append(symbol) processed_symbols.add(symbol) for symbol in stocks_to_fetch: if not symbol: continue try: profile = get_company_logo(symbol, get_full_profile=True) if profile and profile.get('name'): results.append({ "symbol": symbol, "name": profile.get('name'), "logo": profile.get('logo') }) except Exception as e: logging.warning(f"Could not fetch full profile for {symbol}, skipping: {e}") return jsonify(results) except Exception as e: error_message = f"An unexpected error occurred on the server while fetching stocks: {e}" logging.error(error_message, exc_info=True) return jsonify({"error": error_message}), 500 @app.route("/history", methods=["GET"]) def get_history(): symbol = request.args.get("symbol") period = request.args.get("period", "1y") if not symbol: return jsonify({"error": "Stock symbol is required."}), 400 try: stock = yf.Ticker(symbol) hist = stock.history(period=period, interval="1d") if hist.empty: return jsonify({"error": f"No historical data found for symbol {symbol} with period {period}."}), 404 hist.reset_index(inplace=True) hist['Date'] = hist['Date'].dt.strftime('%Y-%m-%d') data = { 'dates': hist['Date'].tolist(), 'prices': hist['Close'].tolist() } return jsonify(data) except Exception as e: logging.error(f"Error fetching history for {symbol}: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 # --- Gemini API --- def call_gemini_api(prompt): gemini_api_key = os.environ.get("GEMINI_API_KEY") if not gemini_api_key: raise ValueError("GEMINI_API_KEY is not configured.") url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-flash-lite-latest:generateContent?key={gemini_api_key}" headers = {'Content-Type': 'application/json'} data = {"contents": [{"parts": [{"text": prompt}]}]} try: response = requests.post(url, headers=headers, json=data) response.raise_for_status() result = response.json() return result['candidates'][0]['content']['parts'][0]['text'].strip() except Exception as e: logging.error(f"Gemini API Error: {e}") return None def analyze_sentiment_with_gemini(news_text: str) -> int: if not news_text or not news_text.strip(): return 50 prompt = f""" You are a financial analyst. Based on the following news, rate the sentiment for the stock on a scale of 0–100. Only output the number. News: {news_text} """ response_text = call_gemini_api(prompt) try: return max(0, min(100, int(response_text))) except: return 50 def summarize_news_with_llm(symbol, articles): if not articles: return "No significant news found." news_text = "\n\n---\n\n".join([f"Headline: {a['title']}\nSummary: {a.get('description', '')}" for a in articles if a.get('title')]) if not news_text.strip(): return "No news content available to summarize." prompt = f""" Summarize these news articles about {symbol} into bullet points. Only include company-specific, financial-relevant information. Articles: {news_text} """ summary = call_gemini_api(prompt) return summary or "News summary could not be generated." def generate_explanation(symbol, last_close, predicted_close, trend_score, sentiment_score, final_score, recommendation, news_text): prompt = f""" Explain in 3–5 bullet points why the recommendation for {symbol} is "{recommendation}". Use ONLY the provided data. Last Close: {last_close} Predicted Close: {predicted_close} Trend Score: {trend_score} Sentiment Score: {sentiment_score} Final Score: {final_score} News: {news_text[:350]} """ explanation = call_gemini_api(prompt) return explanation or "Explanation unavailable." @app.route("/analyze", methods=["GET"]) def analyze(): print("⚠️ ENTERING analyze() route") print("⚠️ Importing stock_service...") from services.stock_service import get_stock_data, predict_next_close print("✅ stock_service imported") print("⚠️ Importing news_service...") from services.news_service import get_company_news print("✅ news_service imported") print("⚠️ Importing finnhub_service...") from services.finnhub_service import get_company_logo print("✅ finnhub_service imported") symbol = request.args.get("symbol", "GOOG") try: logging.info(f"--- Starting full analysis for {symbol} ---") # 1. PRICE PREDICTION df = get_stock_data(symbol) if df.empty: raise ValueError(f"No stock data for {symbol}.") last_close = float(df["close"].iloc[-1]) predicted_close = predict_next_close(df) trend_score = get_price_trend_score(predicted_close, last_close) # 2. NEWS + SENTIMENT company_profile = get_company_logo(symbol, get_full_profile=True) news_query = company_profile.get('name', symbol) if company_profile else symbol articles = get_company_news(news_query) news_text = " ".join([a['title'] + " " + a['description'] for a in articles if a.get('description')]) sentiment_score = analyze_sentiment_with_gemini(news_text) news_summary = summarize_news_with_llm(symbol, articles) # ✅ --- SDG CLASSIFICATION (MINIMAL ADDITION) --- sdg_result = classify_news(news_summary) sdg_score = sdg_result.get("score", 50) sdg_label = sdg_result.get("label", "Neutral") sdg_explanation = sdg_result.get("explanation", "No SDG-related impact detected.") # ✅ ------------------------------------------------ # 3. ADVANCED METRICS advanced_metrics = get_advanced_metrics(symbol) graham_number = advanced_metrics["valuation_models"]["graham_number"] valuation_score = get_valuation_score(last_close, graham_number) # 4. FINAL SCORING (SDG included) final_score = combine_scores( trend_score, sentiment_score, valuation_score, sdg_score=sdg_score ) recommendation = map_to_recommendation(final_score, trend_score, sentiment_score) # 5. EXPLANATION explanation = generate_explanation(symbol, last_close, predicted_close, trend_score, sentiment_score, final_score, recommendation, news_text) # ✅ 6. RESPONSE JSON result = { "symbol": symbol, "last_close": last_close, "predicted_close": predicted_close, "recommendation": recommendation, "final_score": final_score, "scores": { "trend": trend_score, "sentiment": sentiment_score, "valuation": valuation_score, }, "sdg": { "label": sdg_label, "score": sdg_score, "explanation": sdg_explanation }, "explanation": explanation, "latest_news_summary": news_summary, "advanced_metrics": advanced_metrics } return jsonify(result) except Exception as e: logging.error(str(e), exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/predict", methods=["GET"]) def predict(): """ A simple, API-only endpoint to get the next day's prediction. Takes one URL parameter: ?symbol=AAPL """ from services.stock_service import get_stock_data, predict_next_close symbol = request.args.get("symbol") if not symbol: return jsonify({"error": "No symbol provided"}), 400 try: # 1. Get the data (this re-uses your existing function) df = get_stock_data(symbol) # 2. Run the full TF model (this re-uses your existing function) predicted_close = predict_next_close(df) # 3. Return the raw data as JSON return jsonify({"predicted_close": predicted_close}) except Exception as e: # Return any errors as JSON return jsonify({"error": str(e)}), 500 @app.route("/") def home(): return render_template("index.html") if __name__ == "__main__": app.run(debug=True)