Spaces:
Sleeping
Sleeping
File size: 3,479 Bytes
6cf4c9c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
"""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."
|