Spaces:
Sleeping
Sleeping
| """ | |
| pipeline/peer_selector.py -- Automatically select 3-5 peer companies. | |
| Uses FMP's peers endpoint first, then falls back to industry-based hardcoded peers. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| from typing import List, Dict, Optional | |
| import requests | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| import config | |
| from config import FMP_API_KEY, FMP_BASE, REQUEST_TIMEOUT | |
| from pipeline.resolver import CompanyMeta | |
| from pipeline.fetcher import fetch_peer_snapshot, _fmp_get | |
| # yfinance-based peer fetch (no API key needed) | |
| def _yf_peers_from_industry(meta: CompanyMeta) -> list: | |
| """Return peer tickers based on industry using hardcoded maps.""" | |
| return [] # handled by INDIAN_PEERS / GLOBAL_PEERS maps below | |
| # -- Hardcoded industry -> peer tickers (Indian + Global) ---------------------- | |
| INDIAN_PEERS: Dict[str, List[str]] = { | |
| "IT Services": ["TCS.NS", "INFY.NS", "WIPRO.NS", "HCLTECH.NS", "TECHM.NS"], | |
| "Information Technology": ["TCS.NS", "INFY.NS", "WIPRO.NS", "HCLTECH.NS", "LTIM.NS"], | |
| "Banking": ["HDFCBANK.NS", "ICICIBANK.NS", "AXISBANK.NS", "SBIN.NS", "KOTAKBANK.NS"], | |
| "Finance": ["BAJFINANCE.NS", "BAJAJFINSV.NS", "HDFCAMC.NS", "MUTHOOTFIN.NS"], | |
| "Pharmaceuticals": ["SUNPHARMA.NS", "CIPLA.NS", "DRREDDY.NS", "DIVISLAB.NS", "AUROPHARMA.NS"], | |
| "Automobile": ["MARUTI.NS", "TATAMOTORS.NS", "M&M.NS", "HEROMOTOCO.NS", "BAJAJ-AUTO.NS"], | |
| "FMCG": ["HINDUNILVR.NS", "ITC.NS", "NESTLEIND.NS", "DABUR.NS", "MARICO.NS"], | |
| "Cement": ["ULTRACEMCO.NS", "SHREECEM.NS", "AMBUJACEM.NS", "ACC.NS", "GRASIM.NS"], | |
| "Steel": ["TATASTEEL.NS", "JSWSTEEL.NS", "SAIL.NS", "HINDALCO.NS", "VEDL.NS"], | |
| "Telecom": ["BHARTIARTL.NS", "RELIANCE.NS", "IDEA.NS"], | |
| "Oil & Gas": ["RELIANCE.NS", "ONGC.NS", "IOC.NS", "BPCL.NS", "HINDPETRO.NS"], | |
| "Power": ["NTPC.NS", "POWERGRID.NS", "ADANIGREEN.NS", "TATAPOWER.NS", "CESC.NS"], | |
| "Real Estate": ["DLF.NS", "GODREJPROP.NS", "OBEROIRLTY.NS", "PHOENIXLTD.NS"], | |
| "Consumer Electronics": ["DIXON.NS", "VOLTAS.NS", "BLUESTARCO.NS", "HAVELLS.NS"], | |
| "Insurance": ["SBILIFE.NS", "HDFCLIFE.NS", "ICICIPRULI.NS", "LICI.NS"], | |
| "Retail": ["DMART.NS", "TRENT.NS", "SHOPPERSSTOP.NS", "VMART.NS"], | |
| "Paints": ["ASIANPAINT.NS", "BERGEPAINT.NS", "KANSAINER.NS", "INDIGO.NS"], | |
| } | |
| GLOBAL_PEERS: Dict[str, List[str]] = { | |
| "Technology": ["AAPL", "MSFT", "GOOGL", "META", "AMZN"], | |
| "Semiconductors": ["NVDA", "AMD", "INTC", "QCOM", "AVGO"], | |
| "Cloud": ["MSFT", "AMZN", "GOOGL", "ORCL", "CRM"], | |
| "EV/Auto": ["TSLA", "GM", "F", "TM", "STLA"], | |
| "Banking": ["JPM", "BAC", "WFC", "C", "GS"], | |
| "Pharma": ["JNJ", "PFE", "MRK", "ABBV", "LLY"], | |
| "FMCG": ["PG", "KO", "PEP", "UL", "NESN"], | |
| "Oil & Gas": ["XOM", "CVX", "SHEL", "BP", "TTE"], | |
| "Retail": ["WMT", "AMZN", "COST", "TGT", "HD"], | |
| "Telecom": ["VZ", "T", "TMUS", "VOD", "ORAN"], | |
| } | |
| def _fmp_peers(ticker: str) -> List[str]: | |
| """Use FMP's stock-peers endpoint.""" | |
| data = _fmp_get(f"/v4/stock_peers", {"symbol": ticker}) | |
| if data and isinstance(data, list) and data[0].get("peersList"): | |
| return data[0]["peersList"][:5] | |
| return [] | |
| def _parse_tickers_json(text: str) -> Optional[List[str]]: | |
| """Parse JSON array of tickers from LLM response text.""" | |
| import re, json | |
| # Match anything between [ and ] inclusive, across multiple lines | |
| match = re.search(r'\[\s*(.*?)\s*\]', text, re.DOTALL) | |
| if match: | |
| arr_str = match.group(0) | |
| try: | |
| tickers = json.loads(arr_str) | |
| if isinstance(tickers, list) and all(isinstance(t, str) for t in tickers): | |
| return tickers | |
| except Exception: | |
| pass | |
| return None | |
| def _llm_intelligent_peers(meta: CompanyMeta, log_fn=None) -> List[str]: | |
| """Use the first available LLM API (Gemini, Claude, OpenAI, Groq, or Hugging Face) to select peers.""" | |
| def log(msg): | |
| if log_fn: log_fn(msg) | |
| desc = meta.description if meta.description else "No description available." | |
| prompt = ( | |
| f"You are a top-tier financial analyst. Given the following company, break down its exact sub-sector " | |
| f"and identify its 5 closest publicly traded peers.\n" | |
| f"Company Name: {meta.name}\n" | |
| f"Ticker: {meta.ticker}\n" | |
| f"Industry: {meta.industry}\n" | |
| f"Description: {desc}\n\n" | |
| f"Return ONLY a valid JSON array of their Yahoo Finance ticker symbols (e.g., [\"MSFT\", \"GOOGL\"]). " | |
| f"CRITICAL: The peers MUST be traded on the same exchange as the company. " | |
| f"If the company's ticker has a suffix (e.g., .NS for India, .L for London), you MUST append the exact same suffix to all peer tickers.\n" | |
| f"Do not include any other text, markdown, or explanation. Only the JSON array." | |
| ) | |
| # 1. Try Gemini | |
| gemini_key = getattr(config, "GEMINI_API_KEY", "") | |
| if gemini_key: | |
| log(" -> Gemini API key detected. Using Gemini for granular sector breakdown and peer selection...") | |
| try: | |
| gemini_model = getattr(config, "GEMINI_MODEL", "gemini-2.0-flash") | |
| # Try new google-genai SDK | |
| try: | |
| from google import genai | |
| from google.genai import types as gtypes | |
| client = genai.Client(api_key=gemini_key) | |
| response = client.models.generate_content( | |
| model=gemini_model, | |
| contents=prompt, | |
| config=gtypes.GenerateContentConfig(temperature=0.1, max_output_tokens=100), | |
| ) | |
| tickers = _parse_tickers_json(response.text) | |
| if tickers: | |
| log(f" [OK] Gemini successfully identified peers: {tickers}") | |
| return tickers | |
| except Exception: | |
| pass | |
| # Fallback to old SDK | |
| import google.generativeai as genai_old | |
| genai_old.configure(api_key=gemini_key) | |
| model = genai_old.GenerativeModel(model_name=gemini_model) | |
| response = model.generate_content(prompt) | |
| tickers = _parse_tickers_json(response.text) | |
| if tickers: | |
| log(f" [OK] Gemini (old SDK) successfully identified peers: {tickers}") | |
| return tickers | |
| except Exception as e: | |
| log(f" [Warning] Gemini peer selection failed: {e}") | |
| # 2. Try Claude | |
| claude_key = getattr(config, "CLAUDE_API_KEY", "") | |
| if claude_key: | |
| log(" -> Claude API key detected. Using Claude for granular sector breakdown and peer selection...") | |
| try: | |
| import anthropic | |
| client = anthropic.Anthropic(api_key=claude_key) | |
| claude_model = getattr(config, "CLAUDE_MODEL", "claude-3-5-haiku-20241022") | |
| message = client.messages.create( | |
| model=claude_model, | |
| max_tokens=100, | |
| temperature=0.1, | |
| messages=[{"role": "user", "content": prompt}], | |
| ) | |
| tickers = _parse_tickers_json(message.content[0].text) | |
| if tickers: | |
| log(f" [OK] Claude successfully identified peers: {tickers}") | |
| return tickers | |
| except Exception as e: | |
| log(f" [Warning] Claude peer selection failed: {e}") | |
| # 3. Try OpenAI | |
| openai_key = getattr(config, "OPENAI_API_KEY", "") | |
| if openai_key: | |
| log(" -> OpenAI API key detected. Using OpenAI for granular sector breakdown and peer selection...") | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI(api_key=openai_key) | |
| openai_model = getattr(config, "OPENAI_MODEL", "gpt-4o-mini") | |
| resp = client.chat.completions.create( | |
| model=openai_model, | |
| messages=[{"role": "user", "content": prompt}], | |
| temperature=0.1, | |
| max_tokens=100, | |
| ) | |
| tickers = _parse_tickers_json(resp.choices[0].message.content) | |
| if tickers: | |
| log(f" [OK] OpenAI successfully identified peers: {tickers}") | |
| return tickers | |
| except Exception as e: | |
| log(f" [Warning] OpenAI peer selection failed: {e}") | |
| # 4. Try Groq | |
| groq_key = getattr(config, "GROQ_API_KEY", "") | |
| if groq_key: | |
| log(" -> Groq API key detected. Using Groq for granular sector breakdown and peer selection...") | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI(api_key=groq_key, base_url="https://api.groq.com/openai/v1") | |
| groq_model = getattr(config, "GROQ_MODEL", "llama-3.3-70b-versatile") | |
| resp = client.chat.completions.create( | |
| model=groq_model, | |
| messages=[{"role": "user", "content": prompt}], | |
| temperature=0.1, | |
| max_tokens=100, | |
| ) | |
| tickers = _parse_tickers_json(resp.choices[0].message.content) | |
| if tickers: | |
| log(f" [OK] Groq successfully identified peers: {tickers}") | |
| return tickers | |
| except Exception as e: | |
| log(f" [Warning] Groq peer selection failed: {e}") | |
| # 5. Try Hugging Face (Free-tier model Qwen/Qwen2.5-7B-Instruct) | |
| hf_key = getattr(config, "HF_API_KEY", "") | |
| if hf_key: | |
| log(" -> Hugging Face API key detected. Using HF Qwen/Qwen2.5-7B-Instruct for peer selection...") | |
| try: | |
| from huggingface_hub import InferenceClient | |
| client = InferenceClient(api_key=hf_key) | |
| response = client.chat_completion( | |
| model="Qwen/Qwen2.5-7B-Instruct", | |
| messages=[{"role": "user", "content": prompt}], | |
| max_tokens=100, | |
| temperature=0.1 | |
| ) | |
| if response and response.choices: | |
| tickers = _parse_tickers_json(response.choices[0].message.content) | |
| if tickers: | |
| log(f" [OK] Hugging Face successfully identified peers: {tickers}") | |
| return tickers | |
| log(" [Warning] Hugging Face response parsing failed.") | |
| except Exception as e: | |
| log(f" [Warning] Hugging Face peer selection failed: {e}") | |
| return [] | |
| def select_peers( | |
| meta: CompanyMeta, | |
| max_peers: int = 4, | |
| log_fn=None, | |
| ) -> List[Dict]: | |
| """ | |
| Select peer companies for benchmarking. | |
| Returns list of peer snapshot dicts (same format as peers in the JSON schema). | |
| """ | |
| def log(msg): | |
| if log_fn: | |
| log_fn(msg) | |
| ticker = meta.ticker | |
| peers_out: List[Dict] = [] | |
| log(f"Selecting peers for {meta.name} ({meta.industry}) ...") | |
| # 1. Try LLM-based Intelligent Peers (Gemini, Claude, OpenAI, Groq, or HF) | |
| peer_tickers = _llm_intelligent_peers(meta, log_fn) | |
| # 2. Try FMP peers endpoint if HF fails | |
| if not peer_tickers and FMP_API_KEY and meta.is_listed and ticker != "UNLISTED": | |
| peer_tickers = _fmp_peers(ticker) | |
| if peer_tickers: | |
| log(f" -> FMP suggested {len(peer_tickers)} peers") | |
| # 2. Fallback to hardcoded industry/sector map | |
| if not peer_tickers: | |
| industry = meta.industry or "" | |
| # Try specific industry match first | |
| if industry: | |
| for key, tickers in INDIAN_PEERS.items(): | |
| if key.lower() in industry.lower() or industry.lower() in key.lower(): | |
| peer_tickers = [t for t in tickers if t.upper() != ticker.upper()][:max_peers] | |
| log(f" -> Using hardcoded Indian peers for industry '{key}'") | |
| break | |
| if not peer_tickers: | |
| for key, tickers in GLOBAL_PEERS.items(): | |
| if key.lower() in industry.lower() or industry.lower() in key.lower(): | |
| peer_tickers = [t for t in tickers if t.upper() != ticker.upper()][:max_peers] | |
| log(f" -> Using hardcoded global peers for industry '{key}'") | |
| break | |
| # Try general sector match if industry didn't match anything | |
| if not peer_tickers: | |
| sector = meta.sector or "" | |
| if sector: | |
| for key, tickers in INDIAN_PEERS.items(): | |
| if key.lower() in sector.lower() or sector.lower() in key.lower(): | |
| peer_tickers = [t for t in tickers if t.upper() != ticker.upper()][:max_peers] | |
| log(f" -> Using hardcoded Indian peers for sector '{key}'") | |
| break | |
| if not peer_tickers: | |
| for key, tickers in GLOBAL_PEERS.items(): | |
| if key.lower() in sector.lower() or sector.lower() in key.lower(): | |
| peer_tickers = [t for t in tickers if t.upper() != ticker.upper()][:max_peers] | |
| log(f" -> Using hardcoded global peers for sector '{key}'") | |
| break | |
| # Remove self | |
| peer_tickers = [t for t in peer_tickers if t.upper() != ticker.upper()][:max_peers] | |
| if not peer_tickers: | |
| log(" [Warning] No peers found. Peer comparison section will be skipped.") | |
| return [] | |
| # 3. Fetch snapshots for each peer | |
| for pt in peer_tickers: | |
| log(f" -> Fetching peer data: {pt} ...") | |
| snap = fetch_peer_snapshot(pt) | |
| if snap: | |
| peers_out.append(snap) | |
| log(f" [OK] {snap['name']} ({pt})") | |
| else: | |
| log(f" [ERR] Could not fetch {pt}") | |
| log(f" [OK] {len(peers_out)} peers ready") | |
| return peers_out | |