Spaces:
Sleeping
Sleeping
File size: 5,307 Bytes
a78209f | 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 | import pandas as pd
import logging
import os
logger = logging.getLogger(__name__)
class TickerDatabase:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(TickerDatabase, cls).__new__(cls)
cls._instance.df = None
return cls._instance
def load_data(self, filepath: str):
"""Loads and cleans the CSV data into memory."""
if not os.path.exists(filepath):
logger.error(f"Stock data file not found at: {filepath}")
return
try:
# Read CSV with flexible encoding
try:
self.df = pd.read_csv(filepath, encoding='utf-8')
except UnicodeDecodeError:
self.df = pd.read_csv(filepath, encoding='latin1')
# Clean column names
self.df.columns = self.df.columns.str.strip()
# Ensure required columns
if 'Name' not in self.df.columns:
logger.error("CSV missing required 'Name' column.")
self.df = None
return
# Clean Name column
self.df['Name'] = self.df['Name'].astype(str).str.strip()
# Numeric cleaning for all metric columns
# We explicitly clean these columns to be usable floats
numeric_cols = [
'LTP', 'Change(%)', 'Open', 'Volume', 'Market Cap (Cr.)',
'PE Ratio', 'Industry PE', '52W High', '52W Low',
'1M Returns', '3M Returns', '1 Yr Returns', '3 Yr Returns', '5 Yr Returns',
'PB Ratio', 'Dividend', 'ROE', 'ROCE', 'EPS',
'50 DMA', '200 DMA', 'RSI', 'Margin Funding', 'Margin Pledge'
]
for col in numeric_cols:
if col in self.df.columns:
# Remove commas, %, and whitespace, then coerce to numeric
# Force string conversion first to handle mixed types safely
self.df[col] = (
self.df[col]
.astype(str)
.str.replace(',', '', regex=False)
.str.replace('%', '', regex=False)
.str.strip()
)
self.df[col] = pd.to_numeric(self.df[col], errors='coerce').fillna(0.0)
logger.info(f"Loaded {len(self.df)} tickers from {filepath}")
except Exception as e:
logger.error(f"Failed to load stock data: {e}")
self.df = None
def search_names(self, query: str, limit: int = 10):
"""Searches for company names containing the query string (case-insensitive)."""
if self.df is None or not query:
return []
try:
mask = self.df['Name'].str.contains(query, case=False, na=False)
matches = self.df.loc[mask, 'Name'].head(limit).tolist()
return matches
except Exception as e:
logger.error(f"Error filtering names: {e}")
return []
def validate_name(self, name: str) -> bool:
"""Checks if a company name exists in the database (exact match, case-insensitive)."""
if self.df is None or not name:
return False
mask = self.df['Name'].str.lower() == name.strip().lower()
return mask.any()
def get_company_details(self, name: str) -> dict:
"""Returns the full row of data for a given company name."""
if self.df is None or not name:
return None
try:
# Case-insensitive exact match
row = self.df[self.df['Name'].str.lower() == name.strip().lower()]
if row.empty:
return None
# Convert to dictionary and handle Nan/Inf for JSON serialization
data = row.iloc[0].to_dict()
return data
except Exception as e:
logger.error(f"Error getting details for {name}: {e}")
return None
def get_peers_by_industry(self, industry_pe: float, exclude_name: str, limit: int = 5) -> list:
"""
Finds peers with the same or similar Industry PE.
This uses the logic: Same Industry PE = Same Sector/Industry.
"""
if self.df is None or industry_pe == 0:
return []
try:
# Filter by matching Industry PE
# Using a small tolerance just in case of float weirdness, though exact match is requested
peers = self.df[
(self.df['Industry PE'] >= industry_pe - 0.1) &
(self.df['Industry PE'] <= industry_pe + 0.1) &
(self.df['Name'].str.lower() != exclude_name.lower())
]
# Use 'Market Cap (Cr.)' for sorting if available, else random slice
if 'Market Cap (Cr.)' in self.df.columns:
peers = peers.sort_values(by='Market Cap (Cr.)', ascending=False)
return peers.head(limit).to_dict('records')
except Exception as e:
logger.error(f"Error finding peers: {e}")
return []
# Global instance accessor
def get_ticker_db():
return TickerDatabase()
|