Rifqi Hafizuddin
Claude Opus 5
[NOTICKET] tools: one schema mechanism and a description on every input
c0040e4 | """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 pydantic import BaseModel, Field | |
| from src.tools.analytics.agg_vocab import normalise_agg | |
| 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") | |
| # Below this R^2, the straight-line trend explains little of the movement — the | |
| # reported direction is then flagged as weakly supported (D-7). | |
| _WEAK_TREND_R2 = 0.3 | |
| 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. | |
| class TrendInput(BaseModel): | |
| """Planner-facing inputs for `analyze_trend` (spec_builder contract).""" | |
| data: str = Field(..., description="Placeholder ${t<id>} of the upstream table (Pattern A).") | |
| date_column: str = Field( | |
| ..., | |
| description="Column holding dates to bucket on. May be bare month numbers (1-12) " | |
| "only if a `year` column was retrieved alongside it.", | |
| ) | |
| value_column: str = Field(..., description="Numeric column to track over time.") | |
| freq: str = Field( | |
| "month", description="Period granularity: day, week, month, quarter, year." | |
| ) | |
| agg: str = Field( | |
| "sum", | |
| description="How to aggregate the value within each period (sum, mean, count, " | |
| "min, max, median; `avg` accepted as `mean`).", | |
| ) | |
| 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_ab_test | |
| - 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" (endpoints: last vs first) | |
| slope — linear slope across periods, or None if < 2 points | |
| r_squared — R^2 of that linear fit, or None (D-7) | |
| caveats — notes when direction and slope disagree, or the | |
| trend is weak (D-7) | |
| 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)}" | |
| ) | |
| # `avg` is the QueryIR's word for this and planner.md teaches it; the tools | |
| # speak pandas. Accept both, and report whatever the caller actually wrote. | |
| # (A23 Bug 2 -- see agg_vocab.) | |
| requested, agg = agg, normalise_agg(agg) | |
| if agg not in SUPPORTED_AGGS: | |
| raise UnsupportedAggregationError( | |
| f"unsupported aggregation '{requested}'; supported: {list(SUPPORTED_AGGS)}" | |
| ) | |
| # Build a clean datetime-indexed series, then resample into periods. | |
| # date_column may equal value_column — e.g. "count of events per month", | |
| # where the value being aggregated IS the date itself. Selecting | |
| # df[[col, col]] then yields a duplicate-named 2-col frame whose set_index | |
| # key is 2-D ("Index data must be 1-dimensional"), so build the date and | |
| # value series positionally instead of via column selection. | |
| dates = pd.Series(_parse_date_column(df, date_column).to_numpy(), name="_date") | |
| values = pd.Series(df[value_column].to_numpy(), name="_value") | |
| s = pd.concat([dates, values], axis=1).dropna(subset=["_date"]).set_index("_date") | |
| resampled = s["_value"].sort_index().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 | |
| r_squared: float | None | |
| if non_null.empty: | |
| first = last = change_abs = change_pct = slope = r_squared = 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]) | |
| y = non_null.to_numpy(dtype=float) | |
| slope = float(np.polyfit(x, y, 1)[0]) | |
| # R^2 of the straight-line fit = squared Pearson r; undefined for a constant | |
| # series (zero variance), so guard rather than divide by zero. | |
| if float(np.std(y)) > 0: | |
| r = np.corrcoef(x, y)[0, 1] | |
| r_squared = None if np.isnan(r) else round(float(r * r), 4) | |
| else: | |
| r_squared = None | |
| else: | |
| slope = None | |
| r_squared = None | |
| # D-7: `direction` is endpoints-only (last vs first); `slope` is OLS over all points. | |
| # On a noisy series they can disagree, and neither is wrong — so say so rather than | |
| # letting the assembler assert one. R^2 also lets it qualify a weak trend. | |
| caveats: list[str] = [] | |
| if slope is not None and change_abs is not None and change_abs != 0 and slope != 0: | |
| if (change_abs > 0) != (slope > 0): | |
| caveats.append( | |
| f"Mixed signal: first-vs-last goes {direction}, but the overall trend " | |
| f"(slope {slope:+.4g}) points the other way — the series is noisy, so " | |
| f"{direction!r} reflects only the endpoints, not a consistent move." | |
| ) | |
| if r_squared is not None and r_squared < _WEAK_TREND_R2: | |
| caveats.append( | |
| f"The straight-line trend is weak (R^2 = {r_squared:.2f}) — the direction is " | |
| "not well supported by a consistent slope." | |
| ) | |
| return { | |
| "freq": freq, | |
| "agg": agg, | |
| "points": points, | |
| "first": first, | |
| "last": last, | |
| "change_abs": change_abs, | |
| "change_pct": change_pct, | |
| "direction": direction, | |
| "slope": slope, | |
| "r_squared": r_squared, | |
| "caveats": caveats, | |
| } | |