File size: 9,938 Bytes
81e5fe7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
49b0848
 
 
 
 
81e5fe7
 
 
 
 
 
 
 
 
 
 
 
49b0848
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
81e5fe7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
49b0848
 
81e5fe7
 
 
 
 
 
 
 
 
 
 
 
 
 
17e31c6
 
 
 
 
 
 
 
 
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
"""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.
    # 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
    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,
    }