Spaces:
Sleeping
Sleeping
| """ | |
| pipeline/analyst.py -- Multi-model AI narrative generation. | |
| Supports Google Gemini and Anthropic Claude. | |
| Each model generates the full narrative; if both selected, | |
| outputs are combined with a comparison section. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from pathlib import Path | |
| from typing import Dict, List, Optional | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| from config import GEMINI_API_KEY, CLAUDE_API_KEY, GEMINI_MODEL, CLAUDE_MODEL, OPENAI_API_KEY, OPENAI_MODEL, GROQ_API_KEY, GROQ_MODEL, NVIDIA_API_KEY, NVIDIA_MODEL | |
| import config | |
| import time | |
| # -- Formatting helpers -------------------------------------------------------- | |
| def _pct(v): | |
| return f"{v*100:.1f}%" if v is not None else "N/A" | |
| def _num(v, dec=2): | |
| return f"{v:.{dec}f}" if v is not None else "N/A" | |
| def _money(v, currency="INR", units="Crores"): | |
| """Format a pre-scaled monetary value. Data is already in Crores (INR) or Millions (USD).""" | |
| if v is None: | |
| return "N/A" | |
| sym = "\u20b9" if currency == "INR" else "$" | |
| av = abs(v) | |
| if currency == "INR": | |
| # Data already in Crores | |
| if av >= 100000: | |
| return f"{sym}{v/100000:,.2f} L.Cr" | |
| return f"{sym}{v:,.2f} Cr" | |
| else: | |
| # Data already in Millions | |
| if av >= 1000: | |
| return f"{sym}{v/1000:,.2f}B" | |
| return f"{sym}{v:,.2f}M" | |
| # -- Retry Wrapper ------------------------------------------------------------- | |
| def _with_retries(func, log_fn=None, max_retries=3, backoff_factor=2): | |
| """Wrap an API call with exponential backoff for rate limits.""" | |
| def wrapper(*args, **kwargs): | |
| retries = 0 | |
| while retries <= max_retries: | |
| try: | |
| res = func(*args, **kwargs) | |
| if res: | |
| return res | |
| # If None returned (handled internally without raising), assume fail | |
| except Exception as e: | |
| err_str = str(e).lower() | |
| if "429" in err_str or "too many requests" in err_str or "quota" in err_str: | |
| if retries == max_retries: | |
| if log_fn: log_fn(f"[Error] Max retries reached for {func.__name__}") | |
| return None | |
| sleep_time = backoff_factor ** retries | |
| if log_fn: log_fn(f"[Warning] Rate limited in {func.__name__}, retrying in {sleep_time}s...") | |
| time.sleep(sleep_time) | |
| retries += 1 | |
| continue | |
| else: | |
| if log_fn: log_fn(f"[Error] {func.__name__} failed: {e}") | |
| return None | |
| return None | |
| return None | |
| return wrapper | |
| # -- Context builder ----------------------------------------------------------- | |
| def _build_context(data: Dict, meta: Dict) -> str: | |
| """Build a rich textual context string from all financial data.""" | |
| years = data.get("years", []) | |
| inc = data.get("income_statement", {}) | |
| bal = data.get("balance_sheet", {}) | |
| cf = data.get("cash_flow", {}) | |
| rat = data.get("ratios", {}) | |
| gr = data.get("growth_metrics", {}) | |
| dp = data.get("dupont", {}) | |
| cur = data.get("currency", "INR") | |
| def row(label, arr, fmt="money"): | |
| units = data.get("units", "Crores") | |
| vals = " | ".join([ | |
| (_money(v, cur, units) if fmt == "money" else _pct(v) if fmt == "pct" else _num(v)) | |
| for v in (arr or []) | |
| ]) | |
| return f" {label}: {vals}" | |
| lines = [ | |
| f"COMPANY: {meta.get('name', '')} ({meta.get('ticker', '')})", | |
| f"EXCHANGE: {meta.get('exchange', '')} | SECTOR: {meta.get('sector', '')} | INDUSTRY: {meta.get('industry', '')}", | |
| f"COUNTRY: {meta.get('country', '')} | CURRENCY: {cur}", | |
| f"YEARS (oldest->newest): {' | '.join(str(y) for y in years)}", | |
| "", | |
| "=== INCOME STATEMENT ===", | |
| row("Revenue", inc.get("revenue")), | |
| row("Gross Profit", inc.get("gross_profit")), | |
| row("Operating Income", inc.get("operating_income")), | |
| row("EBITDA", inc.get("ebitda")), | |
| row("Net Income", inc.get("net_income")), | |
| row("Diluted EPS", inc.get("diluted_eps"), "num"), | |
| "", | |
| "=== BALANCE SHEET ===", | |
| row("Total Assets", bal.get("total_assets")), | |
| row("Total Equity", bal.get("total_equity")), | |
| row("Total Liabilities", bal.get("total_liabilities")), | |
| row("Long-term Debt", bal.get("long_term_debt")), | |
| row("Cash", bal.get("cash_and_equivalents")), | |
| row("Current Assets", bal.get("total_current_assets")), | |
| row("Current Liabilities",bal.get("total_current_liabilities")), | |
| "", | |
| "=== CASH FLOW ===", | |
| row("Cash from Operations", cf.get("cash_from_operations")), | |
| row("Capex", cf.get("capex")), | |
| row("Free Cash Flow", cf.get("free_cash_flow")), | |
| "", | |
| "=== PROFITABILITY RATIOS ===", | |
| row("Gross Margin", rat.get("gross_margin"), "pct"), | |
| row("Operating Margin", rat.get("operating_margin"), "pct"), | |
| row("Net Margin", rat.get("net_margin"), "pct"), | |
| row("ROE", rat.get("roe"), "pct"), | |
| row("ROA", rat.get("roa"), "pct"), | |
| row("ROIC", rat.get("roic"), "pct"), | |
| "", | |
| "=== LIQUIDITY & SOLVENCY ===", | |
| row("Current Ratio", rat.get("current_ratio"), "num"), | |
| row("Quick Ratio", rat.get("quick_ratio"), "num"), | |
| row("Debt-to-Equity", rat.get("debt_to_equity"), "num"), | |
| row("Net Debt/EBITDA", rat.get("net_debt_to_ebitda"), "num"), | |
| row("Interest Coverage", rat.get("interest_coverage"), "num"), | |
| "", | |
| "=== EFFICIENCY ===", | |
| row("Asset Turnover", rat.get("asset_turnover"), "num"), | |
| row("DSO (days)", rat.get("days_sales_outstanding"), "num"), | |
| row("DPO (days)", rat.get("days_payables"), "num"), | |
| row("Cash Conversion Cycle",rat.get("cash_conversion_cycle"), "num"), | |
| "", | |
| "=== DUPONT DECOMPOSITION ===", | |
| row("Net Margin", dp.get("net_margin"), "pct"), | |
| row("Asset Turnover", dp.get("asset_turnover"), "num"), | |
| row("Equity Multiplier", dp.get("equity_multiplier"), "num"), | |
| row("ROE (DuPont)", dp.get("roe_computed"), "pct"), | |
| "", | |
| "=== GROWTH METRICS ===", | |
| row("Revenue YoY", gr.get("revenue_yoy"), "pct"), | |
| f" Revenue 5yr CAGR: {_pct(gr.get('revenue_cagr_5y'))}", | |
| row("Net Income YoY", gr.get("net_income_yoy"), "pct"), | |
| f" Net Income 5yr CAGR: {_pct(gr.get('net_income_cagr_5y'))}", | |
| row("FCF YoY", gr.get("fcf_yoy"), "pct"), | |
| ] | |
| # Valuation | |
| val = data.get("valuation_snapshot", {}) | |
| if val: | |
| lines += [ | |
| "", | |
| "=== VALUATION SNAPSHOT ===", | |
| f" P/E (TTM): {_num(val.get('pe_ttm'))}", | |
| f" P/B: {_num(val.get('pb'))}", | |
| f" EV/EBITDA: {_num(val.get('ev_ebitda'))}", | |
| f" FCF Yield: {_pct(val.get('fcf_yield'))}", | |
| f" Dividend Yield: {_pct(val.get('dividend_yield'))}", | |
| ] | |
| # Peers | |
| peers = data.get("peers", []) | |
| if peers: | |
| lines += ["", "=== PEER BENCHMARKING ==="] | |
| for p in peers: | |
| lines.append( | |
| f" {p.get('name')} ({p.get('ticker')}): " | |
| f"Rev={_money(p.get('revenue'), p.get('currency','USD'))} | " | |
| f"OpM={_pct(p.get('operating_margin'))} | " | |
| f"ROE={_pct(p.get('roe'))} | " | |
| f"D/E={_num(p.get('debt_to_equity'))}" | |
| ) | |
| # Optional Annual Report Text (MD&A) | |
| ar_text = data.get("annual_report_context", "") | |
| if ar_text: | |
| lines += ["", "=== ANNUAL REPORT EXCERPT (10-K MD&A) ==="] | |
| lines.append(ar_text[:15000]) # Cap at 15k chars to save tokens | |
| return "\n".join(lines) | |
| # -- System prompt ------------------------------------------------------------- | |
| SYSTEM_PROMPT = """You are a senior financial analyst with expertise in Indian and global equity markets. | |
| You write rigorous, opinionated, data-driven financial analysis reports. | |
| Guidelines: | |
| - Be specific: cite exact numbers from the data provided, not generic statements | |
| - Be opinionated: state what the numbers actually mean, not just what they are | |
| - Use Indian financial terminology where appropriate (Crores, Lakhs, SEBI, NSE/BSE) | |
| - Flag data gaps honestly rather than speculating | |
| - Keep commentary professional but readable | |
| - Do NOT use markdown formatting in your response -- plain prose only | |
| - Paragraph breaks only (use double newline to separate paragraphs) | |
| - Numbers should be expressed in appropriate units (Cr for large INR amounts, B/M for USD) | |
| """ | |
| # -- Synthesis & PPT Prompts ---------------------------------------------------------- | |
| SYNTHESIS_PROMPT = """You are a master financial editor. You have been given {num_drafts} different AI-generated drafts for the '{section}' section of a financial report. | |
| Your job is to synthesize these drafts into a single, cohesive, authoritative, and brilliantly written section. | |
| Guidelines: | |
| - Combine the best insights and data points from all drafts. | |
| - Resolve any contradictions by taking the most logical or well-supported viewpoint. | |
| - Write in a professional, direct, and opinionated financial analyst tone. | |
| - Do NOT mention that this is a synthesis of multiple drafts or AI models. | |
| - Output ONLY the final synthesized text, with NO markdown formatting, NO introductory text, and NO commentary. Just the plain text paragraphs. | |
| """ | |
| GENERATE_PPT_JSON = """You are a top-tier Management Consultant (e.g., McKinsey, BCG) tasked with creating a highly visual, insightful, and strategic slide deck for a C-suite presentation. | |
| You have been provided with a detailed textual narrative analysis of the company's financial health, trends, SWOT, and peer benchmarking. | |
| Your task is to synthesize this narrative into a structured PowerPoint presentation. | |
| Output a strict JSON array of slide objects. Each slide object MUST have: | |
| - "title": A short, punchy, action-oriented title for the slide (e.g., "Margins Contract Amidst Pricing Pressure" rather than "Margin Analysis"). | |
| - "bullets": An array of 3-5 strings. These MUST NOT be generic summaries. Each bullet must be a deep, consultant-level insight, data-backed takeaway, or strategic implication extracted directly from the provided narrative. | |
| DO NOT include an Executive Summary slide, as that is handled separately. | |
| Focus on: | |
| 1. Strategic Overview & Market Position | |
| 2. Revenue & Profitability Dynamics | |
| 3. Balance Sheet & Cash Flow Health | |
| 4. Competitive Benchmarking Insights | |
| 5. Key Risks & Strategic Imperatives | |
| Do not include any markdown fences (like ```json), introduction, or conclusion. Output ONLY the raw JSON array. | |
| """ | |
| # -- Section prompts ----------------------------------------------------------- | |
| SECTION_PROMPTS = { | |
| "executive_summary": """ | |
| Write a 250-300 word executive summary of this company's financial health. | |
| Answer these questions directly: | |
| 1. Is the business financially healthy or deteriorating? | |
| 2. What is the most important trend in the numbers? | |
| 3. How does it stack up against peers? | |
| 4. What is the key risk visible in the data? | |
| 5. What is the key opportunity? | |
| Be direct and opinionated. No hedging. | |
| """, | |
| "company_profile": """ | |
| Write a 2-paragraph company profile covering: | |
| - What the company does, its key business segments, and geographic presence | |
| - Its position in the market, key products/services, major customers or clients if known | |
| Use public knowledge about this company; supplement with the financial data provided. | |
| """, | |
| "industry_profile": """ | |
| Write a 2-paragraph industry profile for this company's sector/industry covering: | |
| - Market size and structure (fragmented/consolidated, key players) | |
| - Key growth drivers and headwinds for this industry in the Indian/global context | |
| - Regulatory environment relevant to this industry | |
| """, | |
| "swot": """ | |
| Perform a rigorous SWOT analysis grounded in the financial data and industry knowledge. | |
| Format your response as four sections: | |
| STRENGTHS: | |
| [3-4 specific, data-grounded strengths] | |
| WEAKNESSES: | |
| [3-4 specific, data-grounded weaknesses] | |
| OPPORTUNITIES: | |
| [3-4 specific opportunities for this company/industry] | |
| THREATS: | |
| [3-4 specific threats, including competitive and regulatory] | |
| Be specific -- avoid generic SWOT points that could apply to any company. | |
| """, | |
| "industry_trends": """ | |
| Write 2-3 paragraphs on current and emerging trends in this company's industry, covering: | |
| - Macro trends (digitization, ESG, AI, PLI schemes for Indian companies, etc.) | |
| - Competitive dynamics and consolidation | |
| - Technology disruption if applicable | |
| - Regulatory changes (SEBI, RBI, sector-specific regulators for Indian companies) | |
| Ground trends in the financial data where possible (e.g., if margins are compressing, is it industry-wide?). | |
| """, | |
| "trend_analysis_revenue": """ | |
| Write 2-3 paragraphs analyzing the revenue and profitability trends over the analysis period. | |
| Specifically address: | |
| - Revenue trajectory (acceleration/deceleration/volatility) | |
| - Gross margin trend and what it reveals about pricing power | |
| - Operating leverage (are operating margins improving as revenue grows?) | |
| - Quality of earnings (is net income growing in line with revenue?) | |
| Reference specific numbers from the data. | |
| """, | |
| "trend_analysis_balance_sheet": """ | |
| Write 2 paragraphs analyzing balance sheet strength and capital structure trends. | |
| Address: | |
| - Liquidity position (current ratio, cash levels) | |
| - Leverage trajectory (is debt rising or falling relative to equity/earnings?) | |
| - Asset quality (goodwill, receivables trends) | |
| - Working capital efficiency | |
| """, | |
| "trend_analysis_cashflow": """ | |
| Write 2 paragraphs analyzing cash flow quality and capital allocation. | |
| Address: | |
| - OCF quality (does it consistently exceed net income, or lag it -- and why?) | |
| - Capex intensity and investment cycle | |
| - Free cash flow generation and its trajectory | |
| - How is FCF being allocated (buybacks, dividends, debt repayment, acquisitions)? | |
| """, | |
| "ratio_commentary": """ | |
| Write 2-3 paragraphs interpreting the ratio analysis. | |
| Focus on: | |
| - Which ratios show the most significant trends (positive or negative)? | |
| - Any ratios that are outliers vs. industry norms? | |
| - What do the efficiency ratios (DSO, DPO, CCC) tell us about operational quality? | |
| - DuPont decomposition: which driver is most responsible for ROE trends? | |
| """, | |
| "peer_commentary": """ | |
| Write 2 paragraphs benchmarking this company against its peers. | |
| Address: | |
| - Where does the company rank on the most important metrics for this industry? | |
| - What explains the gaps vs. peers (structural, cyclical, or management decisions)? | |
| - Is the company gaining or losing competitive position? | |
| """, | |
| "key_insights": """ | |
| List 4-6 key insights, one per line. Each insight must: | |
| - Be substantive and specific (not generic) | |
| - Reference a specific number or trend from the data | |
| - Convey an analytical judgment, not just a fact | |
| Format: One insight per line, starting with a dash (-). | |
| """, | |
| "risks": """ | |
| List the top 5-7 risks facing this company. For each risk, use this format: | |
| CATEGORY | Risk description (specific, grounded in data or industry knowledge) | |
| Categories to use: Financial, Business, Industry, Regulatory, ESG | |
| """, | |
| "things_to_watch": """ | |
| List 4-5 specific, monitorable metrics or events that would change the investment picture. | |
| Format: One per line starting with a dash (-). | |
| Be concrete: e.g., 'Watch for gross margin recovery above 30% in FY26' not 'watch margins'. | |
| """, | |
| "limitations": """ | |
| List any material analytical limitations in 3-5 bullet points. Include: | |
| - Data gaps (which line items are missing or estimated) | |
| - Comparability issues (fiscal year changes, acquisitions) | |
| - Data source caveats (scraped vs. filed data) | |
| - Any ratios that could not be computed meaningfully (negative equity, etc.) | |
| Format: One per line starting with a dash (-). | |
| """, | |
| } | |
| # -- Gemini -------------------------------------------------------------------- | |
| def _gemini_generate(context: str, section: str, log_fn=None) -> str: | |
| """Generate using Gemini -- tries API key first, then Google OAuth login.""" | |
| prompt = f"{SYSTEM_PROMPT}\n\nFINANCIAL DATA:\n{context}\n\nTASK:\n{SECTION_PROMPTS[section]}" | |
| # -- Path 1: API Key (new google-genai SDK) ------------------------------ | |
| if GEMINI_API_KEY: | |
| try: | |
| from google import genai | |
| from google.genai import types as gtypes | |
| client = genai.Client(api_key=GEMINI_API_KEY) | |
| response = client.models.generate_content( | |
| model=GEMINI_MODEL, | |
| contents=prompt, | |
| config=gtypes.GenerateContentConfig(temperature=0.2, max_output_tokens=1024), | |
| ) | |
| return response.text.strip() | |
| except Exception: | |
| pass | |
| # Fallback to old SDK | |
| try: | |
| import google.generativeai as genai_old | |
| genai_old.configure(api_key=GEMINI_API_KEY) | |
| model = genai_old.GenerativeModel(model_name=GEMINI_MODEL, system_instruction=SYSTEM_PROMPT) | |
| response = model.generate_content(prompt) | |
| return response.text.strip() | |
| except Exception as e2: | |
| if log_fn: log_fn(f"[Warning] Gemini failed for {section}: {str(e2)[:80]}") | |
| return None | |
| return None | |
| # -- OpenAI / ChatGPT --------------------------------------------------------- | |
| def _openai_generate(context: str, section: str, log_fn=None) -> str: | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI(api_key=OPENAI_API_KEY) | |
| prompt = f"FINANCIAL DATA:\n{context}\n\nTASK:\n{SECTION_PROMPTS[section]}" | |
| resp = client.chat.completions.create( | |
| model=OPENAI_MODEL, | |
| messages=[ | |
| {"role": "system", "content": SYSTEM_PROMPT}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| temperature=0.2, | |
| max_tokens=1024, | |
| ) | |
| return resp.choices[0].message.content.strip() | |
| except Exception as e: | |
| if log_fn: log_fn(f"[Warning] ChatGPT failed for {section}: {str(e)[:80]}") | |
| return None | |
| # -- Groq (free -- Llama 3.3 70B) ---------------------------------------------- | |
| def _groq_generate(context: str, section: str, log_fn=None) -> str: | |
| """Generate using Groq's OpenAI-compatible API (free tier, Llama 3.3 70B).""" | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI( | |
| api_key=GROQ_API_KEY, | |
| base_url="https://api.groq.com/openai/v1", | |
| ) | |
| prompt = f"FINANCIAL DATA:\n{context}\n\nTASK:\n{SECTION_PROMPTS[section]}" | |
| resp = client.chat.completions.create( | |
| model=GROQ_MODEL, | |
| messages=[ | |
| {"role": "system", "content": SYSTEM_PROMPT}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| temperature=0.2, | |
| max_tokens=1024, | |
| ) | |
| return resp.choices[0].message.content.strip() | |
| except Exception as e: | |
| if log_fn: log_fn(f"[Warning] Groq failed for {section}: {str(e)[:80]}") | |
| return None | |
| # -- NVIDIA NIM (Llama 3.1 405B) ---------------------------------------------- | |
| def _nvidia_generate(context: str, section: str, log_fn=None) -> Optional[str]: | |
| """Generate using NVIDIA NIM OpenAI-compatible API.""" | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI( | |
| api_key=NVIDIA_API_KEY, | |
| base_url="https://integrate.api.nvidia.com/v1", | |
| ) | |
| prompt = f"FINANCIAL DATA:\n{context}\n\nTASK:\n{SECTION_PROMPTS[section]}" | |
| resp = client.chat.completions.create( | |
| model=NVIDIA_MODEL, | |
| messages=[ | |
| {"role": "system", "content": SYSTEM_PROMPT}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| temperature=0.2, | |
| max_tokens=1024, | |
| ) | |
| return resp.choices[0].message.content.strip() | |
| except Exception as e: | |
| if log_fn: log_fn(f"[Warning] NVIDIA NIM failed for {section}: {str(e)[:80]}") | |
| return None | |
| # -- Claude -------------------------------------------------------------------- | |
| def _claude_generate(context: str, section: str, log_fn=None) -> str: | |
| try: | |
| import anthropic | |
| client = anthropic.Anthropic(api_key=CLAUDE_API_KEY) | |
| prompt = f"FINANCIAL DATA:\n{context}\n\nTASK:\n{SECTION_PROMPTS[section]}" | |
| message = client.messages.create( | |
| model=CLAUDE_MODEL, | |
| max_tokens=1024, | |
| system=SYSTEM_PROMPT, | |
| messages=[{"role": "user", "content": prompt}], | |
| ) | |
| return message.content[0].text.strip() | |
| except Exception as e: | |
| if log_fn: log_fn(f"[Warning] Claude failed for {section}: {str(e)[:80]}") | |
| return None | |
| # -- Hugging Face (DeepSeek) --------------------------------------------------- | |
| def _hf_generate(context: str, section: str, log_fn=None) -> Optional[str]: | |
| """Generate using Hugging Face Inference API (DeepSeek).""" | |
| hf_key = getattr(config, "HF_API_KEY", "") | |
| if not hf_key: | |
| return None | |
| try: | |
| from huggingface_hub import InferenceClient | |
| client = InferenceClient(api_key=hf_key) | |
| prompt = f"FINANCIAL DATA:\n{context}\n\nTASK:\n{SECTION_PROMPTS[section]}\n\nRespond ONLY with the requested analysis text." | |
| messages = [ | |
| {"role": "system", "content": SYSTEM_PROMPT}, | |
| {"role": "user", "content": prompt} | |
| ] | |
| # We use the recommended chat_completion method | |
| response = client.chat_completion( | |
| model="deepseek-ai/DeepSeek-R1-Distill-Qwen-32B", | |
| messages=messages, | |
| max_tokens=1024, | |
| temperature=0.2 | |
| ) | |
| if response and response.choices: | |
| text = response.choices[0].message.content | |
| # Strip <think> tags if DeepSeek outputs them | |
| import re | |
| text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL).strip() | |
| return text | |
| return None | |
| except Exception as e: | |
| err_str = str(e).lower() | |
| if "429" in err_str or "too many" in err_str or "rate limit" in err_str: | |
| raise Exception("429 Too Many Requests") # Let the retry wrapper catch it | |
| if log_fn: log_fn(f"[Warning] HF DeepSeek failed for {section}: {str(e)[:80]}") | |
| return None | |
| # -- Parse structured sections ------------------------------------------------- | |
| def _parse_swot(text: str) -> Dict: | |
| """Parse SWOT text into a structured dict.""" | |
| result = {"strengths": [], "weaknesses": [], "opportunities": [], "threats": [], "raw": text} | |
| current = None | |
| for line in text.split("\n"): | |
| l = line.strip() | |
| if l.upper().startswith("STRENGTH"): | |
| current = "strengths" | |
| elif l.upper().startswith("WEAKNESS"): | |
| current = "weaknesses" | |
| elif l.upper().startswith("OPPORTUNIT"): | |
| current = "opportunities" | |
| elif l.upper().startswith("THREAT"): | |
| current = "threats" | |
| elif current and l and not l.upper() in ("STRENGTHS:", "WEAKNESSES:", "OPPORTUNITIES:", "THREATS:"): | |
| clean = l.lstrip("-•* ") | |
| if clean: | |
| result[current].append(clean) | |
| return result | |
| def _parse_bullets(text: str) -> List[str]: | |
| """Parse a bulleted list response into a list of strings.""" | |
| items = [] | |
| for line in text.split("\n"): | |
| l = line.strip().lstrip("-•* ") | |
| if l: | |
| items.append(l) | |
| return items | |
| def _parse_risks(text: str) -> List[Dict]: | |
| """Parse risk list into categorized dicts.""" | |
| risks = [] | |
| for line in text.split("\n"): | |
| l = line.strip().lstrip("-•* ") | |
| if "|" in l: | |
| parts = l.split("|", 1) | |
| risks.append({"category": parts[0].strip(), "description": parts[1].strip()}) | |
| elif l: | |
| risks.append({"category": "General", "description": l}) | |
| return risks | |
| # -- Main entry point ---------------------------------------------------------- | |
| def generate_narrative( | |
| data: Dict, | |
| meta: Dict, | |
| models: List[str], | |
| log_fn=None, | |
| ) -> Dict: | |
| """ | |
| Generate full narrative analysis using selected AI models. | |
| Args: | |
| data: Calculated financial data dict. | |
| meta: CompanyMeta.to_dict() | |
| models: List of model names, e.g. ['gemini', 'claude'] or ['gemini'] | |
| log_fn: Optional progress callback. | |
| Returns: | |
| Dict with all narrative fields populated. | |
| """ | |
| def log(msg): | |
| if log_fn: | |
| log_fn(msg) | |
| use_gemini = "gemini" in models and bool(GEMINI_API_KEY) | |
| use_claude = "claude" in models and bool(CLAUDE_API_KEY) | |
| use_openai = "openai" in models and bool(OPENAI_API_KEY) | |
| use_groq = "groq" in models and bool(GROQ_API_KEY) | |
| use_nvidia = "nvidia" in models and bool(NVIDIA_API_KEY) | |
| use_hf = "huggingface" in models and bool(getattr(config, "HF_API_KEY", "")) | |
| if not any([use_gemini, use_claude, use_openai, use_groq, use_nvidia, use_hf]): | |
| log("No AI models configured. Running data-only analysis (no narrative).") | |
| return _empty_narrative() | |
| context = _build_context(data, meta) | |
| active = [m for m, ok in [("Gemini",use_gemini),("Claude",use_claude),("ChatGPT",use_openai),("Groq",use_groq),("NVIDIA",use_nvidia),("HF-DeepSeek",use_hf)] if ok] | |
| log(f"Generating AI narrative ({', '.join(active)}) ...") | |
| def _synthesize_drafts(section: str, drafts: Dict[str, str]) -> str: | |
| """Use the best available model to merge drafts into a single consensus response.""" | |
| # Pick the synthesizer based on perceived quality/capability | |
| synthesizer = None | |
| if use_openai: synthesizer = ("ChatGPT", _openai_generate) | |
| elif use_claude: synthesizer = ("Claude", _claude_generate) | |
| elif use_nvidia: synthesizer = ("NVIDIA", _nvidia_generate) | |
| elif use_gemini: synthesizer = ("Gemini", _gemini_generate) | |
| elif use_groq: synthesizer = ("Groq", _groq_generate) | |
| elif use_hf: synthesizer = ("HF-DeepSeek", _hf_generate) | |
| if not synthesizer: | |
| return list(drafts.values())[0] | |
| name, fn = synthesizer | |
| log(f" > Synthesizing {len(drafts)} drafts using {name}...") | |
| draft_text = "\n\n".join([f"--- Draft from {agent} ---\n{text}" for agent, text in drafts.items()]) | |
| prompt = SYNTHESIS_PROMPT.format(section=section, num_drafts=len(drafts)) + "\n\n" + draft_text | |
| # We reuse the context parameter to pass the prompt, and bypass the section wrapping by mocking it | |
| # Actually, let's just create a custom direct call for synthesis to avoid section wrapping issues. | |
| # But to reuse the functions easily, we can temporarily override the SECTION_PROMPTS. | |
| original_prompt = SECTION_PROMPTS.get(section, "") | |
| SECTION_PROMPTS[section] = prompt | |
| try: | |
| res = fn("", section) # Context empty because prompt contains everything | |
| finally: | |
| SECTION_PROMPTS[section] = original_prompt | |
| if res: return res | |
| # If synthesis fails, fallback to simple concatenation | |
| return "\n\n".join([f"[{a}]\n{t}" for a, t in drafts.items()]) | |
| import concurrent.futures | |
| def gen(section: str) -> str: | |
| tasks = {} | |
| # Max workers set to 3 to avoid extreme rate limits on free tiers (like Groq's 30 RPM) | |
| with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: | |
| if use_gemini: tasks[executor.submit(_with_retries(_gemini_generate, log_fn), context, section)] = "Gemini" | |
| if use_claude: tasks[executor.submit(_with_retries(_claude_generate, log_fn), context, section)] = "Claude" | |
| if use_openai: tasks[executor.submit(_with_retries(_openai_generate, log_fn), context, section)] = "ChatGPT" | |
| if use_groq: tasks[executor.submit(_with_retries(_groq_generate, log_fn), context, section)] = "Groq" | |
| if use_nvidia: tasks[executor.submit(_with_retries(_nvidia_generate, log_fn), context, section)] = "NVIDIA" | |
| if use_hf: tasks[executor.submit(_with_retries(_hf_generate, log_fn), context, section)] = "HF-DeepSeek" | |
| drafts = {} | |
| for future in concurrent.futures.as_completed(tasks): | |
| name = tasks[future] | |
| try: | |
| res = future.result() | |
| if res: drafts[name] = res | |
| except Exception as e: | |
| if log_fn: log_fn(f"[Warning] {name} thread failed: {e}") | |
| if len(drafts) == 0: | |
| return "" | |
| if len(drafts) == 1: | |
| return list(drafts.values())[0] | |
| return _synthesize_drafts(section, drafts) | |
| def _generate_ppt_json() -> str: | |
| # Pick the smartest model available for structured JSON generation (prioritize Anthropic Claude for Consultant output) | |
| synthesizer = None | |
| if use_claude: synthesizer = ("Claude", _claude_generate) | |
| elif use_openai: synthesizer = ("ChatGPT", _openai_generate) | |
| elif use_gemini: synthesizer = ("Gemini", _gemini_generate) | |
| elif use_nvidia: synthesizer = ("NVIDIA", _nvidia_generate) | |
| elif use_groq: synthesizer = ("Groq", _groq_generate) | |
| if not synthesizer: | |
| return "" | |
| name, fn = synthesizer | |
| log(f"\n[PPT] Generating PowerPoint Structure using {name} (Consultant-level Insights) ...") | |
| # Build narrative context to feed the PPT generator | |
| narrative_context = f"=== Company Overview ===\n{company_overview}\n\n=== SWOT ===\n{swot_raw}\n\n=== Trends ===\n{trend_revenue}\n{trend_bs}\n{trend_cf}\n\n=== Peer Benchmarking ===\n{peer_commentary}\n\n=== Risks & Insights ===\n{risks_raw}\n{insights_raw}" | |
| original_prompt = SECTION_PROMPTS.get("ppt_json", "") | |
| SECTION_PROMPTS["ppt_json"] = GENERATE_PPT_JSON | |
| try: | |
| res = fn(narrative_context, "ppt_json") | |
| finally: | |
| if "ppt_json" in SECTION_PROMPTS: | |
| del SECTION_PROMPTS["ppt_json"] | |
| return res if res else "" | |
| log(" > Executing section: Executive Summary") | |
| exec_summary = gen("executive_summary") | |
| log(" Generating company profile ...") | |
| company_overview = gen("company_profile") | |
| log(" Generating industry profile ...") | |
| industry_profile = gen("industry_profile") | |
| log(" Generating SWOT ...") | |
| swot_raw = gen("swot") | |
| swot = _parse_swot(swot_raw) | |
| log(" Generating industry trends ...") | |
| industry_trends = gen("industry_trends") | |
| log(" Generating trend analysis ...") | |
| trend_revenue = gen("trend_analysis_revenue") | |
| trend_bs = gen("trend_analysis_balance_sheet") | |
| trend_cf = gen("trend_analysis_cashflow") | |
| log(" Generating ratio commentary ...") | |
| ratio_commentary = gen("ratio_commentary") | |
| log(" Generating peer commentary ...") | |
| peer_commentary = gen("peer_commentary") | |
| log(" Generating insights ...") | |
| insights_raw = gen("key_insights") | |
| insights = _parse_bullets(insights_raw) | |
| log(" Generating risks ...") | |
| risks_raw = gen("risks") | |
| risks = _parse_risks(risks_raw) | |
| log(" Generating things to watch ...") | |
| watch_raw = gen("things_to_watch") | |
| things_to_watch = _parse_bullets(watch_raw) | |
| log(" Generating limitations ...") | |
| lim_raw = gen("limitations") | |
| limitations = _parse_bullets(lim_raw) | |
| ppt_json = _generate_ppt_json() | |
| log(" [OK] AI narrative generation complete") | |
| return { | |
| "executive_summary": exec_summary, | |
| "company_overview": company_overview, | |
| "industry_profile": industry_profile, | |
| "swot": swot, | |
| "industry_trends": industry_trends, | |
| "trend_analysis": { | |
| "revenue_and_profitability": trend_revenue, | |
| "balance_sheet_strength": trend_bs, | |
| "cash_flow_generation": trend_cf, | |
| }, | |
| "ratio_analysis_commentary": ratio_commentary, | |
| "peer_comparison_commentary": peer_commentary, | |
| "key_insights": insights, | |
| "risks": risks, | |
| "things_to_watch": things_to_watch, | |
| "limitations_and_caveats": limitations, | |
| "models_used": models, | |
| "ppt_json": ppt_json, | |
| } | |
| def _empty_narrative() -> Dict: | |
| return { | |
| "executive_summary": "AI narrative not generated (no API keys configured).", | |
| "company_overview": "", | |
| "industry_profile": "", | |
| "swot": {"strengths": [], "weaknesses": [], "opportunities": [], "threats": []}, | |
| "industry_trends": "", | |
| "trend_analysis": {"revenue_and_profitability": "", "balance_sheet_strength": "", "cash_flow_generation": ""}, | |
| "ratio_analysis_commentary": "", | |
| "peer_comparison_commentary":"", | |
| "key_insights": [], | |
| "risks": [], | |
| "things_to_watch": [], | |
| "limitations_and_caveats": [], | |
| "models_used": [], | |
| "ppt_json": "", | |
| } | |