84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
import pandas as pd
|
|
from tabulate import tabulate
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class Whispers:
|
|
def __init__(self, db_connection):
|
|
self.db = db_connection
|
|
|
|
def get_analyst_ratings(self) -> pd.DataFrame:
|
|
try:
|
|
query = """
|
|
WITH recent_ratings AS (
|
|
SELECT
|
|
t.symbol as ticker,
|
|
r.analyst,
|
|
r.prior_rating,
|
|
r.current_rating,
|
|
r.action,
|
|
r.price_target,
|
|
r.date
|
|
FROM ratings r
|
|
JOIN tickers t ON t.id = r.ticker_id
|
|
WHERE r.date >= CURRENT_DATE - INTERVAL '7 days'
|
|
)
|
|
SELECT *
|
|
FROM recent_ratings
|
|
ORDER BY date DESC, price_target DESC
|
|
LIMIT 20;
|
|
"""
|
|
return self.db.execute_query(query)
|
|
except Exception as e:
|
|
logger.error(f"Error fetching analyst ratings: {e}")
|
|
return pd.DataFrame()
|
|
|
|
def get_rating_consensus(self) -> pd.DataFrame:
|
|
try:
|
|
query = """
|
|
WITH latest_consensus AS (
|
|
SELECT
|
|
t.symbol as ticker,
|
|
rc.consensus_rating,
|
|
rc.total_ratings,
|
|
rc.buy_ratings,
|
|
rc.hold_ratings,
|
|
rc.sell_ratings,
|
|
rc.strong_buy_ratings,
|
|
rc.strong_sell_ratings,
|
|
rc.date
|
|
FROM rating_consensus rc
|
|
JOIN tickers t ON t.id = rc.ticker_id
|
|
WHERE rc.date = (SELECT MAX(date) FROM rating_consensus)
|
|
)
|
|
SELECT *
|
|
FROM latest_consensus
|
|
ORDER BY total_ratings DESC
|
|
LIMIT 10;
|
|
"""
|
|
return self.db.execute_query(query)
|
|
except Exception as e:
|
|
logger.error(f"Error fetching rating consensus: {e}")
|
|
return pd.DataFrame()
|
|
|
|
def display(self):
|
|
print("\n=== Market Sentiment Analysis ===")
|
|
|
|
ratings_data = self.get_analyst_ratings()
|
|
if not ratings_data.empty:
|
|
print("\nRecent Analyst Ratings:")
|
|
print(tabulate(ratings_data, headers='keys', tablefmt='fancy_grid',
|
|
floatfmt=".2f"))
|
|
else:
|
|
print("\nNo recent analyst ratings available")
|
|
|
|
consensus_data = self.get_rating_consensus()
|
|
if not consensus_data.empty:
|
|
print("\nRating Consensus:")
|
|
print(tabulate(consensus_data, headers='keys', tablefmt='fancy_grid',
|
|
floatfmt=".2f"))
|
|
else:
|
|
print("\nNo rating consensus data available")
|