Rifqi Hafizuddin
[KM-644] harden query path: grouped-select rule, int-date trends, sample decode, trace guard
74ad526 | """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, | |
| } | |