Line Graphs for Percentiles — P95 Misreading Tripled Costs

import matplotlib.pyplot as plt
import pandas as pd
from io.thecodeforge.data import DataLoader

def create_production_bar_chart(metrics_df: pd.DataFrame):
    """
    Creates a production-ready bar chart for service latency comparison.

    Bars are conditionally colored to surface SLA violations immediately.
    Value labels are added directly to bars to eliminate axis-reading overhead.
    The SLA threshold line gives context without requiring a separate chart.

    Args:
        metrics_df: DataFrame with columns ['service', 'latency_ms', 'timestamp']

    Returns:
        matplotlib Figure ready for dashboard embedding
    """
    fig, ax = plt.subplots(figsize=(12, 6))

    # Filter to last 24 hours of data
    recent_data = DataLoader.filter_last_n_hours(metrics_df, hours=24)

    # Group by service and calculate p95 latency
    # p95 chosen deliberately: average masks tail behavior in latency data
    service_latency = recent_data.groupby('service')['latency_ms'].quantile(0.95)

    # Sort descending so worst offenders are immediately visible on the left
    service_latency = service_latency.sort_values(ascending=False)

    # Conditional coloring: red above SLA threshold, green below
    # Avoid relying on color alone — add value labels for accessibility
    SLA_THRESHOLD_MS = 500
    colors = ['#e74c3c' if x > SLA_THRESHOLD_MS else '#2ecc71'
              for x in service_latency]

    bars = ax.bar(service_latency.index, service_latency.values, color=colors)

    # Add value labels on bars to eliminate axis-reading overhead
    for bar in bars:
        height = bar.get_height()
        ax.text(
            bar.get_x() + bar.get_width() / 2., height,
            f'{height:.1f}ms',
            ha='center', va='bottom', fontsize=9, fontweight='bold'
        )

    ax.set_ylabel('P95 Latency (ms)')
    ax.set_xlabel('Service')
    ax.set_title(
        'Service P95 Latency — Last 24 Hours\n'
        'Red bars exceed 500ms SLA threshold',
        fontsize=12
    )

    # SLA threshold line provides reference without a separate annotation box
    ax.axhline(
        y=SLA_THRESHOLD_MS, color='orange',
        linestyle='--', alpha=0.7, linewidth=1.5,
        label=f'SLA Threshold ({SLA_THRESHOLD_MS}ms)'
    )

    # Zero baseline is non-negotiable for bar charts
    ax.set_ylim(bottom=0)
    ax.legend()
    plt.xticks(rotation=30, ha='right')
    plt.tight_layout()

    return fig

import plotly.graph_objects as go
from plotly.subplots import make_subplots
from datetime import datetime, timedelta
from io.thecodeforge.monitoring import MetricsCollector

def create_multi_line_dashboard(metrics: dict, sla_thresholds: dict = None):
    """
    Creates a production monitoring dashboard with multiple line graphs.

    Design decisions:
    - Solid lines for primary latency metrics, dotted for secondary signals
    - Unified hover mode so all series values appear at the same timestamp
    - Threshold lines labeled inline to eliminate legend lookups
    - Dark template matches most production monitoring environments

    Args:
        metrics: Dict mapping metric_name -> {'timestamps': [...], 'values': [...]}
        sla_thresholds: Optional dict mapping metric_name -> threshold value

    Returns:
        Plotly Figure ready for dashboard embedding or export
    """
    fig = go.Figure()

    for metric_name, data in metrics.items():
        # Detect and break lines at data gaps
        # Gaps larger than 2x the median interval are treated as missing data
        timestamps = data['timestamps']
        values = data['values']

        # Insert None at gap positions to break the line visually
        # This prevents false continuity across outages or collection failures
        cleaned_values = MetricsCollector.insert_nulls_at_gaps(
            timestamps, values, gap_multiplier=2.0
        )

        fig.add_trace(go.Scatter(
            x=timestamps,
            y=cleaned_values,
            mode='lines',
            name=metric_name,
            line=dict(
                width=2,
                dash='solid' if 'latency' in metric_name else 'dot'
            ),
            connectgaps=False,  # Never bridge gaps — gaps are data too
            hovertemplate=(
                f'<b>{metric_name}</b><br>'
                'Time: %{x}<br>'
                'Value: %{y:.2f}<extra></extra>'
            )
        ))

    # Add threshold lines with inline labels
    if sla_thresholds:
        for metric_name, threshold in sla_thresholds.items():
            fig.add_hline(
                y=threshold,
                line_dash='dash',
                line_color='red',
                annotation_text=f'{metric_name} SLA: {threshold}',
                annotation_position='bottom right'
            )

    fig.update_layout(
        title=dict(
            text='System Health — Last 6 Hours<br>'
                 '<sup>Gaps indicate missing data, not zero values</sup>',
            font=dict(size=14)
        ),
        xaxis_title='Time (UTC)',
        yaxis_title='Value',
        hovermode='x unified',
        template='plotly_dark',
        legend=dict(orientation='h', yanchor='bottom', y=1.02)
    )

    return fig

// Production pie chart with accessibility and data validation
// D3.js v7 — built for cost allocation dashboards

function createAccessiblePieChart(data, containerId, options = {}) {
  const { width = 400, height = 400, innerRadius = 0 } = options;
  const radius = Math.min(width, height) / 2;
  const total = data.reduce((sum, item) => sum + item.value, 0);

  // Enforce slice limit — group small slices into 'Other' automatically
  // Slices below 5% become invisible and mislead readers about their scale
  const MIN_SLICE_PERCENT = 0.05;
  const { primary, grouped } = groupSmallSlices(data, total, MIN_SLICE_PERCENT);
  const chartData = grouped ? [...primary, grouped] : primary;

  // Create SVG with proper ARIA labels for screen reader accessibility
  const svg = d3.select(`#${containerId}`)
    .append('svg')
    .attr('width', width)
    .attr('height', height)
    .attr('role', 'img')
    .attr('aria-label', `Pie chart: ${chartData.map(d =>
      `${d.label} ${((d.value / total) * 100).toFixed(1)}%`
    ).join(', ')}`);

  const g = svg.append('g')
    .attr('transform', `translate(${width / 2}, ${height / 2})`);

  // Generate pie layout — no sorting so caller controls slice order
  // Convention: start largest slice at 12 o'clock (startAngle: -Math.PI/2)
  const pie = d3.pie()
    .value(d => d.value)
    .sort(null)
    .startAngle(-Math.PI / 2);

  const arc = d3.arc()
    .innerRadius(innerRadius)  // Set > 0 for donut variant
    .outerRadius(radius - 20);

  const labelArc = d3.arc()
    .innerRadius(radius * 0.7)
    .outerRadius(radius * 0.7);

  // Add slices with accessible color palette
  // Colors are chosen for contrast at WCAG AA level
  const slices = g.selectAll('path')
    .data(pie(chartData))
    .enter()
    .append('path')
    .attr('d', arc)
    .attr('fill', (d, i) => io.thecodeforge.colors.getAccessibleColor(i))
    .attr('stroke', '#fff')
    .attr('stroke-width', 2)
    .attr('aria-label', d =>
      `${d.data.label}: ${d.data.value} (${((d.data.value / total) * 100).toFixed(1)}%)`
    );

  // Direct percentage labels on slices eliminate legend-lookup overhead
  // Only label slices large enough to hold text (>= 8%)
  g.selectAll('text.slice-label')
    .data(pie(chartData))
    .enter()
    .append('text')
    .attr('class', 'slice-label')
    .attr('transform', d => `translate(${labelArc.centroid(d)})`)
    .attr('text-anchor', 'middle')
    .attr('font-size', '12px')
    .attr('fill', '#fff')
    .text(d => {
      const pct = (d.data.value / total) * 100;
      return pct >= 8 ? `${pct.toFixed(0)}%` : '';
    });

  return svg.node();
}

// Helper: groups slices below threshold into a single 'Other' category
function groupSmallSlices(data, total, threshold) {
  const primary = data.filter(d => d.value / total >= threshold);
  const small = data.filter(d => d.value / total < threshold);

  if (small.length === 0) return { primary, grouped: null };

  const grouped = {
    label: `Other (${small.length} items)`,
    value: small.reduce((sum, d) => sum + d.value, 0),
    drilldown: small  // Preserve detail for drill-down view
  };

  return { primary, grouped };
}

import numpy as np
from scipy import stats
import matplotlib.pyplot as plt
from io.thecodeforge.statistics import DistributionAnalyzer

def create_production_histogram(data: np.ndarray, metric_name: str):
    """
    Creates histogram with statistical annotations for production analysis.

    Design decisions:
    - Freedman-Diaconis bin width: handles heavy-tailed latency distributions
      better than Sturges' or Scott's rule
    - Mean, median, and p95 markers: three numbers tell a richer story
      than any single summary statistic
    - Normality test annotation: tells engineers whether parametric
      statistics (mean, standard deviation) are valid for this data
    - Rug plot overlay: preserves individual data point visibility
      for small-to-medium sample sizes

    Args:
        data: 1D array of continuous numeric values (e.g., latency in ms)
        metric_name: Human-readable metric label for axis and title

    Returns:
        matplotlib Figure ready for dashboard embedding or export
    """
    fig, ax = plt.subplots(figsize=(10, 6))

    # Freedman-Diaconis: bin_width = 2 * IQR * n^(-1/3)
    # More robust than Sturges for skewed or heavy-tailed data
    iqr = stats.iqr(data)
    if iqr == 0:
        # Fallback for near-constant data — avoid zero bin width
        bin_width = (max(data) - min(data)) / 20
    else:
        bin_width = 2 * iqr / (len(data) ** (1 / 3))

    bins = np.arange(min(data), max(data) + bin_width, bin_width)

    # Create histogram
    n, bins_out, patches = ax.hist(
        data, bins=bins, alpha=0.7,
        color='#3498db', edgecolor='white', linewidth=0.5
    )

    # Statistical markers: mean, median, p95
    # Three numbers together reveal skew and tail behavior simultaneously
    mean_val = np.mean(data)
    median_val = np.median(data)
    p95_val = np.percentile(data, 95)

    ax.axvline(mean_val, color='#e74c3c', linestyle='--', linewidth=2,
               label=f'Mean: {mean_val:.2f}ms')
    ax.axvline(median_val, color='#2ecc71', linestyle='-', linewidth=2,
               label=f'Median: {median_val:.2f}ms')
    ax.axvline(p95_val, color='#f39c12', linestyle=':', linewidth=2,
               label=f'P95: {p95_val:.2f}ms')

    # Rug plot: shows individual data points along x-axis
    # Valuable for small-to-medium datasets where bin artifacts can mislead
    if len(data) <= 2000:
        ax.plot(data, np.full_like(data, -0.02 * n.max()),
                '|', color='#2c3e50', alpha=0.3, markersize=5,
                label='Individual values')

    ax.set_xlabel(f'{metric_name} (ms)')
    ax.set_ylabel('Frequency (count)')
    ax.set_title(
        f'Distribution of {metric_name}\n'
        f'n={len(data):,} samples  |  '
        f'Bin width: {bin_width:.1f}ms (Freedman-Diaconis)',
        fontsize=12
    )
    ax.legend()

    # Normality test annotation
    # D'Agostino-Pearson is more reliable than Shapiro-Wilk for n > 5000
    normality_result = stats.normaltest(data)
    normality_p = normality_result.pvalue
    normal_label = 'likely normal' if normality_p > 0.05 else 'not normal'
    annotation_color = '#2ecc71' if normality_p > 0.05 else '#e74c3c'

    ax.text(
        0.02, 0.95,
        f'Normality test: p={normality_p:.4f} ({normal_label})\n'
        f'Skewness: {stats.skew(data):.3f}',
        transform=ax.transAxes,
        fontsize=9,
        verticalalignment='top',
        bbox=dict(facecolor='white', alpha=0.85, edgecolor=annotation_color,
                  linewidth=1.5)
    )

    plt.tight_layout()
    return fig

import seaborn as sns
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
from io.thecodeforge.analysis import CorrelationAnalyzer

def create_correlation_scatter(
    df: pd.DataFrame,
    x_col: str,
    y_col: str,
    hue_col: str = None
):
    """
    Creates scatter plot with correlation analysis for production debugging.

    Design decisions:
    - Dual panel: scatter on left, marginal distribution on right
      Marginal distributions surface the shape of each variable independently,
      which helps distinguish 'no correlation' from 'restricted range'
    - Regression line shown only above r=0.3 threshold
      Below that, a regression line implies a pattern that may not exist
    - Correlation coefficient in title: visible without hunting in annotations
    - Alpha transparency: essential for overplotted production datasets

    Args:
        df: DataFrame containing the variables to correlate
        x_col: Column name for x-axis variable
        y_col: Column name for y-axis variable
        hue_col: Optional column name for categorical grouping

    Returns:
        matplotlib Figure with scatter plot and marginal distributions
    """
    fig, axes = plt.subplots(1, 2, figsize=(14, 6))

    # Main scatter plot
    scatter_kwargs = dict(
        data=df, x=x_col, y=y_col,
        alpha=0.4,   # Transparency reveals density without hexbin complexity
        s=40,        # Point size: visible but not dominant
        ax=axes[0]
    )
    if hue_col:
        scatter_kwargs['hue'] = hue_col

    sns.scatterplot(**scatter_kwargs)

    # Pearson correlation with Spearman as fallback check
    # If Pearson and Spearman differ significantly, the relationship is non-linear
    pearson_r = df[x_col].corr(df[y_col], method='pearson')
    spearman_r = df[x_col].corr(df[y_col], method='spearman')

    title_lines = [f'Pearson r = {pearson_r:.3f}']
    if abs(pearson_r - spearman_r) > 0.1:
        title_lines.append(
            f'Spearman ρ = {spearman_r:.3f} — non-linear relationship suspected'
        )

    axes[0].set_title('\n'.join(title_lines), fontsize=11)

    # Add regression line only if correlation is meaningful
    # A regression line on an uncorrelated scatter plot is misleading
    if abs(pearson_r) > 0.3:
        sns.regplot(
            data=df, x=x_col, y=y_col,
            scatter=False, ax=axes[0],
            line_kws={'color': '#e74c3c', 'alpha': 0.8, 'linewidth': 2},
            ci=95  # Show 95% confidence band around regression line
        )

    axes[0].set_xlabel(x_col)
    axes[0].set_ylabel(y_col)

    # Marginal distribution of x variable
    # Shows whether restricted range or skew might explain correlation patterns
    sns.histplot(df[x_col], kde=True, ax=axes[1], color='#3498db', alpha=0.7)
    axes[1].set_title(
        f'Distribution of {x_col}\n'
        f'(check for restricted range or outliers that may drive correlation)',
        fontsize=10
    )

    plt.tight_layout()
    return fig

// io.thecodeforge — cs-fundamentals tutorial

import heapq

def dijkstra(graph: dict[str, dict[str, int]], start: str) -> dict[str, int]:
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    pq = [(0, start)]

    while pq:
        current_dist, current_node = heapq.heappop(pq)
        if current_dist > distances[current_node]:
            continue
        for neighbor, weight in graph[current_node].items():
            distance = current_dist + weight
            if distance < distances[neighbor]:
                distances[neighbor] = distance
                heapq.heappush(pq, (distance, neighbor))
    return distances

city_roads = {
    'A': {'B': 5, 'C': 2},
    'B': {'D': 4, 'E': 2},
    'C': {'B': 1, 'E': 7},
    'D': {'E': 1},
    'E': {}
}

print(dijkstra(city_roads, 'A'))
# Every weight is seconds of travel time. Negative? Use Bellman-Ford.

// io.thecodeforge — cs-fundamentals tutorial

def has_cycle_directed(graph: dict[str, list[str]]) -> bool:
    visited = set()
    in_stack = set()

    def dfs(node: str) -> bool:
        visited.add(node)
        in_stack.add(node)
        for neighbor in graph.get(node, []):
            if neighbor not in visited:
                if dfs(neighbor):
                    return True
            elif neighbor in in_stack:
                return True
        in_stack.remove(node)
        return False

    for node in graph:
        if node not in visited:
            if dfs(node):
                return True
    return False

job_dag = {
    'build': ['test', 'lint'],
    'test': ['deploy'],
    'lint': ['deploy'],
    'deploy': ['build']  # Bug: deploy -> build creates cycle
}

print(f'Cycle detected: {has_cycle_directed(job_dag)}')

// io.thecodeforge — cs-fundamentals tutorial

from enum import Enum

class EdgeType(Enum):
    DIRECTED = "one-way street"
    UNDIRECTED = "two-way street"
    WEIGHTED = "cost matters"
    UNWEIGHTED = "connection counts"

def classify_graph(edges: list) -> str:
    types = set()
    for u, v, *w in edges:
        types.add(EdgeType.WEIGHTED if w else EdgeType.UNWEIGHTED)
    if any((v, u) in [(e[0], e[1]) for e in edges] for u, v in edges):
        types.add(EdgeType.UNDIRECTED)
    else:
        types.add(EdgeType.DIRECTED)
    return f"Graph type: {', '.join(t.value for t in types)}"

print(classify_graph([(1, 2, 5), (2, 3, 3)]))

// io.thecodeforge — cs-fundamentals tutorial

def validate_consistency(edges: list, expected_type: str) -> bool:
    for u, v, *w in edges:
        cost_matters = bool(w)
        if expected_type == "weighted" and not cost_matters:
            raise TypeError(f"Edge ({u},{v}) lacks weight")
        if expected_type == "unweighted" and cost_matters:
            raise TypeError(f"Edge ({u},{v}) has unexpected weight")
    return True

# Fails: mixed weighted/unweighted edges
validate_consistency([(1, 2), (3, 4, 0.5)], "weighted")

// io.thecodeforge — cs-fundamentals tutorial
from collections import defaultdict

# Multigraph: allow parallel edges
class Multigraph:
    def __init__(self):
        self.adj = defaultdict(list)
    def add_edge(self, u, v):
        self.adj[u].append(v)
        self.adj[v].append(u)

# Forest: count connected components and cycles
class Forest:
    def __init__(self, n):
        self.parent = list(range(n))
    def find(self, x):
        while self.parent[x] != x:
            self.parent[x] = self.parent[self.parent[x]]
            x = self.parent[x]
        return x
    def union(self, x, y):
        px, py = self.find(x), self.find(y)
        if px == py: return False  # cycle detected
        self.parent[px] = py
        return True

// io.thecodeforge — cs-fundamentals tutorial
# Vertex and edge properties, tournament check
def degree_count(adj, vertex):
    return len(adj[vertex])

def is_tournament(adj, n):
    # complete directed graph: exactly one edge per unordered pair
    for i in range(n):
        for j in range(i+1, n):
            if adj[i].count(j) + adj[j].count(i) != 1:
                return False
    return True

# Example: tournament of 3 nodes
tour = [[1,2], [2], [0]]
print(is_tournament(tour, 3))  # True