#!/usr/bin/env python3 """ CJA Dimension Analysis Script v1.0 ==================================== Analyzes CJA dimension data collected by the cja-dimension-analysis skill. Performs cardinality classification, skew/distribution analysis, trend detection, anomaly detection (z-score), data quality scoring, comparisons, and forecasting. Generates both JSON results and comprehensive reports (markdown or HTML). This script is called by the cja-dimension-analysis skill after data collection phases. Usage: python3 cja_dimension_analysis.py \ [output_dir] [--format=markdown|html] [--keep-analyses=N] Arguments: analysis_json - Path to JSON file with dimension data collected by SKILL.md workflow data_view_name - Name of the data view being analyzed data_view_id - ID of the data view being analyzed output_dir - (Optional) Directory for output files (default: current directory) --format=FORMAT - (Optional) Output format: 'markdown' or 'html' (default: html) --keep-analyses=N - (Optional) Keep N recent analyses, delete older ones (default: 0 = keep all) Output Files: - dimension_analysis_results_YYYY-MM-DD_HH-MM.json: Raw analysis data - dimension_analysis_report_YYYY-MM-DD_HH-MM.md or .html: Comprehensive report Expected input JSON structure (collected by SKILL.md workflow): { "analysis_metadata": { "data_view_name": "...", "data_view_id": "...", "date_range": "...", "analyses_run": ["cardinality", "distribution", "trends", "anomalies", "errors"], "generated_at": "ISO8601" }, "dimensions": [ { "id": "variables/page", "name": "Page", "cardinality": {"uniqueValueCount": 1500, "level": "HIGH"}, "distribution": {"gini": 0.72, "skewLabel": "High skew", "topValues": [...]}, "trends": {"period1": "...", "period2": "...", "changes": [...], "newValues": [], "disappearedValues": []}, "anomalies": [...], "errors": {"errorPatterns": [...], "missingDataPct": 3.2} } ], "comparisons": [...], "summary": {} } Version: 1.0.0 Last Updated: March 13, 2026 """ import json import sys import os import math import re from datetime import datetime, timezone from pathlib import Path from collections import defaultdict def esc(s): """HTML-escape a value to prevent XSS in report output.""" return str(s).replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) if s is not None else "" # --------------------------------------------------------------------------- # Cardinality helpers # --------------------------------------------------------------------------- CARDINALITY_LEVELS = [ ("VERY_HIGH", 10000), ("HIGH", 1000), ("MEDIUM", 100), ("LOW", 0), ] CARDINALITY_BADGES = { "LOW": ("🟢", "Low", "< 100 unique values — fast queries"), "MEDIUM": ("🟡", "Medium", "100–1,000 unique values"), "HIGH": ("🟠", "High", "1,000–10,000 unique values — monitor query performance"), "VERY_HIGH": ("🔴", "Very High", "> 10,000 unique values — can slow queries"), } def classify_cardinality(count: int) -> str: for level, threshold in CARDINALITY_LEVELS: if count > threshold: return level return "LOW" # --------------------------------------------------------------------------- # Distribution / skew helpers # --------------------------------------------------------------------------- def compute_gini(values: list[float]) -> float: """Compute Gini coefficient for a list of non-negative values.""" if not values or sum(values) == 0: return 0.0 sorted_vals = sorted(values) n = len(sorted_vals) cumulative = 0.0 for i, v in enumerate(sorted_vals): cumulative += (2 * (i + 1) - n - 1) * v total = sum(sorted_vals) return cumulative / (n * total) if total else 0.0 def classify_skew(top1_pct: float, top5_pct: float, top10_pct: float) -> str: if top1_pct >= 50: return "Extreme skew" if top1_pct >= 30: return "High skew" if top10_pct < 50: return "Long tail" return "Moderate" def enrich_distribution(dim_data: dict) -> dict: """Compute derived distribution metrics from raw topValues list.""" dist = dim_data.get("distribution", {}) top_values = dist.get("topValues", []) if not top_values: return dim_data metrics = [float(v.get("metric", 0)) for v in top_values] total = sum(metrics) if total == 0: return dim_data for i, v in enumerate(top_values): v["pct"] = round(metrics[i] / total * 100, 2) v["cumulative"] = round(sum(metrics[: i + 1]) / total * 100, 2) top1_pct = metrics[0] / total * 100 if metrics else 0 top5_pct = sum(metrics[:5]) / total * 100 if len(metrics) >= 5 else sum(metrics) / total * 100 top10_pct = sum(metrics[:10]) / total * 100 if len(metrics) >= 10 else sum(metrics) / total * 100 dist["top1Pct"] = round(top1_pct, 2) dist["top5Pct"] = round(top5_pct, 2) dist["top10Pct"] = round(top10_pct, 2) dist["gini"] = round(compute_gini(metrics), 3) dist["skewLabel"] = classify_skew(top1_pct, top5_pct, top10_pct) dim_data["distribution"] = dist return dim_data # --------------------------------------------------------------------------- # Anomaly detection (z-score) # --------------------------------------------------------------------------- def detect_anomalies_zscore(time_series: list[dict], threshold: float = 2.0) -> list[dict]: """Detect anomalies in a time series using z-score method.""" if len(time_series) < 3: return [] values = [float(p.get("metric", 0)) for p in time_series] mean = sum(values) / len(values) variance = sum((v - mean) ** 2 for v in values) / len(values) std = math.sqrt(variance) if variance > 0 else 0 anomalies = [] for i, (point, val) in enumerate(zip(time_series, values)): if std == 0: continue z = abs(val - mean) / std if z > threshold: atype = "spike" if val > mean else "drop" anomalies.append({ "date": point.get("date", f"period_{i}"), "value": val, "mean": round(mean, 2), "zScore": round(z, 2), "type": atype, }) return anomalies # --------------------------------------------------------------------------- # Linear regression / forecasting # --------------------------------------------------------------------------- def linear_regression(x: list[float], y: list[float]) -> tuple[float, float, float]: """Simple linear regression. Returns (slope, intercept, r_squared).""" n = len(x) if n < 2: return 0.0, 0.0, 0.0 sum_x = sum(x) sum_y = sum(y) sum_xy = sum(xi * yi for xi, yi in zip(x, y)) sum_x2 = sum(xi ** 2 for xi in x) denom = n * sum_x2 - sum_x ** 2 if denom == 0: return 0.0, sum_y / n, 0.0 slope = (n * sum_xy - sum_x * sum_y) / denom intercept = (sum_y - slope * sum_x) / n # R² y_mean = sum_y / n ss_tot = sum((yi - y_mean) ** 2 for yi in y) ss_res = sum((yi - (slope * xi + intercept)) ** 2 for xi, yi in zip(x, y)) r2 = 1 - ss_res / ss_tot if ss_tot > 0 else 0.0 return round(slope, 4), round(intercept, 4), round(max(0, min(1, r2)), 4) def classify_forecast(slope: float, r2: float, mean: float) -> dict: """Classify forecast trend and confidence.""" if r2 >= 0.7: confidence = "High" elif r2 >= 0.4: confidence = "Medium" else: confidence = "Low" if mean == 0: direction = "Flat" elif slope > mean * 0.05: direction = "Upward" elif slope < -mean * 0.05: direction = "Downward" else: direction = "Flat" return {"direction": direction, "confidence": confidence} # --------------------------------------------------------------------------- # Data quality scoring # --------------------------------------------------------------------------- def score_data_quality(error_data: dict) -> tuple[str, str]: """Return (severity, label) based on missing data percentage.""" pct = error_data.get("missingDataPct", 0) if pct >= 20: return "critical", f"⛔ Critical ({pct:.1f}% missing)" elif pct >= 5: return "warning", f"⚠️ Warning ({pct:.1f}% missing)" else: return "ok", f"✅ OK ({pct:.1f}% missing)" # --------------------------------------------------------------------------- # Main analysis pipeline # --------------------------------------------------------------------------- def analyze(raw: dict) -> dict: """Run all analysis enrichments on the raw input data.""" metadata = raw.get("analysis_metadata", {}) dimensions = raw.get("dimensions", []) enriched_dims = [] summary = { "totalDimensions": len(dimensions), "cardinalityBreakdown": defaultdict(int), "skewBreakdown": defaultdict(int), "totalAnomalies": 0, "criticalDataQualityCount": 0, "warningDataQualityCount": 0, "highCardinalityCount": 0, } for dim in dimensions: # Cardinality card = dim.get("cardinality", {}) count = card.get("uniqueValueCount", 0) card["level"] = classify_cardinality(count) card["badge"] = CARDINALITY_BADGES.get(card["level"], ("⚪", "Unknown", "")) dim["cardinality"] = card level = card["level"] summary["cardinalityBreakdown"][level] += 1 if level in ("HIGH", "VERY_HIGH"): summary["highCardinalityCount"] += 1 # Distribution enrichment dim = enrich_distribution(dim) skew_label = dim.get("distribution", {}).get("skewLabel", "Unknown") summary["skewBreakdown"][skew_label] += 1 # Anomaly count anomalies = dim.get("anomalies", []) summary["totalAnomalies"] += len(anomalies) # Data quality errors = dim.get("errors", {}) severity, label = score_data_quality(errors) errors["severity"] = severity errors["severityLabel"] = label dim["errors"] = errors if severity == "critical": summary["criticalDataQualityCount"] += 1 elif severity == "warning": summary["warningDataQualityCount"] += 1 enriched_dims.append(dim) summary["cardinalityBreakdown"] = dict(summary["cardinalityBreakdown"]) summary["skewBreakdown"] = dict(summary["skewBreakdown"]) return { "analysis_metadata": metadata, "dimensions": enriched_dims, "comparisons": raw.get("comparisons", []), "summary": summary, } # --------------------------------------------------------------------------- # Markdown report generation # --------------------------------------------------------------------------- def generate_markdown(data: dict, data_view_name: str) -> str: now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") meta = data.get("analysis_metadata", {}) dims = data.get("dimensions", []) summary = data.get("summary", {}) lines = [ f"# Dimension Analysis Report", f"**Data View**: {data_view_name}", f"**Date Range**: {meta.get('date_range', 'N/A')}", f"**Generated**: {now}", f"**Dimensions Analyzed**: {summary.get('totalDimensions', 0)}", "", "---", "", "## Summary", "", f"| Metric | Value |", f"|--------|-------|", f"| Total Dimensions | {summary.get('totalDimensions', 0)} |", f"| High / Very High Cardinality | {summary.get('highCardinalityCount', 0)} |", f"| Total Anomalies Detected | {summary.get('totalAnomalies', 0)} |", f"| Critical Data Quality Issues | {summary.get('criticalDataQualityCount', 0)} |", f"| Warning Data Quality Issues | {summary.get('warningDataQualityCount', 0)} |", "", "### Cardinality Breakdown", "", ] card_bd = summary.get("cardinalityBreakdown", {}) for level in ["LOW", "MEDIUM", "HIGH", "VERY_HIGH"]: badge_info = CARDINALITY_BADGES.get(level, ("⚪", level, "")) lines.append(f"- {badge_info[0]} **{badge_info[1]}**: {card_bd.get(level, 0)} dimensions — {badge_info[2]}") lines.append("") for dim in dims: name = dim.get("name", dim.get("id", "Unknown")) dim_id = dim.get("id", "") lines += [ f"---", f"", f"## {name}", f"**ID**: `{dim_id}`", "", ] # Cardinality card = dim.get("cardinality", {}) badge = card.get("badge", ("⚪", "Unknown", "")) lines += [ f"### Cardinality", f"- **Unique Values**: {card.get('uniqueValueCount', 'N/A'):,}", f"- **Level**: {badge[0]} {badge[1]} — {badge[2]}", "", ] # Distribution dist = dim.get("distribution", {}) if dist: lines += [ f"### Distribution & Skew", f"- **Skew**: {dist.get('skewLabel', 'N/A')}", f"- **Gini Coefficient**: {dist.get('gini', 'N/A')}", f"- **Top 1 value share**: {dist.get('top1Pct', 0):.1f}%", f"- **Top 5 value share**: {dist.get('top5Pct', 0):.1f}%", f"- **Top 10 value share**: {dist.get('top10Pct', 0):.1f}%", "", ] top_values = dist.get("topValues", [])[:10] if top_values: lines += [ "| Rank | Value | Metric | % of Total | Cumulative % |", "|------|-------|--------|------------|--------------|", ] for i, v in enumerate(top_values): lines.append( f"| {i+1} | {v.get('value', 'N/A')} | " f"{v.get('metric', 0):,.0f} | {v.get('pct', 0):.1f}% | " f"{v.get('cumulative', 0):.1f}% |" ) lines.append("") # Trends trends = dim.get("trends", {}) if trends: new_vals = trends.get("newValues", []) gone_vals = trends.get("disappearedValues", []) changes = trends.get("changes", []) lines += [ f"### Trends", f"- **Period 1**: {trends.get('period1', 'N/A')}", f"- **Period 2**: {trends.get('period2', 'N/A')}", f"- **New values**: {len(new_vals)}", f"- **Disappeared values**: {len(gone_vals)}", "", ] growing = [c for c in changes if c.get("badge") == "🟢 Growing"] declining = [c for c in changes if c.get("badge") == "🔴 Declining"] if growing: lines.append(f"**Top Growing Values**: {', '.join(c.get('value','') for c in growing[:5])}") if declining: lines.append(f"**Top Declining Values**: {', '.join(c.get('value','') for c in declining[:5])}") lines.append("") # Anomalies anomalies = dim.get("anomalies", []) if anomalies: lines += [ f"### Anomalies ({len(anomalies)} detected)", "", "| Type | Value/Date | Z-Score | Magnitude |", "|------|-----------|---------|-----------|", ] for a in anomalies[:10]: lines.append( f"| {a.get('type', 'N/A')} | {a.get('date', a.get('value', 'N/A'))} | " f"{a.get('zScore', 'N/A')} | {a.get('value', 'N/A')} |" ) lines.append("") # Data quality errors = dim.get("errors", {}) if errors: lines += [ f"### Data Quality", f"- **Status**: {errors.get('severityLabel', 'N/A')}", f"- **Missing data**: {errors.get('missingDataPct', 0):.1f}%", ] patterns = errors.get("errorPatterns", []) if patterns: lines.append("") lines.append("| Error Pattern | Count | % of Total |") lines.append("|--------------|-------|------------|") for p in patterns: lines.append( f"| `{p.get('pattern', 'N/A')}` | {p.get('count', 0):,} | {p.get('pct', 0):.1f}% |" ) lines.append("") # Forecasts forecasts = dim.get("forecasts", []) if forecasts: lines += [ f"### Forecasts", "", "| Value | Direction | Confidence | Trend |", "|-------|-----------|------------|-------|", ] for f in forecasts[:5]: lines.append( f"| {f.get('value', 'N/A')} | {f.get('direction', 'N/A')} | " f"{f.get('confidence', 'N/A')} | slope={f.get('slope', 0):.4f} |" ) lines.append("") lines += [ "---", "", "## Recommendations", "", ] # Generate recommendations based on findings recs = [] for dim in dims: name = dim.get("name", dim.get("id", "Unknown")) card = dim.get("cardinality", {}) if card.get("level") in ("HIGH", "VERY_HIGH"): recs.append( f"🟠 **{name}** — High cardinality ({card.get('uniqueValueCount',0):,} values). " f"Consider whether all values are necessary; high cardinality can slow queries." ) errors = dim.get("errors", {}) if errors.get("severity") == "critical": recs.append( f"⛔ **{name}** — Critical data quality ({errors.get('missingDataPct',0):.1f}% missing). " f"Review data collection for this dimension." ) elif errors.get("severity") == "warning": recs.append( f"⚠️ **{name}** — Data quality warning ({errors.get('missingDataPct',0):.1f}% missing)." ) if len(dim.get("anomalies", [])) > 0: recs.append( f"🔍 **{name}** — {len(dim.get('anomalies', []))} anomalies detected. " f"Review for data spikes or drops." ) if recs: for r in recs: lines.append(f"- {r}") else: lines.append("No critical recommendations — dimensions look healthy.") lines += ["", "---", "", "*Generated by cja-dimension-analysis skill*"] return "\n".join(lines) # --------------------------------------------------------------------------- # HTML report generation # --------------------------------------------------------------------------- def generate_html(data: dict, data_view_name: str, data_view_id: str) -> str: now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") meta = data.get("analysis_metadata", {}) dims = data.get("dimensions", []) summary = data.get("summary", {}) dims_json = json.dumps(dims, default=str) summary_json = json.dumps(summary, default=str) card_bd = summary.get("cardinalityBreakdown", {}) card_labels = json.dumps(list(card_bd.keys())) card_values = json.dumps(list(card_bd.values())) skew_bd = summary.get("skewBreakdown", {}) skew_labels = json.dumps(list(skew_bd.keys())) skew_values = json.dumps(list(skew_bd.values())) nav_items = 'Summary' + "".join( f'{esc(d.get("name", d.get("id","")))}' for i, d in enumerate(dims) ) dim_sections = "" for i, dim in enumerate(dims): name = dim.get("name", dim.get("id", "Unknown")) dim_id = dim.get("id", "") card = dim.get("cardinality", {}) card_count = card.get("uniqueValueCount", 0) card_level = card.get("level", "UNKNOWN") badge_info = CARDINALITY_BADGES.get(card_level, ("⚪", card_level, "")) card_color = {"LOW": "var(--accent-green)", "MEDIUM": "var(--accent-yellow)", "HIGH": "var(--accent-red)", "VERY_HIGH": "var(--accent-red)"}.get(card_level, "var(--ink-muted)") dist = dim.get("distribution", {}) top_values = dist.get("topValues", [])[:10] tv_labels = json.dumps([v.get("value", "N/A") for v in top_values]) tv_values = json.dumps([v.get("metric", 0) for v in top_values]) errors = dim.get("errors", {}) err_severity_color = {"ok": "var(--accent-green)", "warning": "var(--accent-yellow)", "critical": "var(--accent-red)"}.get( errors.get("severity", "ok"), "var(--ink-muted)" ) anomalies = dim.get("anomalies", []) anomaly_rows = "".join( f"{esc(a.get('type','N/A'))}{esc(a.get('date', a.get('value','N/A')))}" f"{esc(a.get('zScore','N/A'))}" for a in anomalies[:5] ) dist_table_rows = "".join( f"{j+1}{esc(v.get('value','N/A'))}" f"{v.get('metric',0):,.0f}" f"{v.get('pct',0):.1f}%" f"{v.get('cumulative',0):.1f}%" for j, v in enumerate(top_values) ) forecasts = dim.get("forecasts", []) forecast_rows = "".join( f"{esc(f.get('value','N/A'))}{esc(f.get('direction','N/A'))}" f"{esc(f.get('confidence','N/A'))}" for f in forecasts[:5] ) dim_sections += f"""

{esc(name)}

{esc(dim_id)}

Cardinality

{badge_info[0]} {badge_info[1]}

{card_count:,} unique values

Skew

{dist.get('skewLabel','N/A')}

Gini: {dist.get('gini','N/A')}

Data Quality

{errors.get('severityLabel','N/A')}

{errors.get('missingDataPct',0):.1f}% missing

Anomalies

{len(anomalies)}

detected

Top Values Distribution

{dist_table_rows}

#	Value	Metric	% of Total	Cumulative

Anomalies (' + str(len(anomalies)) + ')

' + anomaly_rows + '

Type	Date/Value	Z-Score

' if anomalies else ''} {'

Forecasts (' + str(len(forecasts)) + ' values)

' + forecast_rows + '

Value	Direction	Confidence

' if forecasts else ''}

""" return f""" Dimension Analysis — {esc(data_view_name)}

Dimension Analysis

{esc(data_view_name)} Dimension Analysis

Cardinality, skew, anomaly, and data-quality review across the selected dimensions.

📅 {esc(meta.get('date_range','N/A'))} 📊 {esc(data_view_name)} 🕔 Generated {now}

Executive Summary

Dimensions

{summary.get('totalDimensions',0)}

analyzed

High Cardinality

{summary.get('highCardinalityCount',0)}

HIGH or VERY HIGH

Anomalies

{summary.get('totalAnomalies',0)}

detected

Critical DQ

{summary.get('criticalDataQualityCount',0)}

data quality issues

Cardinality Distribution

Skew Distribution

{dim_sections}

""" # --------------------------------------------------------------------------- # File management helpers # --------------------------------------------------------------------------- def cleanup_old_files(output_dir: Path, prefix: str, keep: int): if keep == 0: return files = sorted(output_dir.glob(f"{prefix}*"), key=lambda p: p.stat().st_mtime, reverse=True) for old_file in files[keep:]: try: old_file.unlink() print(f" 🗑️ Removed old file: {old_file.name}") except OSError: pass # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main(): if len(sys.argv) < 4: print(__doc__) sys.exit(1) analysis_json_path = sys.argv[1] data_view_name = sys.argv[2] data_view_id = sys.argv[3] output_dir = Path(".") fmt = "html" keep_analyses = 0 for arg in sys.argv[4:]: if arg.startswith("--format="): fmt = arg.split("=", 1)[1].lower() elif arg.startswith("--keep-analyses="): keep_analyses = int(arg.split("=", 1)[1]) elif not arg.startswith("--"): output_dir = Path(arg) output_dir.mkdir(parents=True, exist_ok=True) print(f"\n🔍 CJA Dimension Analysis") print(f" Data View: {data_view_name}") print(f" Input: {analysis_json_path}") print(f" Output: {output_dir} ({fmt})") print() # Load raw data print("📂 Loading analysis data...") with open(analysis_json_path) as f: raw = json.load(f) # Run analysis enrichment print("⚙️ Running analysis pipeline...") enriched = analyze(raw) # Timestamp for output files ts = datetime.now().strftime("%Y-%m-%d_%H-%M") # Save enriched JSON json_path = output_dir / f"dimension_analysis_results_{ts}.json" with open(json_path, "w") as f: json.dump(enriched, f, indent=2, default=str) print(f"✅ Analysis JSON saved: {json_path} ({json_path.stat().st_size:,} bytes)") # Generate report print(f"📝 Generating {fmt.upper()} report...") if fmt == "markdown": report_content = generate_markdown(enriched, data_view_name) report_ext = "md" report_prefix = "dimension_analysis_report_" else: report_content = generate_html(enriched, data_view_name, data_view_id) report_ext = "html" report_prefix = "dimension_analysis_report_" report_path = output_dir / f"{report_prefix}{ts}.{report_ext}" with open(report_path, "w") as f: f.write(report_content) print(f"✅ Report saved: {report_path} ({report_path.stat().st_size:,} bytes)") # Cleanup old files if keep_analyses > 0: print(f"\n🧹 Cleaning up old analysis files (keeping {keep_analyses} most recent)...") cleanup_old_files(output_dir, "dimension_analysis_results_", keep_analyses) cleanup_old_files(output_dir, report_prefix, keep_analyses) # Clean up input file — fully consumed; enriched results are saved separately try: input_path = Path(analysis_json_path) if input_path.exists(): input_path.unlink() print(f"🧹 Cleaned up input file: {input_path.name}") except Exception as e: print(f"⚠️ Warning: Could not clean up input file: {e}") # Summary summary = enriched.get("summary", {}) dims = enriched.get("dimensions", []) print(f"\n{'='*50}") print(f"📊 Analysis complete — {data_view_name}") print(f" Dimensions analyzed: {summary.get('totalDimensions', 0)}") print(f" High cardinality: {summary.get('highCardinalityCount', 0)}") print(f" Total anomalies: {summary.get('totalAnomalies', 0)}") print(f" Critical DQ issues: {summary.get('criticalDataQualityCount', 0)}") print(f" Warning DQ issues: {summary.get('warningDataQualityCount', 0)}") print(f"{'='*50}\n") return 0 if __name__ == "__main__": sys.exit(main())