Rifqi Hafizuddin
[KM-644] harden query path: grouped-select rule, int-date trends, sample decode, trace guard
74ad526
Raw
History Blame
9.5 kB
"""analyze_trend β€” time-series trend over a period (KM-608).
An analytical "family" tool: in ONE call it buckets rows into time periods
(day/week/month/quarter/year), aggregates a value per period, and summarizes
the movement (first vs last, absolute & percent change, direction, linear
slope). Answers questions like "how did revenue trend month over month?".
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 numpy as np
import pandas as pd
from src.tools.analytics.descriptive import ColumnNotFoundError
# Friendly period name -> pandas resample rule. Using the non-deprecated
# pandas 2.2 codes (ME/QE/YE) avoids FutureWarnings.
FREQ_MAP = {
"day": "D",
"week": "W",
"month": "ME",
"quarter": "QE",
"year": "YE",
}
# How to aggregate the value within each period.
SUPPORTED_AGGS = ("sum", "mean", "count", "min", "max", "median")
class InvalidFrequencyError(ValueError):
"""The requested period is not in FREQ_MAP (maps to error_code INVALID_FREQUENCY)."""
class UnsupportedAggregationError(ValueError):
"""The requested aggregation is not supported (maps to error_code UNSUPPORTED_AGG)."""
class InvalidDateColumnError(ValueError):
"""date_column holds numeric values that aren't a recognizable date/year/month
(maps to error_code INVALID_DATE_COLUMN)."""
def _clean(value: object) -> object:
"""Convert numpy scalars to plain Python; NaN -> None for JSON-clean output."""
if value is None:
return None
if isinstance(value, float) and np.isnan(value):
return None
if hasattr(value, "item"):
value = value.item()
return None if isinstance(value, float) and np.isnan(value) else value
return value
def _parse_date_column(df: pd.DataFrame, date_column: str) -> pd.Series:
"""Parse date_column into datetimes, guarding against numeric epoch misparsing.
pd.to_datetime() treats bare numeric input as epoch-nanoseconds, so bare
month numbers (1-12) or calendar years (e.g. 2025) silently collapse to a
single 1970 timestamp instead of raising. Numeric columns are resolved
explicitly here rather than falling through to pd.to_datetime().
"""
col = df[date_column]
if not pd.api.types.is_numeric_dtype(col):
return pd.to_datetime(col)
non_null = col.dropna()
is_whole = non_null.empty or (non_null == non_null.astype(int)).all()
if is_whole and non_null.between(1, 12).all():
year_col = next((c for c in df.columns if c.lower() == "year"), None)
year_series = df[year_col] if year_col is not None else None
year_non_null = year_series.dropna() if year_series is not None else pd.Series(dtype=float)
year_ok = (
year_series is not None
and pd.api.types.is_numeric_dtype(year_series)
and not year_non_null.empty
and (year_non_null == year_non_null.astype(int)).all()
and year_non_null.between(1900, 2100).all()
)
if not year_ok:
raise InvalidDateColumnError(
f"date_column '{date_column}' holds bare month numbers (1-12) and no "
"'year' column is present in the data β€” retrieve a year column "
"alongside month, or use a real date column."
)
valid = col.notna() & year_series.notna()
result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
result.loc[valid] = pd.to_datetime(
{
"year": year_series.loc[valid].astype(int),
"month": col.loc[valid].astype(int),
"day": 1,
}
)
return result
if is_whole and non_null.between(1900, 2100).all():
result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
valid = col.notna()
result.loc[valid] = pd.to_datetime(
col.loc[valid].astype(int).astype(str), format="%Y"
)
return result
raise InvalidDateColumnError(
f"date_column '{date_column}' is numeric but is not a recognizable date, "
"year, or month column."
)
def _period_label(ts: pd.Timestamp, freq: str) -> str:
"""Human-readable period label keyed off the friendly frequency name."""
if freq == "month":
return str(ts.strftime("%Y-%m"))
if freq == "quarter":
return f"{ts.year}-Q{ts.quarter}"
if freq == "year":
return str(ts.strftime("%Y"))
return str(ts.strftime("%Y-%m-%d")) # day / week
# 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: Time-series trend of one metric over evenly-spaced periods (day, week, \
month, quarter, year). Reports per-period points plus direction, absolute and \
percent change, and a linear slope.
USE WHEN the question is about movement over time β€” growth, decline, trend, \
seasonality. Trigger words: "over time" (dari waktu ke waktu), "trend" (tren), \
"monthly/yearly" (bulanan/tahunan), "growth" (pertumbuhan), "since/last N months".
DON'T USE WHEN:
- it groups by a non-time category -> analyze_aggregate
- it compares two specific groups (A vs B) -> analyze_comparison
- it summarizes a column with no time axis -> analyze_descriptive
Example questions:
- "how did monthly revenue change this year?"
- "show the sales trend over the last 12 months"
- "is the number of signups growing quarter over quarter?"
- "yearly profit from 2019 to 2024"
"""
def analyze_trend(
df: pd.DataFrame,
date_column: str,
value_column: str,
freq: str = "month",
agg: str = "sum",
) -> dict[str, object]:
"""Time-series trend of one value over evenly-spaced periods.
Args:
df: already-materialized data (in the real system the wrapper fetches
this from a source_id).
date_column: column holding dates/timestamps.
value_column: numeric column to aggregate per period.
freq: period granularity β€” one of FREQ_MAP keys (default "month").
agg: how to aggregate within a period β€” one of SUPPORTED_AGGS.
Returns:
dict with:
freq, agg β€” echo of the chosen settings
points β€” [{"period": str, "value": number|None}, ...]
first, last β€” value of the first/last non-empty period
change_abs β€” last - first
change_pct β€” (last - first) / first, or None if first == 0
direction β€” "up" | "down" | "flat"
slope β€” linear slope across periods, or None if < 2 points
Raises:
ColumnNotFoundError: if date_column or value_column is absent.
InvalidFrequencyError: if freq is not a known period.
UnsupportedAggregationError: if agg is not supported.
InvalidDateColumnError: if date_column is numeric but not a recognizable
date, year, or bare month number (needing a companion 'year' column).
"""
missing = [c for c in (date_column, value_column) if c not in df.columns]
if missing:
raise ColumnNotFoundError(f"columns not found: {missing}")
if freq not in FREQ_MAP:
raise InvalidFrequencyError(
f"unknown frequency '{freq}'; supported: {list(FREQ_MAP)}"
)
if agg not in SUPPORTED_AGGS:
raise UnsupportedAggregationError(
f"unsupported aggregation '{agg}'; supported: {list(SUPPORTED_AGGS)}"
)
# Build a clean datetime-indexed series, then resample into periods.
s = df[[date_column, value_column]].copy()
s[date_column] = _parse_date_column(df, date_column)
s = s.dropna(subset=[date_column]).set_index(date_column).sort_index()
resampled = s[value_column].resample(FREQ_MAP[freq]).agg(agg)
points = [
{"period": _period_label(ts, freq), "value": _clean(val)}
for ts, val in resampled.items()
]
# Summary stats are computed over non-empty periods only.
non_null = resampled.dropna()
first: float | None
last: float | None
change_abs: float | None
change_pct: float | None
slope: float | None
if non_null.empty:
first = last = change_abs = change_pct = slope = None
direction = "flat"
else:
first = float(non_null.iloc[0])
last = float(non_null.iloc[-1])
change_abs = last - first
change_pct = (change_abs / first) if first != 0 else None
if change_abs > 0:
direction = "up"
elif change_abs < 0:
direction = "down"
else:
direction = "flat"
if non_null.shape[0] > 1:
x = np.arange(non_null.shape[0])
slope = float(np.polyfit(x, non_null.to_numpy(dtype=float), 1)[0])
else:
slope = None
return {
"freq": freq,
"agg": agg,
"points": points,
"first": first,
"last": last,
"change_abs": change_abs,
"change_pct": change_pct,
"direction": direction,
"slope": slope,
}