tatlock/scripts/benchmark_analysis.py

#!/usr/bin/env python3
"""
Benchmark analysis tool for Steward performance and tool recommendation accuracy.

Usage:
    # View Steward performance over last 24 hours
    python scripts/benchmark_analysis.py --operation steward_analysis --hours 24

    # Analyze tool recommendation accuracy over last 7 days
    python scripts/benchmark_analysis.py --tool-accuracy --days 7

    # Get summary of all operations in last hour
    python scripts/benchmark_analysis.py --summary --hours 1
"""

import sys
from pathlib import Path

# Add project root to path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

import argparse
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List
from collections import defaultdict

from src.core.benchmarks import get_benchmark_store, PerformanceBenchmark


async def analyze_steward_performance(hours: int = 24):
    """
    Analyze Steward analysis performance over time.

    Args:
        hours: Number of hours to look back
    """
    store = get_benchmark_store()

    # Query benchmarks from last N hours
    since = datetime.now() - timedelta(hours=hours)
    benchmarks = await store.query(
        operation="steward_analysis",
        since=since
    )

    if not benchmarks:
        print(f"No Steward analysis benchmarks found in the last {hours} hours.")
        return

    print(f"\n{'='*60}")
    print(f"Steward Analysis Performance (Last {hours} hours)")
    print(f"{'='*60}\n")

    # Calculate statistics
    durations = [b.duration_seconds for b in benchmarks]
    recommendation_counts = [b.recommendation_count for b in benchmarks if b.recommendation_count is not None]

    avg_duration = sum(durations) / len(durations)
    min_duration = min(durations)
    max_duration = max(durations)

    print(f"Total Analyses: {len(benchmarks)}")
    print(f"Success Rate: {sum(1 for b in benchmarks if b.success) / len(benchmarks) * 100:.1f}%")
    print(f"\nLatency Statistics:")
    print(f"  Average: {avg_duration:.3f}s")
    print(f"  Min: {min_duration:.3f}s")
    print(f"  Max: {max_duration:.3f}s")

    if recommendation_counts:
        avg_recommendations = sum(recommendation_counts) / len(recommendation_counts)
        print(f"\nRecommendation Statistics:")
        print(f"  Average recommendations per request: {avg_recommendations:.1f}")
        print(f"  Min recommendations: {min(recommendation_counts)}")
        print(f"  Max recommendations: {max(recommendation_counts)}")

        # Distribution
        print(f"\nRecommendation Count Distribution:")
        distribution = defaultdict(int)
        for count in recommendation_counts:
            distribution[count] += 1
        for count in sorted(distribution.keys()):
            percentage = distribution[count] / len(recommendation_counts) * 100
            print(f"  {count} capabilities: {distribution[count]} ({percentage:.1f}%)")

    # Complexity distribution
    complexities = defaultdict(int)
    for b in benchmarks:
        if b.metadata and "complexity" in b.metadata:
            complexities[b.metadata["complexity"]] += 1

    if complexities:
        print(f"\nComplexity Distribution:")
        for complexity in sorted(complexities.keys()):
            percentage = complexities[complexity] / len(benchmarks) * 100
            print(f"  {complexity}: {complexities[complexity]} ({percentage:.1f}%)")

    print()


async def analyze_tool_accuracy(days: int = 7):
    """
    Analyze tool recommendation accuracy.

    Args:
        days: Number of days to look back
    """
    store = get_benchmark_store()

    # Query tool call benchmarks from last N days
    since = datetime.now() - timedelta(days=days)
    benchmarks = await store.query(
        operation="tool_call",
        since=since
    )

    if not benchmarks:
        print(f"No tool call benchmarks found in the last {days} days.")
        return

    print(f"\n{'='*60}")
    print(f"Tool Recommendation Accuracy (Last {days} days)")
    print(f"{'='*60}\n")

    # Categorize tool calls
    recommended_and_used = []  # True positives
    recommended_not_used = []  # False positives (recommended but not used)
    not_recommended_but_used = []  # False negatives (used but not recommended)

    for b in benchmarks:
        if b.was_recommended and b.was_actually_used:
            recommended_and_used.append(b)
        elif b.was_recommended and not b.was_actually_used:
            recommended_not_used.append(b)
        elif not b.was_recommended and b.was_actually_used:
            not_recommended_but_used.append(b)

    total_recommendations = len(recommended_and_used) + len(recommended_not_used)
    total_tool_calls = len(recommended_and_used) + len(not_recommended_but_used)

    print(f"Total Tool Calls: {total_tool_calls}")
    print(f"Total Recommendations: {total_recommendations}")

    if total_recommendations > 0:
        precision = len(recommended_and_used) / total_recommendations * 100
        print(f"\nPrecision: {precision:.1f}%")
        print(f"  (recommended and actually used / all recommendations)")

    if total_tool_calls > 0:
        recall = len(recommended_and_used) / total_tool_calls * 100
        print(f"\nRecall: {recall:.1f}%")
        print(f"  (recommended and actually used / all tool calls)")

    if total_recommendations > 0 and total_tool_calls > 0:
        f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        print(f"\nF1 Score: {f1:.1f}%")

    print(f"\nBreakdown:")
    print(f"  ✅ Recommended & Used: {len(recommended_and_used)}")
    print(f"  ⚠️  Recommended but Not Used: {len(recommended_not_used)}")
    print(f"  ❌ Not Recommended but Used: {len(not_recommended_but_used)}")

    # Tool-specific accuracy
    tool_usage = defaultdict(lambda: {"recommended_used": 0, "not_recommended_used": 0})

    for b in recommended_and_used:
        if b.tool_name:
            tool_usage[b.tool_name]["recommended_used"] += 1

    for b in not_recommended_but_used:
        if b.tool_name:
            tool_usage[b.tool_name]["not_recommended_used"] += 1

    if tool_usage:
        print(f"\nPer-Tool Accuracy:")
        for tool_name in sorted(tool_usage.keys()):
            stats = tool_usage[tool_name]
            total = stats["recommended_used"] + stats["not_recommended_used"]
            accuracy = stats["recommended_used"] / total * 100 if total > 0 else 0
            print(f"  {tool_name}: {accuracy:.1f}% ({stats['recommended_used']}/{total})")

    # Duration statistics for tool calls
    durations = [b.duration_seconds for b in benchmarks if b.duration_seconds]
    if durations:
        avg_duration = sum(durations) / len(durations)
        print(f"\nTool Call Duration:")
        print(f"  Average: {avg_duration:.3f}s")
        print(f"  Min: {min(durations):.3f}s")
        print(f"  Max: {max(durations):.3f}s")

    print()


async def show_summary(hours: int = 1):
    """
    Show summary of all operations in the specified time window.

    Args:
        hours: Number of hours to look back
    """
    store = get_benchmark_store()

    since = datetime.now() - timedelta(hours=hours)

    # Query all operations
    all_benchmarks = await store.query(since=since)

    if not all_benchmarks:
        print(f"No benchmarks found in the last {hours} hours.")
        return

    print(f"\n{'='*60}")
    print(f"Benchmark Summary (Last {hours} hours)")
    print(f"{'='*60}\n")

    # Group by operation
    by_operation = defaultdict(list)
    for b in all_benchmarks:
        by_operation[b.operation].append(b)

    print(f"Total Operations: {len(all_benchmarks)}\n")

    for operation in sorted(by_operation.keys()):
        benchmarks = by_operation[operation]
        durations = [b.duration_seconds for b in benchmarks if b.duration_seconds]
        avg_duration = sum(durations) / len(durations) if durations else 0
        success_rate = sum(1 for b in benchmarks if b.success) / len(benchmarks) * 100

        print(f"{operation}:")
        print(f"  Count: {len(benchmarks)}")
        print(f"  Success Rate: {success_rate:.1f}%")
        if durations:
            print(f"  Avg Duration: {avg_duration:.3f}s")
        print()


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="Analyze Tatlock benchmark data",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__
    )

    parser.add_argument(
        "--operation",
        choices=["steward_analysis", "tool_call"],
        help="Analyze specific operation type"
    )

    parser.add_argument(
        "--hours",
        type=int,
        default=24,
        help="Number of hours to look back (default: 24)"
    )

    parser.add_argument(
        "--days",
        type=int,
        default=7,
        help="Number of days to look back (default: 7)"
    )

    parser.add_argument(
        "--tool-accuracy",
        action="store_true",
        help="Analyze tool recommendation accuracy"
    )

    parser.add_argument(
        "--summary",
        action="store_true",
        help="Show summary of all operations"
    )

    args = parser.parse_args()

    # Run analysis
    if args.tool_accuracy:
        asyncio.run(analyze_tool_accuracy(args.days))
    elif args.summary:
        asyncio.run(show_summary(args.hours))
    elif args.operation == "steward_analysis":
        asyncio.run(analyze_steward_performance(args.hours))
    elif args.operation == "tool_call":
        # Show tool-specific analysis within the hours window
        asyncio.run(analyze_tool_accuracy(days=args.hours // 24 or 1))
    else:
        # Default: show summary
        asyncio.run(show_summary(args.hours))


if __name__ == "__main__":
    main()