""" news_watch.py — daily news check for held tickers (US version, yfinance.news). Rule requested by the user: for each holding, look only at TODAY's news. If there is news → push a short AI brief. If there is none → ignore (the ticker is listed under "Quiet today" so you know it was checked). """ from __future__ import annotations import datetime as dt import os import traceback import paths HOLDINGS_FILE = os.path.join(paths.OUTPUT_DIR, "holdings.txt") # ── holdings persistence ───────────────────────────────────────────────── def load_holdings() -> list: try: with open(HOLDINGS_FILE, encoding="utf-8") as f: return [x.strip().upper() for x in f if x.strip()] except FileNotFoundError: return [] def save_holdings(tickers: list): seen, out = set(), [] for t in tickers: t = t.strip().upper() if t and t not in seen: seen.add(t) out.append(t) with open(HOLDINGS_FILE, "w", encoding="utf-8") as f: f.write("\n".join(out)) return out # ── news fetch ─────────────────────────────────────────────────────────── def _today_utc() -> dt.date: return dt.datetime.now(dt.timezone.utc).date() def fetch_today_news(ticker: str) -> list: """Return today's news items: [{'title','publisher','time','link'}…].""" try: import yfinance as yf items = yf.Ticker(ticker).news or [] except Exception: traceback.print_exc() return [] today = _today_utc() out = [] for it in items: # yfinance has two schemas: legacy flat dict, or {'content': {...}} c = it.get("content", it) title = c.get("title") or "" ts = it.get("providerPublishTime") when = None if ts: when = dt.datetime.fromtimestamp(ts, dt.timezone.utc) else: pub = c.get("pubDate") or c.get("displayTime") if pub: try: when = dt.datetime.fromisoformat(str(pub).replace("Z", "+00:00")) except ValueError: when = None if when is None or when.date() != today: continue pubr = c.get("publisher") or (c.get("provider") or {}).get("displayName") or "" link = c.get("link") or (c.get("canonicalUrl") or {}).get("url") or "" if title: out.append({"title": title, "publisher": pubr, "time": when.strftime("%H:%M UTC"), "link": link}) return out def _llm_brief(ticker: str, items: list) -> str: heads = "\n".join(f"- [{x['time']}] {x['title']} ({x['publisher']})" for x in items) try: import llm_local if not llm_local.is_loaded(): return "" prompt = ( f"You are an equity news analyst. Today's headlines for {ticker} " f"(a stock the user currently HOLDS):\n{heads}\n\n" "In ENGLISH ONLY, write:\n" "1) **Per-headline:** one short line per headline above — what it says " "and why it matters (or 'noise') for the holding;\n" "2) **Net read:** POSITIVE / NEGATIVE / NEUTRAL with one sentence why;\n" "3) **Action:** one concrete suggestion. ≤180 words, no disclaimers." ) return llm_local.chat(prompt, max_tokens=420, worker="reporter") except Exception: return "" def check_holdings_news_stream(tickers=None): """Generator for the UI: emits cumulative markdown as it goes — each ticker, each headline, and each AI brief appears the moment it's ready, so the user never stares at a frozen screen. AI briefs run on the Reporter sub-agent.""" import llm_local tickers = tickers if tickers is not None else load_holdings() if not tickers: yield ("**No holdings configured.** Add tickers above (e.g. `AAPL, NVDA`) " "and save — they'll be checked for news every day.") return stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M UTC") out = [f"_Checking {len(tickers)} holding(s) · {stamp}_"] quiet = [] def render(): body = "\n".join(out) if quiet: body += f"\n\n**Quiet today (no news):** {', '.join(quiet)}" return body for t in tickers: out.append(f"\n### 📰 {t} …searching today's news") yield render() items = fetch_today_news(t) if not items: out.pop() # drop the "searching" line quiet.append(t) yield render() continue out[-1] = f"\n### 📰 {t} — {len(items)} item(s) today" yield render() # print each headline the moment we have it for x in items[:6]: link = f" · [link]({x['link']})" if x["link"] else "" out.append(f"- **{x['time']}** {x['title']} — *{x['publisher']}*{link}") yield render() # then stream the AI brief for this ticker (Reporter sub-agent) if llm_local.is_loaded("reporter") or llm_local.is_loaded("translator"): wk = "reporter" if llm_local.is_loaded("reporter") else "translator" heads = "\n".join(f"- [{x['time']}] {x['title']} ({x['publisher']})" for x in items[:6]) prompt = ( f"You are an equity news analyst. Today's headlines for {ticker_safe(t)} " f"(a stock the user HOLDS):\n{heads}\n\nIn ENGLISH ONLY:\n" f"1) **Per-headline:** one short line each — what it says and why it " f"matters (or 'noise');\n2) **Net read:** POSITIVE / NEGATIVE / NEUTRAL " f"+ one sentence;\n3) **Action:** one concrete suggestion. ≤180 words.") out.append("\n> 🤖 **Reporter sub-agent brief:** _thinking…_") base = len(out) - 1 for acc in llm_local.chat_stream(prompt, max_tokens=420, worker=wk): out[base] = "> 🤖 **Reporter sub-agent brief:**\n>\n> " + \ acc.replace("\n", "\n> ") yield render() else: out.append("\n> _Model still loading — headlines shown; brief will work shortly._") yield render() out.append(f"\n_Done · {stamp}_") yield render() def ticker_safe(t): return str(t).upper() def check_holdings_news(tickers=None) -> str: """Markdown report: AI brief per holding with today-news; quiet list otherwise.""" tickers = tickers if tickers is not None else load_holdings() if not tickers: return ("**No holdings configured.** Add tickers above (e.g. `AAPL, NVDA`) " "and save — they'll be checked for news every day.") blocks, quiet = [], [] for t in tickers: items = fetch_today_news(t) if not items: quiet.append(t) continue lines = [f"### 📰 {t} — {len(items)} item(s) today"] for x in items[:6]: link = f" · [link]({x['link']})" if x["link"] else "" lines.append(f"- **{x['time']}** {x['title']} — *{x['publisher']}*{link}") brief = _llm_brief(t, items) if brief: lines.append(f"\n> 🤖 **AI brief:** {brief}") else: lines.append("\n> _Load a model in the Model tab for an AI brief._") blocks.append("\n".join(lines)) if quiet: blocks.append(f"**Quiet today (no news, ignored):** {', '.join(quiet)}") stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M UTC") return f"_Checked {stamp}_\n\n" + ("\n\n---\n\n".join(blocks) if blocks else "No output.")