File size: 5,225 Bytes
81e5fe7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """analyze_contribution — share-of-total per category (KM-608).
An analytical "family" tool: in ONE call it breaks a total down into each
category's contribution (absolute value, share of total, and running
cumulative share), sorted largest-first. This is a single snapshot — "who
makes up the total right now" — as opposed to analyze_comparison (two groups)
or analyze_trend (movement over time). Answers questions like "which region
drives most of revenue?" and supports Pareto (80/20) reasoning.
STATUS: compute layer only — the function takes an already-materialized
DataFrame. The wrapper layer (fetching data from the catalog via source_id,
the ToolOutput envelope, ToolSpec registration) is added once the Planner
seam (KM-418) is settled. Keeping compute separate from data-fetching makes
this function easy to unit-test in isolation and stable when wrapped.
"""
from __future__ import annotations
import pandas as pd
from src.tools.analytics.descriptive import ColumnNotFoundError
# Share-of-total is most meaningful for additive aggregates (sum/count), but
# mean/min/max are allowed; "total" is then the sum of the group aggregates.
SUPPORTED_AGGS = ("sum", "mean", "count", "min", "max", "median")
class UnsupportedAggregationError(ValueError):
"""The requested aggregation is not supported (maps to error_code UNSUPPORTED_AGG)."""
def _clean(value: object) -> object:
"""Convert numpy scalars to plain Python so the output is JSON-clean."""
if hasattr(value, "item"):
return value.item()
return value
# Prompt-style description read by the Planner to decide WHEN to pick this tool.
# Final destination is ToolSpec.description once the wrapper layer is built.
DESCRIPTION = """\
Summary: Breaks a single total into per-category contributions (share of total) \
in one snapshot, largest first, with cumulative share. Supports Pareto (80/20) \
reasoning. Optional top_n folds the long tail into "Others".
USE WHEN the question is about how a whole splits into parts, or which \
categories dominate. Trigger words: "contribution" (kontribusi), "share" \
(porsi/proporsi), "% of total" (persen dari total), "Pareto/80-20", "top \
contributors", "which ... make up most".
DON'T USE WHEN:
- it pits two specific groups against each other -> analyze_comparison
- it tracks change over time -> analyze_trend
- it just aggregates per group without share-of-total -> analyze_aggregate
Example questions:
- "which products contribute most to total sales?"
- "what share of revenue comes from each region?"
- "top 5 customers by contribution to profit"
- "do 20% of items make up 80% of revenue?"
"""
def analyze_contribution(
df: pd.DataFrame,
dimension: str,
value_column: str,
agg: str = "sum",
top_n: int | None = None,
) -> dict[str, object]:
"""Contribution (share of total) of each category, largest first.
Args:
df: already-materialized data (in the real system the wrapper fetches
this from a source_id).
dimension: categorical column to break the total down by.
value_column: numeric column to aggregate per category.
agg: how to aggregate within each category — one of SUPPORTED_AGGS.
top_n: if set, keep the top N categories and lump the remainder into a
single "Others" row.
Returns:
dict with:
dimension, value_column, agg — echo of the chosen settings
total — sum of all category aggregates
items — [{"category", "value", "share",
"cumulative_share"}] largest first
Raises:
ColumnNotFoundError: if dimension or value_column is absent.
UnsupportedAggregationError: if agg is not supported.
"""
missing = [c for c in (dimension, value_column) if c not in df.columns]
if missing:
raise ColumnNotFoundError(f"columns not found: {missing}")
if agg not in SUPPORTED_AGGS:
raise UnsupportedAggregationError(
f"unsupported aggregation '{agg}'; supported: {list(SUPPORTED_AGGS)}"
)
grouped = df.groupby(dimension, dropna=False)[value_column].agg(agg)
grouped = grouped.sort_values(ascending=False)
pairs = list(grouped.items())
# Optionally collapse the long tail into a single "Others" bucket.
if top_n is not None and len(pairs) > top_n:
head = pairs[:top_n]
others_value = sum(val for _, val in pairs[top_n:])
head.append(("Others", others_value))
pairs = head
total = sum(val for _, val in pairs)
items = []
cumulative = 0.0
for cat, val in pairs:
share = (val / total) if total else None
if share is not None:
cumulative += share
items.append(
{
"category": _clean(cat),
"value": _clean(val),
"share": share,
"cumulative_share": cumulative if total else None,
}
)
return {
"dimension": dimension,
"value_column": value_column,
"agg": agg,
"total": _clean(total),
"items": items,
}
|