| """analyze_merge — combine TWO upstream tables on shared keys (KM-608). |
| |
| The only analytics "family" tool with a SECOND data input. In ONE call it joins |
| two already-materialized tables (`data` = LEFT, `data_right` = RIGHT) on one or |
| more shared key columns and returns the combined rows. This is what unlocks the |
| "which X has BOTH the worst A and the biggest B" question shape: A and B come |
| from two separate `retrieve_data` pulls (e.g. PA-by-section and backlog-by-section) |
| that must be aligned per X before either can be judged against the other. Without |
| a two-input combine the run dies with ColumnNotFoundError because no single tool |
| ever sees both metrics. |
| |
| Pattern A, extended: it takes TWO `"${t<id>}"` placeholders. The invoker |
| materializes BOTH into DataFrames before calling this function (no self-fetch); |
| the `on` key(s) reference the column aliases the upstream queries produced. |
| |
| STATUS: compute layer only — takes two already-materialized DataFrames. The |
| wrapper layer (the ToolOutput envelope, dual-arg materialization, ToolSpec |
| registration) lives in src/tools/invoker.py + registry.py. Keeping compute |
| separate from data-fetching keeps this easy to unit-test and stable when wrapped. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import pandas as pd |
|
|
| from src.tools.analytics.descriptive import ColumnNotFoundError |
|
|
| |
| |
| |
| SUPPORTED_HOWS = ("inner", "left", "right", "outer") |
|
|
|
|
| class UnsupportedJoinError(ValueError): |
| """Requested join type is not in SUPPORTED_HOWS (maps to error_code UNSUPPORTED_JOIN).""" |
|
|
|
|
| def _clean(value: object) -> object: |
| """Coerce a scalar to a JSON-clean Python value. |
| |
| An outer/left/right join introduces `NaN` for non-matching rows, and numpy / |
| pandas scalars (numpy.int64, pandas.Timestamp) are not JSON-serializable — |
| normalise all three so the returned rows are clean. |
| """ |
| if isinstance(value, pd.Timestamp): |
| return value.isoformat() |
| if value is None: |
| return None |
| try: |
| if pd.isna(value): |
| return None |
| except (TypeError, ValueError): |
| pass |
| if hasattr(value, "item"): |
| return value.item() |
| return value |
|
|
|
|
| |
| DESCRIPTION = """\ |
| Summary: Combine TWO upstream tables into one by joining on shared key column(s) \ |
| (a pandas merge). `data` is the LEFT table, `data_right` is the RIGHT table; `on` \ |
| is the shared column alias(es) present in BOTH. Returns the combined rows, one \ |
| per matched key (join type controlled by `how`, default inner). |
| |
| USE WHEN a question needs TWO different metrics per the SAME entity and those \ |
| metrics come from two separate pulls — the tell-tale shape is "which X has BOTH \ |
| A and B" (e.g. "which section has the worst PA AND the biggest backlog", "top \ |
| customers by revenue that also have the most complaints"). Plan it as two \ |
| retrieve_data tasks (one per metric, each keyed by X), then analyze_merge on X. |
| |
| SETTING KEYS: `on` must be column alias(es) that exist in BOTH tables (the entity \ |
| you align on, e.g. section_id). Use `suffixes` (default ["_left","_right"]) to \ |
| disambiguate non-key columns that share a name across the two tables. `how`: \ |
| inner (only matched keys), left/right (keep one side), outer (keep all). |
| |
| DON'T USE WHEN: |
| - both metrics can be pulled in ONE retrieve_data query -> just retrieve_data |
| - it groups/aggregates a single table -> analyze_aggregate |
| |
| Example questions: |
| - "which section has the worst PA and the biggest maintenance backlog" |
| - "regions in the top 10 for sales that are also bottom 10 for margin" |
| - "products low on stock that also have high demand" |
| """ |
|
|
|
|
| def analyze_merge( |
| df: pd.DataFrame, |
| data_right: pd.DataFrame, |
| on: list[str] | str, |
| how: str = "inner", |
| suffixes: tuple[str, str] | list[str] = ("_left", "_right"), |
| ) -> list[dict[str, object]]: |
| """Join two already-materialized tables on shared key column(s). |
| |
| Args: |
| df: LEFT table (in the real system the invoker materializes this from the |
| `data` placeholder). |
| data_right: RIGHT table (materialized from the `data_right` placeholder). |
| on: shared key column alias(es) present in BOTH tables. A bare string is |
| treated as a single key. |
| how: join type — one of SUPPORTED_HOWS (default "inner"). |
| suffixes: 2-element (left, right) suffixes applied to non-key columns that |
| collide by name across the two tables. |
| |
| Returns: |
| list[dict]: one row per merged record, values JSON-clean (NaN -> None). |
| |
| Raises: |
| ColumnNotFoundError: if `on` is empty or a key is absent from either side. |
| UnsupportedJoinError: if `how` is not supported. |
| ValueError: if `suffixes` is not a 2-element sequence. |
| """ |
| keys = [on] if isinstance(on, str) else list(on) |
| if not keys: |
| raise ColumnNotFoundError("merge 'on' must name at least one shared key column") |
| if how not in SUPPORTED_HOWS: |
| raise UnsupportedJoinError( |
| f"unsupported join '{how}'; supported: {list(SUPPORTED_HOWS)}" |
| ) |
|
|
| missing_left = [c for c in keys if c not in df.columns] |
| missing_right = [c for c in keys if c not in data_right.columns] |
| if missing_left or missing_right: |
| raise ColumnNotFoundError( |
| f"join key(s) not found — left missing {missing_left}, " |
| f"right missing {missing_right}" |
| ) |
|
|
| suf = tuple(suffixes) |
| if len(suf) != 2: |
| raise ValueError(f"suffixes must be a 2-element (left, right) sequence, got {suffixes!r}") |
|
|
| merged = df.merge(data_right, on=keys, how=how, suffixes=suf) |
| return [{k: _clean(v) for k, v in rec.items()} for rec in merged.to_dict("records")] |
|
|
