File size: 12,963 Bytes
81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 49b0848 81e5fe7 49b0848 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 49b0848 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 17e31c6 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 e201ae4 81e5fe7 e201ae4 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 | """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,
}
|