Blame for src/tools/analytics/temporal.py

"""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.charts import (
    _LINE_STYLE,
    COLORWAY,
    LAYOUT_PRESET,
    _axis,
    envelope,
)
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

# A period with fewer rows than this share of the median period is a stub: a month
# the data only partly covers. Audit P0-8: a September with 9 rows (other months
# 63-66) set the first/last comparison and the answer said "turun".
_STUB_SHARE = 0.5
# Below this many non-empty periods a median says too little to call one short.
_STUB_MIN_PERIODS = 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 _resample(series: pd.Series, freq: str, agg: str) -> pd.Series:
    """Aggregate a datetime-indexed series per period; a period with no rows stays empty.

    Plain `resample().agg("sum")` turns a month with no rows into 0, which reads as a
    collapse in the trend and is fitted as a real value by the forecast. `count` keeps
    its 0 — zero events in a month is the true count.
    """
    resampler = series.sort_index().resample(FREQ_MAP[freq])
    if agg == "sum":
        return resampler.sum(min_count=1)
    return resampler.agg(agg)


def _stub_periods(counts: pd.Series) -> tuple[set[pd.Timestamp], float | None]:
    """Periods whose row count is below `_STUB_SHARE` of the median non-empty period.

    Returns (stub period timestamps, that median). Empty periods are not stubs — they
    have no value to mislead with.
    """
    filled = counts[counts > 0]
    if filled.shape[0] < _STUB_MIN_PERIODS:
        return set(), None
    median = float(filled.median())
    return set(filled[filled < median * _STUB_SHARE].index), median


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


def _trend_chart(points: list[dict[str, object]], value_column: str) -> dict[str, object]:
    """Line chart of the per-period points; an empty period is a gap, not a zero."""
    trace = {
        "type": "scatter",
        "mode": "lines+markers",
        "x": [p["period"] for p in points],
        "y": [p["value"] for p in points],
        "name": value_column,
        "line": {**_LINE_STYLE, "color": COLORWAY[0]},
        "marker": {"size": 6},
    }
    title = f"Trend of {value_column}"
    layout = {
        **LAYOUT_PRESET,
        "title": {"text": title},
        "showlegend": False,
        "xaxis": _axis("period"),
        "yaxis": _axis(value_column),
    }
    return envelope("line", title, [trace], layout)


# 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, value_column — echo of the chosen settings
            points             — [{"period": str, "value": number|None, "n": rows,
                                   "stub": bool}, ...]; a stub has fewer than half
                                   the median period's rows
            first, last        — value of the first/last non-empty, non-stub 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)
            charts             — a line chart of the points (reserved auto-chart key)

    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 = _resample(s["_value"], freq, agg)
    counts = _resample(s["_value"], freq, "count")
    stubs, median_n = _stub_periods(counts)

    points = [
        {
            "period": _period_label(ts, freq),
            "value": _clean(val),
            "n": int(counts[ts]),
            "stub": ts in stubs,
        }
        for ts, val in resampled.items()
    ]

    # Summary stats are computed over non-empty periods only, and leave stubs out so a
    # thinly covered month cannot set the direction (unless nothing else is left).
    non_null = resampled.dropna()
    full = non_null[~non_null.index.isin(list(stubs))]
    if not full.empty:
        non_null = full
    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 stubs and median_n is not None:
        thin = ", ".join(
            f"{_period_label(ts, freq)} (n={int(counts[ts])})" for ts in sorted(stubs)
        )
        caveats.append(
            f"Partly covered period(s): {thin}, against a median of {median_n:g} rows per "
            "period. Shown in the points, but left out of first/last, direction and slope."
        )
    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,
        "value_column": value_column,
        "points": points,
        "first": first,
        "last": last,
        "change_abs": change_abs,
        "change_pct": change_pct,
        "direction": direction,
        "slope": slope,
        "r_squared": r_squared,
        "caveats": caveats,
        # Reserved auto-chart key (the invoker moves it to meta["charts"]): a trend
        # answer always came without a chart, and a "show it as a chart" follow-up
        # cannot draw one from series output (validator Check 10) — demo test
        # 2026-11-02.
        "charts": [] if non_null.empty else [_trend_chart(points, value_column)],
    }