kussssh/IPO-Analyzer

1"""2Chunk- and module-level diff helpers for document version comparisons.3"""4from __future__ import annotations5 6from typing import Any, Dict, List, Optional, Set7 8 9def compute_chunk_diff(previous_chunks: List[Dict[str, Any]], current_chunks: List[Dict[str, Any]]) -> Dict[str, Any]:10    previous_hashes: Set[str] = {chunk["chunk_hash"] for chunk in previous_chunks}11    current_hashes: Set[str] = {chunk["chunk_hash"] for chunk in current_chunks}12 13    new_hashes = current_hashes - previous_hashes14    removed_hashes = previous_hashes - current_hashes15 16    new_chunks = [chunk for chunk in current_chunks if chunk["chunk_hash"] in new_hashes]17    removed_chunks = [chunk for chunk in previous_chunks if chunk["chunk_hash"] in removed_hashes]18 19    section_changes = sorted(20        {21            chunk.get("section") or "General"22            for chunk in [*new_chunks, *removed_chunks]23        }24    )25 26    return {27        "new_chunks": len(new_chunks),28        "removed_chunks": len(removed_chunks),29        "changed_chunks": len(new_chunks) + len(removed_chunks),30        "section_changes": section_changes,31        "new_chunk_hashes": sorted(new_hashes),32        "removed_chunk_hashes": sorted(removed_hashes),33    }34 35 36def build_analysis_diff(37    company_name: str,38    from_doc_type: str,39    to_doc_type: str,40    base_result: Dict[str, Any],41    later_result: Dict[str, Any],42    chunk_diff: Dict[str, Any],43) -> Dict[str, Any]:44    module_changes = []45    red_flags = []46 47    base_modules = (base_result or {}).get("modules", {})48    later_modules = (later_result or {}).get("modules", {})49    module_keys = [50        "business",51        "financials",52        "growth_quality",53        "valuation",54        "promoter_ofs",55        "use_of_proceeds",56        "risks",57        "institutional",58    ]59 60    for key in module_keys:61        previous_module = base_modules.get(key)62        current_module = later_modules.get(key)63        if not previous_module or not current_module:64            continue65 66        changes = []67        previous_signal = previous_module.get("signal")68        current_signal = current_module.get("signal")69        if previous_signal != current_signal:70            changes.append(f"Signal changed from {previous_signal} to {current_signal}")71 72        if key == "risks":73            previous_titles = {risk.get("risk_title") for risk in previous_module.get("high_risks", []) + previous_module.get("medium_risks", [])}74            current_titles = {risk.get("risk_title") for risk in current_module.get("high_risks", []) + current_module.get("medium_risks", [])}75            added_risks = sorted(title for title in current_titles - previous_titles if title)76            if added_risks:77                red_flags.append(f"New risk factor added: {added_risks[0]}")78                changes.append(f"{len(added_risks)} new risk factor(s) added")79 80        if key == "promoter_ofs":81            previous_holding = previous_module.get("promoter_post_ipo_pct")82            current_holding = current_module.get("promoter_post_ipo_pct")83            if previous_holding is not None and current_holding is not None and current_holding < previous_holding:84                red_flags.append(f"Promoter stake dropped from {previous_holding}% to {current_holding}%")85                changes.append("Promoter post-issue holding reduced")86 87        if key == "financials":88            prev_year = _latest_financial_year(previous_module)89            current_year = _latest_financial_year(current_module)90            prev_pat = prev_year.get("pat_cr")91            current_pat = current_year.get("pat_cr")92            if prev_pat is not None and current_pat is not None and current_pat < prev_pat and current_pat < 0:93                red_flags.append(f"Loss increased from {prev_pat} Cr to {current_pat} Cr")94                changes.append("Loss increased in latest financial year")95 96        if key == "use_of_proceeds":97            prev_allocs = _purpose_map(previous_module)98            curr_allocs = _purpose_map(current_module)99            if prev_allocs != curr_allocs:100                red_flags.append("Fund use revised")101                changes.append("Objects of issue changed")102 103        if key == "valuation":104            previous_peers = {peer.get("company_name") for peer in previous_module.get("peers", []) if peer.get("company_name")}105            current_peers = {peer.get("company_name") for peer in current_module.get("peers", []) if peer.get("company_name")}106            if previous_peers != current_peers:107                red_flags.append("Peer list changed")108                changes.append("Listed peer set changed")109 110        if not changes and previous_module != current_module:111            changes.append("Module data updated")112 113        if changes:114            module_changes.append(115                {116                    "module": key,117                    "changes": changes,118                    "status": "flagged" if any(change in red_flags for change in changes) else "changed",119                }120            )121 122    unique_red_flags = []123    seen_flags = set()124    for item in red_flags:125        if item in seen_flags:126            continue127        seen_flags.add(item)128        unique_red_flags.append(item)129 130    report_lines = [131        f"# {company_name}: {from_doc_type.upper()} -> {to_doc_type.upper()} Diff Report",132        "",133        f"- Changed chunks: {chunk_diff['changed_chunks']}",134        f"- New chunks: {chunk_diff['new_chunks']}",135        f"- Removed chunks: {chunk_diff['removed_chunks']}",136        f"- Sections touched: {', '.join(chunk_diff['section_changes']) if chunk_diff['section_changes'] else 'None'}",137        "",138    ]139 140    if unique_red_flags:141        report_lines.append("## Red Flags")142        for flag in unique_red_flags:143            report_lines.append(f"- {flag}")144        report_lines.append("")145 146    if module_changes:147        report_lines.append("## Module Changes")148        for change in module_changes:149            report_lines.append(f"- {change['module']}: {'; '.join(change['changes'])}")150 151    return {152        "company": company_name,153        "from_doc_type": from_doc_type,154        "to_doc_type": to_doc_type,155        "chunk_diff": chunk_diff,156        "module_changes": module_changes,157        "red_flags": unique_red_flags,158        "report_markdown": "\n".join(report_lines).strip(),159    }160 161 162def _latest_financial_year(module: Dict[str, Any]) -> Dict[str, Any]:163    financials = module.get("financials", [])164    return financials[-1] if financials else {}165 166 167def _purpose_map(module: Dict[str, Any]) -> Dict[str, Any]:168    return {169        allocation.get("purpose"): round(float(allocation.get("amount_cr") or 0), 2)170        for allocation in module.get("allocations", [])171        if allocation.get("purpose")172    }173