Clari Core Workflow B

Clari Core Workflow: Revenue Analytics

Overview

Build revenue analytics from Clari export data: forecast accuracy tracking, pipeline coverage analysis, rep performance dashboards, and forecast call change detection.

Prerequisites

• Completed clari-core-workflow-a (export pipeline)
• Historical forecast exports for accuracy tracking
• Pandas/SQL for data analysis

Instructions

Step 1: Forecast Accuracy Analysis

import pandas as pd

def calculate_forecast_accuracy(
    forecasts: list[dict], actuals: list[dict]
) -> pd.DataFrame:
    df_forecast = pd.DataFrame(forecasts)
    df_actual = pd.DataFrame(actuals)

    merged = df_forecast.merge(
        df_actual[["ownerEmail", "crmClosed"]],
        on="ownerEmail",
        suffixes=("_forecast", "_actual"),
    )

    merged["accuracy_pct"] = (
        1 - abs(merged["forecastAmount"] - merged["crmClosed_actual"])
        / merged["forecastAmount"]
    ) * 100

    merged["variance"] = merged["crmClosed_actual"] - merged["forecastAmount"]

    return merged[["ownerName", "forecastAmount", "crmClosed_actual",
                    "accuracy_pct", "variance"]].sort_values("accuracy_pct")

Step 2: Pipeline Coverage Report

def pipeline_coverage_report(entries: list[dict]) -> dict:
    df = pd.DataFrame(entries)

    return {
        "total_pipeline": df["crmTotal"].sum(),
        "total_closed": df["crmClosed"].sum(),
        "total_quota": df["quotaAmount"].sum(),
        "total_forecast": df["forecastAmount"].sum(),
        "coverage_ratio": df["crmTotal"].sum() / df["quotaAmount"].sum()
            if df["quotaAmount"].sum() > 0 else 0,
        "close_rate": df["crmClosed"].sum() / df["crmTotal"].sum()
            if df["crmTotal"].sum() > 0 else 0,
        "attainment_pct": df["crmClosed"].sum() / df["quotaAmount"].sum() * 100
            if df["quotaAmount"].sum() > 0 else 0,
        "at_risk_reps": len(df[df["forecastAmount"] < df["quotaAmount"] * 0.7]),
        "on_track_reps": len(df[df["forecastAmount"] >= df["quotaAmount"] * 0.9]),
    }

Step 3: Forecast Change Detection

def detect_forecast_changes(
    current: list[dict], previous: list[dict], threshold_pct: float = 10.0
) -> list[dict]:
    curr = {e["ownerEmail"]: e for e in current}
    prev = {e["ownerEmail"]: e for e in previous}

    changes = []
    for email, curr_entry in curr.items():
        prev_entry = prev.get(email)
        if not prev_entry:
            continue

        prev_amount = prev_entry["forecastAmount"]
        curr_amount = curr_entry["forecastAmount"]

        if prev_amount == 0:
            continue

        change_pct = ((curr_amount - prev_amount) / prev_amount) * 100

        if abs(change_pct) >= threshold_pct:
            changes.append({
                "rep": curr_entry["ownerName"],
                "previous_forecast": prev_amount,
                "current_forecast": curr_amount,
                "change_pct": round(change_pct, 1),
                "direction": "up" if change_pct > 0 else "down",
            })

    return sorted(changes, key=lambda x: abs(x["change_pct"]), reverse=True)

Step 4: SQL Analytics Queries

-- Forecast accuracy by quarter
SELECT
    time_period,
    owner_name,
    forecast_amount,
    crm_closed AS actual_closed,
    ROUND((1 - ABS(forecast_amount - crm_closed) / NULLIF(forecast_amount, 0)) * 100, 1) AS accuracy_pct
FROM clari_forecasts
WHERE time_period = '2025_Q4'
ORDER BY accuracy_pct DESC;

-- Pipeline coverage trend
SELECT
    time_period,
    SUM(crm_total) / NULLIF(SUM(quota_amount), 0) AS coverage_ratio,
    SUM(crm_closed) / NULLIF(SUM(quota_amount), 0) AS attainment
FROM clari_forecasts
GROUP BY time_period
ORDER BY time_period;

Error Handling

Output

Return a time-bounded analytics result with source export timestamp, period, calculation version, aggregate counts, and any suppression applied for small or unauthorized cohorts. Treat forecast accuracy and rep-level variance as sensitive commercial data; distribute detailed views only to authorized roles.

Examples

Compare two certified quarterly exports and flag forecast changes over the approved threshold, while omitting individual values from the shared summary. If the prior period is missing or stale, return an explicit unavailable result and request a fresh workflow-A export rather than inferring a change from incompatible data.

Resources

• Clari API Reference
• Pandas Documentation

Next Steps

For error troubleshooting, see clari-common-errors.