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: (x[]), reverse=)
Step 4: SQL Analytics Queries
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;
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 |
Resources
Next Steps
For error troubleshooting, see clari-common-errors.
