RivaTerminal/modules/signals/markets_view.py

78 lines
2.5 KiB
Python

import pandas as pd
from tabulate import tabulate
import logging
from datetime import datetime, timedelta
logger = logging.getLogger(__name__)
class MarketsView:
def __init__(self, db_connection):
self.db = db_connection
def get_major_indices(self) -> pd.DataFrame:
try:
query = """
WITH latest_date AS (
SELECT MAX(trade_date) as max_date FROM index_quotes
)
SELECT
mi.symbol,
mi.name,
iq.price,
iq.change,
iq.changes_percentage,
iq.volume,
iq.trade_date
FROM market_indices mi
JOIN index_quotes iq ON iq.index_id = mi.id
WHERE iq.trade_date = (SELECT max_date FROM latest_date)
ORDER BY iq.volume DESC;
"""
return self.db.execute_query(query)
except Exception as e:
logger.error(f"Error fetching major indices: {e}")
return pd.DataFrame()
def get_market_movers(self) -> pd.DataFrame:
try:
query = """
WITH latest_date AS (
SELECT MAX(trade_date) as max_date FROM market_movers
)
SELECT
symbol,
name,
price,
change,
changes_percentage,
mover_type,
trade_date
FROM market_movers
WHERE trade_date = (SELECT max_date FROM latest_date)
ORDER BY ABS(changes_percentage) DESC
LIMIT 10;
"""
return self.db.execute_query(query)
except Exception as e:
logger.error(f"Error fetching market movers: {e}")
return pd.DataFrame()
def display(self):
print("\n=== Global Markets Overview ===")
indices_data = self.get_major_indices()
if not indices_data.empty:
print("\nMajor Market Indices:")
print(tabulate(indices_data, headers='keys', tablefmt='fancy_grid',
floatfmt=".2f"))
else:
print("\nNo recent market indices data available")
movers_data = self.get_market_movers()
if not movers_data.empty:
print("\nMarket Movers:")
print(tabulate(movers_data, headers='keys', tablefmt='fancy_grid',
floatfmt=".2f"))
else:
print("\nNo recent market movers data available")