research.py

"""
research.py — Research Note (beta): a first slice of "Auto Research / auto report".

V1 scope (what this file does today):
    input ticker → pull fundamentals via yfinance .info + recent headlines
    → local LLM writes a structured English research note covering
      valuation, moat / supply-chain position, bull vs bear case, key risks.

V2 (not built yet, see Automation tab description):
    multi-step agentic research (filings, peers, news crawl) and an automatic
    trigger that generates a report whenever a NEW ticker enters the pool.
"""
from __future__ import annotations

import traceback


_FIELDS = [
    ("longName", "Name"), ("sector", "Sector"), ("industry", "Industry"),
    ("marketCap", "Market cap"), ("trailingPE", "P/E (ttm)"),
    ("forwardPE", "P/E (fwd)"), ("priceToSalesTrailing12Months", "P/S (ttm)"),
    ("priceToBook", "P/B"), ("enterpriseToEbitda", "EV/EBITDA"),
    ("profitMargins", "Net margin"), ("grossMargins", "Gross margin"),
    ("operatingMargins", "Operating margin"), ("returnOnEquity", "ROE"),
    ("revenueGrowth", "Revenue growth (yoy)"), ("earningsGrowth", "Earnings growth (yoy)"),
    ("freeCashflow", "Free cash flow"), ("totalCash", "Total cash"),
    ("totalDebt", "Total debt"), ("dividendYield", "Dividend yield"),
    ("beta", "Beta"), ("fiftyTwoWeekHigh", "52w high"), ("fiftyTwoWeekLow", "52w low"),
    ("currentPrice", "Price"), ("targetMeanPrice", "Analyst mean target"),
    ("recommendationKey", "Street view"),
]


def _fmt(key: str, v):
    if v is None:
        return None
    try:
        if key in ("marketCap", "freeCashflow", "totalCash", "totalDebt"):
            v = float(v)
            return f"${v/1e9:,.1f}B" if abs(v) >= 1e9 else f"${v/1e6:,.0f}M"
        if key in ("profitMargins", "grossMargins", "operatingMargins", "returnOnEquity",
                   "revenueGrowth", "earningsGrowth", "dividendYield"):
            return f"{float(v):.1%}"
        if isinstance(v, float):
            return f"{v:,.2f}"
    except (TypeError, ValueError):
        pass
    return str(v)


def gather_facts(ticker: str) -> tuple:
    """Return (facts_markdown, facts_plain, error)."""
    try:
        import yfinance as yf
        tk = yf.Ticker(ticker)
        info = tk.info or {}
    except Exception as e:
        traceback.print_exc()
        return "", "", f"Could not fetch fundamentals for {ticker}: {e}"
    if not info or info.get("regularMarketPrice") is None and info.get("currentPrice") is None \
            and not info.get("longName"):
        return "", "", f"No fundamentals returned for {ticker} — check the symbol."
    rows, plain = [], []
    for key, label in _FIELDS:
        val = _fmt(key, info.get(key))
        if val is None:
            continue
        rows.append(f"| {label} | {val} |")
        plain.append(f"{label}: {val}")
    summary = (info.get("longBusinessSummary") or "")[:900]
    heads = []
    try:
        for x in (tk.news or [])[:6]:
            c = x.get("content", x)
            t = c.get("title")
            if t:
                heads.append(t)
    except Exception:
        pass
    md = f"**{info.get('longName', ticker)}** ({ticker})\n\n"
    md += "| Metric | Value |\n|---|---|\n" + "\n".join(rows)
    if heads:
        md += "\n\n**Recent headlines:** " + " · ".join(heads[:5])
    plain_txt = "\n".join(plain)
    if summary:
        plain_txt += f"\nBusiness: {summary}"
    if heads:
        plain_txt += "\nRecent headlines: " + " | ".join(heads)
    return md, plain_txt, ""