Spaces:
Sleeping
Sleeping
| """Generate comparative tables and commentary for two portfolios.""" | |
| from typing import Dict, List | |
| import pandas as pd | |
| from infrastructure.cache import CacheUnavailableError | |
| from infrastructure.llm_client import llm_service | |
| from infrastructure.output_api import extract_portfolio_id, fetch_metrics_cached | |
| from presentation.components.analysis_formatter import ( | |
| render_analysis_html, | |
| render_status_html, | |
| ) | |
| from prompts.system_prompts import COMPARISON_SYSTEM_PROMPT | |
| def show_comparison_table(portfolio_a: str, portfolio_b: str): | |
| """Return the comparison DataFrame along with commentary.""" | |
| pid_a = extract_portfolio_id(portfolio_a) | |
| pid_b = extract_portfolio_id(portfolio_b) | |
| if not pid_a or not pid_b: | |
| message = "❌ Invalid portfolio IDs." | |
| return _message_df(message), render_status_html(message) | |
| try: | |
| df, commentary = _build_comparison_with_comment(pid_a, pid_b) | |
| return df, render_analysis_html(commentary) | |
| except CacheUnavailableError as e: | |
| wait = int(e.retry_in) + 1 | |
| message = f"⚠️ Metrics temporarily unavailable. Retry in ~{wait} seconds." | |
| return _message_df(message), render_status_html(message) | |
| except Exception: | |
| message = "❌ Unable to build comparison right now. Please try again later." | |
| return _message_df(message), render_status_html(message) | |
| def _build_comparison_with_comment(p1: str, p2: str): | |
| m1 = fetch_metrics_cached(p1) | |
| m2 = fetch_metrics_cached(p2) | |
| if not m1 or not m2: | |
| raise ValueError("Metrics unavailable for one or both portfolios.") | |
| rows = _rows_from_metrics(m1, m2) | |
| df = pd.DataFrame(rows, columns=["Δ Difference", "Portfolio A", "Portfolio B", "Metric"]) | |
| commentary = _collect_commentary(rows) | |
| return df, commentary | |
| def _rows_from_metrics(m1: Dict, m2: Dict) -> List[Dict]: | |
| all_keys = sorted(set(m1.keys()) | set(m2.keys())) | |
| rows: List[Dict] = [] | |
| for k in all_keys: | |
| v1 = m1.get(k, 0) | |
| v2 = m2.get(k, 0) | |
| diff = v1 - v2 | |
| symbol = "▲" if diff > 0 else "▼" if diff < 0 else "—" | |
| rows.append( | |
| { | |
| "Δ Difference": f"{symbol} {diff:+.3f}", | |
| "Portfolio A": round(v1, 3), | |
| "Portfolio B": round(v2, 3), | |
| "Metric": k, | |
| } | |
| ) | |
| return rows | |
| def _message_df(message: str) -> pd.DataFrame: | |
| return pd.DataFrame({"Message": [message]}) | |
| def _collect_commentary(rows: List[Dict]) -> str: | |
| commentary = "" | |
| for partial in _commentary_stream(rows): | |
| commentary = partial | |
| return commentary | |
| def _commentary_stream(rows: List[Dict]): | |
| summary = "\n".join(f"{r['Metric']}: {r['Δ Difference']}" for r in rows) | |
| prompt = ( | |
| f"{COMPARISON_SYSTEM_PROMPT}\n" | |
| f"Compare and explain the differences between Portfolio A and B:\n{summary}\n" | |
| f"Write your insights as a concise professional commentary." | |
| ) | |
| partial = "" | |
| try: | |
| iterator = llm_service.stream_chat( | |
| messages=[ | |
| {"role": "system", "content": "You are an investment portfolio analyst."}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| model="meta-llama/Meta-Llama-3.1-8B-Instruct", | |
| ) | |
| for delta in iterator: | |
| partial += delta | |
| yield partial | |
| except Exception: | |
| yield "❌ LLM analysis is unavailable right now. Please try again later." | |