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

Error	Cause	Solution
Division by zero	Zero quota or forecast	Add `NULLIF` guards
Missing previous period	First export run	Skip change detection
Accuracy > 100%	Overachievement	Cap at 100% or allow for analysis
Stale data	Export not refreshed	Run `clari-core-workflow-a` first

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

Next Steps

For error troubleshooting, see clari-common-errors.