Stage 11: The automated QBR

{
  "goal": "Cut the time to build the weekly ops report",
  "baseline": "18 minutes per report, measured at kickoff",
  "target": "under 5 minutes",
  "current": "2 minutes",
  "status": "met"
}

python -m venv venv
source venv/bin/activate   # on Windows: venv\Scripts\activate
pip install pandas python-pptx pydantic pytest anthropic

"""Make mock accounts and weekly usage. Run once: python make_data.py"""
import numpy as np
import pandas as pd

rng = np.random.default_rng(42)
accounts = pd.DataFrame([
    ("acme", "Acme Logistics", 40, 6000, 4000, 18, 2, 0.30, 45.0),
    ("globex", "Globex Health", 25, 4500, 3000, 25, 3, 0.25, 52.0),
], columns=["account_id", "name", "seats", "license_per_quarter", "onboarding_fee",
            "baseline_min", "tool_min", "displaced_share", "wage_per_hour"])

weeks = pd.date_range("2026-04-06", "2026-09-28", freq="W-MON")  # Q2 and Q3 2026
rows = []
for a in accounts.itertuples():
    ramp = np.linspace(0.45, 0.80, len(weeks))  # adoption grows
    for w, r in zip(weeks, ramp):
        wau = int(a.seats * r + rng.integers(-2, 3))
        rows.append((a.account_id, w.date().isoformat(), wau,
                     int(wau * rng.integers(8, 14))))  # tasks per active user per week
usage = pd.DataFrame(rows, columns=["account_id", "week_start", "weekly_active_users", "tasks_completed"])
accounts.to_csv("accounts.csv", index=False)
usage.to_csv("usage.csv", index=False)
print(len(accounts), "accounts,", len(usage), "usage rows")

"""All numbers in the deck come from this file. The model never computes anything."""
from dataclasses import dataclass
import pandas as pd

# BLS ECEC, private industry, June 2026: $46.89 total per hour, $32.82 of it wages.
LOAD_FACTOR = 46.89 / 32.82   # about 1.43; replace with the customer's own number if known
REALIZATION = 0.80            # haircut: assume only 80% of measured time savings becomes real value


@dataclass
class Fact:
    id: str
    label: str
    value: float
    fmt: str  # "int", "usd", "pct", "hours", "months"

    def show(self) -> str:
        v = self.value
        if v != v:  # NaN, for example payback when the net benefit is not positive
            return "not reached"
        return {"int": f"{v:,.0f}", "usd": f"${v:,.0f}", "pct": f"{v:.0%}",
                "hours": f"{v:,.0f} hours", "months": f"{v:.1f} months"}[self.fmt]


def in_quarter(usage: pd.DataFrame, q: pd.Period) -> pd.DataFrame:
    d = pd.to_datetime(usage["week_start"])
    return usage[(d >= q.start_time) & (d <= q.end_time)]


def compute(account: pd.Series, usage: pd.DataFrame, q: pd.Period) -> dict[str, Fact]:
    u = usage[usage.account_id == account.account_id]
    cur, prev = in_quarter(u, q), in_quarter(u, q - 1)
    tasks, tasks_prev = cur.tasks_completed.sum(), prev.tasks_completed.sum()
    wau = cur.weekly_active_users.mean()
    minutes_saved = account.baseline_min - account.tool_min
    hours = tasks * account.displaced_share * minutes_saved / 60 * REALIZATION
    loaded = account.wage_per_hour * LOAD_FACTOR
    value = hours * loaded
    cost = account.license_per_quarter
    net_monthly = (value - cost) / 3
    payback = account.onboarding_fee / net_monthly if net_monthly > 0 else float("nan")
    f = [
        Fact("tasks", "Tasks completed this quarter", tasks, "int"),
        Fact("tasks_change", "Change in tasks vs last quarter", tasks / tasks_prev - 1, "pct"),
        Fact("utilization", "Average weekly active users / licensed seats", wau / account.seats, "pct"),
        Fact("hours_saved", "Hours saved (after 80% realization haircut)", hours, "hours"),
        Fact("value", "Labor value of hours saved", value, "usd"),
        Fact("cost", "License cost this quarter", cost, "usd"),
        Fact("roi", "ROI: (value - cost) / cost", (value - cost) / cost, "pct"),
        Fact("payback", "Payback on onboarding fee", payback, "months"),
    ]
    return {x.id: x for x in f}

"""The model writes words. Code supplies every number."""
import json
import re
from pydantic import BaseModel
from metrics import Fact

MODEL = "claude-sonnet-5-5"
TOKEN = re.compile(r"\{(\w+)\}")
SYSTEM = (
    "You write the executive summary slide of a customer business review. "
    "You get a list of facts with ids. Refer to a number ONLY by writing its id in braces, "
    "for example {hours_saved}. Never write a digit or a spelled-out number yourself. "
    "Use only the facts given. If the facts do not support a claim, leave it out. "
    "Write one headline sentence and exactly 3 short bullets. Plain words, no hype."
)


class Summary(BaseModel):
    headline: str
    bullets: list[str]


def check(s: Summary, facts: dict[str, Fact]) -> None:
    for text in [s.headline, *s.bullets]:
        unknown = set(TOKEN.findall(text)) - set(facts)
        if unknown:
            raise ValueError(f"unknown fact ids: {unknown}")
        rest = TOKEN.sub("", text)
        if re.search(r"\d", rest):
            raise ValueError(f"model wrote its own digits: {text!r}")
        if "{" in rest or "}" in rest:
            raise ValueError(f"malformed placeholder: {text!r}")
    if len(s.bullets) != 3:
        raise ValueError("need exactly 3 bullets")


def render(s: Summary, facts: dict[str, Fact]) -> Summary:
    def fill(t: str) -> str:
        return TOKEN.sub(lambda m: facts[m.group(1)].show(), t)
    return Summary(headline=fill(s.headline), bullets=[fill(b) for b in s.bullets])


def fallback() -> Summary:
    return Summary(
        headline="{hours_saved} saved this quarter, a {roi} return on the license.",
        bullets=["{tasks} tasks completed, {tasks_change} versus last quarter.",
                 "{utilization} of licensed seats were active in an average week.",
                 "The investment pays back its onboarding fee in {payback}."])


def write_summary(facts: dict[str, Fact], client=None, tries: int = 2) -> Summary:
    if client is not None:
        table = json.dumps([{"id": f.id, "label": f.label, "value": round(float(f.value), 3)}
                            for f in facts.values()])
        for _ in range(tries):
            resp = client.messages.parse(
                model=MODEL, max_tokens=800, system=SYSTEM,
                messages=[{"role": "user", "content": f"Facts:\n{table}"}],
                output_format=Summary)
            try:
                check(resp.parsed_output, facts)
                return render(resp.parsed_output, facts)
            except ValueError as e:
                print("rejected model output:", e)
    return render(fallback(), facts)  # fallback is checked by the tests, not trusted blindly

import pandas as pd
import pytest
import metrics
import narrative
from narrative import Summary


@pytest.fixture
def facts():
    acct = pd.read_csv("accounts.csv").iloc[0]
    return metrics.compute(acct, pd.read_csv("usage.csv"), pd.Period("2026Q3"))


def test_fallback_passes_its_own_check(facts):
    narrative.check(narrative.fallback(), facts)


def test_model_digits_are_rejected(facts):
    bad = Summary(headline="You saved 500 hours.", bullets=["{tasks}", "{roi}", "{cost}"])
    with pytest.raises(ValueError):
        narrative.check(bad, facts)


def test_unknown_fact_id_is_rejected(facts):
    bad = Summary(headline="{made_up} saved.", bullets=["{tasks}", "{roi}", "{cost}"])
    with pytest.raises(ValueError):
        narrative.check(bad, facts)


def test_malformed_placeholder_is_rejected(facts):
    bad = Summary(headline="{hours-saved} saved.", bullets=["{tasks}", "{roi}", "{cost}"])
    with pytest.raises(ValueError):
        narrative.check(bad, facts)

"""Build the branded deck with python-pptx. Charts are native, so the CSM can edit them."""
from pptx import Presentation
from pptx.chart.data import CategoryChartData
from pptx.dml.color import RGBColor
from pptx.enum.chart import XL_CHART_TYPE
from pptx.util import Inches, Pt

BRAND = {"navy": RGBColor(0x1F, 0x2A, 0x44), "teal": RGBColor(0x0E, 0x7C, 0x86),
         "grey": RGBColor(0x6B, 0x72, 0x80), "font": "Calibri"}


def text(slide, x, y, w, h, s, size=18, bold=False, color="navy"):
    tf = slide.shapes.add_textbox(Inches(x), Inches(y), Inches(w), Inches(h)).text_frame
    tf.word_wrap = True
    for i, line in enumerate(s.split("\n")):
        p = tf.paragraphs[0] if i == 0 else tf.add_paragraph()
        p.text = line
        p.font.size, p.font.bold, p.font.name = Pt(size), bold, BRAND["font"]
        p.font.color.rgb = BRAND[color]
    return tf


def new_slide(prs, title, footer, draft):
    s = prs.slides.add_slide(prs.slide_layouts[6])  # layout 6 is blank in the default template
    text(s, 0.6, 0.4, 12.1, 1.2, title, size=28, bold=True)  # the title states the finding
    text(s, 0.6, 6.95, 10, 0.4, footer, size=11, color="grey")
    if draft:
        text(s, 10.5, 6.95, 2.4, 0.4, "DRAFT: CSM review pending", size=11, bold=True, color="teal")
    return s


def build(account_name, quarter_label, weekly, facts, summary, assumptions, out_path, draft=True):
    prs = Presentation()
    prs.slide_width, prs.slide_height = Inches(13.333), Inches(7.5)  # 16:9
    foot = f"{account_name} | {quarter_label} business review | Source: product usage data"

    s = prs.slides.add_slide(prs.slide_layouts[6])
    text(s, 0.8, 2.4, 11.5, 1.2, f"{account_name}: {quarter_label} business review", size=40, bold=True)
    text(s, 0.8, 3.8, 11.5, 0.6, "Value delivered and plan for next quarter", size=20, color="teal")

    s = new_slide(prs, summary.headline, foot, draft)
    text(s, 0.8, 2.0, 11.5, 4, "\n".join(summary.bullets), size=22)

    ch = facts["tasks_change"].value
    s = new_slide(prs, f"Tasks completed {'grew' if ch >= 0 else 'fell'} {abs(ch):.0%} quarter on quarter",
                  foot, draft)
    cd = CategoryChartData(number_format="#,##0")
    cd.categories = list(weekly["week_start"])
    cd.add_series("Tasks completed per week", list(weekly["tasks_completed"]))
    chart = s.shapes.add_chart(XL_CHART_TYPE.LINE, Inches(0.8), Inches(1.9), Inches(11.7),
                               Inches(4.8), cd).chart
    chart.has_legend = False  # one series, so no legend: the title carries the message
    chart.value_axis.has_major_gridlines = False
    chart.series[0].format.line.color.rgb = BRAND["teal"]
    chart.series[0].format.line.width = Pt(3)
    chart.series[0].smooth = False

    s = new_slide(prs, f"{facts['hours_saved'].show()} saved: {facts['value'].show()} of labor value "
                       f"against {facts['cost'].show()} of license", foot, draft)
    keys = ("tasks", "hours_saved", "value", "cost", "roi", "payback")
    rows = [("Measure", "This quarter")] + [(facts[k].label, facts[k].show()) for k in keys]
    t = s.shapes.add_table(len(rows), 2, Inches(0.8), Inches(1.9), Inches(11.7),
                           Inches(0.5 * len(rows))).table
    for i, row in enumerate(rows):
        for j, v in enumerate(row):
            t.cell(i, j).text = v
            t.cell(i, j).text_frame.paragraphs[0].font.size = Pt(18)

    s = new_slide(prs, "How we calculated this, so you can challenge it", foot, draft)
    text(s, 0.8, 1.9, 11.5, 4.5, "\n".join(f"- {a}" for a in assumptions), size=18)

    s = new_slide(prs, "Next quarter: two decisions for you", foot, draft)
    text(s, 0.8, 1.9, 11.5, 4, "[CSM: write the two recommendations and the ask]", size=22, color="grey")
    s.notes_slide.notes_text_frame.text = "Fill in from the last call. Delete this note before sending."

    prs.save(out_path)

"""python build_qbr.py acme 2026Q3   -> out/acme-2026Q3-DRAFT.pptx plus out/review.md"""
import argparse
from pathlib import Path
import pandas as pd
import deck
import metrics
import narrative

p = argparse.ArgumentParser()
p.add_argument("account_id")
p.add_argument("quarter")
p.add_argument("--llm", action="store_true")
a = p.parse_args()

accounts, usage = pd.read_csv("accounts.csv"), pd.read_csv("usage.csv")
acct = accounts[accounts.account_id == a.account_id].iloc[0]
q = pd.Period(a.quarter)
facts = metrics.compute(acct, usage, q)
client = None
if a.llm:
    import anthropic
    client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from the environment
summary = narrative.write_summary(facts, client)
weekly = metrics.in_quarter(usage[usage.account_id == a.account_id], q)
assumptions = [
    f"Minutes per task: {acct.baseline_min} before, {acct.tool_min} with the tool (measured at kickoff).",
    f"{acct.displaced_share:.0%} of tasks replace manual work (sampled with your team).",
    f"Hourly cost: ${acct.wage_per_hour:.2f} wage x {metrics.LOAD_FACTOR:.2f} load (BLS, June 2026).",
    f"We count {metrics.REALIZATION:.0%} of measured savings.",
    "Payback = onboarding fee / (monthly value - monthly license).",
]
Path("out").mkdir(exist_ok=True)
name = f"out/{a.account_id}-{a.quarter}-DRAFT.pptx"
deck.build(acct["name"], f"{q.year} Q{q.quarter}", weekly, facts, summary, assumptions, name)
Path("out/review.md").write_text("# Review checklist\n" + "\n".join(
        f"- [ ] {f.label}: {f.show()}" for f in facts.values()) + "\n- [ ] Next-quarter slide written\n")
print("wrote", name)

"""python finalize.py out/acme-2026Q3-DRAFT.pptx
Run after the CSM edited the draft in PowerPoint. Refuses to ship a draft with open placeholders."""
import sys
from pathlib import Path
from pptx import Presentation

src = Path(sys.argv[1])
prs = Presentation(src)
open_items = []
for n, slide in enumerate(prs.slides, 1):
    for shape in list(slide.shapes):
        if not shape.has_text_frame:
            continue
        t = shape.text_frame.text
        if "[CSM:" in t:
            open_items.append(f"slide {n}: placeholder still present")
        if t.startswith("DRAFT:"):
            shape._element.getparent().remove(shape._element)  # drop the draft stamp
    if slide.has_notes_slide:
        slide.notes_slide.notes_text_frame.text = ""  # internal notes never go to the customer
if "- [ ]" in (src.parent / "review.md").read_text():
    open_items.append("review.md has unchecked boxes")
if open_items:
    raise SystemExit("Not final:\n" + "\n".join(open_items))
final = src.with_name(src.name.replace("-DRAFT", ""))
prs.save(final)
print("wrote", final)

"""Fill a Google Slides template that holds tags like {{hours_saved}} and {{headline}}."""
import google.auth
from googleapiclient.discovery import build

SCOPES = ["https://www.googleapis.com/auth/presentations",
          "https://www.googleapis.com/auth/drive.file"]


def requests_for(facts, summary, chart_url=None):
    tags = {f"{{{{{k}}}}}": f.show() for k, f in facts.items()}
    tags["{{headline}}"] = summary.headline
    tags["{{bullets}}"] = "\n".join(summary.bullets)
    reqs = [{"replaceAllText": {"containsText": {"text": tag, "matchCase": True},
                                "replaceText": value}} for tag, value in tags.items()]
    if chart_url:  # must be a public URL: PNG, JPEG or GIF, under 50 MB and 25 megapixels
        reqs.append({"replaceAllShapesWithImage": {
            "containsText": {"text": "{{chart}}", "matchCase": True},
            "imageUrl": chart_url, "imageReplaceMethod": "CENTER_INSIDE"}})
    return reqs


def make_deck(template_id, title, facts, summary, chart_url=None):
    creds, _ = google.auth.default(scopes=SCOPES)
    drive = build("drive", "v3", credentials=creds)
    slides = build("slides", "v1", credentials=creds)
    copy = drive.files().copy(fileId=template_id, body={"name": title}).execute()
    deck_id = copy["id"]  # never edit the template itself
    slides.presentations().batchUpdate(
        presentationId=deck_id,
        body={"requests": requests_for(facts, summary, chart_url)}).execute()
    return f"https://docs.google.com/presentation/d/{deck_id}/edit"

name: quarterly-qbr-drafts
on:
  schedule:
    - cron: "17 13 3 1,4,7,10 *"   # 13:17 UTC on the 3rd of Jan, Apr, Jul, Oct
  workflow_dispatch:
jobs:
  draft:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v7
        with:
          python-version: "3.13"
      - run: pip install pandas python-pptx pydantic anthropic
      - run: python make_data.py   # mock data only; swap for your real export
      - run: |
          Q=$(python -c "import pandas as pd; print(pd.Period(pd.Timestamp.now(), 'Q') - 1)")
          python build_qbr.py acme "$Q" --llm
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
      - uses: actions/upload-artifact@v7
        with:
          name: qbr-drafts
          path: out/