← Files VeraARCHIVED FILE
modules/variance-analysis/scripts/variance_core.py
210 KB · Oct 2, 2026 · 00:29 UTC
"""Deterministic variance analysis helpers for the Codex plugin."""
from __future__ import annotations
import argparse
import calendar
import itertools
import json
import logging
import re
import shutil
import sys
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Iterable, Sequence
import polars as pl
from accounting_controls import (
default_accounting_review,
evaluate_accounting_readiness,
normalize_accounting_review,
)
from ibcs_titles import build_ibcs_title
from legacy_adapter import (
cleanup_legacy_imports,
legacy_date_period_context,
run_legacy_variable_dimension_bridge,
run_legacy_variable_dimension_component_bridge,
run_legacy_variance,
)
from legacy_plotting import write_pvm_decomposition_ladder_png, write_waterfall_png
SCRIPT_DIR = Path(__file__).resolve().parent
def _ensure_local_review_session_import() -> None:
"""Use this plugin's review-session module in multi-plugin test runs."""
script_dir = str(SCRIPT_DIR)
if script_dir in sys.path:
sys.path.remove(script_dir)
sys.path.insert(0, script_dir)
module = sys.modules.get("review_session")
module_file = getattr(module, "__file__", None) if module is not None else None
if module_file and Path(module_file).resolve().is_relative_to(SCRIPT_DIR.resolve()):
return
if module is not None:
del sys.modules["review_session"]
_ensure_local_review_session_import()
from exploded_variance_bridge_chart import (
DEFAULT_CHILD_TOP_N as EXPLODED_BRIDGE_DEFAULT_CHILD_TOP_N,
)
from exploded_variance_bridge_chart import (
DEFAULT_MAX_DRILLDOWNS as EXPLODED_BRIDGE_DEFAULT_MAX_DRILLDOWNS,
)
from exploded_variance_bridge_chart import (
write_exploded_variance_bridge_artifacts,
)
from model_use import write_model_use_manifest
from review_session import write_review_session_artifacts, write_run_intake
from root_cause_bridge_chart import write_root_cause_bridge_png
from root_cause_client_report import write_root_cause_client_report
from total_by_dimension_bridge_chart import (
DEFAULT_TOP_N as TOTAL_BY_DIMENSION_DEFAULT_TOP_N,
)
from total_by_dimension_bridge_chart import (
write_total_by_dimension_bridge_artifacts,
)
PLUGIN_ROOT = Path(__file__).resolve().parents[1]
VENDOR_ROOT = PLUGIN_ROOT / "vendor"
REPO_ROOT = Path(__file__).resolve().parents[3]
SHARED_VENDOR_ROOT = REPO_ROOT / "plugins" / "_shared" / "vendor"
def is_vera_managed_host(plugin_root: Path | None = None) -> bool:
"""Return whether this component is embedded in a Vera package.
Host detection is deterministic because it enforces a package-layout and
package-manifest security contract: an embedded Vera component must never
silently fall back to the standalone execution boundary.
"""
root = (plugin_root or PLUGIN_ROOT).resolve()
layout_is_vera = (
root.name == "variance-analysis"
and root.parent.name == "modules"
and root.parent.parent.name == "vera"
)
host_manifest = root.parent.parent / ".codex-plugin" / "plugin.json"
manifest_is_vera = False
if host_manifest.is_file():
try:
manifest = json.loads(host_manifest.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
manifest = {}
manifest_is_vera = manifest.get("name") == "vera"
return layout_is_vera or manifest_is_vera
def _ensure_shared_modules_path() -> None:
"""Make shared plugin harness modules importable in repo and ZIP runs."""
shared_parent = SHARED_VENDOR_ROOT if SHARED_VENDOR_ROOT.exists() else VENDOR_ROOT
shared_text = str(shared_parent)
if shared_text in sys.path:
sys.path.remove(shared_text)
sys.path.insert(0, shared_text)
module_root = (shared_parent / "modules").resolve()
for name, module in list(sys.modules.items()):
if name == "modules" or name.startswith("modules."):
module_file = getattr(module, "__file__", None)
if not module_file or not Path(module_file).resolve().is_relative_to(
module_root
):
del sys.modules[name]
_ensure_shared_modules_path()
from modules.chart_harness import ( # noqa: E402 # isort: skip
apply_recipe_filters,
apply_recipe_cohorts,
available_analysis_context,
default_scenario_comparison_pair,
PERIOD_GRAIN_MONTH,
PERIOD_GRAIN_QUARTER,
PERIOD_GRAIN_WEEK,
PERIOD_GRAIN_YEAR,
PERIOD_TYPE_CUSTOM,
PERIOD_TYPE_FISCAL,
PERIOD_TYPE_ROLLING,
PERIOD_TYPE_TO_DATE,
period_contract_options,
preserve_recipe_cohorts,
preserve_recipe_filters,
recipe_cohort_dimension_names,
recipe_cohort_source_dimensions,
scenario_column_kind,
write_prepared_data_manifest,
)
__all__ = [
"InspectionResult",
"VarianceRunResult",
"add_common_args",
"configure_logging",
"inspect_variance_inputs",
"is_vera_managed_host",
"run_variance_analysis",
]
LOGGER = logging.getLogger(__name__)
LEGACY_RENDER_ERRORS = (
AttributeError,
IndexError,
ImportError,
KeyError,
ModuleNotFoundError,
NameError,
OSError,
RuntimeError,
TypeError,
ValueError,
pl.exceptions.PolarsError,
)
SCHEMA_VERSION = "1.0"
ARTIFACT_MODE_DATA_ONLY = "data_only"
ARTIFACT_MODE_DATA_AND_RENDER = "data_and_render"
ARTIFACT_MODES = {
ARTIFACT_MODE_DATA_ONLY,
ARTIFACT_MODE_DATA_AND_RENDER,
}
VARIANCE_CHART_STANDARD_WATERFALL = "standard_variance_waterfall"
VARIANCE_CHART_PVM_LADDER = "pvm_decomposition_ladder"
VARIANCE_CHART_SMALL_MULTIPLES = "standard_variance_small_multiples"
VARIANCE_CHART_TOTAL_BY_DIMENSION = "total_by_dimension_bridge"
VARIANCE_CHART_EXPLODED_BRIDGE = "exploded_variance_bridge"
VARIANCE_CHART_ROOT_CAUSE = "root_cause_total_bridge"
TOTAL_DIMENSION = "__total"
ROOT_CAUSE_BRIDGE_MEASURE_COLUMNS = {
"bridge_level",
"bridge_dimensions",
"variance_type",
"variance_amount",
"amount_baseline",
"amount_comparison",
"units_baseline",
"units_comparison",
"bridge_unique_value_weight",
}
ROOT_CAUSE_AUTO_DRILLDOWN_VALUES = {
"none",
"single_row",
"dominant_row",
"all_selected",
}
COMPARISON_BASIS_SCENARIO = "scenario"
COMPARISON_BASIS_PERIOD = "period"
COMPARISON_BASIS_VALUES = {COMPARISON_BASIS_SCENARIO, COMPARISON_BASIS_PERIOD}
PERIOD_MODE_NOT_APPLICABLE = "not_applicable"
PERIOD_MODE_CALENDAR = "calendar_period"
PERIOD_MODE_YEAR_TO_DATE = "year_to_date"
PERIOD_MODE_ROLLING = "rolling_period"
PERIOD_MODE_CUSTOM = "custom"
PERIOD_COMPARISON_MODE_VALUES = {
PERIOD_MODE_NOT_APPLICABLE,
PERIOD_MODE_CALENDAR,
PERIOD_MODE_YEAR_TO_DATE,
PERIOD_MODE_ROLLING,
PERIOD_MODE_CUSTOM,
}
SYNTHETIC_PERIOD_COLUMN = "__variance_period_bucket"
DATE_WORK_COLUMN = "__variance_date"
ROLLING_COMPARISON_PRIOR_YEAR = "prior_year"
ROLLING_COMPARISON_PREVIOUS_WINDOW = "previous_window"
ROLLING_COMPARISON_VALUES = {
ROLLING_COMPARISON_PRIOR_YEAR,
ROLLING_COMPARISON_PREVIOUS_WINDOW,
}
DATE_COLUMN_HINTS = (
"date",
"order date",
"orderdate",
"sales date",
"invoice date",
"transaction date",
"posting date",
)
DIMENSION_NAME_PRIORITY = (
"category",
"subcategory",
"productline",
"product line",
"region",
"country",
"group",
"segment",
"channel",
"brand",
"product",
"productname",
"product name",
"customer segment",
)
SMALL_MULTIPLES_DIMENSION_NAME_PRIORITY = (
"productline",
"product line",
"category",
"subcategory",
"region",
"country",
"market",
"territory",
"group",
"segment",
"customer segment",
"channel",
"brand",
"product",
"customer",
"productname",
"product name",
)
SMALL_MULTIPLES_GRANULAR_NAME_HINTS = (
"sku",
"item",
"productname",
"product name",
"customer",
"customer name",
"account",
"id",
"code",
"number",
)
SMALL_MULTIPLES_MAX_REASONABLE_CARDINALITY = 50
SMALL_MULTIPLES_IDEAL_CARDINALITY = 12
SMALL_MULTIPLES_MIN_DIMENSION_SCORE = 30.0
SMALL_MULTIPLES_PANEL_LIMIT = 12
SMALL_MULTIPLES_NULL_LABEL = "N/A"
SMALL_MULTIPLES_OTHER_LABEL = "Others aggregated"
SMALL_MULTIPLES_WORK_COLUMN = "__small_multiples_dimension_value"
TOTAL_BY_DIMENSION_MIN_SCORE = 10.0
TOTAL_BY_DIMENSION_MAX_REASONABLE_CARDINALITY = 250
TOTAL_BY_DIMENSION_NAME_PRIORITY = (
"company",
"brand",
"region",
"market",
"country",
"channel",
"customer segment",
"segment",
"category",
"subcategory",
"productline",
"product line",
"retailer",
)
STANDARD_VARIANCE_COMPONENT_COLUMNS = (
("Price", "price_variance"),
("volume_or_units", "volume_variance"),
("Mix", "mix_variance"),
)
TOLERANCE = 1e-9
CALCULATION_GRAIN_NAME_PRIORITY = (
"sku",
"item",
"product",
"productname",
"product name",
"customer",
"customer name",
"account",
)
COLUMN_HINTS: dict[str, tuple[str, ...]] = {
"period_column": (
"period",
"month",
"year",
"fiscal period",
"date",
"time",
),
"amount_column": (
"sales",
"revenue",
"turnover",
"amount",
"net sales",
"value",
),
"units_column": (
"units",
"unit",
"qty",
"quantity",
"volume",
"pieces",
),
"discount_column": (
"discount",
"rebate",
"promo",
"markdown",
),
"cogs_column": (
"cogs",
"cost of goods",
"cost",
"costs",
"unit cost",
),
}
@dataclass(frozen=True)
class InspectionResult:
"""Inspection payload and suggested recipe."""
payload: dict[str, Any]
recipe: dict[str, Any]
@dataclass(frozen=True)
class VarianceRunResult:
"""Variance output payloads returned after a deterministic run."""
frame: pl.DataFrame
audit: dict[str, Any]
summary_markdown: str
artifact_paths: list[str]
def configure_logging(verbose: bool = False) -> None:
"""Configure CLI logging."""
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(level=level, format="%(levelname)s: %(message)s")
def add_common_args(parser: argparse.ArgumentParser) -> None:
"""Add common plugin CLI arguments."""
parser.add_argument(
"--language",
default="en",
choices=["it", "en", "fr", "de", "es"],
help="Working/output language for Codex-facing summaries.",
)
parser.add_argument(
"--verbose",
action="store_true",
help="Show debug logging.",
)
parser.add_argument(
"--artifact-mode",
choices=sorted(ARTIFACT_MODES),
default=ARTIFACT_MODE_DATA_AND_RENDER,
help=("Write chart data only or keep the legacy data-and-render behavior."),
)
def utc_now() -> str:
"""Return an ISO timestamp for audit outputs."""
return datetime.now(timezone.utc).isoformat()
def get_schema_and_column_names(df: pl.DataFrame) -> tuple[list[str], dict[str, str]]:
"""Return column names and JSON-safe schema."""
return list(df.schema.keys()), {
name: str(dtype) for name, dtype in df.schema.items()
}
def _collect_csv_scan(path: Path, *, separator: str) -> pl.DataFrame:
"""Read delimited input through a lazy scan and collect once."""
lf = pl.scan_csv(path, separator=separator, infer_schema_length=10000)
try:
return lf.collect(engine="streaming")
except pl.exceptions.PolarsError:
return lf.collect()
def read_table(path: Path) -> pl.DataFrame:
"""Read a supported CSV or Excel file."""
suffix = path.suffix.lower()
if suffix in {".csv", ".txt", ".tsv", ".psv"}:
separator = {
".tsv": "\t",
".psv": "|",
}.get(suffix, ",")
return _collect_csv_scan(path, separator=separator)
if suffix in {".xlsx", ".xlsm"}:
return pl.read_excel(path)
raise ValueError(
f"Unsupported input file type '{suffix}'. Use CSV, TSV, PSV, XLSX, or XLSM."
)
def write_json(path: Path, payload: dict[str, Any]) -> None:
"""Write a stable JSON file."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(json_safe(payload), ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def warn_if_output_dir_has_existing_files(output_dir: Path, context: str) -> None:
"""Log a warning when a run may mix with stale output files."""
if not output_dir.exists() or not output_dir.is_dir():
return
existing_files = sorted(
path.name for path in output_dir.iterdir() if path.is_file()
)
if not existing_files:
return
visible_files = ", ".join(existing_files[:8])
if len(existing_files) > 8:
visible_files = f"{visible_files}, ..."
LOGGER.warning(
"%s output directory already contains files; this run may overwrite "
"matching artifacts and leave unrelated stale artifacts in place: %s "
"(existing files: %s)",
context,
output_dir,
visible_files,
)
def json_safe(value: Any) -> Any:
"""Return a JSON-serializable representation of common analysis values."""
if isinstance(value, dict):
return {str(key): json_safe(item) for key, item in value.items()}
if isinstance(value, list):
return [json_safe(item) for item in value]
if isinstance(value, tuple):
return [json_safe(item) for item in value]
if isinstance(value, (datetime, date, Path)):
return str(value)
return value
def load_json(path: Path) -> dict[str, Any]:
"""Load a JSON object from disk."""
payload = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise ValueError(f"Expected JSON object in recipe: {path}")
return payload
def normalize_name(name: str) -> str:
"""Normalize a column name for deterministic matching."""
return " ".join(name.replace("_", " ").replace("-", " ").lower().split())
def compact_normalized_name(name: str) -> str:
"""Return a compact normalized name for underscore/spacing-insensitive matching."""
return "".join(token for token in normalize_name(name).split())
def first_matching_column(columns: Iterable[str], hints: Iterable[str]) -> str | None:
"""Return the first column whose normalized name matches the hints."""
normalized_columns = [(column, normalize_name(column)) for column in columns]
normalized_hints = [normalize_name(hint) for hint in hints]
for hint in normalized_hints:
for column, normalized in normalized_columns:
if normalized == hint:
return column
for hint in normalized_hints:
for column, normalized in normalized_columns:
if hint in normalized:
return column
return None
def is_numeric_dtype(dtype: pl.DataType) -> bool:
"""Return whether a Polars dtype is numeric enough for variance math."""
return dtype.is_numeric()
def is_temporal_dtype(dtype: pl.DataType) -> bool:
"""Return whether a Polars dtype is temporal."""
return dtype.is_temporal()
def suggested_date_column(df: pl.DataFrame, columns: list[str]) -> str | None:
"""Return the best date column for rolling/YTD bucket preparation."""
hinted = first_matching_column(columns, DATE_COLUMN_HINTS)
if hinted and is_temporal_dtype(df.schema[hinted]):
return hinted
temporal_columns = [
column for column in columns if is_temporal_dtype(df.schema[column])
]
if hinted:
return hinted
return temporal_columns[0] if temporal_columns else None
def distinct_text_values(df: pl.DataFrame, column: str) -> list[str]:
"""Return distinct non-empty text values preserving source labels."""
values = (
df.select(pl.col(column).cast(pl.Utf8).drop_nulls().unique().sort())
.to_series(0)
.to_list()
)
return [str(value).strip() for value in values if str(value).strip()]
def is_scenario_comparison_column(df: pl.DataFrame, column: str | None) -> bool:
"""Return whether a column contains an AC-vs-plan-like scenario comparison.
This deterministic rule is limited to exact IBCS scenario roles: Actual can
be compared with Plan, Forecast, or Budget. AC/PY alone remains a period
comparison, not a scenario comparison.
"""
if not column or column not in df.schema:
return False
values = distinct_text_values(df, column)
if scenario_column_kind(column, values) != COMPARISON_BASIS_SCENARIO:
return False
baseline, comparison = default_scenario_comparison_pair(values)
return bool(baseline and comparison)
def is_plan_actual_scenario_column(df: pl.DataFrame, column: str | None) -> bool:
"""Return whether a column is a supported Actual-vs-scenario code column."""
return is_scenario_comparison_column(df, column)
def plan_actual_scenario_column(df: pl.DataFrame, columns: Iterable[str]) -> str | None:
"""Return the AC-vs-plan/forecast/budget scenario column when present."""
for column in columns:
if is_plan_actual_scenario_column(df, column):
return column
return None
def ranked_dimension_candidates(candidates: list[str]) -> list[str]:
"""Return dimension candidates with known business dimensions first."""
selected: list[str] = []
normalized = [(column, normalize_name(column)) for column in candidates]
for hint in DIMENSION_NAME_PRIORITY:
normalized_hint = normalize_name(hint)
for column, normalized_name in normalized:
if normalized_name == normalized_hint and column not in selected:
selected.append(column)
selected.extend(column for column in candidates if column not in selected)
return selected
def dimension_priority_score(column: str, priority: tuple[str, ...]) -> float:
"""Return a deterministic business-readability score for a dimension name."""
normalized_name = normalize_name(column)
compact_name = compact_normalized_name(column)
for index, hint in enumerate(priority):
normalized_hint = normalize_name(hint)
compact_hint = compact_normalized_name(hint)
base_score = max(12.0, 55.0 - (index * 3.0))
if normalized_name == normalized_hint or compact_name == compact_hint:
return base_score
if compact_hint in compact_name or compact_name in compact_hint:
return max(10.0, base_score - 8.0)
return 8.0
def name_contains_hint(column: str, hint: str) -> bool:
"""Return whether a normalized column name contains a safe name hint."""
normalized_name = normalize_name(column)
normalized_hint = normalize_name(hint)
compact_hint = compact_normalized_name(hint)
if len(compact_hint) <= 2:
return normalized_hint in normalized_name.split()
if normalized_hint in normalized_name:
return True
return compact_hint in compact_normalized_name(column)
def cardinality_score(cardinality: int) -> float:
"""Return a small-multiples score for the number of panels needed."""
if cardinality < 2:
return -100.0
if cardinality <= 4:
return 26.0
if cardinality <= SMALL_MULTIPLES_IDEAL_CARDINALITY:
return 32.0
if cardinality <= 24:
return 16.0
if cardinality <= SMALL_MULTIPLES_MAX_REASONABLE_CARDINALITY:
return 4.0
return -25.0
def granularity_penalty(column: str, cardinality: int) -> float:
"""Return penalty for columns that usually create noisy small multiples."""
normalized_name = normalize_name(column)
penalty = 0.0
if any(
name_contains_hint(normalized_name, hint)
for hint in SMALL_MULTIPLES_GRANULAR_NAME_HINTS
):
penalty += 18.0
if cardinality > SMALL_MULTIPLES_IDEAL_CARDINALITY:
penalty += 8.0
if cardinality > SMALL_MULTIPLES_MAX_REASONABLE_CARDINALITY:
penalty += 20.0
return penalty
def select_waterfall_small_multiples_dimension(
result: pl.DataFrame,
recipe: dict[str, Any],
) -> dict[str, Any]:
"""Choose a presentation dimension for standard-variance small multiples."""
options = recipe.get("options") or {}
mappings = recipe.get("mappings") or {}
dimensions = [
str(column)
for column in mappings.get("dimensions") or []
if str(column) in result.schema
]
explicit_dimension = options.get("waterfall_small_multiples_dimension")
if explicit_dimension:
explicit_dimension = str(explicit_dimension)
return {
"status": (
"selected_explicit"
if explicit_dimension in result.schema
else "not_selected_missing_explicit_dimension"
),
"dimension": (
explicit_dimension if explicit_dimension in result.schema else None
),
"reason": "explicit_recipe_or_cli_dimension",
"candidates": [],
}
if not dimensions:
return {
"status": "not_selected_no_mapped_dimensions",
"dimension": None,
"reason": "standard_variance_has_no_reporting_dimension",
"candidates": [],
}
candidates: list[dict[str, Any]] = []
for dimension in dimensions:
grouped = (
result.group_by(dimension)
.agg(pl.col("total_delta").sum().alias("_delta"))
.with_columns(pl.col("_delta").abs().alias("_abs_delta"))
.sort("_abs_delta", descending=True)
)
cardinality = grouped.height
total_abs_delta = float(
grouped.select(pl.col("_abs_delta").sum()).item() or 0.0
)
top_abs_delta = float(grouped.select(pl.col("_abs_delta").max()).item() or 0.0)
non_zero_count = grouped.filter(pl.col("_abs_delta") > 0.000001).height
top_share = top_abs_delta / total_abs_delta if total_abs_delta else 0.0
name_score = dimension_priority_score(
dimension,
SMALL_MULTIPLES_DIMENSION_NAME_PRIORITY,
)
candidate_cardinality_score = cardinality_score(cardinality)
spread_score = min(14.0, float(non_zero_count) * 2.0)
if 0.25 <= top_share <= 0.85:
spread_score += 8.0
elif top_share > 0.95:
spread_score -= 4.0
penalty = granularity_penalty(dimension, cardinality)
score = name_score + candidate_cardinality_score + spread_score - penalty
candidates.append(
{
"dimension": dimension,
"score": round(score, 3),
"name_score": round(name_score, 3),
"cardinality_score": round(candidate_cardinality_score, 3),
"spread_score": round(spread_score, 3),
"granularity_penalty": round(penalty, 3),
"cardinality": cardinality,
"non_zero_members": non_zero_count,
"top_member_share_of_abs_delta": round(top_share, 6),
}
)
candidates.sort(
key=lambda item: (
float(item["score"]),
float(item["name_score"]),
-int(item["cardinality"]),
),
reverse=True,
)
best = candidates[0]
if float(best["score"]) < SMALL_MULTIPLES_MIN_DIMENSION_SCORE:
return {
"status": "not_selected_no_clear_dimension",
"dimension": None,
"reason": "no_candidate_met_business_readability_threshold",
"candidates": candidates,
}
return {
"status": "selected_ranked_candidate",
"dimension": best["dimension"],
"reason": (
"highest score from business-readability, cardinality, and variance "
"spread heuristic"
),
"candidates": candidates,
}
def select_total_by_dimension_bridge_dimension(
result: pl.DataFrame,
recipe: dict[str, Any],
) -> dict[str, Any]:
"""Choose one fixed dimension for the total variance bridge."""
options = recipe.get("options") or {}
mappings = recipe.get("mappings") or {}
dimensions = [
str(column)
for column in mappings.get("dimensions") or []
if str(column) in result.schema
]
explicit_dimension = options.get("total_by_dimension_bridge_dimension")
if explicit_dimension:
explicit_dimension = str(explicit_dimension)
return {
"status": (
"selected_explicit"
if explicit_dimension in result.schema
else "not_selected_missing_explicit_dimension"
),
"dimension": (
explicit_dimension if explicit_dimension in result.schema else None
),
"reason": "explicit_recipe_or_cli_dimension",
"candidates": [],
}
if not dimensions:
return {
"status": "not_selected_no_mapped_dimensions",
"dimension": None,
"reason": "total_by_dimension_bridge_requires_one_dimension",
"candidates": [],
}
candidates: list[dict[str, Any]] = []
for dimension in dimensions:
grouped = (
result.group_by(dimension)
.agg(pl.col("total_delta").sum().alias("_delta"))
.with_columns(pl.col("_delta").abs().alias("_abs_delta"))
.sort("_abs_delta", descending=True)
)
cardinality = grouped.height
total_abs_delta = float(
grouped.select(pl.col("_abs_delta").sum()).item() or 0.0
)
top_abs_delta = float(grouped.select(pl.col("_abs_delta").max()).item() or 0.0)
non_zero_count = grouped.filter(pl.col("_abs_delta") > TOLERANCE).height
top_share = top_abs_delta / total_abs_delta if total_abs_delta else 0.0
name_score = dimension_priority_score(
dimension,
TOTAL_BY_DIMENSION_NAME_PRIORITY,
)
if 2 <= cardinality <= 12:
cardinality_component = 28.0
elif cardinality <= 36:
cardinality_component = 18.0
elif cardinality <= TOTAL_BY_DIMENSION_MAX_REASONABLE_CARDINALITY:
cardinality_component = 8.0
else:
cardinality_component = -20.0
spread_score = min(16.0, float(non_zero_count) * 2.0)
if 0.2 <= top_share <= 0.9:
spread_score += 6.0
elif top_share > 0.98:
spread_score -= 6.0
penalty = granularity_penalty(dimension, cardinality)
score = name_score + cardinality_component + spread_score - penalty
candidates.append(
{
"dimension": dimension,
"score": round(score, 3),
"name_score": round(name_score, 3),
"cardinality_score": round(cardinality_component, 3),
"spread_score": round(spread_score, 3),
"granularity_penalty": round(penalty, 3),
"cardinality": cardinality,
"non_zero_members": non_zero_count,
"top_member_share_of_abs_delta": round(top_share, 6),
}
)
candidates.sort(
key=lambda item: (
float(item["score"]),
float(item["name_score"]),
-int(item["cardinality"]),
),
reverse=True,
)
best = candidates[0]
if float(best["score"]) < TOTAL_BY_DIMENSION_MIN_SCORE:
fallback_dimension = dimensions[0]
return {
"status": "selected_fallback_first_dimension",
"dimension": fallback_dimension,
"reason": "no_candidate_met_threshold_but_single_dimension_view_is_lightweight",
"candidates": candidates,
}
return {
"status": "selected_ranked_candidate",
"dimension": best["dimension"],
"reason": (
"highest score from business-readability, cardinality, and total "
"variance spread heuristic"
),
"candidates": candidates,
}
def select_exploded_variance_bridge_dimensions(
result: pl.DataFrame,
recipe: dict[str, Any],
*,
fallback_parent_dimension: str | None = None,
) -> dict[str, Any]:
"""Choose parent and child dimensions for the exploded bridge."""
options = recipe.get("options") or {}
mappings = recipe.get("mappings") or {}
dimensions = [
str(column)
for column in mappings.get("dimensions") or []
if str(column) in result.schema
]
if len(dimensions) < 2:
return {
"status": "not_selected_insufficient_dimensions",
"parent_dimension": None,
"child_dimension": None,
"reason": "exploded_variance_bridge_requires_two_dimensions",
}
explicit_parent = options.get("exploded_variance_bridge_parent_dimension")
parent_dimension = (
str(explicit_parent)
if explicit_parent
else fallback_parent_dimension
or options.get("total_by_dimension_bridge_dimension")
)
if parent_dimension:
parent_dimension = str(parent_dimension)
if parent_dimension not in result.schema:
return {
"status": "not_selected_missing_parent_dimension",
"parent_dimension": None,
"child_dimension": None,
"reason": "configured_parent_dimension_missing",
}
else:
parent_selection = select_total_by_dimension_bridge_dimension(result, recipe)
parent_dimension = parent_selection.get("dimension")
if not parent_dimension:
return {
"status": "not_selected_no_parent_dimension",
"parent_dimension": None,
"child_dimension": None,
"reason": parent_selection.get("reason"),
"parent_selection": parent_selection,
}
explicit_child = options.get("exploded_variance_bridge_child_dimension")
if explicit_child:
child_dimension = str(explicit_child)
if child_dimension not in result.schema:
return {
"status": "not_selected_missing_child_dimension",
"parent_dimension": parent_dimension,
"child_dimension": None,
"reason": "configured_child_dimension_missing",
}
if child_dimension == parent_dimension:
return {
"status": "not_selected_child_matches_parent",
"parent_dimension": parent_dimension,
"child_dimension": None,
"reason": "parent_and_child_dimensions_must_differ",
}
return {
"status": "selected_explicit",
"parent_dimension": parent_dimension,
"child_dimension": child_dimension,
"reason": "explicit_recipe_or_cli_dimensions",
}
for dimension in dimensions:
if dimension != parent_dimension:
return {
"status": "selected_parent_plus_next_dimension",
"parent_dimension": parent_dimension,
"child_dimension": dimension,
"reason": "parent_dimension_plus_first_other_mapped_dimension",
}
return {
"status": "not_selected_no_child_dimension",
"parent_dimension": parent_dimension,
"child_dimension": None,
"reason": "no_mapped_dimension_remaining_after_parent",
}
def _numeric_column_sum(frame: pl.DataFrame, column: str) -> float:
"""Return a stable numeric sum for a result column that may be absent."""
if frame.is_empty() or column not in frame.schema:
return 0.0
value = frame.select(pl.col(column).sum()).item()
return float(value or 0.0)
def standard_variance_component_columns(
recipe: dict[str, Any],
) -> list[tuple[str, str]]:
"""Return standard variance components using the same labels as charts."""
if not recipe["mappings"].get("units_column"):
return [("Total variance", "total_delta")]
volume_label = "Units"
return [
(volume_label if label == "volume_or_units" else label, column)
for label, column in STANDARD_VARIANCE_COMPONENT_COLUMNS
]
def _small_multiples_panel_slices(
result: pl.DataFrame,
dimension: str,
*,
limit: int = SMALL_MULTIPLES_PANEL_LIMIT,
) -> tuple[pl.DataFrame, list[dict[str, Any]], bool]:
"""Return chart-aligned panel filters, including an Other member panel."""
working = result.with_columns(
pl.col(dimension)
.cast(pl.Utf8)
.fill_null(SMALL_MULTIPLES_NULL_LABEL)
.alias(SMALL_MULTIPLES_WORK_COLUMN)
)
ranked = (
working.group_by(SMALL_MULTIPLES_WORK_COLUMN)
.agg(pl.col("total_delta").abs().sum().alias("_abs_delta"))
.sort("_abs_delta", descending=True)
)
all_values = [str(value) for value in ranked[SMALL_MULTIPLES_WORK_COLUMN].to_list()]
selected_values = all_values[:limit]
include_other = len(all_values) > limit
if include_other:
selected_values = all_values[: max(1, limit - 1)]
panels = [
{
"panel_number": index + 1,
"dimension_value": value,
"panel_type": "member",
"selected_values": [value],
}
for index, value in enumerate(selected_values)
]
if include_other:
panels.append(
{
"panel_number": len(panels) + 1,
"dimension_value": SMALL_MULTIPLES_OTHER_LABEL,
"panel_type": "other_members",
"selected_values": [
value for value in all_values if value not in selected_values
],
}
)
return working, panels, include_other
def _small_multiples_panel_frame(
working: pl.DataFrame,
panel: dict[str, Any],
) -> pl.DataFrame:
"""Return the result rows included in one small-multiples panel."""
selected_values = [str(value) for value in panel["selected_values"]]
if panel["panel_type"] == "other_members":
return working.filter(
pl.col(SMALL_MULTIPLES_WORK_COLUMN).is_in(selected_values)
)
return working.filter(pl.col(SMALL_MULTIPLES_WORK_COLUMN) == selected_values[0])
def _panel_component_values(
panel_frame: pl.DataFrame,
recipe: dict[str, Any],
) -> tuple[list[dict[str, Any]], dict[str, float]]:
"""Return component rows and aggregate totals for one panel."""
amount_baseline = _numeric_column_sum(panel_frame, "amount_baseline")
amount_comparison = _numeric_column_sum(panel_frame, "amount_comparison")
total_delta = _numeric_column_sum(panel_frame, "total_delta")
if recipe["mappings"].get("units_column"):
unit_mix_label = "Units & mix"
components = [
{
"variance_type": "Price",
"variance_amount": _numeric_column_sum(panel_frame, "price_variance"),
"is_residual_other": False,
"is_residual_balance": False,
"displayed_in_chart": True,
},
{
"variance_type": unit_mix_label,
"variance_amount": _numeric_column_sum(panel_frame, "volume_variance")
+ _numeric_column_sum(panel_frame, "mix_variance"),
"is_residual_other": False,
"is_residual_balance": False,
"displayed_in_chart": True,
},
]
else:
components = [
{
"variance_type": "Total variance",
"variance_amount": total_delta,
"is_residual_other": False,
"is_residual_balance": False,
"displayed_in_chart": True,
}
]
component_sum = sum(item["variance_amount"] for item in components)
residual = total_delta - component_sum
components.append(
{
"variance_type": "Balance",
"variance_amount": residual,
"is_residual_other": True,
"is_residual_balance": True,
"displayed_in_chart": abs(residual) > TOLERANCE,
}
)
return components, {
"amount_baseline": amount_baseline,
"amount_comparison": amount_comparison,
"total_delta": total_delta,
"total_abs_delta": _numeric_column_sum(
panel_frame.with_columns(pl.col("total_delta").abs().alias("_abs_delta")),
"_abs_delta",
),
"component_sum": component_sum,
"other_residual": residual,
}
def _dominant_component(components: list[dict[str, Any]]) -> dict[str, Any]:
"""Return the largest component by absolute amount."""
if not components:
return {"variance_type": "", "variance_amount": 0.0}
return max(components, key=lambda item: abs(float(item["variance_amount"] or 0.0)))
def _output_language(value: object | None) -> str:
text = str(value or "en").strip().lower().replace("_", "-")
return "es" if text.split("-", 1)[0] == "es" else "en"
def _small_multiples_summary_markdown(context: dict[str, Any]) -> str:
"""Return a markdown chart-data block for Codex's business interpretation."""
if context.get("status") != "written":
return ""
spanish = _output_language(context.get("language")) == "es"
lines = [
"",
(
"## Múltiples gráficos de variaciones estándar"
if spanish
else "## Standard Variance Small Multiples"
),
"",
f"- {'Dimensión' if spanish else 'Dimension'}: `{context.get('dimension')}`",
f"- {'Número de paneles' if spanish else 'Panel count'}: `{context.get('panel_count')}`",
f"- {'Panel de Otros' if spanish else 'Other member panel'}: `{context.get('has_other_member_panel')}`",
f"- {'Componente Saldo residual' if spanish else 'Any residual Balance component'}: `{context.get('has_residual_balance_component')}`",
f"- {'Archivos fuente' if spanish else 'Source files'}: `waterfall_small_multiples.png`, "
"`waterfall_small_multiples_summary.csv`, "
"`waterfall_small_multiples_context.json`",
"",
(
"Paneles principales por variación absoluta:"
if spanish
else "Top panels by absolute variance:"
),
]
for panel in context.get("panels", [])[:6]:
dominant = panel.get("dominant_component") or {}
lines.append(
"- "
f"{panel.get('dimension_value')}: total_delta="
f"{float(panel.get('total_delta') or 0.0):,.2f}; "
f"dominant={dominant.get('variance_type')} "
f"{float(dominant.get('variance_amount') or 0.0):,.2f}"
)
lines.extend(
[
"",
(
"Codex debe utilizar estos datos fuente para explicar si la variación estándar se concentra en elementos concretos o se distribuye por la dimensión seleccionada."
if spanish
else "Codex must use this source data when explaining whether the standard variance story is concentrated in specific dimension members or spread across the selected dimension."
),
]
)
return "\n".join(lines) + "\n"
def write_waterfall_small_multiples_chart_data(
result: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
) -> tuple[list[str], dict[str, Any], str]:
"""Write data context for the standard-variance small-multiples chart."""
options = recipe.get("options") or {}
if not bool(options.get("waterfall_chart", True)):
return [], {"enabled": False, "status": "disabled_waterfall_chart"}, ""
if not bool(options.get("waterfall_small_multiples", False)):
return [], {"enabled": False, "status": "disabled_no_selected_dimension"}, ""
dimension = options.get("waterfall_small_multiples_dimension")
if not dimension or str(dimension) not in result.schema:
return (
[],
{
"enabled": True,
"status": "not_written_no_dimension",
"dimension": str(dimension or ""),
},
"",
)
dimension = str(dimension)
working, panel_specs, has_other_member_panel = _small_multiples_panel_slices(
result,
dimension,
)
if not panel_specs:
return (
[],
{
"enabled": True,
"status": "not_written_no_panel_values",
"dimension": dimension,
},
"",
)
rows: list[dict[str, Any]] = []
panels: list[dict[str, Any]] = []
total_abs_delta = _numeric_column_sum(
working.with_columns(pl.col("total_delta").abs().alias("_abs_delta")),
"_abs_delta",
)
has_residual_balance_component = False
for panel in panel_specs:
panel_frame = _small_multiples_panel_frame(working, panel)
components, totals = _panel_component_values(panel_frame, recipe)
dominant = _dominant_component(components)
has_residual_balance_component = (
has_residual_balance_component or abs(totals["other_residual"]) > TOLERANCE
)
included_member_count = len(panel["selected_values"])
panel_payload = {
"panel_number": panel["panel_number"],
"dimension": dimension,
"dimension_value": panel["dimension_value"],
"panel_type": panel["panel_type"],
"included_member_count": included_member_count,
"source_result_rows": panel_frame.height,
"amount_baseline": totals["amount_baseline"],
"amount_comparison": totals["amount_comparison"],
"total_delta": totals["total_delta"],
"total_abs_delta": totals["total_abs_delta"],
"share_of_all_panel_abs_delta": (
totals["total_abs_delta"] / total_abs_delta if total_abs_delta else 0.0
),
"other_residual": totals["other_residual"],
"dominant_component": {
"variance_type": dominant.get("variance_type"),
"variance_amount": dominant.get("variance_amount"),
},
"components": components,
}
panels.append(panel_payload)
abs_component_total = sum(
abs(float(component["variance_amount"] or 0.0)) for component in components
)
for component in components:
amount = float(component["variance_amount"] or 0.0)
rows.append(
{
"panel_number": panel["panel_number"],
"dimension": dimension,
"dimension_value": panel["dimension_value"],
"panel_type": panel["panel_type"],
"included_member_count": included_member_count,
"source_result_rows": panel_frame.height,
"amount_baseline": totals["amount_baseline"],
"amount_comparison": totals["amount_comparison"],
"total_delta": totals["total_delta"],
"total_abs_delta": totals["total_abs_delta"],
"share_of_all_panel_abs_delta": (
totals["total_abs_delta"] / total_abs_delta
if total_abs_delta
else 0.0
),
"variance_type": component["variance_type"],
"variance_amount": amount,
"share_of_panel_abs_components": (
abs(amount) / abs_component_total
if abs_component_total
else 0.0
),
"is_residual_other": component["is_residual_other"],
"is_residual_balance": component.get(
"is_residual_balance",
False,
),
"displayed_in_chart": component["displayed_in_chart"],
}
)
panels.sort(key=lambda item: abs(float(item["total_delta"] or 0.0)), reverse=True)
summary_path = output_dir / "waterfall_small_multiples_summary.csv"
context_path = output_dir / "waterfall_small_multiples_context.json"
pl.DataFrame(rows).write_csv(summary_path)
context = {
"analysis_type": "standard_variance_small_multiples",
"status": "written",
"language": _output_language(recipe.get("language")),
"dimension": dimension,
"panel_limit": SMALL_MULTIPLES_PANEL_LIMIT,
"panel_count": len(panel_specs),
"has_other_member_panel": has_other_member_panel,
"has_residual_other_component": has_residual_balance_component,
"has_residual_balance_component": has_residual_balance_component,
"chart_component_mode": "legacy_price_units_mix_balance",
"chart_artifact": "waterfall_small_multiples.png",
"summary_csv": summary_path.name,
"context_json": context_path.name,
"selection": options.get("waterfall_small_multiples_dimension_selection"),
**_title_contract_payload(
recipe,
chart_kind="standard_small_multiples",
dimension=str(dimension),
),
"codex_interpretation_contract": {
"must_review_when_written": True,
"required_points": [
"Use panel labels and values to explain concentration or spread by dimension member.",
"Explain the dominant variance type within the largest panels.",
"Call out the Others aggregated member panel when it exists.",
"Call out residual Balance components when they are material.",
"Do not treat panel members as additional bridge rows; each panel repeats the same standard variance bridge.",
],
},
"panels": panels,
}
write_json(context_path, context)
audit = {
"enabled": True,
"status": "written",
"dimension": dimension,
"summary_csv": summary_path.name,
"context_json": context_path.name,
"panel_count": len(panel_specs),
"has_other_member_panel": has_other_member_panel,
"has_residual_other_component": has_residual_balance_component,
"has_residual_balance_component": has_residual_balance_component,
"chart_component_mode": "legacy_price_units_mix_balance",
}
return (
[str(summary_path), str(context_path)],
audit,
_small_multiples_summary_markdown(context),
)
def suggested_calculation_grain(
candidates: list[str], dimensions: list[str]
) -> list[str]:
"""Return lower-level PVM grain when product or customer columns exist."""
selected: list[str] = []
normalized = [(column, normalize_name(column)) for column in candidates]
for hint in CALCULATION_GRAIN_NAME_PRIORITY:
normalized_hint = normalize_name(hint)
for column, normalized_name in normalized:
if normalized_name == normalized_hint and column not in selected:
selected.append(column)
return selected or dimensions
def infer_mappings(df: pl.DataFrame) -> dict[str, Any]:
"""Infer a suggested recipe mapping from column names and dtypes."""
columns, _schema = get_schema_and_column_names(df)
mapping: dict[str, Any] = {}
for key, hints in COLUMN_HINTS.items():
mapping[key] = first_matching_column(columns, hints)
mapping["date_column"] = suggested_date_column(df, columns)
mapping["period_column"] = (
plan_actual_scenario_column(df, columns) or mapping["period_column"]
)
schema = df.schema
period_col = mapping["period_column"]
metric_cols = {
value
for value in (
mapping["amount_column"],
mapping["units_column"],
mapping["discount_column"],
mapping["cogs_column"],
mapping["date_column"],
period_col,
)
if value
}
dimension_candidates = [
column
for column in columns
if column not in metric_cols
and not is_numeric_dtype(schema[column])
and not is_temporal_dtype(schema[column])
]
if not dimension_candidates:
dimension_candidates = [
column
for column in columns
if column not in metric_cols and column != period_col
][:2]
dimensions = ranked_dimension_candidates(dimension_candidates)[:4]
mapping["dimensions"] = dimensions
mapping["calculation_grain"] = suggested_calculation_grain(
dimension_candidates, dimensions
)
return mapping
def period_values(df: pl.DataFrame, period_col: str | None) -> list[str]:
"""Return sorted non-null period values as strings."""
if not period_col or period_col not in df.schema:
return []
values = (
df.select(pl.col(period_col).cast(pl.Utf8).drop_nulls().unique().sort())
.to_series(0)
.to_list()
)
return [str(value) for value in values]
def suggested_period_pair(
df: pl.DataFrame, period_col: str | None, periods: list[str]
) -> tuple[str | None, str | None]:
"""Return suggested baseline and comparison period values."""
if is_scenario_comparison_column(df, period_col):
return default_scenario_comparison_pair(periods)
return (
periods[0] if len(periods) >= 1 else None,
periods[1] if len(periods) >= 2 else None,
)
def suggested_comparison_options(
df: pl.DataFrame, period_col: str | None
) -> dict[str, str]:
"""Return advisory comparison metadata for Codex intake and review."""
if is_scenario_comparison_column(df, period_col):
return {
"comparison_basis": COMPARISON_BASIS_SCENARIO,
"period_comparison_mode": PERIOD_MODE_NOT_APPLICABLE,
}
return {
"comparison_basis": COMPARISON_BASIS_PERIOD,
"period_comparison_mode": PERIOD_MODE_CALENDAR,
}
def legacy_period_mode_from_period_type(period_type: str) -> str:
"""Return the legacy variance period mode for a normalized period type."""
if period_type == PERIOD_TYPE_ROLLING:
return PERIOD_MODE_ROLLING
if period_type in {PERIOD_TYPE_TO_DATE, PERIOD_TYPE_FISCAL}:
return PERIOD_MODE_YEAR_TO_DATE
if period_type == PERIOD_TYPE_CUSTOM:
return PERIOD_MODE_CUSTOM
return PERIOD_MODE_CALENDAR
def build_recipe(
source_path: Path,
df: pl.DataFrame,
*,
language: str,
existing_recipe: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build a suggested or merged variance recipe."""
mapping = infer_mappings(df)
periods = period_values(df, mapping["period_column"])
baseline_period, comparison_period = suggested_period_pair(
df, mapping["period_column"], periods
)
suggested_small_multiples_dimension = (
mapping["dimensions"][0] if len(mapping["dimensions"]) == 1 else None
)
suggested_comparison = suggested_comparison_options(df, mapping["period_column"])
suggested_period_contract = period_contract_options(
suggested_comparison,
default_type=(
PERIOD_TYPE_CUSTOM
if suggested_comparison["comparison_basis"] == COMPARISON_BASIS_SCENARIO
else "calendar"
),
)
suggested = {
"schema_version": SCHEMA_VERSION,
"language": language,
"source_file": str(source_path),
"accounting_review": default_accounting_review(),
"mappings": {
"period_column": mapping["period_column"],
"baseline_period": baseline_period,
"comparison_period": comparison_period,
"amount_column": mapping["amount_column"],
"units_column": mapping["units_column"],
"discount_column": mapping["discount_column"],
"cogs_column": mapping["cogs_column"],
"date_column": mapping["date_column"],
"dimensions": mapping["dimensions"],
"calculation_grain": mapping["calculation_grain"],
},
"options": {
**suggested_comparison,
**suggested_period_contract,
"currency": "EUR",
"unit_label": "units",
"root_cause_bridge": True,
"root_cause_bridge_alternative_result": 1,
"root_cause_bridge_drilldown_rows": [],
"root_cause_bridge_drilldown_all": False,
"root_cause_bridge_move_rows": {},
"root_cause_bridge_alternative_sweep": True,
"root_cause_bridge_alternative_sweep_start": 1,
"root_cause_bridge_alternative_sweep_end": 10,
"root_cause_bridge_auto_drilldown": "all_selected",
"root_cause_bridge_auto_drilldown_min_share": 0.75,
"root_cause_component_bridge": False,
"root_cause_component_bridge_alternative_result": 1,
"waterfall_chart": True,
"pvm_decomposition_ladder": True,
"waterfall_small_multiples": bool(mapping["dimensions"]),
"waterfall_small_multiples_dimension": suggested_small_multiples_dimension,
"total_by_dimension_bridge": bool(mapping["dimensions"]),
"total_by_dimension_bridge_dimension": None,
"total_by_dimension_bridge_top_n": TOTAL_BY_DIMENSION_DEFAULT_TOP_N,
"exploded_variance_bridge": len(mapping["dimensions"]) >= 2,
"exploded_variance_bridge_parent_dimension": None,
"exploded_variance_bridge_child_dimension": None,
"exploded_variance_bridge_parent_top_n": TOTAL_BY_DIMENSION_DEFAULT_TOP_N,
"exploded_variance_bridge_child_top_n": EXPLODED_BRIDGE_DEFAULT_CHILD_TOP_N,
"exploded_variance_bridge_max_drilldowns": (
EXPLODED_BRIDGE_DEFAULT_MAX_DRILLDOWNS
),
},
}
if not existing_recipe:
return suggested
merged = dict(suggested)
merged.update(
{k: v for k, v in existing_recipe.items() if k not in {"mappings", "options"}}
)
merged["mappings"] = {
**suggested["mappings"],
**existing_recipe.get("mappings", {}),
}
merged["options"] = {**suggested["options"], **existing_recipe.get("options", {})}
merged["accounting_review"] = normalize_accounting_review(
existing_recipe.get("accounting_review")
)
return merged
def _source_reference(
input_path: Path,
client_engagement: dict[str, Any] | None,
) -> str:
"""Return an unmanaged path or a portable managed-run source reference."""
if client_engagement is None:
return input_path.as_posix()
run_root = Path(str(client_engagement["run_root"])).resolve(strict=True)
try:
return input_path.resolve(strict=True).relative_to(run_root).as_posix()
except ValueError as exc:
raise ValueError("Variance input is outside the managed run.") from exc
def inspect_variance_inputs(
input_path: Path,
output_dir: Path,
recipe_path: Path | None = None,
*,
language: str = "en",
client_engagement: dict[str, Any] | None = None,
) -> InspectionResult:
"""Inspect a variance input file and write inspection and recipe JSON."""
if is_vera_managed_host() and client_engagement is None:
raise ValueError(
"The Vera-packaged variance inspector requires --client-engagement."
)
df = read_table(input_path)
columns, schema = get_schema_and_column_names(df)
existing_recipe = load_json(recipe_path) if recipe_path else None
recipe = build_recipe(
input_path, df, language=language, existing_recipe=existing_recipe
)
mappings = recipe["mappings"]
candidate_columns = list(
dict.fromkeys(
value
for value in (
mappings.get("period_column"),
mappings.get("amount_column"),
mappings.get("units_column"),
mappings.get("discount_column"),
mappings.get("cogs_column"),
mappings.get("date_column"),
*(mappings.get("dimensions") or []),
*(mappings.get("calculation_grain") or []),
)
if isinstance(value, str) and value in columns
)
)
source_file = _source_reference(input_path, client_engagement)
if client_engagement is not None:
recipe["source_file"] = source_file
payload = {
"schema_version": SCHEMA_VERSION,
**(
{"path_reference": "run_root_relative"}
if client_engagement is not None
else {}
),
"generated_at": utc_now(),
"language": language,
"source_file": source_file,
"row_count": df.height,
"columns": [{"name": column, "dtype": schema[column]} for column in columns],
"available_analysis_context": available_analysis_context(df),
"period_values": period_values(df, mappings.get("period_column")),
"suggested_mappings": mappings,
"sampled_columns": candidate_columns,
"omitted_sample_columns": [
column for column in columns if column not in candidate_columns
],
"sample_row_limit": 10,
"sample_behavior": "first_rows_candidate_columns_only",
"sample_rows": df.select(candidate_columns).head(10).to_dicts(),
"warnings": inspection_warnings(df, mappings),
}
warn_if_output_dir_has_existing_files(output_dir, "Inspection")
output_dir.mkdir(parents=True, exist_ok=True)
write_json(output_dir / "inspection.json", payload)
write_json(output_dir / "suggested_recipe.json", recipe)
return InspectionResult(payload=payload, recipe=recipe)
def inspection_warnings(df: pl.DataFrame, mappings: dict[str, Any]) -> list[str]:
"""Return deterministic warnings about missing or incomplete mappings."""
warnings: list[str] = []
required = {
"period_column": "Period column is required.",
"baseline_period": "Baseline period is required.",
"comparison_period": "Comparison period is required.",
"amount_column": "Amount/sales column is required.",
}
for key, message in required.items():
if not mappings.get(key):
warnings.append(message)
columns, _schema = get_schema_and_column_names(df)
for key in (
"period_column",
"amount_column",
"units_column",
"discount_column",
"cogs_column",
"date_column",
):
value = mappings.get(key)
if value and value not in columns:
warnings.append(f"Mapped {key} does not exist in input columns: {value}")
for dimension in mappings.get("dimensions") or []:
if dimension not in columns:
warnings.append(
f"Mapped dimension does not exist in input columns: {dimension}"
)
return warnings
def parsed_date_expression(column: str) -> pl.Expr:
"""Return a date expression for common sales-export date representations."""
text = pl.col(column).cast(pl.Utf8)
return pl.coalesce(
[
pl.col(column).cast(pl.Date, strict=False),
text.str.strptime(pl.Date, "%Y-%m-%d", strict=False),
text.str.strptime(pl.Date, "%Y/%m/%d", strict=False),
text.str.strptime(pl.Date, "%d/%m/%Y", strict=False),
text.str.strptime(pl.Date, "%m/%d/%Y", strict=False),
text.str.strptime(pl.Date, "%Y-%m", strict=False),
text.str.strptime(pl.Date, "%b-%Y", strict=False),
]
)
def as_date(value: Any) -> date:
"""Return a Python date from legacy or Polars date values."""
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
raise ValueError(f"Expected a date value, got {value!r}")
def add_months(value: date, months: int) -> date:
"""Shift a date by whole months, preserving the closest valid day."""
month_index = value.month - 1 + months
year = value.year + month_index // 12
month = month_index % 12 + 1
day = min(value.day, calendar.monthrange(year, month)[1])
return date(year, month, day)
def fiscal_year_start_for(value: date, fiscal_start_month: int) -> date:
"""Return the fiscal/calendar year start for a cutoff date."""
year = value.year if value.month >= fiscal_start_month else value.year - 1
return date(year, fiscal_start_month, 1)
def int_option(options: dict[str, Any], key: str, default: int) -> int:
"""Return a positive integer option with a conservative fallback."""
try:
value = int(options.get(key, default))
except (TypeError, ValueError):
value = default
return value if value > 0 else default
def rolling_window_length_options(
options: dict[str, Any], period_grain: str
) -> dict[str, int]:
"""Return a rolling window length from explicit options or period grain.
This rule is deterministic because the grain-to-window mapping is mechanical:
week=7 days, month=1 month, quarter=3 months, year=12 months.
"""
if options.get("rolling_window_days") is not None:
return {"days": int_option(options, "rolling_window_days", 7)}
if options.get("rolling_window_months") is not None:
return {"months": int_option(options, "rolling_window_months", 12)}
if period_grain == PERIOD_GRAIN_WEEK:
return {"days": 7}
if period_grain == PERIOD_GRAIN_MONTH:
return {"months": 1}
if period_grain == PERIOD_GRAIN_QUARTER:
return {"months": 3}
if period_grain == PERIOD_GRAIN_YEAR:
return {"months": 12}
return {"months": 12}
def bounded_int_option(
options: dict[str, Any], key: str, default: int, *, minimum: int, maximum: int
) -> int:
"""Return an integer option clamped to an inclusive range."""
value = int_option(options, key, default)
return min(max(value, minimum), maximum)
def bounded_float_option(
options: dict[str, Any], key: str, default: float, *, minimum: float, maximum: float
) -> float:
"""Return a float option clamped to an inclusive range."""
try:
value = float(options.get(key, default))
except (TypeError, ValueError):
value = default
return min(max(value, minimum), maximum)
def positive_int_list_option(options: dict[str, Any], key: str) -> list[int]:
"""Return unique positive integers from an option list."""
raw_value = options.get(key, [])
if raw_value is None:
return []
values = raw_value if isinstance(raw_value, list) else [raw_value]
result: list[int] = []
for value in values:
try:
parsed = int(value)
except (TypeError, ValueError):
continue
if parsed > 0 and parsed not in result:
result.append(parsed)
return result
def move_rows_option(options: dict[str, Any], key: str) -> dict[str, list[int]]:
"""Return normalized legacy move-row mappings with string keys."""
raw_value = options.get(key, {})
if not isinstance(raw_value, dict):
return {}
result: dict[str, list[int]] = {}
for main_row, drilldown_rows in raw_value.items():
try:
parsed_main = int(main_row)
except (TypeError, ValueError):
continue
if parsed_main <= 0:
continue
values = (
drilldown_rows if isinstance(drilldown_rows, list) else [drilldown_rows]
)
parsed_rows: list[int] = []
for row in values:
try:
parsed_row = int(row)
except (TypeError, ValueError):
continue
if parsed_row > 0 and parsed_row not in parsed_rows:
parsed_rows.append(parsed_row)
if parsed_rows:
result[str(parsed_main)] = parsed_rows
return result
def root_cause_auto_drilldown_option(options: dict[str, Any]) -> str:
"""Return a supported automatic root-cause drilldown mode."""
mode = (
str(options.get("root_cause_bridge_auto_drilldown") or "none")
.strip()
.lower()
.replace("-", "_")
)
return mode if mode in ROOT_CAUSE_AUTO_DRILLDOWN_VALUES else "none"
def between_dates(column: str, start: date, end: date) -> pl.Expr:
"""Return an inclusive date-window predicate."""
return (pl.col(column) >= pl.lit(start)) & (pl.col(column) <= pl.lit(end))
def period_window_metadata(
*,
mode: str,
date_column: str,
baseline_label: str,
baseline_start: date,
baseline_end: date,
comparison_label: str,
comparison_start: date,
comparison_end: date,
legacy_context: dict[str, Any],
extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build an auditable period-window payload for the used recipe."""
payload = {
"mode": mode,
"date_column": date_column,
"source": "legacy_date_period_context",
"most_recent_date": as_date(legacy_context["most_recent_date"]),
"least_recent_date": as_date(legacy_context["least_recent_date"]),
"period_length_months": legacy_context["period_length_months"],
"baseline": {
"label": baseline_label,
"start_date": baseline_start,
"end_date": baseline_end,
},
"comparison": {
"label": comparison_label,
"start_date": comparison_start,
"end_date": comparison_end,
},
}
if extra:
payload.update(extra)
return payload
def prepare_period_comparison_buckets(
df: pl.DataFrame,
recipe: dict[str, Any],
) -> tuple[pl.DataFrame, dict[str, Any]]:
"""Create legacy-backed YTD/rolling buckets before variance arithmetic.
This deterministic step is justified because date-window boundaries are
mechanically verifiable and must be reproducible for audit review.
"""
options = dict(recipe.get("options") or {})
normalized_period_contract = period_contract_options(options)
if options.get("period_type") and options.get("period_comparison_mode") in {
None,
PERIOD_MODE_CALENDAR,
}:
mode = legacy_period_mode_from_period_type(
str(normalized_period_contract["period_type"])
)
elif options.get("period_comparison_mode"):
mode = str(options.get("period_comparison_mode"))
if mode not in PERIOD_COMPARISON_MODE_VALUES:
mode = legacy_period_mode_from_period_type(
str(normalized_period_contract["period_type"])
)
else:
mode = legacy_period_mode_from_period_type(
str(normalized_period_contract["period_type"])
)
if mode not in {PERIOD_MODE_YEAR_TO_DATE, PERIOD_MODE_ROLLING}:
return df, recipe
mappings = dict(recipe["mappings"])
date_column = mappings.get("date_column") or mappings.get("period_column")
if not date_column or date_column not in df.schema:
raise ValueError(
"A date_column mapping is required for year-to-date or rolling periods."
)
prepared = df.with_columns(
parsed_date_expression(date_column).alias(DATE_WORK_COLUMN)
)
valid_date_rows = prepared.filter(pl.col(DATE_WORK_COLUMN).is_not_null())
if valid_date_rows.is_empty():
raise ValueError(
f"Could not parse any dates from mapped date column: {date_column}"
)
legacy_context = legacy_date_period_context(valid_date_rows, DATE_WORK_COLUMN)
comparison_end = as_date(legacy_context["most_recent_date"])
period_specific_options: dict[str, Any] = {}
if mode == PERIOD_MODE_YEAR_TO_DATE:
fiscal_start_month = int_option(options, "fiscal_start_month", 1)
if fiscal_start_month > 12:
fiscal_start_month = 1
comparison_start = fiscal_year_start_for(comparison_end, fiscal_start_month)
baseline_start = add_months(comparison_start, -12)
baseline_end = add_months(comparison_end, -12)
baseline_label = str(legacy_context["ytd_baseline_label"])
comparison_label = str(legacy_context["ytd_label"])
period_window = period_window_metadata(
mode=mode,
date_column=date_column,
baseline_label=baseline_label,
baseline_start=baseline_start,
baseline_end=baseline_end,
comparison_label=comparison_label,
comparison_start=comparison_start,
comparison_end=comparison_end,
legacy_context=legacy_context,
extra={"fiscal_start_month": fiscal_start_month},
)
else:
window_length = rolling_window_length_options(
options, str(normalized_period_contract["period_grain"])
)
rolling_comparison = str(
options.get("rolling_comparison") or ROLLING_COMPARISON_PRIOR_YEAR
)
if rolling_comparison not in ROLLING_COMPARISON_VALUES:
rolling_comparison = ROLLING_COMPARISON_PRIOR_YEAR
if "days" in window_length:
window_days = window_length["days"]
comparison_start = comparison_end - timedelta(days=window_days - 1)
if rolling_comparison == ROLLING_COMPARISON_PREVIOUS_WINDOW:
baseline_end = comparison_start - timedelta(days=1)
baseline_start = baseline_end - timedelta(days=window_days - 1)
baseline_label = (
f"previous_{window_days}d_"
f"{baseline_start.isoformat()}_{baseline_end.isoformat()}"
)
else:
baseline_end = add_months(comparison_end, -12)
baseline_start = baseline_end - timedelta(days=window_days - 1)
baseline_label = (
f"prior_year_{window_days}d_"
f"{baseline_start.isoformat()}_{baseline_end.isoformat()}"
)
comparison_label = (
f"rolling_{window_days}d_"
f"{comparison_start.isoformat()}_{comparison_end.isoformat()}"
)
rolling_extra = {
"rolling_window_days": window_days,
"rolling_comparison": rolling_comparison,
}
else:
window_months = window_length["months"]
comparison_start = add_months(comparison_end, -(window_months - 1)).replace(
day=1
)
if rolling_comparison == ROLLING_COMPARISON_PREVIOUS_WINDOW:
baseline_start = add_months(comparison_start, -window_months)
baseline_end = comparison_start - timedelta(days=1)
baseline_label = (
f"previous_{window_months}m_"
f"{baseline_start.isoformat()}_{baseline_end.isoformat()}"
)
else:
baseline_start = add_months(comparison_start, -12)
baseline_end = add_months(comparison_end, -12)
baseline_label = str(legacy_context["rolling_baseline_label"])
comparison_label = str(legacy_context["rolling_label"])
rolling_extra = {
"rolling_window_months": window_months,
"rolling_comparison": rolling_comparison,
}
period_specific_options = rolling_extra
period_window = period_window_metadata(
mode=mode,
date_column=date_column,
baseline_label=baseline_label,
baseline_start=baseline_start,
baseline_end=baseline_end,
comparison_label=comparison_label,
comparison_start=comparison_start,
comparison_end=comparison_end,
legacy_context=legacy_context,
extra=rolling_extra,
)
prepared = prepared.with_columns(
pl.when(between_dates(DATE_WORK_COLUMN, baseline_start, baseline_end))
.then(pl.lit(baseline_label))
.when(between_dates(DATE_WORK_COLUMN, comparison_start, comparison_end))
.then(pl.lit(comparison_label))
.otherwise(None)
.alias(SYNTHETIC_PERIOD_COLUMN)
).drop(DATE_WORK_COLUMN)
bucket_counts = {
row[SYNTHETIC_PERIOD_COLUMN]: row["len"]
for row in prepared.group_by(SYNTHETIC_PERIOD_COLUMN)
.len()
.filter(pl.col(SYNTHETIC_PERIOD_COLUMN).is_not_null())
.to_dicts()
}
if not bucket_counts.get(baseline_label) or not bucket_counts.get(comparison_label):
raise ValueError(
"Prepared period windows do not contain both baseline and comparison rows."
)
mappings.update(
{
"period_column": SYNTHETIC_PERIOD_COLUMN,
"baseline_period": baseline_label,
"comparison_period": comparison_label,
}
)
options.update(
{
"comparison_basis": COMPARISON_BASIS_PERIOD,
"period_comparison_mode": mode,
"period_type": normalized_period_contract["period_type"],
"period_grain": normalized_period_contract["period_grain"],
"fiscal_start_month": (
fiscal_start_month
if mode == PERIOD_MODE_YEAR_TO_DATE
else normalized_period_contract["fiscal_start_month"]
),
"period_window": {
**period_window,
"row_counts": bucket_counts,
},
**period_specific_options,
}
)
updated_recipe = {**recipe, "mappings": mappings, "options": options}
return prepared, updated_recipe
def validate_recipe(df: pl.DataFrame, recipe: dict[str, Any]) -> dict[str, Any]:
"""Validate and normalize the recipe used for deterministic calculations."""
mappings = dict(recipe.get("mappings") or {})
options = dict(recipe.get("options") or {})
columns, schema = get_schema_and_column_names(df)
required_keys = (
"period_column",
"baseline_period",
"comparison_period",
"amount_column",
)
missing = [key for key in required_keys if not mappings.get(key)]
if missing:
raise ValueError(f"Missing required recipe mappings: {', '.join(missing)}")
for key in (
"period_column",
"amount_column",
"units_column",
"discount_column",
"cogs_column",
"date_column",
):
column = mappings.get(key)
if column and column not in columns:
raise ValueError(
f"Recipe mapping '{key}' references missing column '{column}'"
)
for key in ("amount_column", "units_column", "discount_column", "cogs_column"):
column = mappings.get(key)
if column and not is_numeric_dtype(df.schema[column]):
raise ValueError(
f"Recipe mapping '{key}' must be numeric: {column} ({schema[column]})"
)
dimensions = mappings.get("dimensions") or []
if not isinstance(dimensions, list):
raise ValueError("Recipe mapping 'dimensions' must be a list of column names.")
cohort_source_dimensions = recipe_cohort_source_dimensions(recipe)
missing_cohort_sources = [
column for column in cohort_source_dimensions if column not in columns
]
if missing_cohort_sources:
raise ValueError(
"Recipe cohort source columns are missing: "
+ ", ".join(missing_cohort_sources)
)
cohort_dimension_names = recipe_cohort_dimension_names(recipe)
missing_dimensions = [
column
for column in dimensions
if column not in columns and column not in cohort_dimension_names
]
if missing_dimensions:
raise ValueError(
"Recipe dimensions reference missing columns: "
+ ", ".join(missing_dimensions)
)
calculation_grain = mappings.get("calculation_grain")
if calculation_grain is None:
calculation_grain = dimensions
if not isinstance(calculation_grain, list):
raise ValueError(
"Recipe mapping 'calculation_grain' must be a list of column names."
)
missing_grain = [
column
for column in calculation_grain
if column not in columns and column not in cohort_dimension_names
]
if missing_grain:
raise ValueError(
"Recipe calculation_grain references missing columns: "
+ ", ".join(missing_grain)
)
mappings["dimensions"] = dimensions
mappings["calculation_grain"] = calculation_grain
default_comparison_options = suggested_comparison_options(
df, mappings.get("period_column")
)
comparison_basis = str(
options.get("comparison_basis")
or default_comparison_options["comparison_basis"]
)
if comparison_basis not in COMPARISON_BASIS_VALUES:
comparison_basis = default_comparison_options["comparison_basis"]
default_period_mode = default_comparison_options["period_comparison_mode"]
normalized_period_contract = period_contract_options(
options,
default_type=(
PERIOD_TYPE_CUSTOM
if comparison_basis == COMPARISON_BASIS_SCENARIO
else "calendar"
),
)
requested_period_mode = options.get("period_comparison_mode")
if options.get("period_type") and requested_period_mode in {
None,
PERIOD_MODE_CALENDAR,
}:
period_comparison_mode = legacy_period_mode_from_period_type(
str(normalized_period_contract["period_type"])
)
elif requested_period_mode:
period_comparison_mode = str(requested_period_mode)
elif options.get("period_type") or options.get("fiscal_start_month"):
period_comparison_mode = legacy_period_mode_from_period_type(
str(normalized_period_contract["period_type"])
)
else:
period_comparison_mode = default_period_mode
if period_comparison_mode not in PERIOD_COMPARISON_MODE_VALUES:
period_comparison_mode = legacy_period_mode_from_period_type(
str(normalized_period_contract["period_type"])
)
if (
comparison_basis == COMPARISON_BASIS_PERIOD
and period_comparison_mode == PERIOD_MODE_NOT_APPLICABLE
):
period_comparison_mode = PERIOD_MODE_CALENDAR
root_cause_enabled = bool(options.get("root_cause_bridge", True))
root_cause_auto_drilldown = root_cause_auto_drilldown_option(options)
if root_cause_enabled and root_cause_auto_drilldown == "none":
root_cause_auto_drilldown = "all_selected"
default_waterfall_small_multiples = bool(dimensions)
waterfall_small_multiples = bool(
options.get("waterfall_small_multiples", default_waterfall_small_multiples)
)
waterfall_small_multiples_dimension = options.get(
"waterfall_small_multiples_dimension"
)
total_by_dimension_bridge = bool(
options.get("total_by_dimension_bridge", bool(dimensions))
)
total_by_dimension_bridge_dimension = options.get(
"total_by_dimension_bridge_dimension"
)
exploded_variance_bridge = bool(
options.get("exploded_variance_bridge", len(dimensions) >= 2)
)
exploded_variance_bridge_parent_dimension = options.get(
"exploded_variance_bridge_parent_dimension"
)
exploded_variance_bridge_child_dimension = options.get(
"exploded_variance_bridge_child_dimension"
)
normalized = dict(recipe)
normalized["schema_version"] = recipe.get("schema_version") or SCHEMA_VERSION
normalized["mappings"] = mappings
period_window = options.get("period_window")
normalized["options"] = {
"comparison_basis": comparison_basis,
"period_comparison_mode": period_comparison_mode,
"period_type": normalized_period_contract["period_type"],
"period_grain": normalized_period_contract["period_grain"],
"fiscal_start_month": normalized_period_contract["fiscal_start_month"],
"period_window": period_window if isinstance(period_window, dict) else {},
"currency": str(options.get("currency") or "EUR"),
"unit_label": options.get("unit_label") or "units",
"root_cause_bridge": root_cause_enabled,
"root_cause_bridge_alternative_result": bounded_int_option(
options,
"root_cause_bridge_alternative_result",
1,
minimum=1,
maximum=10,
),
"root_cause_bridge_drilldown_rows": positive_int_list_option(
options,
"root_cause_bridge_drilldown_rows",
),
"root_cause_bridge_drilldown_all": bool(
options.get("root_cause_bridge_drilldown_all", False)
),
"root_cause_bridge_move_rows": move_rows_option(
options,
"root_cause_bridge_move_rows",
),
"root_cause_bridge_alternative_sweep": (
root_cause_enabled
and bool(options.get("root_cause_bridge_alternative_sweep", True))
),
"root_cause_bridge_alternative_sweep_start": bounded_int_option(
options,
"root_cause_bridge_alternative_sweep_start",
1,
minimum=1,
maximum=10,
),
"root_cause_bridge_alternative_sweep_end": bounded_int_option(
options,
"root_cause_bridge_alternative_sweep_end",
10,
minimum=1,
maximum=10,
),
"root_cause_bridge_auto_drilldown": root_cause_auto_drilldown,
"root_cause_bridge_auto_drilldown_min_share": bounded_float_option(
options,
"root_cause_bridge_auto_drilldown_min_share",
0.75,
minimum=0.0,
maximum=1.0,
),
"root_cause_component_bridge": root_cause_enabled
and bool(options.get("root_cause_component_bridge", False)),
"root_cause_component_bridge_alternative_result": bounded_int_option(
options,
"root_cause_component_bridge_alternative_result",
1,
minimum=1,
maximum=10,
),
"waterfall_chart": bool(options.get("waterfall_chart", True)),
"pvm_decomposition_ladder": bool(options.get("pvm_decomposition_ladder", True)),
"waterfall_small_multiples": waterfall_small_multiples,
"waterfall_small_multiples_dimension": waterfall_small_multiples_dimension,
"total_by_dimension_bridge": total_by_dimension_bridge,
"total_by_dimension_bridge_dimension": total_by_dimension_bridge_dimension,
"total_by_dimension_bridge_top_n": bounded_int_option(
options,
"total_by_dimension_bridge_top_n",
TOTAL_BY_DIMENSION_DEFAULT_TOP_N,
minimum=1,
maximum=50,
),
"exploded_variance_bridge": exploded_variance_bridge,
"exploded_variance_bridge_parent_dimension": (
exploded_variance_bridge_parent_dimension
),
"exploded_variance_bridge_child_dimension": (
exploded_variance_bridge_child_dimension
),
"exploded_variance_bridge_parent_top_n": bounded_int_option(
options,
"exploded_variance_bridge_parent_top_n",
TOTAL_BY_DIMENSION_DEFAULT_TOP_N,
minimum=1,
maximum=20,
),
"exploded_variance_bridge_child_top_n": bounded_int_option(
options,
"exploded_variance_bridge_child_top_n",
EXPLODED_BRIDGE_DEFAULT_CHILD_TOP_N,
minimum=1,
maximum=8,
),
"exploded_variance_bridge_max_drilldowns": bounded_int_option(
options,
"exploded_variance_bridge_max_drilldowns",
EXPLODED_BRIDGE_DEFAULT_MAX_DRILLDOWNS,
minimum=1,
maximum=EXPLODED_BRIDGE_DEFAULT_MAX_DRILLDOWNS,
),
}
if options.get("reporting_entity_label"):
normalized["options"]["reporting_entity_label"] = str(
options["reporting_entity_label"]
)
for key in (
"cohorts",
"cohort_definition",
"cohort_contract",
"derived_dimensions",
"like_for_like",
"cohort_current_period",
"cohort_previous_period",
"current_period_label",
"previous_period_label",
"recipe_cohort_audit",
):
if key in options:
normalized["options"][key] = options[key]
sweep_start = normalized["options"]["root_cause_bridge_alternative_sweep_start"]
sweep_end = normalized["options"]["root_cause_bridge_alternative_sweep_end"]
if sweep_end < sweep_start:
normalized["options"]["root_cause_bridge_alternative_sweep_start"] = sweep_end
normalized["options"]["root_cause_bridge_alternative_sweep_end"] = sweep_start
waterfall_dimension = normalized["options"]["waterfall_small_multiples_dimension"]
if waterfall_dimension:
waterfall_dimension = str(waterfall_dimension)
if (
waterfall_dimension not in columns
and waterfall_dimension not in cohort_dimension_names
):
raise ValueError(
"Recipe option 'waterfall_small_multiples_dimension' references "
f"missing column: {waterfall_dimension}"
)
normalized["options"][
"waterfall_small_multiples_dimension"
] = waterfall_dimension
total_bridge_dimension = normalized["options"][
"total_by_dimension_bridge_dimension"
]
if total_bridge_dimension:
total_bridge_dimension = str(total_bridge_dimension)
if (
total_bridge_dimension not in columns
and total_bridge_dimension not in cohort_dimension_names
):
raise ValueError(
"Recipe option 'total_by_dimension_bridge_dimension' references "
f"missing column: {total_bridge_dimension}"
)
normalized["options"][
"total_by_dimension_bridge_dimension"
] = total_bridge_dimension
for option_key, role in (
("exploded_variance_bridge_parent_dimension", "parent"),
("exploded_variance_bridge_child_dimension", "child"),
):
configured_dimension = normalized["options"][option_key]
if not configured_dimension:
continue
configured_dimension = str(configured_dimension)
if (
configured_dimension not in columns
and configured_dimension not in cohort_dimension_names
):
raise ValueError(
f"Recipe option '{option_key}' references missing {role} "
f"column: {configured_dimension}"
)
normalized["options"][option_key] = configured_dimension
if (
normalized["options"].get("exploded_variance_bridge_parent_dimension")
and normalized["options"].get("exploded_variance_bridge_child_dimension")
and normalized["options"]["exploded_variance_bridge_parent_dimension"]
== normalized["options"]["exploded_variance_bridge_child_dimension"]
):
raise ValueError(
"Recipe options 'exploded_variance_bridge_parent_dimension' and "
"'exploded_variance_bridge_child_dimension' must differ."
)
if (
period_comparison_mode == PERIOD_MODE_ROLLING
or options.get("rolling_window_days") is not None
or options.get("rolling_window_months") is not None
):
rolling_window = rolling_window_length_options(
options, str(normalized_period_contract["period_grain"])
)
if "days" in rolling_window:
normalized["options"]["rolling_window_days"] = rolling_window["days"]
normalized["options"].pop("rolling_window_months", None)
else:
normalized["options"]["rolling_window_months"] = rolling_window["months"]
normalized["options"].pop("rolling_window_days", None)
if options.get("rolling_comparison") is not None:
normalized["options"]["rolling_comparison"] = str(
options.get("rolling_comparison")
)
if options.get("fiscal_start_month") is not None:
normalized["options"]["fiscal_start_month"] = normalized_period_contract[
"fiscal_start_month"
]
return normalized
def add_total_dimension(
df: pl.DataFrame, dimensions: list[str]
) -> tuple[pl.DataFrame, list[str]]:
"""Add a deterministic total dimension when no dimensions are mapped."""
if dimensions:
return df, dimensions
return df.with_columns(pl.lit("Total").alias(TOTAL_DIMENSION)), [TOTAL_DIMENSION]
def aggregate_period(
df: pl.DataFrame,
dimensions: list[str],
period_col: str,
period_value: str,
value_columns: list[str],
suffix: str,
) -> pl.DataFrame:
"""Aggregate one period snapshot and suffix value columns."""
filtered = df.filter(pl.col(period_col).cast(pl.Utf8) == str(period_value))
if filtered.is_empty():
raise ValueError(f"No rows found for period '{period_value}'.")
grouped = (
filtered.lazy()
.with_columns(
[
pl.col(column).fill_null("").cast(pl.Utf8).alias(column)
for column in dimensions
]
)
.group_by(dimensions)
.agg(
[
pl.col(column).sum().alias(f"{column}_{suffix}")
for column in value_columns
]
)
.collect()
)
return grouped
def joined_period_frame(
df: pl.DataFrame,
recipe: dict[str, Any],
dimensions: list[str],
) -> tuple[pl.DataFrame, dict[str, str]]:
"""Return a full outer joined baseline/comparison frame."""
mappings = recipe["mappings"]
period_col = mappings["period_column"]
baseline_period = str(mappings["baseline_period"])
comparison_period = str(mappings["comparison_period"])
amount_col = mappings["amount_column"]
optional_cols = [
mappings.get("units_column"),
mappings.get("discount_column"),
mappings.get("cogs_column"),
]
value_columns = [amount_col] + [column for column in optional_cols if column]
value_columns = list(dict.fromkeys(value_columns))
baseline = aggregate_period(
df, dimensions, period_col, baseline_period, value_columns, "baseline"
)
comparison = aggregate_period(
df, dimensions, period_col, comparison_period, value_columns, "comparison"
)
joined = baseline.join(comparison, on=dimensions, how="full", coalesce=True)
fill_columns = [
f"{column}_{suffix}"
for column in value_columns
for suffix in ("baseline", "comparison")
]
joined = joined.with_columns(
[pl.col(column).fill_null(0).alias(column) for column in fill_columns]
)
names = {
"amount0": f"{amount_col}_baseline",
"amount1": f"{amount_col}_comparison",
}
for key, source_key in (
("units", "units_column"),
("discount", "discount_column"),
("cogs", "cogs_column"),
):
column = mappings.get(source_key)
if column:
names[f"{key}0"] = f"{column}_baseline"
names[f"{key}1"] = f"{column}_comparison"
return joined, names
def optional_col(names: dict[str, str], key: str) -> pl.Expr:
"""Return an optional numeric expression, defaulting to zero."""
column = names.get(key)
if column:
return pl.col(column)
return pl.lit(0.0)
def safe_ratio(numerator: pl.Expr, denominator: pl.Expr) -> pl.Expr:
"""Return numerator / denominator, using zero when denominator is zero."""
return pl.when(denominator == 0).then(0.0).otherwise(numerator / denominator)
def ordered_unique(values: Iterable[str]) -> list[str]:
"""Return values once, preserving their first-seen order."""
return list(dict.fromkeys(values))
def calculation_grain_for(
recipe: dict[str, Any], report_dimensions: list[str]
) -> list[str]:
"""Return the bottom-up grain used for legacy-compatible variance math."""
mappings = recipe["mappings"]
grain = list(mappings.get("calculation_grain") or report_dimensions)
return ordered_unique([*report_dimensions, *grain])
def changed_mask(amount0: pl.Expr, amount1: pl.Expr) -> pl.Expr:
"""Return rows present in both periods for legacy price-volume-mix allocation."""
return (amount0 > 0) & (amount1 > 0)
def new_or_lost_contribution(
amount0: pl.Expr, amount1: pl.Expr, contribution: pl.Expr
) -> pl.Expr:
"""Put new/lost rows entirely into the volume-style contribution."""
return (
pl.when(amount0 == 0)
.then(contribution)
.when(amount1 == 0)
.then(contribution)
.otherwise(0.0)
)
def legacy_price_component(
amount0: pl.Expr, amount1: pl.Expr, units0: pl.Expr, units1: pl.Expr
) -> pl.Expr:
"""Legacy price effect from variance_formulas.calculate_residual_variance."""
price0 = safe_ratio(amount0, units0)
price1 = safe_ratio(amount1, units1)
units_change = units1 - units0
price_change = price1 - price0
return (
pl.when(changed_mask(amount0, amount1))
.then(price_change * (units0 + units_change / 2))
.otherwise(0.0)
)
def legacy_volume_component(
amount0: pl.Expr, amount1: pl.Expr, units0: pl.Expr, units1: pl.Expr
) -> pl.Expr:
"""Legacy volume effect with new/lost rows treated as volume variance."""
price0 = safe_ratio(amount0, units0)
price1 = safe_ratio(amount1, units1)
units_change = units1 - units0
price_change = price1 - price0
total_delta = amount1 - amount0
changed_volume = units_change * (price0 + price_change / 2)
return (
pl.when(changed_mask(amount0, amount1))
.then(changed_volume)
.otherwise(total_delta)
)
def numeric_sum_expressions(
df: pl.DataFrame, group_columns: list[str]
) -> list[pl.Expr]:
"""Return sum expressions for numeric non-grouping columns."""
return [
pl.col(column).sum().alias(column)
for column, dtype in df.schema.items()
if column not in group_columns and dtype.is_numeric()
]
def add_report_total_dimension(
df: pl.DataFrame, report_dimensions: list[str]
) -> tuple[pl.DataFrame, list[str]]:
"""Add a total reporting dimension when the user asks for total output."""
if report_dimensions:
return df, report_dimensions
return df.with_columns(pl.lit("Total").alias(TOTAL_DIMENSION)), [TOTAL_DIMENSION]
def calculate_variance_frame(
df: pl.DataFrame,
recipe: dict[str, Any],
dimensions: list[str],
) -> pl.DataFrame:
"""Calculate deterministic variance output using legacy bottom-up price-volume-mix."""
reporting_dimensions = list(dimensions)
leaf_dimensions = calculation_grain_for(recipe, reporting_dimensions)
working_df, grouping_dimensions = add_total_dimension(df, leaf_dimensions)
wide, names = joined_period_frame(working_df, recipe, grouping_dimensions)
amount0 = pl.col(names["amount0"])
amount1 = pl.col(names["amount1"])
units0 = optional_col(names, "units0")
units1 = optional_col(names, "units1")
discount0 = optional_col(names, "discount0")
discount1 = optional_col(names, "discount1")
cogs0 = optional_col(names, "cogs0")
cogs1 = optional_col(names, "cogs1")
has_units = "units0" in names and "units1" in names
has_discount = "discount0" in names and "discount1" in names
has_cogs = "cogs0" in names and "cogs1" in names
result = wide.with_columns(
(amount1 - amount0).alias("total_delta"),
pl.when(amount0 == 0)
.then(None)
.otherwise((amount1 - amount0) / amount0 * 100)
.alias("amount_pct_change"),
amount0.alias("_legacy_amount0"),
amount1.alias("_legacy_amount1"),
)
if has_units:
price0 = safe_ratio(amount0, units0)
price1 = safe_ratio(amount1, units1)
changed = changed_mask(amount0, amount1)
result = result.with_columns(
price0.alias("price_baseline"),
price1.alias("price_comparison"),
legacy_price_component(amount0, amount1, units0, units1).alias(
"price_variance"
),
legacy_volume_component(amount0, amount1, units0, units1).alias(
"volume_variance"
),
pl.when(changed)
.then(amount0)
.otherwise(0.0)
.alias("_legacy_changed_amount0"),
pl.when(changed)
.then(amount1)
.otherwise(0.0)
.alias("_legacy_changed_amount1"),
pl.when(changed)
.then(units0)
.otherwise(0.0)
.alias("_legacy_changed_units0"),
pl.when(changed)
.then(units1)
.otherwise(0.0)
.alias("_legacy_changed_units1"),
units0.alias("_legacy_units0"),
units1.alias("_legacy_units1"),
new_or_lost_contribution(amount0, amount1, amount1 - amount0).alias(
"_legacy_new_lost_volume"
),
)
result = result.with_columns(
pl.col("volume_variance").alias("_legacy_bottom_up_volume"),
pl.lit(0.0).alias("mix_variance"),
)
if has_discount:
net0 = amount0 - discount0
net1 = amount1 - discount1
result = result.with_columns(
net0.alias("net_baseline"),
net1.alias("net_comparison"),
(net1 - net0).alias("net_delta"),
pl.when(net0 == 0)
.then(None)
.otherwise((net1 - net0) / net0 * 100)
.alias("net_pct_change"),
(-(discount1 - discount0)).alias("discount_variance"),
discount0.alias("_legacy_discount0"),
discount1.alias("_legacy_discount1"),
)
if has_cogs:
margin0 = amount0 - discount0 - cogs0
margin1 = amount1 - discount1 - cogs1
result = result.with_columns(
(cogs1 - cogs0).alias("cogs_delta"),
(-(cogs1 - cogs0)).alias("cogs_variance"),
cogs0.alias("_legacy_cogs0"),
cogs1.alias("_legacy_cogs1"),
margin0.alias("margin_baseline"),
margin1.alias("margin_comparison"),
(margin1 - margin0).alias("margin_delta"),
pl.when(margin0 == 0)
.then(None)
.otherwise((margin1 - margin0) / margin0 * 100)
.alias("margin_pct_change"),
)
if has_units:
changed = changed_mask(amount0, amount1)
price0 = safe_ratio(amount0, units0)
price1 = safe_ratio(amount1, units1)
discount_per_unit0 = safe_ratio(discount0, units0)
discount_per_unit1 = safe_ratio(discount1, units1)
cogs_per_unit0 = safe_ratio(cogs0, units0)
cogs_per_unit1 = safe_ratio(cogs1, units1)
cost_per_unit0 = discount_per_unit0 + cogs_per_unit0
cost_per_unit1 = discount_per_unit1 + cogs_per_unit1
units_change = units1 - units0
price_change = price1 - price0
discount_change = discount_per_unit1 - discount_per_unit0
cogs_change = cogs_per_unit1 - cogs_per_unit0
cost_change = cost_per_unit1 - cost_per_unit0
changed_weight = units0 + units_change / 2
margin_volume = units_change * (
price0 - cost_per_unit0 + price_change / 2 - cost_change / 2
)
result = result.with_columns(
pl.when(changed)
.then(price_change * (units0 + units_change / 2))
.otherwise(0.0)
.alias("margin_price_variance"),
pl.when(changed)
.then(-discount_change * changed_weight)
.otherwise(0.0)
.alias("discount_variance"),
pl.when(changed)
.then(-cogs_change * changed_weight)
.otherwise(0.0)
.alias("cogs_variance"),
pl.when(changed)
.then(margin_volume)
.otherwise(margin1 - margin0)
.alias("margin_volume_variance"),
pl.when(changed)
.then(discount0)
.otherwise(0.0)
.alias("_legacy_changed_discount0"),
pl.when(changed)
.then(discount1)
.otherwise(0.0)
.alias("_legacy_changed_discount1"),
pl.when(changed)
.then(cogs0)
.otherwise(0.0)
.alias("_legacy_changed_cogs0"),
pl.when(changed)
.then(cogs1)
.otherwise(0.0)
.alias("_legacy_changed_cogs1"),
new_or_lost_contribution(amount0, amount1, margin1 - margin0).alias(
"_legacy_new_lost_margin_volume"
),
)
result = result.with_columns(
pl.col("margin_volume_variance").alias(
"_legacy_bottom_up_margin_volume"
),
pl.lit(0.0).alias("margin_mix_variance"),
)
result = result.with_columns(
(
pl.col("margin_delta")
- pl.col("margin_price_variance")
- pl.col("margin_volume_variance")
- pl.col("margin_mix_variance")
).alias("margin_component_reconciliation_delta")
)
result = aggregate_legacy_components(
result,
reporting_dimensions,
has_units=has_units,
has_discount=has_discount,
has_cogs=has_cogs,
)
if TOTAL_DIMENSION in result.schema:
result = result.rename({TOTAL_DIMENSION: "segment"})
internal_columns = [
column for column in result.columns if column.startswith("_legacy_")
]
if internal_columns:
result = result.drop(internal_columns)
return result.sort(
"total_delta",
descending=True,
)
def aggregate_legacy_components(
leaf: pl.DataFrame,
report_dimensions: list[str],
*,
has_units: bool,
has_discount: bool,
has_cogs: bool,
) -> pl.DataFrame:
"""Aggregate bottom-up leaf calculations and recompute pure mix effects."""
report_df, grouping_dimensions = add_report_total_dimension(leaf, report_dimensions)
grouped = (
report_df.lazy()
.group_by(grouping_dimensions)
.agg(numeric_sum_expressions(report_df, grouping_dimensions))
.collect()
)
amount0 = pl.col("_legacy_amount0")
amount1 = pl.col("_legacy_amount1")
grouped = grouped.with_columns(
pl.when(amount0 == 0)
.then(None)
.otherwise((amount1 - amount0) / amount0 * 100)
.alias("amount_pct_change")
)
if has_units:
changed_amount0 = pl.col("_legacy_changed_amount0")
changed_amount1 = pl.col("_legacy_changed_amount1")
changed_units0 = pl.col("_legacy_changed_units0")
changed_units1 = pl.col("_legacy_changed_units1")
pure_changed_volume = legacy_volume_component(
changed_amount0, changed_amount1, changed_units0, changed_units1
)
grouped = grouped.with_columns(
safe_ratio(amount0, pl.col("_legacy_units0")).alias("price_baseline"),
safe_ratio(amount1, pl.col("_legacy_units1")).alias("price_comparison"),
(pure_changed_volume + pl.col("_legacy_new_lost_volume")).alias(
"volume_variance"
),
)
grouped = grouped.with_columns(
(pl.col("_legacy_bottom_up_volume") - pl.col("volume_variance")).alias(
"mix_variance"
),
)
grouped = grouped.with_columns(
(
pl.col("total_delta")
- pl.col("price_variance")
- pl.col("volume_variance")
- pl.col("mix_variance")
).alias("component_reconciliation_delta"),
)
if has_discount:
net0 = amount0 - pl.col("_legacy_discount0")
net1 = amount1 - pl.col("_legacy_discount1")
if "discount_variance" in grouped.schema:
grouped = grouped.with_columns(
net0.alias("net_baseline"),
net1.alias("net_comparison"),
(net1 - net0).alias("net_delta"),
pl.when(net0 == 0)
.then(None)
.otherwise((net1 - net0) / net0 * 100)
.alias("net_pct_change"),
)
if has_cogs:
discount0 = (
pl.col("_legacy_discount0")
if "_legacy_discount0" in grouped.schema
else pl.lit(0.0)
)
discount1 = (
pl.col("_legacy_discount1")
if "_legacy_discount1" in grouped.schema
else pl.lit(0.0)
)
cogs0 = pl.col("_legacy_cogs0")
cogs1 = pl.col("_legacy_cogs1")
margin0 = amount0 - discount0 - cogs0
margin1 = amount1 - discount1 - cogs1
grouped = grouped.with_columns(
margin0.alias("margin_baseline"),
margin1.alias("margin_comparison"),
(margin1 - margin0).alias("margin_delta"),
pl.when(margin0 == 0)
.then(None)
.otherwise((margin1 - margin0) / margin0 * 100)
.alias("margin_pct_change"),
)
if has_units:
changed_amount0 = pl.col("_legacy_changed_amount0")
changed_amount1 = pl.col("_legacy_changed_amount1")
changed_units0 = pl.col("_legacy_changed_units0")
changed_units1 = pl.col("_legacy_changed_units1")
changed_discount0 = pl.col("_legacy_changed_discount0")
changed_discount1 = pl.col("_legacy_changed_discount1")
changed_cogs0 = pl.col("_legacy_changed_cogs0")
changed_cogs1 = pl.col("_legacy_changed_cogs1")
price0 = safe_ratio(changed_amount0, changed_units0)
price1 = safe_ratio(changed_amount1, changed_units1)
cost0 = safe_ratio(changed_discount0 + changed_cogs0, changed_units0)
cost1 = safe_ratio(changed_discount1 + changed_cogs1, changed_units1)
units_change = changed_units1 - changed_units0
price_change = price1 - price0
cost_change = cost1 - cost0
pure_margin_volume = units_change * (
price0 - cost0 + price_change / 2 - cost_change / 2
)
grouped = grouped.with_columns(
(pure_margin_volume + pl.col("_legacy_new_lost_margin_volume")).alias(
"margin_volume_variance"
)
)
grouped = grouped.with_columns(
(
pl.col("_legacy_bottom_up_margin_volume")
- pl.col("margin_volume_variance")
).alias("margin_mix_variance"),
)
margin_component_sum = (
pl.col("margin_price_variance")
+ pl.col("margin_volume_variance")
+ pl.col("margin_mix_variance")
)
if "discount_variance" in grouped.schema:
margin_component_sum = margin_component_sum + pl.col(
"discount_variance"
)
if "cogs_variance" in grouped.schema:
margin_component_sum = margin_component_sum + pl.col("cogs_variance")
grouped = grouped.with_columns(
(pl.col("margin_delta") - margin_component_sum).alias(
"margin_component_reconciliation_delta"
)
)
return grouped
def bridge_dimension_sets(dimensions: list[str], enabled: bool) -> list[list[str]]:
"""Return dimension combinations for root-cause variance output."""
if not enabled or len(dimensions) < 2:
return []
return [
list(combo)
for size in range(1, len(dimensions) + 1)
for combo in itertools.combinations(dimensions, size)
]
def write_outputs(
result: pl.DataFrame,
audit: dict[str, Any],
summary: str,
output_dir: Path,
*,
artifact_paths: list[str],
) -> None:
"""Write deterministic variance outputs."""
output_dir.mkdir(parents=True, exist_ok=True)
result.write_csv(output_dir / "variance_results.csv")
# Legacy component columns remain available for audit, but an amount-only
# workbook must not present them as measured price/volume/mix effects.
hidden_components = (
[
name
for name in (
"price_baseline",
"price_comparison",
"price_variance",
"volume_variance",
"mix_variance",
)
if name in result.schema
]
if not audit["recipe"]["mappings"]["units_column"]
else []
)
try:
result.write_excel(
output_dir / "variance_results.xlsx",
autofit=True,
freeze_panes=(1, 1),
hide_gridlines=True,
hidden_columns=hidden_components,
header_format={
"bold": True,
"font_color": "#FFFFFF",
"bg_color": "#183B56",
},
)
audit["outputs"]["variance_results.xlsx"] = "written"
except (ImportError, ModuleNotFoundError, OSError, ValueError) as exc:
audit["outputs"]["variance_results.xlsx"] = f"not_written: {exc}"
write_json(output_dir / "variance_audit.json", audit)
(output_dir / "variance_summary.md").write_text(summary, encoding="utf-8")
audit["outputs"]["variance_results.csv"] = "written"
audit["outputs"]["variance_audit.json"] = "written"
audit["outputs"]["variance_summary.md"] = "written"
for path in artifact_paths:
audit["outputs"][Path(path).name] = "written"
write_json(output_dir / "variance_audit.json", audit)
def _relative_artifact_path(path: Path, base: Path) -> str:
"""Return a stable artifact path relative to ``base`` when possible."""
try:
return path.relative_to(base).as_posix()
except ValueError:
return path.as_posix()
def _artifact_kind(path: Path) -> str:
"""Return the artifact group for a generated file."""
suffix = path.suffix.lower()
if suffix in {".png", ".html", ".htm"}:
return "charts"
if suffix in {".csv", ".xlsx"}:
return "tables"
if suffix == ".json":
return "contexts"
if suffix == ".md":
return "briefs"
if suffix == ".docx":
return "reports"
return "files"
def _chart_type_from_name(name: str) -> str:
"""Return a stable chart type from a generated PNG name."""
stem = Path(name).stem
if stem == "waterfall":
return "standard_variance_waterfall"
if stem == "waterfall_small_multiples":
return "standard_variance_small_multiples"
if stem == "total_by_dimension_bridge":
return "total_by_dimension_bridge"
if stem == "exploded_variance_bridge":
return "exploded_variance_bridge"
if stem == "pvm_decomposition_ladder":
return "pvm_decomposition_ladder"
if stem.startswith("root_cause_client_report_"):
return "client_report_chart"
if stem.startswith("root_cause_component_bridge"):
return "root_cause_component_bridge"
if stem.startswith("root_cause_total_bridge"):
return "root_cause_total_bridge"
if "drilldown" in stem:
return "root_cause_drilldown"
if stem.startswith("root_cause_bridge_alt_"):
return "root_cause_alternative_bridge"
if stem.startswith("root_cause_bridge"):
return "root_cause_bridge"
return "chart"
def _root_cause_bridge_variant(stem: str) -> str | None:
"""Return the formal root-cause bridge variant for a generated artifact."""
if stem.startswith("root_cause_bridge_alt_") and "drilldown" in stem:
return "alternative_drilldown"
if stem.startswith("root_cause_bridge_alt_"):
return "alternative_sequence"
if stem == "root_cause_component_bridge":
return "component_sequence"
if stem == "root_cause_total_bridge":
return "main_sequence"
if "drilldown" in stem:
return "drilldown"
if stem == "root_cause_bridge_moved_rows":
return "moved_rows"
if stem == "root_cause_bridge":
return "main_sequence"
return None
def _root_cause_alternative_result(stem: str) -> int | None:
"""Return the legacy alternative number embedded in a root-cause artifact."""
prefix = "root_cause_bridge_alt_"
if not stem.startswith(prefix):
return None
candidate = stem.removeprefix(prefix).split("_", 1)[0]
return int(candidate) if candidate.isdecimal() else None
def _root_cause_drilldown_row(stem: str) -> int | None:
"""Return the parent row number embedded in a root-cause drilldown artifact."""
marker = "_drilldown_row_"
if marker not in stem:
return None
candidate = stem.split(marker, 1)[1].split("_", 1)[0]
return int(candidate) if candidate.isdecimal() else None
def _root_cause_parent_artifact_id(stem: str) -> str | None:
"""Return the parent sequence artifact id for a root-cause drilldown."""
drilldown_row = _root_cause_drilldown_row(stem)
if drilldown_row is None:
return None
alternative_result = _root_cause_alternative_result(stem)
if alternative_result is not None:
return f"root_cause_bridge_alt_{alternative_result}"
if stem.startswith("root_cause_total_bridge_drilldown_row_"):
return "root_cause_total_bridge"
if stem.startswith("root_cause_bridge_drilldown_row_"):
return "root_cause_bridge"
return None
def _root_cause_bridge_frame_identity(source: Path) -> dict[str, Any]:
"""Return the analytical row sequence behind a root-cause bridge artifact."""
bridge_path = source.with_suffix(".csv")
if not bridge_path.exists():
return {}
try:
frame = _collect_csv_scan(bridge_path, separator=",")
except (OSError, pl.exceptions.PolarsError):
return {}
dimensions = _root_cause_dimension_columns(frame)
sequence: list[dict[str, Any]] = []
for index, row in enumerate(frame.to_dicts(), start=1):
filters = {
dimension: row.get(dimension)
for dimension in dimensions
if row.get(dimension) not in (None, "", "All")
}
sequence.append(
{
"row_number": index,
"bridge_level": row.get("bridge_level"),
"bridge_dimensions": row.get("bridge_dimensions"),
"filters": filters,
"variance_type": row.get("variance_type"),
"variance_amount": float(row.get("variance_amount") or 0.0),
}
)
variant = _root_cause_bridge_variant(source.stem)
alternative_result = _root_cause_alternative_result(source.stem)
drilldown_row = _root_cause_drilldown_row(source.stem)
identity = {
"root_cause_bridge_variant": _root_cause_bridge_variant(source.stem),
"alternative_result": alternative_result,
"drilldown_row": drilldown_row,
"root_cause_sequence": sequence,
}
parent_artifact_id = _root_cause_parent_artifact_id(source.stem)
if parent_artifact_id is not None:
identity["parent_artifact_id"] = parent_artifact_id
identity["parent_row_number"] = drilldown_row
if variant == "alternative_drilldown":
identity["child_artifact_type"] = "alternative_drilldown"
identity["parent_capability_id"] = "variance.root_cause_alternative_sweep"
identity["parent_child_type"] = "alternative_sequence"
identity["parent_alternative_result"] = alternative_result
return identity
def _total_by_dimension_bridge_frame_identity(source: Path) -> dict[str, Any]:
"""Return the analytical row sequence behind a total-by-dimension chart."""
bridge_path = source.with_suffix(".csv")
if not bridge_path.exists():
return {}
try:
frame = _collect_csv_scan(bridge_path, separator=",")
except (OSError, pl.exceptions.PolarsError):
return {}
if frame.is_empty() or "dimension_value" not in frame.columns:
return {}
first_dimension = (
str(frame["dimension"][0]) if "dimension" in frame.columns else None
)
sequence: list[dict[str, Any]] = []
for row in frame.to_dicts():
sequence.append(
{
"row_number": row.get("row_number"),
"dimension_value": row.get("dimension_value"),
"row_type": row.get("row_type"),
"amount_baseline": float(row.get("amount_baseline") or 0.0),
"amount_comparison": float(row.get("amount_comparison") or 0.0),
"total_delta": float(row.get("total_delta") or 0.0),
"percent_delta": (
float(row["percent_delta"])
if row.get("percent_delta") is not None
else None
),
}
)
identity: dict[str, Any] = {
"total_by_dimension_bridge_dimension": first_dimension,
"total_by_dimension_sequence": sequence,
}
return identity
def _exploded_variance_bridge_identity(source: Path) -> dict[str, Any]:
"""Return the native parent/child bridge identity from its spec sidecar."""
spec_path = source.with_name("exploded_variance_bridge_spec.json")
if not spec_path.exists():
return {}
try:
payload = load_json(spec_path)
except (OSError, json.JSONDecodeError):
return {}
selection = payload.get("selection") or {}
parent = payload.get("parent") or {}
children = payload.get("children") or []
return {
"exploded_variance_bridge_parent_dimension": selection.get("parent_dimension"),
"exploded_variance_bridge_child_dimension": selection.get("child_dimension"),
"exploded_variance_bridge_max_drilldowns": selection.get("max_drilldowns"),
"exploded_variance_bridge_parent_rows": parent.get("rows") or [],
"exploded_variance_bridge_drilldowns": [
{
"drilldown_id": child.get("drilldown_id"),
"parent_row_number": child.get("parent_row_number"),
"parent_dimension_value": child.get("parent_dimension_value"),
"child_dimension": child.get("child_dimension"),
"rows": child.get("rows") or [],
}
for child in children
],
}
def _movement_driver_dimension(
chart_type: str,
recipe: dict[str, Any],
) -> str | list[str] | None:
"""Return the deterministic driver dimension represented by a variance chart."""
mappings = recipe.get("mappings") or {}
options = recipe.get("options") or {}
if chart_type == "standard_variance_waterfall":
return "variance_component"
if chart_type == "standard_variance_small_multiples":
return options.get("waterfall_small_multiples_dimension")
if chart_type == "total_by_dimension_bridge":
return options.get("total_by_dimension_bridge_dimension")
if chart_type == "exploded_variance_bridge":
return [
value
for value in [
options.get("exploded_variance_bridge_parent_dimension"),
options.get("exploded_variance_bridge_child_dimension"),
]
if value
]
if chart_type == "pvm_decomposition_ladder":
return "pvm_component"
if chart_type.startswith("root_cause"):
dimensions = mappings.get("dimensions") or []
return [str(value) for value in dimensions] if dimensions else "root_cause"
return None
def _normalize_artifact_mode(artifact_mode: str) -> str:
"""Return a supported artifact mode or raise for invalid contract input."""
normalized = str(artifact_mode or ARTIFACT_MODE_DATA_AND_RENDER).strip().lower()
if normalized not in ARTIFACT_MODES:
allowed = ", ".join(sorted(ARTIFACT_MODES))
raise ValueError(f"Unsupported artifact_mode {artifact_mode!r}; use {allowed}.")
return normalized
def _variance_data_chart_enabled(
chart_type: str,
available_chart_types: Sequence[str],
) -> bool:
return chart_type in set(available_chart_types)
def _variance_chart_selected(
chart_type: str,
artifact_mode: str,
available_chart_types: Sequence[str],
) -> bool:
if chart_type not in set(available_chart_types):
return False
return artifact_mode != ARTIFACT_MODE_DATA_ONLY
def _data_written_chart_audit(
*,
chart_type: str,
artifact_name: str,
source_functions: Sequence[str],
) -> dict[str, Any]:
return {
"status": "data_written",
"artifact": artifact_name,
"rendered": False,
"source_functions": list(source_functions),
"chart_type": chart_type,
}
def _skipped_chart_audit(chart_type: str, reason: str) -> dict[str, Any]:
return {
"status": "skipped",
"chart_type": chart_type,
"reason": reason,
}
def _comparison_payload(recipe: dict[str, Any]) -> dict[str, Any]:
"""Return comparison metadata for reporting agents."""
mappings = recipe.get("mappings") or {}
options = recipe.get("options") or {}
return {
"baseline": mappings.get("baseline_period"),
"comparison": mappings.get("comparison_period"),
"basis": options.get("comparison_basis"),
"mode": options.get("period_comparison_mode"),
"period_window": options.get("period_window"),
}
def _title_contract_payload(
recipe: dict[str, Any],
*,
chart_kind: str,
dimension: str | None = None,
) -> dict[str, Any]:
"""Return the visible three-row title contract for a variance chart."""
title = build_ibcs_title(recipe, chart_kind=chart_kind, dimension=dimension)
return {
"chart_title_lines": title.lines(),
"title_contract": {
"who": title.who,
"what": title.what,
"when": title.when,
},
}
def _standard_variance_context(
result: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
) -> dict[str, Any]:
"""Return data context behind the standard variance chart."""
components = [
{
"variance_type": label,
"variance_amount": _numeric_column_sum(result, column),
"source_column": column,
"is_residual_other": False,
}
for label, column in standard_variance_component_columns(recipe)
]
total_delta = _numeric_column_sum(result, "total_delta")
component_sum = sum(float(item["variance_amount"] or 0.0) for item in components)
residual = total_delta - component_sum
components.append(
{
"variance_type": "Other",
"variance_amount": residual,
"source_column": "component_reconciliation_delta",
"is_residual_other": True,
}
)
dominant = _dominant_component(components)
mappings = recipe.get("mappings") or {}
return {
"schema_version": SCHEMA_VERSION,
"chart_id": "standard_variance_waterfall",
"chart_family": "variance_analysis",
"chart_type": "standard_variance_waterfall",
"chart_path": (
"charts/waterfall.png" if (output_dir / "waterfall.png").exists() else None
),
"metric": mappings.get("amount_column"),
"pvm_available": bool(mappings.get("units_column")),
"unit": (recipe.get("options") or {}).get("currency") or "EUR",
"comparison": _comparison_payload(recipe),
"recipe_filters": (recipe.get("options") or {}).get("recipe_filter_audit"),
"recipe_cohorts": (recipe.get("options") or {}).get("recipe_cohort_audit"),
"dimensions": mappings.get("dimensions") or [],
"calculation_grain": mappings.get("calculation_grain") or [],
**_title_contract_payload(recipe, chart_kind="standard_variance"),
"totals": {
"amount_baseline": _numeric_column_sum(result, "amount_baseline"),
"amount_comparison": _numeric_column_sum(result, "amount_comparison"),
"total_delta": total_delta,
"component_sum": component_sum,
"other_residual": residual,
},
"components": components,
"dominant_component": {
"variance_type": dominant.get("variance_type"),
"variance_amount": dominant.get("variance_amount"),
},
"top_rows_by_abs_delta": top_driver_rows(result, limit=10),
"codex_interpretation_contract": {
"must_review_when_written": True,
"required_points": [
"Explain the dominant standard variance component.",
"State whether Other/residual is material.",
"Use small-multiples context when available before claiming concentration or spread.",
"Distinguish deterministic variance facts from business interpretation.",
],
},
}
def _pvm_ladder_levels(
result: pl.DataFrame,
recipe: dict[str, Any],
) -> list[dict[str, Any]]:
"""Return the structured chart data behind the PVM decomposition ladder."""
total_delta = _numeric_column_sum(result, "total_delta")
price = _numeric_column_sum(result, "price_variance")
volume = _numeric_column_sum(result, "volume_variance")
mix = _numeric_column_sum(result, "mix_variance")
residual = total_delta - price - volume - mix
unit_label = "Units" if recipe["mappings"].get("units_column") else "Volume"
return [
{
"level": 1,
"calculation": f"Price & {unit_label.lower()} & mix",
"components": [
{
"variance_type": f"Price & {unit_label.lower()} & mix",
"variance_amount": total_delta,
"source_columns": ["total_delta"],
"is_residual_other": False,
}
],
},
{
"level": 2,
"calculation": f"Price, {unit_label.lower()} & mix",
"components": [
{
"variance_type": "Price",
"variance_amount": price,
"source_columns": ["price_variance"],
"is_residual_other": False,
},
{
"variance_type": f"{unit_label} & mix",
"variance_amount": total_delta - price,
"source_columns": [
"volume_variance",
"mix_variance",
"component_reconciliation_delta",
],
"is_residual_other": False,
},
],
},
{
"level": 3,
"calculation": f"Price, {unit_label.lower()}, mix",
"components": [
{
"variance_type": "Price",
"variance_amount": price,
"source_columns": ["price_variance"],
"is_residual_other": False,
},
{
"variance_type": unit_label,
"variance_amount": volume,
"source_columns": ["volume_variance"],
"is_residual_other": False,
},
{
"variance_type": "Mix",
"variance_amount": mix,
"source_columns": ["mix_variance"],
"is_residual_other": False,
},
{
"variance_type": "Other",
"variance_amount": residual,
"source_columns": ["component_reconciliation_delta"],
"is_residual_other": True,
},
],
},
]
def _pvm_decomposition_ladder_context(
result: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
) -> dict[str, Any]:
"""Return model-facing context for the PVM decomposition ladder."""
levels = _pvm_ladder_levels(result, recipe)
totals = {
"amount_baseline": _numeric_column_sum(result, "amount_baseline"),
"amount_comparison": _numeric_column_sum(result, "amount_comparison"),
"total_delta": _numeric_column_sum(result, "total_delta"),
"price_variance": _numeric_column_sum(result, "price_variance"),
"volume_variance": _numeric_column_sum(result, "volume_variance"),
"mix_variance": _numeric_column_sum(result, "mix_variance"),
"component_reconciliation_delta": _numeric_column_sum(
result, "component_reconciliation_delta"
),
}
baseline_total = float(totals["amount_baseline"] or 0.0)
totals["delta_percent"] = (
(float(totals["total_delta"] or 0.0) / baseline_total) * 100
if abs(baseline_total) > TOLERANCE
else None
)
return {
"schema_version": SCHEMA_VERSION,
"chart_id": "pvm_decomposition_ladder",
"chart_family": "variance_analysis",
"chart_type": "pvm_decomposition_ladder",
"chart_path": (
"charts/pvm_decomposition_ladder.png"
if (output_dir / "pvm_decomposition_ladder.png").exists()
else None
),
"metric": (recipe.get("mappings") or {}).get("amount_column"),
"unit": (recipe.get("options") or {}).get("currency") or "EUR",
"comparison": _comparison_payload(recipe),
**_title_contract_payload(recipe, chart_kind="pvm_decomposition_ladder"),
"levels": levels,
"totals": totals,
"codex_interpretation_contract": {
"must_review_when_written": True,
"purpose": (
"Compare the same total variance under progressively richer "
"Price / Units / Mix decompositions before deciding which "
"breakdown is useful for the client story."
),
"required_points": [
"State whether the one-line total variance hides offsetting components.",
"Explain what changes when Price is separated from Units & Mix.",
"Explain what changes when Units and Mix are separated.",
"Mention Other only when the reconciliation residual is material.",
],
},
}
def write_pvm_decomposition_ladder_chart_data(
result: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
) -> tuple[list[str], dict[str, Any], str]:
"""Write CSV and JSON data for the PVM decomposition ladder."""
if not bool(recipe.get("options", {}).get("pvm_decomposition_ladder", True)):
return [], {"status": "disabled"}, ""
if not recipe.get("mappings", {}).get("units_column"):
return (
[],
{
"status": "not_written_missing_units",
"reason": "PVM decomposition ladder requires a units column.",
},
"",
)
context = _pvm_decomposition_ladder_context(result, recipe, output_dir)
context_path = output_dir / "pvm_decomposition_ladder_context.json"
write_json(context_path, context)
rows = []
for level in context["levels"]:
for component in level["components"]:
rows.append(
{
"level": level["level"],
"calculation": level["calculation"],
"variance_type": component["variance_type"],
"variance_amount": component["variance_amount"],
"is_residual_other": component["is_residual_other"],
"source_columns": " | ".join(component["source_columns"]),
}
)
table_path = output_dir / "pvm_decomposition_ladder.csv"
pl.DataFrame(rows).write_csv(table_path)
spanish = _output_language(recipe.get("language")) == "es"
summary = [
"",
(
"## Escalera de descomposición PVM"
if spanish
else "## PVM Decomposition Ladder"
),
"",
(
"Este gráfico compara la misma variación total en tres niveles de cálculo: Precio/Unidades/Mix combinados, Precio más Unidades y Mix, y Precio/Unidades/Mix por separado."
if spanish
else "This chart compares the same total variance at three calculation depths: combined Price/Units/Mix, Price plus Units & Mix, and Price/Units/Mix separately."
),
"",
]
for level in context["levels"]:
components = ", ".join(
f"{item['variance_type']} `{float(item['variance_amount']):,.2f}`"
for item in level["components"]
if abs(float(item["variance_amount"] or 0.0)) > TOLERANCE
)
summary.append(
f"- {'Nivel' if spanish else 'Level'} {level['level']} "
f"`{level['calculation']}`: {components}"
)
return (
[str(table_path), str(context_path)],
{
"status": "written",
"table": table_path.name,
"context": context_path.name,
"level_count": len(context["levels"]),
"component_row_count": len(rows),
},
"\n".join(summary) + "\n",
)
def _load_optional_json(path: Path) -> dict[str, Any] | None:
"""Load a JSON file when present and valid."""
if not path.exists():
return None
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
return payload if isinstance(payload, dict) else None
def max_abs(df: pl.DataFrame, column: str) -> float | None:
"""Return the maximum absolute value for a column when it exists."""
if column not in df.schema or df.is_empty():
return None
value = df.select(pl.col(column).abs().max()).item()
return float(value) if value is not None else None
def build_audit(
input_path: Path,
recipe: dict[str, Any],
result: pl.DataFrame,
artifact_paths: list[str],
*,
client_engagement: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build audit metadata for a variance run."""
return {
"schema_version": SCHEMA_VERSION,
"generated_at": utc_now(),
"source_file": _source_reference(input_path, client_engagement),
"row_count": result.height,
"recipe": recipe,
"accounting_readiness": recipe.get("accounting_readiness"),
"checks": {
"max_abs_component_reconciliation_delta": max_abs(
result, "component_reconciliation_delta"
),
"max_abs_margin_reconciliation_delta": max_abs(
result, "margin_mix_variance"
),
"max_abs_margin_component_reconciliation_delta": max_abs(
result, "margin_component_reconciliation_delta"
),
},
"outputs": {Path(path).name: "written" for path in artifact_paths},
}
def top_driver_rows(result: pl.DataFrame, limit: int = 10) -> list[dict[str, Any]]:
"""Return top variance rows by absolute total delta."""
if result.is_empty():
return []
return (
result.with_columns(pl.col("total_delta").abs().alias("_abs_delta"))
.sort("_abs_delta", descending=True)
.drop("_abs_delta")
.head(limit)
.to_dicts()
)
def build_summary_markdown(result: pl.DataFrame, recipe: dict[str, Any]) -> str:
"""Build deterministic markdown for Codex to interpret."""
mappings = recipe["mappings"]
options = recipe.get("options") or {}
spanish = _output_language(recipe.get("language")) == "es"
labels = (
{
"title": "Datos fuente del análisis de variaciones",
"comparison_basis": "Base de comparación",
"period_mode": "Modo de comparación de periodos",
"baseline": "Periodo base",
"comparison": "Periodo de comparación",
"amount": "Columna de importe",
"dimensions": "Dimensiones",
"grain": "Nivel de cálculo",
"rows": "Filas de resultados",
"totals": "Totales",
"date_column": "Columna de fecha",
"baseline_window": "Ventana de fechas base",
"comparison_window": "Ventana de fechas de comparación",
"drivers": "Principales factores por valor absoluto",
"total": "Total",
"boundary": "Límite de interpretación",
"boundary_text": "Las cifras anteriores son cálculos deterministas. Codex debe utilizarlas como datos fuente, identificar los factores principales, señalar asignaciones ausentes o salvedades por denominadores cero y no modificar silenciosamente los resultados deterministas.",
"to": "a",
"unknown": "desconocido",
}
if spanish
else {
"title": "Variance Analysis Source Data",
"comparison_basis": "Comparison basis",
"period_mode": "Period comparison mode",
"baseline": "Baseline period",
"comparison": "Comparison period",
"amount": "Amount column",
"dimensions": "Dimensions",
"grain": "Calculation grain",
"rows": "Result rows",
"totals": "Totals",
"date_column": "Date column",
"baseline_window": "Baseline date window",
"comparison_window": "Comparison date window",
"drivers": "Largest Absolute Drivers",
"total": "Total",
"boundary": "Interpretation Boundary",
"boundary_text": "The figures above are deterministic calculations. Codex should use them as source data, identify the largest drivers, call out missing mappings or zero-denominator caveats, and avoid changing deterministic results silently.",
"to": "to",
"unknown": "unknown",
}
)
lines = [
f"# {labels['title']}",
"",
f"- {labels['comparison_basis']}: `{options.get('comparison_basis') or labels['unknown']}`",
f"- {labels['period_mode']}: `{options.get('period_comparison_mode') or labels['unknown']}`",
f"- {labels['baseline']}: `{mappings['baseline_period']}`",
f"- {labels['comparison']}: `{mappings['comparison_period']}`",
f"- {labels['amount']}: `{mappings['amount_column']}`",
f"- {labels['dimensions']}: `{', '.join(mappings.get('dimensions') or [labels['total']])}`",
f"- {labels['grain']}: `{', '.join(mappings.get('calculation_grain') or mappings.get('dimensions') or [labels['total']])}`",
f"- {labels['rows']}: `{result.height}`",
"",
f"## {labels['totals']}",
"",
]
period_window = options.get("period_window") or {}
if period_window:
baseline = period_window.get("baseline", {})
comparison = period_window.get("comparison", {})
lines[7:7] = [
f"- {labels['date_column']}: `{period_window.get('date_column')}`",
(
f"- {labels['baseline_window']}: "
f"`{baseline.get('start_date')}` {labels['to']} `{baseline.get('end_date')}`"
),
(
f"- {labels['comparison_window']}: "
f"`{comparison.get('start_date')}` {labels['to']} `{comparison.get('end_date')}`"
),
]
totals = result.select(
[
pl.col("total_delta").sum().alias("total_delta"),
*[
pl.col(column).sum().alias(column)
for column in (
"price_variance",
"volume_variance",
"mix_variance",
"net_delta",
"margin_delta",
"cogs_delta",
)
if column in result.schema
],
]
).to_dicts()[0]
for key, value in totals.items():
lines.append(
f"- `{key}`: {value:,.2f}"
if isinstance(value, (int, float))
else f"- `{key}`: {value}"
)
lines.extend(["", f"## {labels['drivers']}", ""])
for row in top_driver_rows(result, 10):
dims = [
f"{dimension}={row.get(dimension)}"
for dimension in mappings.get("dimensions") or ["segment"]
if dimension in row
]
label = ", ".join(dims) if dims else labels["total"]
delta = row.get("total_delta")
lines.append(f"- {label}: total_delta={float(delta):,.2f}")
lines.extend(
[
"",
f"## {labels['boundary']}",
"",
labels["boundary_text"],
]
)
return "\n".join(lines) + "\n"
def append_bridge_dimension_summary(
summary: str,
bridge_audit: dict[str, Any] | None,
*,
language: str = "en",
) -> str:
"""Append requested-versus-emitted root-cause dimensions to the summary."""
if not bridge_audit or not bridge_audit.get("enabled"):
return summary
spanish = _output_language(language) == "es"
def formatted(values: Any) -> str:
if not values:
return "`ninguna`" if spanish else "`none`"
return "`" + ", ".join(values) + "`"
lines = [
"",
(
"## Análisis de las causas de la variación"
if spanish
else "## Root-Cause Variance Analysis"
),
"",
(
"- Dimensiones solicitadas del informe: "
if spanish
else "- Requested report dimensions: "
)
+ formatted(bridge_audit.get("requested_report_dimensions")),
(
"- Nivel de cálculo solicitado: "
if spanish
else "- Requested calculation grain: "
)
+ formatted(bridge_audit.get("requested_calculation_grain")),
(
"- Dimensiones efectivas de causa: "
if spanish
else "- Effective root-cause dimensions: "
)
+ formatted(bridge_audit.get("effective_bridge_dimensions")),
(
"- Dimensiones de causa generadas: "
if spanish
else "- Emitted root-cause dimensions: "
)
+ formatted(bridge_audit.get("emitted_bridge_dimensions")),
(
"- Dimensiones de causa descartadas: "
if spanish
else "- Dropped root-cause dimensions: "
)
+ formatted(bridge_audit.get("dropped_bridge_dimensions")),
(
"- Dimensiones de causa añadidas internamente: "
if spanish
else "- Internally added root-cause dimensions: "
)
+ formatted(bridge_audit.get("internally_added_bridge_dimensions")),
]
return summary.rstrip() + "\n" + "\n".join(lines) + "\n"
def _root_cause_dimension_columns(frame: pl.DataFrame) -> list[str]:
"""Return active dimension columns in a normalized root-cause bridge frame."""
return [
column
for column in frame.columns
if column not in ROOT_CAUSE_BRIDGE_MEASURE_COLUMNS
]
def _root_cause_row_label(row: dict[str, Any], dimensions: list[str]) -> str:
"""Return a compact business label for a selected root-cause row."""
inactive_values = {"", "All", "N/A", "None", "null"}
values = [
str(row.get(dimension))
for dimension in dimensions
if row.get(dimension) is not None
and str(row.get(dimension)) not in inactive_values
]
label = " / ".join(values) if values else "Total"
variance_type = row.get("variance_type")
if variance_type is not None and str(variance_type):
label = f"{label} - {variance_type}"
return label
def _root_cause_total_delta(result: pl.DataFrame) -> float:
"""Return the overall comparison minus baseline delta for bridge residuals."""
if result.is_empty():
return 0.0
values = result.select(
(pl.col("amount_comparison").sum() - pl.col("amount_baseline").sum()).alias(
"total_delta"
)
).to_dicts()[0]
return float(values["total_delta"] or 0.0)
def _root_cause_selected_sum(frame: pl.DataFrame) -> float:
"""Return the selected legacy sequence variance sum."""
if frame.is_empty() or "variance_amount" not in frame.schema:
return 0.0
value = frame.select(pl.col("variance_amount").sum()).item()
return float(value or 0.0)
def _root_cause_sequence_payload(
*,
alternative_result: int,
frame: pl.DataFrame,
audit: dict[str, Any],
result: pl.DataFrame,
bridge_path: Path | None,
chart_path: Path | None,
) -> dict[str, Any]:
"""Return model-facing metadata for one root-cause alternative."""
dimensions = _root_cause_dimension_columns(frame)
rows = [
{
"row_number": index + 1,
"label": _root_cause_row_label(row, dimensions),
"variance_amount": float(row.get("variance_amount") or 0.0),
"bridge_dimensions": row.get("bridge_dimensions"),
"variance_type": row.get("variance_type"),
}
for index, row in enumerate(frame.to_dicts())
]
drilldown_row_numbers: list[int] = []
for audit_key in ("drilldown_requested_rows", "automatic_drilldown_rows"):
for value in audit.get(audit_key, []) or []:
try:
row_number = int(value)
except (TypeError, ValueError):
continue
if row_number > 0 and row_number not in drilldown_row_numbers:
drilldown_row_numbers.append(row_number)
child_artifacts = []
for row_number in drilldown_row_numbers:
parent_row = rows[row_number - 1] if row_number <= len(rows) else {}
child_artifacts.append(
{
"child_type": "alternative_drilldown",
"capability_id": "variance.root_cause_alternative_sweep",
"parent_capability_id": "variance.root_cause_alternative_sweep",
"parent_child_type": "alternative_sequence",
"parent_artifact_id": f"root_cause_bridge_alt_{alternative_result}",
"alternative_result": alternative_result,
"drilldown_row": row_number,
"parent_row_number": row_number,
"parent_row_label": parent_row.get("label", ""),
"parent_row_bridge_dimensions": parent_row.get("bridge_dimensions"),
"parent_row_variance_type": parent_row.get("variance_type"),
"parent_row_variance_amount": parent_row.get("variance_amount"),
"artifact_id": (
f"root_cause_bridge_alt_{alternative_result}"
f"_drilldown_row_{row_number}"
),
"chart_artifact": (
f"root_cause_bridge_alt_{alternative_result}"
f"_drilldown_row_{row_number}.png"
),
"table_artifact": (
f"root_cause_bridge_alt_{alternative_result}"
f"_drilldown_row_{row_number}.csv"
),
}
)
total_delta = _root_cause_total_delta(result)
selected_sum = _root_cause_selected_sum(frame)
return {
"alternative_result": alternative_result,
"row_count": frame.height,
"selected_rows": rows,
"selected_sequence_bridge_dimensions": audit.get(
"selected_sequence_bridge_dimensions", []
),
"selected_sequence_unique_bridge_dimensions": audit.get(
"selected_sequence_unique_bridge_dimensions", []
),
"selected_sequence_has_mixed_dimensions": bool(
audit.get("selected_sequence_has_mixed_dimensions", False)
),
"total_delta": total_delta,
"selected_sequence_sum": selected_sum,
"other_residual": total_delta - selected_sum,
"drilldown_status": audit.get("drilldown_status"),
"drilldown_requested_rows": audit.get("drilldown_requested_rows", []),
"automatic_drilldown_mode": audit.get("automatic_drilldown_mode", "none"),
"automatic_drilldown_rows": audit.get("automatic_drilldown_rows", []),
"child_artifacts": child_artifacts,
"bridge_path": str(bridge_path) if bridge_path else "",
"chart_path": str(chart_path) if chart_path else "",
}
def _root_cause_resolved_parameters(
recipe: dict[str, Any],
audit: dict[str, Any],
) -> dict[str, Any]:
"""Return the formal request parameters resolved by the root-cause runtime."""
mappings = recipe.get("mappings") or {}
options = recipe.get("options") or {}
return {
key: json_safe(value)
for key, value in {
"metric": mappings.get("amount_column"),
"units_metric": mappings.get("units_column"),
"comparison_basis": options.get("comparison_basis"),
"baseline_period": mappings.get("baseline_period"),
"comparison_period": mappings.get("comparison_period"),
"period_column": mappings.get("period_column"),
"date_column": mappings.get("date_column"),
"dimensions": mappings.get("dimensions") or [],
"calculation_grain": mappings.get("calculation_grain") or [],
"effective_bridge_dimensions": audit.get("effective_bridge_dimensions"),
"emitted_bridge_dimensions": audit.get("emitted_bridge_dimensions"),
"dropped_bridge_dimensions": audit.get("dropped_bridge_dimensions"),
"alternative_result": audit.get("alternative_result"),
"filters": options.get("recipe_filter_audit")
or recipe.get("filters")
or options.get("filters"),
"population_preparation": options.get("recipe_cohort_audit")
or options.get("cohort_definition")
or recipe.get("cohorts")
or options.get("cohorts"),
}.items()
if value is not None and value != "" and value != [] and value != {}
}
def _write_root_cause_bridge_context(
*,
frame: pl.DataFrame,
result: pl.DataFrame,
recipe: dict[str, Any],
audit: dict[str, Any],
output_dir: Path,
bridge_path: Path,
chart_path: Path | None,
context_name: str = "root_cause_total_bridge_context.json",
analysis_type: str = "root_cause_total_bridge",
capability_id: str = "variance.root_cause_total_bridge",
chart_type: str = "root_cause_total_bridge",
) -> Path:
"""Write the structured payload behind the main root-cause bridge."""
variance_mode = str(audit.get("root_cause_variance_mode") or "total_variance")
if variance_mode == "component_variance":
deterministic_boundary = (
"The plugin selected and reconciled component-by-dimension rows. "
"Codex may interpret whether the component bridge is useful, but "
"must not change calculated rows, component types, amounts, "
"dimensions, or residuals."
)
required_points = [
"Treat this as second-order diagnostic source data.",
"Read each row as a variance component for a selected slice.",
"Do not describe component rows as having meaningful initial/final row bars.",
"Do not invent causes outside the generated component rows.",
]
else:
deterministic_boundary = (
"The plugin selected and reconciled total-variance root-cause rows. "
"Codex may interpret whether the bridge is useful, but must not "
"change calculated rows, amounts, dimensions, or residuals."
)
required_points = [
"State that each selected row is total variance for that slice.",
"Use baseline and comparison row values where they are present.",
"Call out when a selected sequence mixes dimensions.",
"Use generated drilldown artifacts only when they exist.",
"Do not invent causes outside the generated bridge rows.",
]
payload = {
"schema_version": SCHEMA_VERSION,
"analysis_type": analysis_type,
"capability_id": capability_id,
"chart_family": "variance_analysis",
"chart_type": chart_type,
"root_cause_variance_mode": variance_mode,
"metric": (recipe.get("mappings") or {}).get("amount_column"),
"comparison": _comparison_payload(recipe),
"resolved_parameters": _root_cause_resolved_parameters(recipe, audit),
"main_sequence": _root_cause_sequence_payload(
alternative_result=int(audit.get("alternative_result") or 1),
frame=frame,
audit=audit,
result=result,
bridge_path=bridge_path,
chart_path=chart_path,
),
"runtime_audit": {
"variable_bridge_source": audit.get("variable_bridge_source"),
"legacy_processing_choice": audit.get("legacy_processing_choice"),
"legacy_variance_aggregation": audit.get("legacy_variance_aggregation"),
"plugin_variance_aggregation": audit.get("plugin_variance_aggregation"),
"root_cause_variance_mode": audit.get("root_cause_variance_mode"),
"selected_sequence_has_mixed_dimensions": audit.get(
"selected_sequence_has_mixed_dimensions"
),
"selected_sequence_unique_bridge_dimensions": audit.get(
"selected_sequence_unique_bridge_dimensions"
),
"drilldown_status": audit.get("drilldown_status"),
"moved_rows_status": audit.get("moved_rows_status"),
"alternative_sweep_enabled": audit.get("alternative_sweep_enabled"),
"alternative_sweep_values": audit.get("alternative_sweep_values"),
},
"codex_interpretation_contract": {
"must_review_when_written": True,
"deterministic_boundary": deterministic_boundary,
"required_points": required_points,
},
}
context_path = output_dir / context_name
write_json(context_path, payload)
return context_path
def _root_cause_sweep_interpretation_contract() -> dict[str, Any]:
"""Return the model-facing contract for root-cause sweep interpretation."""
return {
"codex_output_file": "codex_root_cause_sweep_analysis.md",
"deterministic_boundary": (
"The plugin has already selected and reconciled the alternatives. "
"Codex may interpret, compare, and recommend drilldowns, but must "
"not change calculated rows, amounts, dimensions, or residuals."
),
"required_interpretation_points": [
"Separate deterministic facts from business interpretation.",
"Identify which alternatives are genuinely mixed-dimension.",
"Explain that each row is residual after earlier selected rows.",
"Rank alternatives as useful, possibly useful, or misleading/noisy.",
"Recommend drilldowns only from generated rows and detail outputs.",
"State caveats where labels, currency assumptions, units, or hierarchy meaning are missing.",
],
"forbidden_claims": [
"Do not describe same-dimension alternatives as mixed variable-dimension bridges.",
"Do not interpret a later mixed row as the total for that dimension.",
"Do not fabricate rows or causes that are not in the generated files.",
],
}
def _format_root_cause_brief_amount(value: Any) -> str:
"""Format a root-cause amount for a markdown interpretation brief."""
try:
return f"{float(value):,.1f}"
except (TypeError, ValueError):
return str(value)
def _root_cause_sweep_interpretation_brief(
*,
context_payload: dict[str, Any],
summary_rows: list[dict[str, Any]],
) -> str:
"""Return a Codex-facing markdown brief for root-cause sweep analysis."""
contract = context_payload["codex_interpretation_contract"]
spanish = _output_language(context_payload.get("language")) == "es"
required_points = (
[
"Separe los hechos deterministas de la interpretación empresarial.",
"Identifique qué alternativas combinan realmente varias dimensiones.",
"Explique que cada fila es el residual después de las filas seleccionadas anteriormente.",
"Clasifique las alternativas como útiles, posiblemente útiles o engañosas/ruidosas.",
"Recomiende desgloses solo a partir de las filas y los detalles generados.",
"Indique las salvedades cuando falten etiquetas, supuestos de moneda, unidades o el significado de la jerarquía.",
]
if spanish
else contract["required_interpretation_points"]
)
forbidden_claims = (
[
"No describa alternativas de una sola dimensión como puentes de dimensiones variables.",
"No interprete una fila mixta posterior como el total de esa dimensión.",
"No invente filas ni causas que no figuren en los archivos generados.",
]
if spanish
else contract["forbidden_claims"]
)
lines = [
(
"# Guía de Codex para interpretar el barrido de causas"
if spanish
else "# Codex Root-Cause Sweep Interpretation Brief"
),
"",
(
"Este archivo sirve de entrega al modelo. Las cifras son deterministas; Codex debe realizar la interpretación narrativa."
if spanish
else "This file is a model-facing handoff. The figures are deterministic; the narrative interpretation is Codex's job."
),
"",
"## Salida que debe redactarse" if spanish else "## Output To Write",
"",
f"- {'Redacte' if spanish else 'Write'}: `{contract['codex_output_file']}`",
(
"- Utilice: `root_cause_sweep_model_context.json`, `root_cause_sweep_summary.csv` y los gráficos `root_cause_bridge_alt_<n>.png` generados."
if spanish
else "- Use: `root_cause_sweep_model_context.json`, `root_cause_sweep_summary.csv`, and the generated `root_cause_bridge_alt_<n>.png` charts."
),
"",
"## Límite de interpretación" if spanish else "## Interpretation Boundary",
"",
(
"- El plugin ya ha seleccionado y conciliado las alternativas. Codex puede interpretarlas, compararlas y recomendar desgloses, pero no debe modificar las filas, los importes, las dimensiones ni los residuales calculados."
if spanish
else f"- {contract['deterministic_boundary']}"
),
(
"- Las filas posteriores son residuales después de las filas seleccionadas anteriormente, no totales independientes de esa dimensión."
if spanish
else "- Later rows are residual after earlier selected rows, not standalone totals for that dimension."
),
"",
"## Análisis requerido" if spanish else "## Required Analysis",
"",
]
lines.extend(f"- {item}" for item in required_points)
lines.extend(
[
"",
"## Afirmaciones prohibidas" if spanish else "## Forbidden Claims",
"",
]
)
lines.extend(f"- {item}" for item in forbidden_claims)
lines.extend(
[
"",
"## Resumen de alternativas" if spanish else "## Alternative Overview",
"",
(
"| Alternativa | Filas | Dimensiones mixtas | Residual Otros | Desglose | Filas seleccionadas |"
if spanish
else "| Alternative | Rows | Mixed dimensions | Other residual | Drilldown | Selected rows |"
),
"| --- | ---: | --- | ---: | --- | --- |",
]
)
for row in summary_rows:
selected_labels = str(row.get("selected_labels") or "").replace("|", "/")
lines.append(
"| "
f"{row.get('alternative_result')} | "
f"{row.get('row_count')} | "
f"{row.get('selected_sequence_has_mixed_dimensions')} | "
f"{_format_root_cause_brief_amount(row.get('other_residual'))} | "
f"{row.get('drilldown_status')} | "
f"{selected_labels} |"
)
lines.extend(
[
"",
"## Estructura recomendada" if spanish else "## Recommended Structure",
"",
(
"1. Lectura ejecutiva: qué alternativas son más útiles y por qué."
if spanish
else "1. Executive read: which alternatives are most useful and why."
),
(
"2. Hechos deterministas: variación total, filas seleccionadas, residuales e indicadores de dimensiones mixtas."
if spanish
else "2. Deterministic facts: total delta, selected rows, residuals, and mixed-dimension flags."
),
(
"3. Interpretación: qué sugieren las alternativas útiles, explicando la lógica residual."
if spanish
else "3. Interpretation: what the useful alternatives suggest, with residual logic explained."
),
(
"4. Recomendaciones de desglose: qué filas generadas requieren seguimiento."
if spanish
else "4. Drilldown recommendations: which generated rows deserve follow-up."
),
(
"5. Salvedades: etiquetas, supuestos de moneda, significado ausente de la jerarquía y alternativas ruidosas o engañosas."
if spanish
else "5. Caveats: labels, currency assumptions, missing hierarchy meaning, and alternatives that are noisy or misleading."
),
"",
]
)
return "\n".join(lines)
def _legacy_chart_not_written_audit(
exc: BaseException,
*,
artifact_name: str,
) -> dict[str, Any]:
"""Describe a non-blocking legacy chart render failure for audit output."""
return {
"status": "not_written_legacy_error",
"artifact": artifact_name,
"error": str(exc),
"exception_type": exc.__class__.__name__,
}
def _write_root_cause_sweep_outputs(
bridge_result: Any,
result: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
*,
render: bool = True,
) -> tuple[list[str], dict[str, Any]]:
"""Write root-cause alternative sweep data and optional PNG artifacts."""
if not bridge_result.sweep_runs:
return [], {}
paths: list[str] = []
chart_audits: dict[str, Any] = {}
alternatives: list[dict[str, Any]] = []
for alternative, alternative_run in sorted(bridge_result.sweep_runs.items()):
sequence = alternative_run.sequence
bridge_path: Path | None = None
chart_path: Path | None = None
if not sequence.frame.is_empty():
bridge_path = output_dir / f"root_cause_bridge_alt_{alternative}.csv"
sequence.frame.write_csv(bridge_path)
paths.append(str(bridge_path))
chart_artifact = f"root_cause_bridge_alt_{alternative}.png"
if render:
try:
chart_export = write_root_cause_bridge_png(
sequence.frame,
result,
recipe,
output_dir,
artifact_name=chart_artifact,
legacy_frame=sequence.legacy_frame,
legacy_param=sequence.param,
legacy_chart=bridge_result.chart,
legacy_index_cols=bridge_result.bridge_dimensions,
)
paths.extend(chart_export.paths)
chart_audits[str(alternative)] = chart_export.audit
chart_path = output_dir / chart_artifact
except LEGACY_RENDER_ERRORS as exc:
chart_audits[str(alternative)] = _legacy_chart_not_written_audit(
exc,
artifact_name=chart_artifact,
)
else:
chart_audits[str(alternative)] = _data_written_chart_audit(
chart_type=f"root_cause_bridge_alt_{alternative}",
artifact_name=chart_artifact,
source_functions=[
"legacy_adapter.run_legacy_variable_dimension_bridge"
],
)
if (
alternative_run.audit.get("drilldown_requested_rows")
and not sequence.details_frame.is_empty()
):
details_path = (
output_dir / f"root_cause_bridge_alt_{alternative}_details.csv"
)
sequence.details_frame.write_csv(details_path)
paths.append(str(details_path))
if (
alternative_run.audit.get("drilldown_requested_rows")
and not sequence.snapshot_frame.is_empty()
):
snapshot_path = (
output_dir / f"root_cause_bridge_alt_{alternative}_snapshot.csv"
)
sequence.snapshot_frame.write_csv(snapshot_path)
paths.append(str(snapshot_path))
drilldown_chart_audits: dict[str, Any] = {}
for row, drilldown_run in sorted(alternative_run.drilldown_runs.items()):
if drilldown_run.frame.is_empty():
continue
drilldown_path = (
output_dir
/ f"root_cause_bridge_alt_{alternative}_drilldown_row_{row}.csv"
)
drilldown_run.frame.write_csv(drilldown_path)
paths.append(str(drilldown_path))
parent_result = sequence.frame.slice(row - 1, 1)
if parent_result.is_empty():
continue
drilldown_artifact = (
f"root_cause_bridge_alt_{alternative}_drilldown_row_{row}.png"
)
if render:
try:
drilldown_chart = write_root_cause_bridge_png(
drilldown_run.frame,
parent_result,
recipe,
output_dir,
artifact_name=drilldown_artifact,
legacy_frame=drilldown_run.legacy_frame,
legacy_param=drilldown_run.param,
legacy_chart=bridge_result.chart,
legacy_index_cols=bridge_result.bridge_dimensions,
)
paths.extend(drilldown_chart.paths)
drilldown_chart_audits[str(row)] = drilldown_chart.audit
except LEGACY_RENDER_ERRORS as exc:
drilldown_chart_audits[str(row)] = _legacy_chart_not_written_audit(
exc,
artifact_name=drilldown_artifact,
)
else:
drilldown_chart_audits[str(row)] = _data_written_chart_audit(
chart_type=(
f"root_cause_bridge_alt_{alternative}_drilldown_row_{row}"
),
artifact_name=drilldown_artifact,
source_functions=[
"legacy_adapter.run_legacy_variable_dimension_bridge"
],
)
if drilldown_chart_audits:
chart_audits[f"{alternative}_drilldowns"] = drilldown_chart_audits
alternatives.append(
_root_cause_sequence_payload(
alternative_result=alternative,
frame=sequence.frame,
audit=alternative_run.audit,
result=result,
bridge_path=bridge_path,
chart_path=chart_path,
)
)
summary_rows = [
{
"alternative_result": alternative["alternative_result"],
"row_count": alternative["row_count"],
"selected_labels": " | ".join(
row["label"] for row in alternative["selected_rows"]
),
"selected_amounts": " | ".join(
f"{row['variance_amount']:.6f}" for row in alternative["selected_rows"]
),
"selected_sequence_bridge_dimensions": " | ".join(
str(value)
for value in alternative["selected_sequence_bridge_dimensions"]
),
"selected_sequence_has_mixed_dimensions": alternative[
"selected_sequence_has_mixed_dimensions"
],
"other_residual": alternative["other_residual"],
"drilldown_status": alternative["drilldown_status"],
"automatic_drilldown_rows": " | ".join(
str(value) for value in alternative["automatic_drilldown_rows"]
),
"bridge_path": alternative["bridge_path"],
"chart_path": alternative["chart_path"],
}
for alternative in alternatives
]
if summary_rows:
summary_path = output_dir / "root_cause_sweep_summary.csv"
pl.DataFrame(summary_rows).write_csv(summary_path)
paths.append(str(summary_path))
context_payload = {
"analysis_type": "root_cause_alternative_sweep",
"language": _output_language(recipe.get("language")),
"codex_interpretation_contract": _root_cause_sweep_interpretation_contract(),
"interpretation_rule": (
"Each selected row is residual after all prior selected rows have "
"been removed; mixed dimensions are valid but require this reading."
),
"alternatives": alternatives,
}
client_report_paths, client_report_audit = write_root_cause_client_report(
summary_rows=summary_rows,
recipe=recipe,
output_dir=output_dir,
)
paths.extend(client_report_paths)
context_payload["client_report"] = client_report_audit
chart_audits["client_report"] = client_report_audit
context_path = output_dir / "root_cause_sweep_model_context.json"
write_json(context_path, context_payload)
paths.append(str(context_path))
summary_json_path = output_dir / "root_cause_sweep_summary.json"
write_json(summary_json_path, {"alternatives": summary_rows})
paths.append(str(summary_json_path))
brief_path = output_dir / "root_cause_sweep_interpretation_brief.md"
brief_path.write_text(
_root_cause_sweep_interpretation_brief(
context_payload=context_payload,
summary_rows=summary_rows,
),
encoding="utf-8",
)
paths.append(str(brief_path))
return paths, chart_audits
def write_root_cause_bridge(
df: pl.DataFrame,
result: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
*,
render: bool = True,
) -> tuple[list[str], dict[str, Any] | None, dict[str, Any] | None]:
"""Write root-cause data and optionally render bridge PNG artifacts."""
if not bool(recipe["options"].get("root_cause_bridge")):
return [], None, None
try:
bridge_result = run_legacy_variable_dimension_bridge(df, recipe)
except LEGACY_RENDER_ERRORS as exc:
LOGGER.warning("Root-cause bridge was not written: %s", exc)
return (
[],
{
"enabled": True,
"status": "not_written_legacy_error",
"reason": str(exc),
"exception_type": exc.__class__.__name__,
},
None,
)
paths: list[str] = []
drilldown_requested = bool(
bridge_result.audit.get("drilldown_requested_rows")
or bridge_result.audit.get("moved_rows_requested")
)
if not bridge_result.candidate_frame.is_empty():
candidate_path = output_dir / "root_cause_bridge_candidates.csv"
bridge_result.candidate_frame.write_csv(candidate_path)
paths.append(str(candidate_path))
if drilldown_requested and not bridge_result.details_frame.is_empty():
details_path = output_dir / "root_cause_bridge_details.csv"
bridge_result.details_frame.write_csv(details_path)
paths.append(str(details_path))
if drilldown_requested and not bridge_result.snapshot_frame.is_empty():
snapshot_path = output_dir / "root_cause_bridge_snapshot.csv"
bridge_result.snapshot_frame.write_csv(snapshot_path)
paths.append(str(snapshot_path))
if bridge_result.frame.is_empty():
return paths, bridge_result.audit, None
bridge_path = output_dir / "root_cause_total_bridge.csv"
bridge_result.frame.write_csv(bridge_path)
paths.append(str(bridge_path))
chart_paths: list[str] = []
chart_path: Path | None = None
if render:
try:
chart_export = write_root_cause_bridge_png(
bridge_result.frame,
result,
recipe,
output_dir,
artifact_name="root_cause_total_bridge.png",
variance_mode="total_variance",
legacy_frame=bridge_result.legacy_frame,
legacy_param=bridge_result.param,
legacy_chart=bridge_result.chart,
legacy_index_cols=bridge_result.bridge_dimensions,
)
chart_paths = chart_export.paths
chart_audit = chart_export.audit
chart_path_value = chart_audit.get("path")
chart_path = Path(str(chart_path_value)) if chart_path_value else None
except LEGACY_RENDER_ERRORS as exc:
LOGGER.warning("Root-cause bridge chart was not written: %s", exc)
chart_audit = _legacy_chart_not_written_audit(
exc,
artifact_name="root_cause_total_bridge.png",
)
else:
chart_audit = _data_written_chart_audit(
chart_type=VARIANCE_CHART_ROOT_CAUSE,
artifact_name="root_cause_total_bridge.png",
source_functions=["legacy_adapter.run_legacy_variable_dimension_bridge"],
)
bridge_result.audit["status"] = "data_written"
bridge_result.audit["rendered"] = False
drilldown_chart_audits: dict[str, Any] = {}
for row, drilldown_run in sorted(bridge_result.drilldown_runs.items()):
if drilldown_run.frame.is_empty():
continue
drilldown_path = output_dir / f"root_cause_total_bridge_drilldown_row_{row}.csv"
drilldown_run.frame.write_csv(drilldown_path)
paths.append(str(drilldown_path))
parent_result = bridge_result.frame.slice(row - 1, 1)
if parent_result.is_empty():
continue
drilldown_artifact = f"root_cause_total_bridge_drilldown_row_{row}.png"
if render:
try:
drilldown_chart = write_root_cause_bridge_png(
drilldown_run.frame,
parent_result,
recipe,
output_dir,
artifact_name=drilldown_artifact,
variance_mode="total_variance",
legacy_frame=drilldown_run.legacy_frame,
legacy_param=drilldown_run.param,
legacy_chart=bridge_result.chart,
legacy_index_cols=bridge_result.bridge_dimensions,
)
except LEGACY_RENDER_ERRORS as exc:
drilldown_chart_audits[str(row)] = _legacy_chart_not_written_audit(
exc,
artifact_name=drilldown_artifact,
)
continue
paths.extend(drilldown_chart.paths)
drilldown_chart_audits[str(row)] = drilldown_chart.audit
else:
drilldown_chart_audits[str(row)] = _data_written_chart_audit(
chart_type=f"root_cause_total_bridge_drilldown_row_{row}",
artifact_name=drilldown_artifact,
source_functions=[
"legacy_adapter.run_legacy_variable_dimension_bridge"
],
)
if drilldown_chart_audits:
bridge_result.audit["drilldown_chart_audits"] = drilldown_chart_audits
if (
bridge_result.moved_run is not None
and not bridge_result.moved_run.frame.is_empty()
):
moved_path = output_dir / "root_cause_total_bridge_moved_rows.csv"
bridge_result.moved_run.frame.write_csv(moved_path)
paths.append(str(moved_path))
if render:
try:
moved_chart = write_root_cause_bridge_png(
bridge_result.moved_run.frame,
result,
recipe,
output_dir,
artifact_name="root_cause_total_bridge_moved_rows.png",
variance_mode="total_variance",
legacy_frame=bridge_result.moved_run.legacy_frame,
legacy_param=bridge_result.moved_run.param,
legacy_chart=bridge_result.chart,
legacy_index_cols=bridge_result.bridge_dimensions,
)
paths.extend(moved_chart.paths)
bridge_result.audit["moved_rows_chart_audit"] = moved_chart.audit
except LEGACY_RENDER_ERRORS as exc:
bridge_result.audit["moved_rows_chart_audit"] = (
_legacy_chart_not_written_audit(
exc,
artifact_name="root_cause_total_bridge_moved_rows.png",
)
)
else:
bridge_result.audit["moved_rows_chart_audit"] = _data_written_chart_audit(
chart_type="root_cause_total_bridge_moved_rows",
artifact_name="root_cause_total_bridge_moved_rows.png",
source_functions=[
"legacy_adapter.run_legacy_variable_dimension_bridge"
],
)
sweep_paths, sweep_chart_audits = _write_root_cause_sweep_outputs(
bridge_result,
result,
recipe,
output_dir,
render=render,
)
paths.extend(sweep_paths)
if sweep_chart_audits:
bridge_result.audit["alternative_sweep_chart_audits"] = sweep_chart_audits
context_path = _write_root_cause_bridge_context(
frame=bridge_result.frame,
result=result,
recipe=recipe,
audit=bridge_result.audit,
output_dir=output_dir,
bridge_path=bridge_path,
chart_path=chart_path,
context_name="root_cause_total_bridge_context.json",
analysis_type="root_cause_total_bridge",
capability_id="variance.root_cause_total_bridge",
chart_type="root_cause_total_bridge",
)
paths.append(str(context_path))
if bool(recipe.get("options", {}).get("root_cause_component_bridge")):
try:
component_result = run_legacy_variable_dimension_component_bridge(
df,
recipe,
)
except LEGACY_RENDER_ERRORS as exc:
bridge_result.audit["component_bridge_audit"] = {
"enabled": True,
"status": "not_written_legacy_error",
"reason": str(exc),
"exception_type": exc.__class__.__name__,
"root_cause_variance_mode": "component_variance",
}
else:
bridge_result.audit["component_bridge_audit"] = component_result.audit
if not component_result.candidate_frame.is_empty():
component_candidate_path = (
output_dir / "root_cause_component_bridge_candidates.csv"
)
component_result.candidate_frame.write_csv(component_candidate_path)
paths.append(str(component_candidate_path))
if not component_result.frame.is_empty():
component_path = output_dir / "root_cause_component_bridge.csv"
component_result.frame.write_csv(component_path)
paths.append(str(component_path))
component_chart_path: Path | None = None
try:
component_chart = write_root_cause_bridge_png(
component_result.frame,
result,
recipe,
output_dir,
artifact_name="root_cause_component_bridge.png",
variance_mode="component_variance",
legacy_frame=component_result.legacy_frame,
legacy_param=component_result.param,
legacy_chart=component_result.chart,
legacy_index_cols=component_result.bridge_dimensions,
)
paths.extend(component_chart.paths)
bridge_result.audit["component_bridge_chart_audit"] = (
component_chart.audit
)
chart_path_value = component_chart.audit.get("path")
component_chart_path = (
Path(str(chart_path_value)) if chart_path_value else None
)
except LEGACY_RENDER_ERRORS as exc:
bridge_result.audit["component_bridge_chart_audit"] = (
_legacy_chart_not_written_audit(
exc,
artifact_name="root_cause_component_bridge.png",
)
)
component_context = _write_root_cause_bridge_context(
frame=component_result.frame,
result=result,
recipe=recipe,
audit=component_result.audit,
output_dir=output_dir,
bridge_path=component_path,
chart_path=component_chart_path,
context_name="root_cause_component_bridge_context.json",
analysis_type="root_cause_component_bridge",
capability_id="variance.root_cause_component_bridge",
chart_type="root_cause_component_bridge",
)
paths.append(str(component_context))
return (
[*paths, *chart_paths],
bridge_result.audit,
chart_audit,
)
def run_variance_analysis(
input_path: Path,
output_dir: Path,
recipe_path: Path | None = None,
*,
root_cause_bridge: bool | None = None,
root_cause_bridge_alternative_result: int | None = None,
root_cause_bridge_drilldown_rows: list[int] | None = None,
root_cause_bridge_drilldown_all: bool | None = None,
root_cause_bridge_move_rows: dict[int | str, list[int]] | None = None,
root_cause_bridge_alternative_sweep: bool | None = None,
root_cause_bridge_alternative_sweep_start: int | None = None,
root_cause_bridge_alternative_sweep_end: int | None = None,
root_cause_bridge_auto_drilldown: str | None = None,
root_cause_bridge_auto_drilldown_min_share: float | None = None,
root_cause_component_bridge: bool | None = None,
root_cause_component_bridge_alternative_result: int | None = None,
waterfall_chart: bool | None = None,
waterfall_small_multiples: bool | None = None,
waterfall_small_multiples_dimension: str | None = None,
total_by_dimension_bridge: bool | None = None,
total_by_dimension_bridge_dimension: str | None = None,
total_by_dimension_bridge_top_n: int | None = None,
exploded_variance_bridge: bool | None = None,
exploded_variance_bridge_parent_dimension: str | None = None,
exploded_variance_bridge_child_dimension: str | None = None,
exploded_variance_bridge_parent_top_n: int | None = None,
exploded_variance_bridge_child_top_n: int | None = None,
exploded_variance_bridge_max_drilldowns: int | None = None,
currency: str | None = None,
language: str = "en",
artifact_mode: str = ARTIFACT_MODE_DATA_AND_RENDER,
client_engagement: dict[str, Any] | None = None,
) -> VarianceRunResult:
"""Run deterministic variance analysis and write output artifacts."""
vera_managed_host = is_vera_managed_host()
if vera_managed_host and client_engagement is None:
raise ValueError(
"The Vera-packaged variance runner requires --client-engagement."
)
if (vera_managed_host or client_engagement is not None) and not str(
currency or ""
).strip():
raise ValueError("Managed Vera variance runs require an explicit currency.")
artifact_mode = _normalize_artifact_mode(artifact_mode)
try:
df = read_table(input_path)
source_population_rows = df.height
recipe = build_recipe(input_path, df, language=language)
existing_recipe = None
if recipe_path:
existing_recipe = load_json(recipe_path)
recipe = build_recipe(
input_path,
df,
language=language,
existing_recipe=existing_recipe,
)
recipe = preserve_recipe_filters(recipe, existing_recipe)
recipe = preserve_recipe_cohorts(recipe, existing_recipe)
recipe["accounting_review"] = normalize_accounting_review(
recipe.get("accounting_review")
)
recipe["source_file"] = _source_reference(input_path, client_engagement)
recipe["execution_context"] = {
"host_mode": (
"vera_managed"
if vera_managed_host or client_engagement is not None
else "standalone_or_clara"
)
}
if currency is not None:
recipe.setdefault("options", {})["currency"] = currency
recipe = validate_recipe(df, recipe)
if root_cause_bridge is not None:
recipe["options"]["root_cause_bridge"] = root_cause_bridge
if root_cause_bridge_alternative_result is not None:
recipe["options"][
"root_cause_bridge_alternative_result"
] = root_cause_bridge_alternative_result
if root_cause_bridge_drilldown_rows is not None:
recipe["options"][
"root_cause_bridge_drilldown_rows"
] = root_cause_bridge_drilldown_rows
if root_cause_bridge_drilldown_all is not None:
recipe["options"][
"root_cause_bridge_drilldown_all"
] = root_cause_bridge_drilldown_all
if root_cause_bridge_move_rows is not None:
recipe["options"][
"root_cause_bridge_move_rows"
] = root_cause_bridge_move_rows
if root_cause_bridge_alternative_sweep is not None:
recipe["options"][
"root_cause_bridge_alternative_sweep"
] = root_cause_bridge_alternative_sweep
if root_cause_bridge_alternative_sweep:
recipe["options"]["root_cause_bridge"] = True
if root_cause_bridge_alternative_sweep_start is not None:
recipe["options"][
"root_cause_bridge_alternative_sweep_start"
] = root_cause_bridge_alternative_sweep_start
if root_cause_bridge_alternative_sweep_end is not None:
recipe["options"][
"root_cause_bridge_alternative_sweep_end"
] = root_cause_bridge_alternative_sweep_end
if root_cause_bridge_auto_drilldown is not None:
recipe["options"][
"root_cause_bridge_auto_drilldown"
] = root_cause_bridge_auto_drilldown
if root_cause_bridge_auto_drilldown_min_share is not None:
recipe["options"][
"root_cause_bridge_auto_drilldown_min_share"
] = root_cause_bridge_auto_drilldown_min_share
if root_cause_component_bridge is not None:
recipe["options"][
"root_cause_component_bridge"
] = root_cause_component_bridge
if root_cause_component_bridge:
recipe["options"]["root_cause_bridge"] = True
if root_cause_component_bridge_alternative_result is not None:
recipe["options"][
"root_cause_component_bridge_alternative_result"
] = root_cause_component_bridge_alternative_result
if waterfall_chart is not None:
recipe["options"]["waterfall_chart"] = waterfall_chart
if waterfall_small_multiples is not None:
recipe["options"]["waterfall_small_multiples"] = waterfall_small_multiples
if waterfall_small_multiples_dimension:
recipe["options"][
"waterfall_small_multiples_dimension"
] = waterfall_small_multiples_dimension
if total_by_dimension_bridge is not None:
recipe["options"]["total_by_dimension_bridge"] = total_by_dimension_bridge
if total_by_dimension_bridge_dimension:
recipe["options"][
"total_by_dimension_bridge_dimension"
] = total_by_dimension_bridge_dimension
if total_by_dimension_bridge_top_n is not None:
recipe["options"][
"total_by_dimension_bridge_top_n"
] = total_by_dimension_bridge_top_n
if exploded_variance_bridge is not None:
recipe["options"]["exploded_variance_bridge"] = exploded_variance_bridge
if exploded_variance_bridge_parent_dimension:
recipe["options"][
"exploded_variance_bridge_parent_dimension"
] = exploded_variance_bridge_parent_dimension
if exploded_variance_bridge_child_dimension:
recipe["options"][
"exploded_variance_bridge_child_dimension"
] = exploded_variance_bridge_child_dimension
if exploded_variance_bridge_parent_top_n is not None:
recipe["options"][
"exploded_variance_bridge_parent_top_n"
] = exploded_variance_bridge_parent_top_n
if exploded_variance_bridge_child_top_n is not None:
recipe["options"][
"exploded_variance_bridge_child_top_n"
] = exploded_variance_bridge_child_top_n
if exploded_variance_bridge_max_drilldowns is not None:
recipe["options"][
"exploded_variance_bridge_max_drilldowns"
] = exploded_variance_bridge_max_drilldowns
df, recipe = prepare_period_comparison_buckets(df, recipe)
df, filter_audit = apply_recipe_filters(df, recipe)
recipe.setdefault("options", {})["recipe_filter_audit"] = filter_audit
df, cohort_audit = apply_recipe_cohorts(
df,
recipe,
period_column=str(recipe["mappings"]["period_column"]),
value_column=str(recipe["mappings"]["amount_column"]),
current_period=str(recipe["mappings"]["comparison_period"]),
previous_period=str(recipe["mappings"]["baseline_period"]),
)
recipe = validate_recipe(df, recipe)
warn_if_output_dir_has_existing_files(output_dir, "Variance")
output_dir.mkdir(parents=True, exist_ok=True)
run_intake = write_run_intake(
output_dir,
input_path,
recipe_path=recipe_path,
recipe=recipe,
source_row_count=df.height,
client_engagement=client_engagement,
)
legacy_result = run_legacy_variance(
df, recipe, list(recipe["mappings"].get("dimensions") or [])
)
result = legacy_result.frame
amount_baseline = float(
result.select(pl.col("amount_baseline").sum()).item() or 0.0
)
amount_comparison = float(
result.select(pl.col("amount_comparison").sum()).item() or 0.0
)
recipe["accounting_readiness"] = evaluate_accounting_readiness(
recipe["accounting_review"],
amount_baseline=amount_baseline,
amount_comparison=amount_comparison,
max_abs_component_reconciliation_delta=max_abs(
result, "component_reconciliation_delta"
),
)
if bool(recipe["options"].get("waterfall_small_multiples")):
small_multiples_selection = select_waterfall_small_multiples_dimension(
result,
recipe,
)
else:
small_multiples_selection = {
"status": "disabled",
"reason": "waterfall_small_multiples is false",
"dimension": None,
}
recipe["options"][
"waterfall_small_multiples_dimension_selection"
] = small_multiples_selection
selected_small_multiples_dimension = small_multiples_selection.get("dimension")
recipe["options"]["waterfall_small_multiples"] = bool(
selected_small_multiples_dimension
)
recipe["options"][
"waterfall_small_multiples_dimension"
] = selected_small_multiples_dimension
if bool(recipe["options"].get("total_by_dimension_bridge")):
total_bridge_selection = select_total_by_dimension_bridge_dimension(
result,
recipe,
)
else:
total_bridge_selection = {
"status": "disabled",
"reason": "total_by_dimension_bridge is false",
"dimension": None,
}
recipe["options"][
"total_by_dimension_bridge_dimension_selection"
] = total_bridge_selection
selected_total_bridge_dimension = total_bridge_selection.get("dimension")
recipe["options"]["total_by_dimension_bridge"] = bool(
selected_total_bridge_dimension
)
recipe["options"][
"total_by_dimension_bridge_dimension"
] = selected_total_bridge_dimension
if bool(recipe["options"].get("exploded_variance_bridge")):
exploded_bridge_selection = select_exploded_variance_bridge_dimensions(
result,
recipe,
fallback_parent_dimension=(
str(selected_total_bridge_dimension)
if selected_total_bridge_dimension
else None
),
)
else:
exploded_bridge_selection = {
"status": "disabled",
"reason": "exploded_variance_bridge is false",
"parent_dimension": None,
"child_dimension": None,
}
recipe["options"][
"exploded_variance_bridge_dimension_selection"
] = exploded_bridge_selection
selected_exploded_parent_dimension = exploded_bridge_selection.get(
"parent_dimension"
)
selected_exploded_child_dimension = exploded_bridge_selection.get(
"child_dimension"
)
recipe["options"]["exploded_variance_bridge"] = bool(
selected_exploded_parent_dimension and selected_exploded_child_dimension
)
recipe["options"][
"exploded_variance_bridge_parent_dimension"
] = selected_exploded_parent_dimension
recipe["options"][
"exploded_variance_bridge_child_dimension"
] = selected_exploded_child_dimension
available_chart_types = []
if bool(recipe["options"].get("waterfall_chart", True)):
available_chart_types.append(VARIANCE_CHART_STANDARD_WATERFALL)
if recipe.get("mappings", {}).get("units_column") and bool(
recipe.get("options", {}).get("pvm_decomposition_ladder", True)
):
available_chart_types.append(VARIANCE_CHART_PVM_LADDER)
if bool(recipe["options"].get("waterfall_small_multiples")):
available_chart_types.append(VARIANCE_CHART_SMALL_MULTIPLES)
if selected_total_bridge_dimension:
available_chart_types.append(VARIANCE_CHART_TOTAL_BY_DIMENSION)
if selected_exploded_parent_dimension and selected_exploded_child_dimension:
available_chart_types.append(VARIANCE_CHART_EXPLODED_BRIDGE)
if bool(recipe["options"].get("root_cause_bridge", True)):
available_chart_types.append(VARIANCE_CHART_ROOT_CAUSE)
render_standard_waterfall = _variance_chart_selected(
VARIANCE_CHART_STANDARD_WATERFALL,
artifact_mode,
available_chart_types,
)
render_small_multiples = _variance_chart_selected(
VARIANCE_CHART_SMALL_MULTIPLES,
artifact_mode,
available_chart_types,
)
if artifact_mode == ARTIFACT_MODE_DATA_ONLY and _variance_data_chart_enabled(
VARIANCE_CHART_STANDARD_WATERFALL,
available_chart_types,
):
waterfall_export = argparse.Namespace(
paths=[],
audit=_data_written_chart_audit(
chart_type=VARIANCE_CHART_STANDARD_WATERFALL,
artifact_name="waterfall.png",
source_functions=[
"modules.charting.draw_waterfall.draw_vertical_waterfall_chart"
],
),
)
if _variance_data_chart_enabled(
VARIANCE_CHART_SMALL_MULTIPLES,
available_chart_types,
):
waterfall_export.audit["small_multiples_status"] = "data_written"
waterfall_export.audit["small_multiples_artifact"] = (
"waterfall_small_multiples.png"
)
elif artifact_mode == ARTIFACT_MODE_DATA_ONLY:
waterfall_export = argparse.Namespace(
paths=[],
audit=_skipped_chart_audit(
VARIANCE_CHART_STANDARD_WATERFALL,
"not enabled in recipe",
),
)
elif render_standard_waterfall or render_small_multiples:
waterfall_export = write_waterfall_png(
result,
recipe,
output_dir,
legacy_frame=legacy_result.legacy_frame,
render_standard=render_standard_waterfall,
render_small_multiples=render_small_multiples,
)
else:
waterfall_export = argparse.Namespace(
paths=[],
audit=_skipped_chart_audit(
VARIANCE_CHART_STANDARD_WATERFALL,
"not enabled in recipe",
),
)
render_pvm_ladder = _variance_chart_selected(
VARIANCE_CHART_PVM_LADDER,
artifact_mode,
available_chart_types,
)
if artifact_mode == ARTIFACT_MODE_DATA_ONLY and _variance_data_chart_enabled(
VARIANCE_CHART_PVM_LADDER,
available_chart_types,
):
pvm_ladder_export = argparse.Namespace(
paths=[],
audit=_data_written_chart_audit(
chart_type=VARIANCE_CHART_PVM_LADDER,
artifact_name="pvm_decomposition_ladder.png",
source_functions=[
"modules.charting.draw_waterfall.draw_vertical_waterfall_chart"
],
),
)
elif render_pvm_ladder:
pvm_ladder_export = write_pvm_decomposition_ladder_png(
result,
recipe,
output_dir,
legacy_frame=legacy_result.legacy_frame,
)
else:
pvm_ladder_export = argparse.Namespace(
paths=[],
audit=_skipped_chart_audit(
VARIANCE_CHART_PVM_LADDER,
"not enabled in recipe",
),
)
(
pvm_ladder_chart_data_paths,
pvm_ladder_chart_data_audit,
pvm_ladder_summary,
) = write_pvm_decomposition_ladder_chart_data(result, recipe, output_dir)
(
small_multiples_chart_data_paths,
small_multiples_chart_data_audit,
small_multiples_summary,
) = write_waterfall_small_multiples_chart_data(result, recipe, output_dir)
render_total_by_dimension = _variance_chart_selected(
VARIANCE_CHART_TOTAL_BY_DIMENSION,
artifact_mode,
available_chart_types,
)
if selected_total_bridge_dimension and artifact_mode == ARTIFACT_MODE_DATA_ONLY:
total_by_dimension_export = write_total_by_dimension_bridge_artifacts(
result,
recipe,
output_dir,
dimension=str(selected_total_bridge_dimension),
top_n=int(recipe["options"]["total_by_dimension_bridge_top_n"]),
render=False,
)
elif selected_total_bridge_dimension and render_total_by_dimension:
total_by_dimension_export = write_total_by_dimension_bridge_artifacts(
result,
recipe,
output_dir,
dimension=str(selected_total_bridge_dimension),
top_n=int(recipe["options"]["total_by_dimension_bridge_top_n"]),
)
else:
total_by_dimension_export = argparse.Namespace(
paths=[],
audit={
"enabled": bool(recipe["options"].get("total_by_dimension_bridge")),
"status": total_bridge_selection["status"],
"dimension": None,
"selection": total_bridge_selection,
},
summary_markdown="",
)
render_exploded_bridge = _variance_chart_selected(
VARIANCE_CHART_EXPLODED_BRIDGE,
artifact_mode,
available_chart_types,
)
if (
selected_exploded_parent_dimension
and selected_exploded_child_dimension
and artifact_mode == ARTIFACT_MODE_DATA_ONLY
):
exploded_bridge_export = write_exploded_variance_bridge_artifacts(
result,
recipe,
output_dir,
parent_dimension=str(selected_exploded_parent_dimension),
child_dimension=str(selected_exploded_child_dimension),
parent_top_n=int(
recipe["options"]["exploded_variance_bridge_parent_top_n"]
),
child_top_n=int(
recipe["options"]["exploded_variance_bridge_child_top_n"]
),
max_drilldowns=int(
recipe["options"]["exploded_variance_bridge_max_drilldowns"]
),
render=False,
)
elif (
selected_exploded_parent_dimension
and selected_exploded_child_dimension
and render_exploded_bridge
):
exploded_bridge_export = write_exploded_variance_bridge_artifacts(
result,
recipe,
output_dir,
parent_dimension=str(selected_exploded_parent_dimension),
child_dimension=str(selected_exploded_child_dimension),
parent_top_n=int(
recipe["options"]["exploded_variance_bridge_parent_top_n"]
),
child_top_n=int(
recipe["options"]["exploded_variance_bridge_child_top_n"]
),
max_drilldowns=int(
recipe["options"]["exploded_variance_bridge_max_drilldowns"]
),
)
else:
exploded_bridge_export = argparse.Namespace(
paths=[],
audit={
"enabled": bool(recipe["options"].get("exploded_variance_bridge")),
"status": exploded_bridge_selection["status"],
"parent_dimension": None,
"child_dimension": None,
"selection": exploded_bridge_selection,
},
summary_markdown="",
)
render_root_cause = _variance_chart_selected(
VARIANCE_CHART_ROOT_CAUSE,
artifact_mode,
available_chart_types,
)
if artifact_mode == ARTIFACT_MODE_DATA_ONLY and _variance_data_chart_enabled(
VARIANCE_CHART_ROOT_CAUSE,
available_chart_types,
):
bridge_paths, bridge_audit, bridge_chart_audit = write_root_cause_bridge(
df,
result,
recipe,
output_dir,
render=False,
)
elif render_root_cause:
bridge_paths, bridge_audit, bridge_chart_audit = write_root_cause_bridge(
df,
result,
recipe,
output_dir,
)
else:
bridge_paths = []
bridge_audit = None
bridge_chart_audit = bridge_audit
summary = build_summary_markdown(result, recipe)
summary += pvm_ladder_summary
summary += small_multiples_summary
summary += total_by_dimension_export.summary_markdown
summary += exploded_bridge_export.summary_markdown
summary = append_bridge_dimension_summary(
summary,
bridge_audit,
language=str(recipe.get("language") or "en"),
)
artifact_paths = [
*bridge_paths,
*waterfall_export.paths,
*pvm_ladder_export.paths,
*pvm_ladder_chart_data_paths,
*small_multiples_chart_data_paths,
*total_by_dimension_export.paths,
*exploded_bridge_export.paths,
]
audit = build_audit(
input_path,
recipe,
result,
artifact_paths,
client_engagement=client_engagement,
)
audit["legacy_runtime"] = legacy_result.audit
audit["legacy_runtime"]["variable_dimension_bridge"] = bridge_audit
audit["legacy_runtime"]["variable_dimension_bridge_chart"] = bridge_chart_audit
audit["legacy_runtime"]["waterfall_chart"] = waterfall_export.audit
audit["legacy_runtime"][
"pvm_decomposition_ladder_chart"
] = pvm_ladder_export.audit
audit["legacy_runtime"][
"pvm_decomposition_ladder_chart_data"
] = pvm_ladder_chart_data_audit
audit["legacy_runtime"][
"waterfall_small_multiples_dimension_selection"
] = small_multiples_selection
audit["legacy_runtime"][
"waterfall_small_multiples_chart_data"
] = small_multiples_chart_data_audit
audit["legacy_runtime"][
"total_by_dimension_bridge_dimension_selection"
] = total_bridge_selection
audit["legacy_runtime"][
"total_by_dimension_bridge"
] = total_by_dimension_export.audit
audit["legacy_runtime"][
"exploded_variance_bridge_dimension_selection"
] = exploded_bridge_selection
audit["legacy_runtime"][
"exploded_variance_bridge"
] = exploded_bridge_export.audit
if waterfall_export.audit.get("status") == "not_written":
audit["outputs"]["waterfall.png"] = "not_written: " + str(
waterfall_export.audit.get("error", "unknown")
)
if waterfall_export.audit.get("small_multiples_status") == "not_written":
audit["outputs"]["waterfall_small_multiples.png"] = "not_written: " + str(
waterfall_export.audit.get("small_multiples_error", "unknown")
)
if pvm_ladder_export.audit.get("status", "").startswith("not_written"):
audit["outputs"]["pvm_decomposition_ladder.png"] = "not_written: " + str(
pvm_ladder_export.audit.get(
"error", pvm_ladder_export.audit.get("reason", "unknown")
)
)
audit["legacy_runtime"]["artifact_mode"] = artifact_mode
chart_statuses = {
VARIANCE_CHART_STANDARD_WATERFALL: waterfall_export.audit.get("status"),
VARIANCE_CHART_PVM_LADDER: pvm_ladder_export.audit.get("status"),
VARIANCE_CHART_SMALL_MULTIPLES: (
waterfall_export.audit.get("small_multiples_status")
or small_multiples_chart_data_audit.get("status")
),
VARIANCE_CHART_TOTAL_BY_DIMENSION: (
total_by_dimension_export.audit.get("status")
),
VARIANCE_CHART_EXPLODED_BRIDGE: (
exploded_bridge_export.audit.get("status")
),
VARIANCE_CHART_ROOT_CAUSE: (
bridge_audit.get("status") if bridge_audit else None
),
}
audit["checks"]["legacy_chart_attempt_count"] = sum(
1
for status in chart_statuses.values()
if status not in {None, "skipped", "disabled"}
)
audit["checks"]["legacy_chart_written_count"] = sum(
1 for status in chart_statuses.values() if status == "written"
)
audit["checks"]["legacy_chart_data_count"] = sum(
1
for status in chart_statuses.values()
if status in {"written", "data_written"}
)
write_json(output_dir / "used_recipe.json", recipe)
audit["outputs"]["used_recipe.json"] = "written"
write_outputs(result, audit, summary, output_dir, artifact_paths=artifact_paths)
prepared_manifest_path = write_prepared_data_manifest(
output_dir=output_dir,
plugin="variance-analysis",
chart_family="variance_analysis",
source_file=str(recipe["source_file"]),
prepared_path=output_dir / "variance_results.csv",
frame=result,
recipe=recipe,
stage="variance_result",
preparation_audit={
"status": "prepared",
"recipe_filters": filter_audit,
"recipe_cohorts": cohort_audit,
"legacy_runtime": legacy_result.audit,
},
)
audit["outputs"][
_relative_artifact_path(prepared_manifest_path, output_dir)
] = "written"
standard_context_path = output_dir / "standard_variance_context.json"
write_json(
standard_context_path,
_standard_variance_context(result, recipe, output_dir),
)
audit["outputs"][
_relative_artifact_path(standard_context_path, output_dir)
] = "written"
write_json(output_dir / "variance_audit.json", audit)
default_model_paths = [
output_dir / "variance_results.csv",
output_dir / "variance_summary.md",
output_dir / "used_recipe.json",
prepared_manifest_path,
standard_context_path,
*(Path(path) for path in artifact_paths),
]
model_use_manifest_path = write_model_use_manifest(
input_path=input_path,
output_dir=output_dir,
recipe=recipe,
recipe_path=output_dir / "used_recipe.json",
source_population_rows=source_population_rows,
in_scope_rows=df.height,
default_artifact_paths=default_model_paths,
)
audit["outputs"][
_relative_artifact_path(model_use_manifest_path, output_dir)
] = "written"
artifact_paths = [
*artifact_paths,
str(prepared_manifest_path),
str(standard_context_path),
str(model_use_manifest_path),
]
review_session = write_review_session_artifacts(
output_dir,
input_path,
run_id=run_intake.run_id,
run_intake_path=run_intake.path,
recipe_path=recipe_path,
recipe=recipe,
result_rows=result.to_dicts(),
audit=audit,
client_engagement=client_engagement,
)
audit["review_session"] = {
"run_intake_path": review_session.run_intake_path.name,
"review_payload_path": review_session.review_payload_path.name,
"ui_decisions_path": review_session.ui_decisions_path.name,
"final_artifacts_path": review_session.final_artifacts_path.name,
"review_item_count": review_session.review_item_count,
}
for path in (
review_session.run_intake_path,
review_session.review_payload_path,
review_session.ui_decisions_path,
review_session.final_artifacts_path,
):
audit["outputs"][_relative_artifact_path(path, output_dir)] = "written"
artifact_paths = [
*artifact_paths,
str(review_session.run_intake_path),
str(review_session.review_payload_path),
str(review_session.ui_decisions_path),
str(review_session.final_artifacts_path),
]
write_json(output_dir / "variance_audit.json", audit)
return VarianceRunResult(
frame=result,
audit=audit,
summary_markdown=summary,
artifact_paths=artifact_paths,
)
finally:
cleanup_legacy_imports()
SHA-256: f11fb496b17f827621a030ef4832930173c22ec35a20eff4f2ced325c57c54ef