Update src/utils.py

src/utils.py

@@ -1,296 +1,40 @@
 import yfinance as yf
 import pandas as pd
-# Using curl_cffi for session robustness to bypass potential anti-bot measures
-from curl_cffi import requests 
 
-class StockAnalyzer:
-    def __init__(self, ticker_symbol):
-        # --- FIX 1: Auto-correct Ticker for Indian Market ---
-        ticker_symbol = ticker_symbol.upper().strip()
-        if not (ticker_symbol.endswith('.NS') or ticker_symbol.endswith('.BO')) and ticker_symbol.isalpha():
-            self.ticker_symbol = f"{ticker_symbol}.NS"
-        else:
-            self.ticker_symbol = ticker_symbol
-        
-        # --- ROBUST SESSION ---
-        self.session = requests.Session(impersonate="chrome")
-        self.ticker = yf.Ticker(self.ticker_symbol, session=self.session)
-        
-        # --- FIX 2: Safe Fetching of ALL financial data (Annual Data) ---
-        self.info = self.ticker.info or {} # Info is already wrapped in original code
-        
-        # Wrap statement fetching in try/except blocks to ensure object initialization doesn't fail
-        try:
-            self.income_stmt = self.ticker.income_stmt
-        except Exception:
-            self.income_stmt = pd.DataFrame()
-            
-        try:
-            self.balance_sheet = self.ticker.balance_sheet
-        except Exception:
-            self.balance_sheet = pd.DataFrame()
-            
-        try:
-            self.cashflow = self.ticker.cashflow
-        except Exception:
-            self.cashflow = pd.DataFrame()
-
-    def _get_data(self, df, possible_keys):
-        """
-        Smart Search: Tries to find data using a list of possible key names.
-        Returns None if not found.
-        """
-        if df is None or df.empty:
-            return None
-
-        # Create a normalized map for case-insensitive lookup, stripping whitespace
-        index_map = {str(idx).strip().lower(): idx for idx in df.index}
-
-        for key in possible_keys:
-            # 1. Exact Match (case-sensitive)
-            if key in df.index:
-                return df.loc[key]
-            
-            # 2. Case-Insensitive Match
-            normalized_key = key.strip().lower()
-            if normalized_key in index_map:
-                real_key = index_map[normalized_key]
-                return df.loc[real_key]
-        
-        return None
-
-    def get_financial_trends(self):
-        try:
-            # --- 1. INCOME STATEMENT (Includes Gross Profit for Piotroski Score) ---
-            rev = self._get_data(self.income_stmt, [
-                'Total Revenue', 'Revenue', 'Operating Revenue', 'TotalOperatingRevenue'
-            ])
-            ni = self._get_data(self.income_stmt, [
-                'Net Income', 'Net Income Common Stockholders', 'NetIncomeContinuousOperations', 'NetIncome'
-            ])
-            gp = self._get_data(self.income_stmt, [
-                'Gross Profit', 'GrossProfit'
-            ]) # NEW: Needed for Gross Margin Change
-            
-            # --- 2. BALANCE SHEET (Includes Current Assets/Liabs, Debt, and Shares for Piotroski Score) ---
-            assets = self._get_data(self.balance_sheet, [
-                'Total Assets', 'Assets', 'TotalAssets'
-            ])
-            liabs = self._get_data(self.balance_sheet, [
-                'Total Liabilities Net Minority Interest', 'Total Liabilities', 'TotalLiabilities'
-            ])
-            equity = self._get_data(self.balance_sheet, [
-                'Stockholders Equity', 'Total Equity Gross Minority Interest', 'TotalEquityGrossMinorityInterest'
-            ])
-            current_assets = self._get_data(self.balance_sheet, [
-                'Current Assets', 'Total Current Assets', 'CurrentAssets'
-            ]) # NEW: Needed for Current Ratio
-            current_liabilities = self._get_data(self.balance_sheet, [
-                'Current Liabilities', 'Total Current Liabilities', 'CurrentLiabilities'
-            ]) # NEW: Needed for Current Ratio
-            total_debt = self._get_data(self.balance_sheet, [
-                'Total Debt', 'Long Term Debt', 'TotalDebt', 'LongTermDebt'
-            ]) # Using total debt for comprehensive leverage check
-            shares = self._get_data(self.balance_sheet, [
-                'Share Issued', 'Common Stock', 'CommonStock'
-            ]) # Proxy for shares outstanding/issuance
-
-            # --- 3. CASH FLOW ---
-            ocf = self._get_data(self.cashflow, [
-                'Operating Cash Flow', 'Total Cash From Operating Activities', 'OperatingCashFlow'
-            ])
-            
-            # Smart FCF Calculation: If 'Free Cash Flow' is missing, calculate it (OCF + CapEx)
-            fcf = self._get_data(self.cashflow, ['Free Cash Flow', 'FreeCashFlow'])
-            if fcf is None and ocf is not None:
-                capex = self._get_data(self.cashflow, ['Capital Expenditure', 'Capital Expenditures', 'CapitalExpenditure'])
-                if capex is not None:
-                    fcf = ocf + capex
-
-            # Combine into DataFrame
-            data = {}
-            if rev is not None: data['Revenue'] = rev
-            if ni is not None: data['Net Income'] = ni
-            if gp is not None: data['Gross Profit'] = gp # ADDED
-            if assets is not None: data['Total Assets'] = assets
-            if liabs is not None: data['Total Liabilities'] = liabs
-            if equity is not None: data['Equity'] = equity
-            if current_assets is not None: data['Current Assets'] = current_assets # ADDED
-            if current_liabilities is not None: data['Current Liabilities'] = current_liabilities # ADDED
-            if total_debt is not None: data['Total Debt'] = total_debt # ADDED
-            if shares is not None: data['Shares Issued'] = shares # ADDED
-            if ocf is not None: data['Operating Cash Flow'] = ocf
-            if fcf is not None: data['Free Cash Flow'] = fcf
-
-            if not data:
-                return pd.DataFrame()
-
-            df = pd.DataFrame(data)
-            
-            # Ensure Index is DateTime and Sorted
-            df.index = pd.to_datetime(df.index)
-            df = df.sort_index() 
-            
-            return df
-            
-        except Exception as e:
-            # print(f"Error in get_financial_trends: {e}") # Suppress console output in final app
-            return pd.DataFrame()
-
-    def get_summary_metrics(self):
-        # Unchanged from original
-        i = self.info
-        return {
-            "current_price": i.get("currentPrice") or i.get("regularMarketPrice"),
-            "market_cap": i.get("marketCap"),
-            "pe_ratio": i.get("trailingPE"),
-            "dividend_yield": i.get("dividendYield"),
-            "currency": i.get("currency", "USD"),
-            "summary": i.get("longBusinessSummary", "No summary available."),
-            "sector": i.get("sector", "Unknown"),
-            "industry": i.get("industry", "Unknown"),
-            "website": i.get("website", "#")
-        }
-
-    def check_red_flags(self):
-        flags = []
-        try:
-            df = self.get_financial_trends()
-            if df.empty: 
-                return [{"type": "warning", "msg": "Insufficient data for Red Flag analysis"}]
-            
-            latest = df.iloc[-1]
-            
-            # 1. Debt Check (Using Total Debt from trends)
-            debt = latest.get('Total Debt', 0)
-            equity_val = latest.get('Equity', 0)
-            
-            if equity_val > 0:
-                de_ratio = debt / equity_val
-                if de_ratio > 2.0:
-                    flags.append({"type": "danger", "msg": f"High Debt (D/E: {de_ratio:.2f})"})
-                else:
-                    flags.append({"type": "success", "msg": f"Healthy Debt (D/E: {de_ratio:.2f})"})
-            else:
-                 flags.append({"type": "warning", "msg": "Cannot calculate D/E (Negative/Zero Equity)"})
-
-
-            # 2. Revenue Trend
-            if len(df) > 1 and 'Revenue' in df:
-                if df['Revenue'].iloc[-1] < df['Revenue'].iloc[-2]:
-                     flags.append({"type": "danger", "msg": "Declining Revenue (YoY)"})
-                else:
-                     flags.append({"type": "success", "msg": "Revenue Growing"})
-
-            # 3. Free Cash Flow
-            fcf_val = latest.get('Free Cash Flow', 0)
-            if fcf_val < 0:
-                flags.append({"type": "danger", "msg": "Negative Free Cash Flow"})
-            else:
-                flags.append({"type": "success", "msg": "Positive Free Cash Flow"})
-            
-            # 4. NEW: Current Ratio Check (Liquidity check)
-            ca_val = latest.get('Current Assets', 0)
-            cl_val = latest.get('Current Liabilities', 0)
-            if cl_val > 0:
-                current_ratio = ca_val / cl_val
-                if current_ratio < 1.0:
-                    flags.append({"type": "danger", "msg": f"Poor Liquidity (Current Ratio: {current_ratio:.2f} < 1.0)"})
-                else:
-                    flags.append({"type": "success", "msg": f"Good Liquidity (Current Ratio: {current_ratio:.2f})"})
-            
-
-        except Exception as e:
-            flags.append({"type": "warning", "msg": f"Error calculating flags: {str(e)}"})
-        
-        return flags
-
-    def calculate_piotroski_score(self):
-        score = 0
-        breakdown = []
-        try:
-            df = self.get_financial_trends()
-            
-            # We need at least 2 periods for year-over-year comparison (8 of 9 points)
-            if len(df) < 2: 
-                return 0, ["Insufficient Historical Data (Need 2+ periods)"]
-            
-            curr = df.iloc[-1]
-            prev = df.iloc[-2]
-
-            # --- PROFITABILITY (F1 - F4) ---
-            # F1. Positive Net Income
-            if curr.get('Net Income', 0) > 0: score+=1; breakdown.append("✅ F1. Positive Net Income")
-            else: breakdown.append("❌ F1. Negative Net Income")
-            
-            # F2. Positive Operating Cash Flow
-            if curr.get('Operating Cash Flow', 0) > 0: score+=1; breakdown.append("��� F2. Positive OCF")
-            else: breakdown.append("❌ F2. Negative OCF")
-            
-            # F3. ROA Increasing (Calculated using Net Income / Total Assets)
-            if 'Total Assets' in curr and 'Net Income' in curr:
-                # Use a small epsilon to prevent division by zero for total assets
-                epsilon = 1 
-                roa_curr = curr.get('Net Income', 0) / curr.get('Total Assets', epsilon)
-                roa_prev = prev.get('Net Income', 0) / prev.get('Total Assets', epsilon)
-                if roa_curr > roa_prev: score+=1; breakdown.append("✅ F3. ROA Increasing")
-                else: breakdown.append("❌ F3. ROA Decreasing")
-            else: breakdown.append("⚠️ F3. ROA Change cannot be calculated")
-            
-            # F4. Quality of Earnings (OCF > Net Income)
-            if curr.get('Operating Cash Flow', 0) > curr.get('Net Income', 0): score+=1; breakdown.append("✅ F4. Quality Earnings (OCF > Net Income)")
-            else: breakdown.append("❌ F4. Low Quality Earnings")
-            
-            # --- LEVERAGE, LIQUIDITY, SOURCE OF FUNDS (F5 - F7) ---
-            
-            # F5. Decreased Leverage (Change in Total Debt / Total Assets)
-            if 'Total Debt' in curr and 'Total Assets' in curr:
-                epsilon = 1
-                leverage_curr = curr.get('Total Debt', 0) / curr.get('Total Assets', epsilon)
-                leverage_prev = prev.get('Total Debt', 0) / prev.get('Total Assets', epsilon)
-                if leverage_curr < leverage_prev: score+=1; breakdown.append("✅ F5. Decreased Leverage (Debt/Assets Ratio)")
-                else: breakdown.append("❌ F5. Increased Leverage")
-            else: breakdown.append("⚠️ F5. Leverage Change cannot be calculated")
-            
-            # F6. Increased Current Ratio (Current Assets / Current Liabilities)
-            if 'Current Assets' in curr and 'Current Liabilities' in curr:
-                epsilon = 1
-                cr_curr = curr.get('Current Assets', 0) / curr.get('Current Liabilities', epsilon)
-                cr_prev = prev.get('Current Assets', 0) / prev.get('Current Liabilities', epsilon)
-                if cr_curr > cr_prev: score+=1; breakdown.append("✅ F6. Increased Current Ratio (Liquidity)")
-                else: breakdown.append("❌ F6. Decreased Current Ratio")
-            else: breakdown.append("⚠️ F6. Current Ratio Change cannot be calculated")
-
-            # F7. No New Shares Issued (Shares Issued <= Previous Period Shares Issued)
-            if 'Shares Issued' in curr:
-                 if curr.get('Shares Issued', 0) <= prev.get('Shares Issued', 0): score+=1; breakdown.append("✅ F7. No Share Dilution (Shares <= Prior Period)")
-                 else: breakdown.append("❌ F7. Share Dilution")
-            else: breakdown.append("⚠️ F7. Share Dilution cannot be assessed")
-
-
-            # --- OPERATING EFFICIENCY (F8 - F9) ---
-            
-            # F8. Increased Gross Margin (Gross Profit / Revenue)
-            if 'Gross Profit' in curr and 'Revenue' in curr:
-                epsilon = 1
-                gm_curr = curr.get('Gross Profit', 0) / curr.get('Revenue', epsilon)
-                gm_prev = prev.get('Gross Profit', 0) / prev.get('Revenue', epsilon)
-                if gm_curr > gm_prev: score+=1; breakdown.append("✅ F8. Increased Gross Margin")
-                else: breakdown.append("❌ F8. Decreased Gross Margin")
-            else: breakdown.append("⚠️ F8. Gross Margin Change cannot be calculated")
-            
-            # F9. Increased Asset Turnover (Revenue / Total Assets)
-            if 'Revenue' in curr and 'Total Assets' in curr:
-                epsilon = 1
-                at_curr = curr.get('Revenue', 0) / curr.get('Total Assets', epsilon)
-                at_prev = prev.get('Revenue', 0) / prev.get('Total Assets', epsilon)
-                if at_curr > at_prev: score+=1; breakdown.append("✅ F9. Asset Turnover Up (Efficiency)")
-                else: breakdown.append("❌ F9. Asset Turnover Down")
-            else: breakdown.append("⚠️ F9. Asset Turnover Change cannot be calculated")
-
-            return score, breakdown
-        except Exception as e:
-            # print(f"Error in Piotroski score: {e}") # Suppress console output
-            return score, breakdown or ["Calculation Error: " + str(e)]
-
-# End of utils.py
+def fetch_financials(ticker_symbol: str, freq: str = "annual"):
+    """Fetch income statement and cashflow for a given ticker."""
+    ticker = yf.Ticker(ticker_symbol)
+
+    # Use get_* methods if available
+    try:
+        income = ticker.get_income_stmt(freq=freq)
+    except AttributeError:
+        # fallback to older API
+        income = ticker.financials
+
+    try:
+        cash = ticker.get_cashflow(freq=freq)
+    except AttributeError:
+        cash = ticker.cashflow
+
+    # If they return dict (as_dict=True), convert to DataFrame
+    if isinstance(income, dict):
+        income = pd.DataFrame(income)
+    if isinstance(cash, dict):
+        cash = pd.DataFrame(cash)
+
+    return income, cash
+
+
+def clean_financials(df: pd.DataFrame) -> pd.DataFrame:
+    """Clean / transform the financials DataFrame for better plotting."""
+    # Transpose so dates become rows
+    df2 = df.T.copy()
+    # Optionally sort by date
+    try:
+        # If columns are strings of dates or period
+        df2.index = pd.to_datetime(df2.index)
+        df2 = df2.sort_index()
+    except Exception:
+        pass
+    return df2