multi_agent_demo/deviations/otel_parser.py

"""
OpenTelemetry Data Parser
Extracts business metrics and attributes from OTEL traces
"""

from typing import Dict, List, Any, Set
from datetime import datetime
from collections import defaultdict
import re


def parse_otel_data(otel_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Parse OTEL data and extract structured metrics for analysis

    Args:
        otel_data: Raw OTEL data (can be in various formats)

    Returns:
        Dictionary containing:
        - traces: List of parsed trace objects
        - metrics: Extracted numeric metrics by name
        - attributes: All unique attributes found
        - temporal_groups: Traces grouped by time periods
        - parameter_groups: Traces grouped by categorical parameters
    """
    # Handle different OTEL formats
    traces = _extract_traces(otel_data)

    if not traces:
        raise ValueError("No traces found in OTEL data")

    # Aggregate spans by trace ID to merge attributes from related spans
    traces = _aggregate_spans_by_trace(traces)

    # Extract metrics and attributes
    metrics = _extract_metrics(traces)
    attributes = _extract_attributes(traces)
    temporal_groups = _group_by_time(traces)
    parameter_groups = _group_by_parameters(traces, attributes)

    return {
        "traces": traces,
        "metrics": metrics,
        "attributes": attributes,
        "temporal_groups": temporal_groups,
        "parameter_groups": parameter_groups,
        "trace_count": len(traces)
    }


def _extract_traces(otel_data: Dict[str, Any]) -> List[Dict[str, Any]]:
    """
    Extract traces from OTEL data (handles various formats)

    Supports:
    - {"traces": [...]}
    - {"resourceSpans": [...]} (OTLP format)
    - Direct list of traces
    """
    traces = []

    if isinstance(otel_data, list):
        traces = otel_data
    elif "traces" in otel_data:
        traces = otel_data["traces"]
    elif "resourceSpans" in otel_data:
        # OTLP format (supports both scopeSpans and instrumentationLibrarySpans)
        for resource_span in otel_data["resourceSpans"]:
            # Try scopeSpans first (newer format)
            scope_spans = resource_span.get("scopeSpans", [])
            # Fall back to instrumentationLibrarySpans (older format)
            if not scope_spans:
                scope_spans = resource_span.get("instrumentationLibrarySpans", [])

            for scope_span in scope_spans:
                for span in scope_span.get("spans", []):
                    trace = _convert_otlp_span(span, resource_span.get("resource", {}))
                    traces.append(trace)
    elif "traceId" in otel_data or "spanId" in otel_data:
        # Single trace/span
        traces = [otel_data]

    return traces


def _convert_otlp_span(span: Dict, resource: Dict) -> Dict[str, Any]:
    """Convert OTLP span format to simplified trace format"""
    attributes = {}

    # Extract span attributes
    for attr in span.get("attributes", []):
        key = attr.get("key", "")
        value = attr.get("value", {})
        # Handle different value types
        if "stringValue" in value:
            attributes[key] = value["stringValue"]
        elif "intValue" in value:
            attributes[key] = int(value["intValue"])
        elif "doubleValue" in value:
            attributes[key] = float(value["doubleValue"])
        elif "boolValue" in value:
            attributes[key] = value["boolValue"]

    # Extract resource attributes
    for attr in resource.get("attributes", []):
        key = attr.get("key", "")
        value = attr.get("value", {})
        if "stringValue" in value:
            attributes[f"resource.{key}"] = value["stringValue"]

    # Convert decision/status outcomes to numeric values for bias detection
    # This ensures traces have numeric decision values that _calculate_group_statistics can use
    for key in ['decision', 'status', 'outcome', 'result']:
        if key in attributes and isinstance(attributes[key], str):
            value_lower = attributes[key].lower()
            if value_lower in ['approved', 'accepted', 'granted', 'yes', 'pass', 'success']:
                attributes[key] = 1.0
            elif value_lower in ['rejected', 'denied', 'declined', 'no', 'fail', 'failure']:
                attributes[key] = 0.0

    return {
        "trace_id": span.get("traceId", ""),
        "span_id": span.get("spanId", ""),
        "parent_span_id": span.get("parentSpanId", ""),  # Include parent for aggregation
        "span_name": span.get("name", ""),
        "timestamp": span.get("startTimeUnixNano", ""),
        "attributes": attributes
    }


def _extract_metrics(traces: List[Dict[str, Any]]) -> Dict[str, List[float]]:
    """
    Extract numeric metrics from traces

    Note: Decision/status values are already converted to 1.0/0.0 during parsing.
    Boolean values are converted to 1.0/0.0 here.

    Returns dictionary mapping metric names to lists of values
    """
    metrics = defaultdict(list)

    for trace in traces:
        attributes = trace.get("attributes", {})
        for key, value in attributes.items():
            # Extract numeric values (including pre-converted decision values)
            if isinstance(value, (int, float)):
                metrics[key].append(float(value))
            # Convert boolean values to numeric
            elif isinstance(value, bool):
                metrics[key].append(1.0 if value else 0.0)

    return dict(metrics)


def _extract_attributes(traces: List[Dict[str, Any]]) -> Dict[str, Set[Any]]:
    """
    Extract all unique attribute values

    Returns dictionary mapping attribute names to sets of unique values
    """
    attributes = defaultdict(set)

    for trace in traces:
        trace_attrs = trace.get("attributes", {})
        for key, value in trace_attrs.items():
            attributes[key].add(value)

    return {k: v for k, v in attributes.items()}


def _group_by_time(traces: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
    """
    Group traces by time periods (weeks, days, hours)

    Returns dictionary with keys: 'by_week', 'by_day', 'by_hour'
    """
    by_week = defaultdict(list)
    by_day = defaultdict(list)
    by_hour = defaultdict(list)

    for trace in traces:
        timestamp = _parse_timestamp(trace.get("timestamp", ""))
        if timestamp:
            # Week number (ISO week)
            week_key = timestamp.strftime("%Y-W%W")
            by_week[week_key].append(trace)

            # Day
            day_key = timestamp.strftime("%Y-%m-%d")
            by_day[day_key].append(trace)

            # Hour
            hour_key = timestamp.strftime("%Y-%m-%d %H:00")
            by_hour[hour_key].append(trace)

        # Also handle explicit week/day attributes
        attributes = trace.get("attributes", {})
        if "week" in attributes:
            week_key = f"week_{attributes['week']}"
            by_week[week_key].append(trace)

    return {
        "by_week": dict(by_week),
        "by_day": dict(by_day),
        "by_hour": dict(by_hour)
    }


def _group_by_parameters(traces: List[Dict[str, Any]], attributes: Dict[str, Set[Any]]) -> Dict[str, Dict[Any, List[Dict[str, Any]]]]:
    """
    Group traces by categorical parameters (for bias detection)

    Returns dictionary mapping parameter names to groups
    """
    parameter_groups = {}

    for attr_name, unique_values in attributes.items():
        # Check if this is a numeric attribute that should be binned
        is_numeric = all(isinstance(v, (int, float)) for v in unique_values)

        if is_numeric:
            # Check if this is a protected numeric attribute (age, income, etc.)
            attr_lower = attr_name.lower()
            needs_binning = any(keyword in attr_lower for keyword in ['age', 'income', 'salary', 'tenure', 'experience', 'years'])

            if needs_binning:
                # Bin numeric values into categorical groups
                groups = _bin_numeric_attribute(traces, attr_name, unique_values)
                if len(groups) > 1:
                    parameter_groups[f"{attr_name}_group"] = groups
            # Skip other numeric attributes (handled as metrics)
            continue

        # Only group by categorical attributes with reasonable cardinality
        if len(unique_values) > 50:
            continue

        groups = defaultdict(list)
        for trace in traces:
            trace_attrs = trace.get("attributes", {})
            if attr_name in trace_attrs:
                param_value = trace_attrs[attr_name]
                groups[param_value].append(trace)

        if len(groups) > 1:  # Only include if there are multiple groups
            parameter_groups[attr_name] = dict(groups)

    return parameter_groups


def _bin_numeric_attribute(traces: List[Dict[str, Any]], attr_name: str, unique_values: Set) -> Dict[str, List[Dict[str, Any]]]:
    """
    Bin numeric attribute values into categorical groups for bias detection

    For age: uses common age brackets (under_40, 40_and_over)
    For other numeric attributes: uses quartiles or median split
    """
    attr_lower = attr_name.lower()
    groups = defaultdict(list)

    # Special handling for age
    if 'age' in attr_lower:
        for trace in traces:
            attrs = trace.get("attributes", {})
            if attr_name in attrs:
                age = attrs[attr_name]
                # Common age discrimination threshold
                if age < 40:
                    groups["under_40"].append(trace)
                else:
                    groups["40_and_over"].append(trace)

    # Special handling for income/salary
    elif 'income' in attr_lower or 'salary' in attr_lower:
        # Use median split for income
        values = sorted(unique_values)
        median = values[len(values) // 2] if values else 0

        for trace in traces:
            attrs = trace.get("attributes", {})
            if attr_name in attrs:
                value = attrs[attr_name]
                if value < median:
                    groups["below_median"].append(trace)
                else:
                    groups["above_median"].append(trace)

    # Special handling for experience/tenure (years)
    elif 'years' in attr_lower or 'tenure' in attr_lower or 'experience' in attr_lower:
        for trace in traces:
            attrs = trace.get("attributes", {})
            if attr_name in attrs:
                years = attrs[attr_name]
                if years < 5:
                    groups["0-5_years"].append(trace)
                elif years < 10:
                    groups["5-10_years"].append(trace)
                else:
                    groups["10+_years"].append(trace)

    # Default: use quartile split
    else:
        values = sorted(unique_values)
        if len(values) >= 4:
            q1 = values[len(values) // 4]
            q3 = values[3 * len(values) // 4]

            for trace in traces:
                attrs = trace.get("attributes", {})
                if attr_name in attrs:
                    value = attrs[attr_name]
                    if value <= q1:
                        groups["low"].append(trace)
                    elif value >= q3:
                        groups["high"].append(trace)
                    else:
                        groups["medium"].append(trace)
        else:
            # Too few values, use median split
            median = values[len(values) // 2] if values else 0
            for trace in traces:
                attrs = trace.get("attributes", {})
                if attr_name in attrs:
                    value = attrs[attr_name]
                    if value < median:
                        groups["below_median"].append(trace)
                    else:
                        groups["above_median"].append(trace)

    return dict(groups)


def _parse_timestamp(timestamp: Any) -> datetime:
    """
    Parse timestamp from various formats

    Supports:
    - ISO 8601 strings
    - Unix timestamps (seconds)
    - Unix timestamps (nanoseconds)
    """
    if not timestamp:
        return None

    try:
        # ISO 8601 string
        if isinstance(timestamp, str):
            # Handle various ISO formats
            timestamp = timestamp.replace('Z', '+00:00')
            if 'T' in timestamp:
                return datetime.fromisoformat(timestamp)

        # Unix timestamp (seconds)
        if isinstance(timestamp, (int, float)):
            if timestamp > 1e12:  # Likely nanoseconds
                return datetime.fromtimestamp(timestamp / 1e9)
            else:  # Seconds
                return datetime.fromtimestamp(timestamp)

        # String representation of Unix timestamp
        if isinstance(timestamp, str) and timestamp.isdigit():
            ts = int(timestamp)
            if ts > 1e12:
                return datetime.fromtimestamp(ts / 1e9)
            else:
                return datetime.fromtimestamp(ts)

    except Exception:
        pass

    return None


def identify_business_metrics(parsed_data: Dict[str, Any], agent_purpose: str = "") -> List[Dict[str, Any]]:
    """
    Identify which metrics are likely business-relevant based on:
    - Metric names (semantic analysis)
    - Agent purpose
    - Statistical properties

    Returns list of metric metadata with relevance scores
    """
    metrics = parsed_data["metrics"]
    business_metrics = []

    # Keywords indicating business-relevant metrics
    business_keywords = [
        "amount", "cost", "price", "revenue", "profit", "refund", "fee", "commission",
        "score", "rating", "satisfaction", "count", "duration", "time", "delay",
        "success", "failure", "error", "rate", "percentage", "approved", "rejected"
    ]

    for metric_name, values in metrics.items():
        if len(values) < 5:  # Skip metrics with too few data points
            continue

        # Calculate relevance score
        relevance_score = 0.0

        # Name-based relevance
        metric_lower = metric_name.lower()
        for keyword in business_keywords:
            if keyword in metric_lower:
                relevance_score += 0.5

        # Agent purpose relevance
        if agent_purpose:
            purpose_lower = agent_purpose.lower()
            words = re.findall(r'\w+', metric_lower)
            for word in words:
                if len(word) > 3 and word in purpose_lower:
                    relevance_score += 0.3

        # Statistical properties (variability indicates interesting metric)
        import statistics
        if len(values) > 1:
            mean = statistics.mean(values)
            stdev = statistics.stdev(values) if len(values) > 1 else 0
            cv = stdev / mean if mean != 0 else 0  # Coefficient of variation
            if cv > 0.1:  # Some variability
                relevance_score += 0.2

        business_metrics.append({
            "name": metric_name,
            "relevance_score": relevance_score,
            "sample_count": len(values),
            "mean": statistics.mean(values),
            "stdev": statistics.stdev(values) if len(values) > 1 else 0
        })

    # Sort by relevance
    business_metrics.sort(key=lambda x: x["relevance_score"], reverse=True)

    return business_metrics


def _aggregate_spans_by_trace(traces: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    Aggregate spans that belong to the same trace to merge attributes

    This is useful when attributes are split across child spans
    (e.g., age in one span, score in another span of the same trace)

    Strategy:
    1. If spans share the same parent_span_id, aggregate them together
    2. Otherwise, group by trace_id + span_id combinations

    Args:
        traces: List of individual span traces

    Returns:
        List of aggregated traces with merged attributes
    """
    # First, try to group by parent_span_id (for child spans)
    by_parent = defaultdict(list)
    standalone = []

    for trace in traces:
        parent_id = trace.get("parent_span_id")
        if parent_id:
            # This is a child span, group with siblings
            key = f"{trace.get('trace_id', '')}:{parent_id}"
            by_parent[key].append(trace)
        else:
            # No parent, standalone span
            standalone.append(trace)

    # Aggregate child spans by parent
    aggregated = []

    for parent_key, sibling_spans in by_parent.items():
        # Merge all attributes from sibling spans
        merged_attributes = {}
        earliest_timestamp = None

        for span in sibling_spans:
            # Merge attributes
            attrs = span.get("attributes", {})
            merged_attributes.update(attrs)

            # Use earliest timestamp
            ts = span.get("timestamp")
            if ts:
                if earliest_timestamp is None:
                    earliest_timestamp = ts
                # Simple string comparison works for ISO timestamps
                elif isinstance(ts, str) and isinstance(earliest_timestamp, str):
                    if ts < earliest_timestamp:
                        earliest_timestamp = ts

        # Create aggregated trace from sibling spans
        aggregated_trace = {
            "trace_id": sibling_spans[0].get("trace_id", ""),
            "parent_span_id": sibling_spans[0].get("parent_span_id", ""),
            "timestamp": earliest_timestamp or sibling_spans[0].get("timestamp"),
            "span_name": sibling_spans[0].get("span_name", ""),
            "attributes": merged_attributes,
            "span_count": len(sibling_spans)
        }
        aggregated.append(aggregated_trace)

    # Add standalone spans as-is
    for trace in standalone:
        aggregated.append(trace)

    return aggregated