""" 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 tags if DeepSeek outputs them import re text = re.sub(r'.*?', '', 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": "", }