Earnings Call Transcript Monitor

name: earnings_transcript_monitor
schedule: "0 */4 * * *"
steps:
  - fetch_rss:
      sources:
        - url: "https://seekingalpha.com/earnings/earnings-call-transcripts"
          type: seeking_alpha
        - url: "https://www.fool.com/earnings-call-transcripts/"
          type: motley_fool
      filter_tickers: "{{ watchlist_tickers }}"
  - check_seen:
      dedup_key: transcript_url
      store: seen_transcripts.json
  - extract:
      sections:
        - prepared_remarks
        - qa_section
      phrases_to_flag:
        - "guidance"
        - "raised"
        - "lowered"
        - "headwinds"
        - "macro"
        - "tariff"
        - "pricing power"
  - llm:
      prompt: |
        Analyze this earnings call transcript and produce a structured brief:

        1. GUIDANCE: What guidance was given? Raised, lowered, or maintained?
        2. KEY THEMES: Top 3 topics management emphasized
        3. ANALYST CONCERNS: What did analysts probe hardest in Q&A?
        4. TONE: Rate management confidence 1-5. Note any hedging language.
        5. NOTABLE QUOTES: 2-3 verbatim quotes worth flagging

        Transcript: {{ transcript_text }}
  - notify:
      channel: email
      subject: "📞 Earnings Transcript: {{ company }} Q{{ quarter }} {{ year }}"

import httpx
import feedparser
import json
import os
from datetime import datetime

SEEN_FILE = "seen_transcripts.json"

def load_seen() -> set:
    if os.path.exists(SEEN_FILE):
        with open(SEEN_FILE) as f:
            return set(json.load(f))
    return set()

def save_seen(seen: set):
    with open(SEEN_FILE, "w") as f:
        json.dump(list(seen), f)

def fetch_new_transcripts(rss_url: str, watchlist: list) -> list:
    seen = load_seen()
    feed = feedparser.parse(rss_url)
    new_transcripts = []
    for entry in feed.entries:
        if entry.link in seen:
            continue
        title_upper = entry.title.upper()
        for ticker in watchlist:
            if ticker.upper() in title_upper or "TRANSCRIPT" in title_upper:
                new_transcripts.append({
                    "ticker": ticker,
                    "title": entry.title,
                    "url": entry.link,
                    "published": entry.get("published", ""),
                })
                seen.add(entry.link)
                break
    save_seen(seen)
    return new_transcripts

def extract_guidance_language(text: str) -> dict:
    """Find guidance-related sentences in transcript text."""
    import re
    guidance_patterns = [
        r"(?:we (?:raise|lower|maintain|reiterate|withdraw)|guidance).{0,200}",
        r"(?:full[- ]year|fiscal \d{4}|Q[1-4]).{0,100}(?:expect|guide|anticipate).{0,100}",
        r"(?:EPS|revenue|margin).{0,50}(?:between|of|approximately).{0,100}",
    ]
    results = []
    for pattern in guidance_patterns:
        matches = re.findall(pattern, text, re.IGNORECASE)
        results.extend(matches[:3])
    return {"guidance_sentences": results[:10]}