239 lines
6.9 KiB
Python
239 lines
6.9 KiB
Python
import math
|
|
from datetime import datetime, timedelta
|
|
|
|
import matplotlib
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from beyond_metrics.dimensions.OperationalPerformance import OperationalPerformanceMetrics
|
|
|
|
matplotlib.use("Agg")
|
|
|
|
|
|
def _sample_df() -> pd.DataFrame:
|
|
"""
|
|
Dataset sintético pequeño para probar la dimensión de rendimiento operacional.
|
|
|
|
Incluye:
|
|
- varios skills
|
|
- FCR, abandonos, transferencias
|
|
- reincidencia <7 días
|
|
- logged_time para occupancy
|
|
"""
|
|
base = datetime(2024, 1, 1, 10, 0, 0)
|
|
|
|
rows = [
|
|
# cliente C1, resolved, no abandon, voz, ventas
|
|
{
|
|
"interaction_id": "id1",
|
|
"datetime_start": base,
|
|
"queue_skill": "ventas",
|
|
"channel": "voz",
|
|
"duration_talk": 600,
|
|
"hold_time": 60,
|
|
"wrap_up_time": 30,
|
|
"agent_id": "A1",
|
|
"transfer_flag": 0,
|
|
"is_resolved": 1,
|
|
"abandoned_flag": 0,
|
|
"customer_id": "C1",
|
|
"logged_time": 900,
|
|
},
|
|
# C1 vuelve en 3 días mismo canal/skill
|
|
{
|
|
"interaction_id": "id2",
|
|
"datetime_start": base + timedelta(days=3),
|
|
"queue_skill": "ventas",
|
|
"channel": "voz",
|
|
"duration_talk": 700,
|
|
"hold_time": 30,
|
|
"wrap_up_time": 40,
|
|
"agent_id": "A1",
|
|
"transfer_flag": 1,
|
|
"is_resolved": 1,
|
|
"abandoned_flag": 0,
|
|
"customer_id": "C1",
|
|
"logged_time": 900,
|
|
},
|
|
# cliente C2, soporte, chat, no resuelto, transferido
|
|
{
|
|
"interaction_id": "id3",
|
|
"datetime_start": base + timedelta(hours=1),
|
|
"queue_skill": "soporte",
|
|
"channel": "chat",
|
|
"duration_talk": 400,
|
|
"hold_time": 20,
|
|
"wrap_up_time": 30,
|
|
"agent_id": "A2",
|
|
"transfer_flag": 1,
|
|
"is_resolved": 0,
|
|
"abandoned_flag": 0,
|
|
"customer_id": "C2",
|
|
"logged_time": 800,
|
|
},
|
|
# cliente C3, abandonado
|
|
{
|
|
"interaction_id": "id4",
|
|
"datetime_start": base + timedelta(hours=2),
|
|
"queue_skill": "soporte",
|
|
"channel": "voz",
|
|
"duration_talk": 100,
|
|
"hold_time": 50,
|
|
"wrap_up_time": 10,
|
|
"agent_id": "A2",
|
|
"transfer_flag": 0,
|
|
"is_resolved": 0,
|
|
"abandoned_flag": 1,
|
|
"customer_id": "C3",
|
|
"logged_time": 600,
|
|
},
|
|
# cliente C4, una sola interacción, email
|
|
{
|
|
"interaction_id": "id5",
|
|
"datetime_start": base + timedelta(days=10),
|
|
"queue_skill": "ventas",
|
|
"channel": "email",
|
|
"duration_talk": 300,
|
|
"hold_time": 0,
|
|
"wrap_up_time": 20,
|
|
"agent_id": "A1",
|
|
"transfer_flag": 0,
|
|
"is_resolved": 1,
|
|
"abandoned_flag": 0,
|
|
"customer_id": "C4",
|
|
"logged_time": 700,
|
|
},
|
|
]
|
|
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Inicialización y validación básica
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_init_and_required_columns():
|
|
df = _sample_df()
|
|
op = OperationalPerformanceMetrics(df)
|
|
assert not op.is_empty
|
|
|
|
# Falta columna obligatoria -> ValueError
|
|
df_missing = df.drop(columns=["duration_talk"])
|
|
try:
|
|
OperationalPerformanceMetrics(df_missing)
|
|
assert False, "Debería lanzar ValueError si falta duration_talk"
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# AHT y distribución
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_aht_distribution_basic():
|
|
df = _sample_df()
|
|
op = OperationalPerformanceMetrics(df)
|
|
|
|
dist = op.aht_distribution()
|
|
assert "p10" in dist and "p50" in dist and "p90" in dist and "p90_p50_ratio" in dist
|
|
|
|
# Comprobamos que el ratio P90/P50 es razonable (>1)
|
|
assert dist["p90_p50_ratio"] >= 1.0
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# FCR, escalación, abandono
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_fcr_escalation_abandonment_rates():
|
|
df = _sample_df()
|
|
op = OperationalPerformanceMetrics(df)
|
|
|
|
fcr = op.fcr_rate()
|
|
esc = op.escalation_rate()
|
|
aband = op.abandonment_rate()
|
|
|
|
# FCR: interacciones resueltas / total
|
|
# is_resolved=1 en id1, id2, id5 -> 3 de 5
|
|
assert math.isclose(fcr, 60.0, rel_tol=1e-6)
|
|
|
|
# Escalación: transfer_flag=1 en id2, id3 -> 2 de 5
|
|
assert math.isclose(esc, 40.0, rel_tol=1e-6)
|
|
|
|
# Abandono: abandoned_flag=1 en id4 -> 1 de 5
|
|
assert math.isclose(aband, 20.0, rel_tol=1e-6)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Reincidencia y repetición de canal
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_recurrence_and_repeat_channel():
|
|
df = _sample_df()
|
|
op = OperationalPerformanceMetrics(df)
|
|
|
|
rec = op.recurrence_rate_7d()
|
|
rep = op.repeat_channel_rate()
|
|
|
|
# Clientes: C1, C2, C3, C4 -> 4 clientes
|
|
# Recurrente: C1 (tiene 2 contactos en 3 días). Solo 1 de 4 -> 25%
|
|
assert math.isclose(rec, 25.0, rel_tol=1e-6)
|
|
|
|
# Reincidencias (<7d):
|
|
# Solo el par de C1: voz -> voz, mismo canal => 100%
|
|
assert math.isclose(rep, 100.0, rel_tol=1e-6)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Occupancy
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_occupancy_rate():
|
|
df = _sample_df()
|
|
op = OperationalPerformanceMetrics(df)
|
|
|
|
occ = op.occupancy_rate()
|
|
|
|
# handle_time = (600+60+30) + (700+30+40) + (400+20+30) + (100+50+10) + (300+0+20)
|
|
# = 690 + 770 + 450 + 160 + 320 = 2390
|
|
# logged_time total = 900 + 900 + 800 + 600 + 700 = 3900
|
|
expected_occ = 2390 / 3900 * 100
|
|
assert math.isclose(occ, round(expected_occ, 2), rel_tol=1e-6)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Performance Score
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_performance_score_structure_and_range():
|
|
df = _sample_df()
|
|
op = OperationalPerformanceMetrics(df)
|
|
|
|
score_info = op.performance_score()
|
|
assert "score" in score_info
|
|
assert 0.0 <= score_info["score"] <= 10.0
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Plots
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_plot_methods_return_axes():
|
|
df = _sample_df()
|
|
op = OperationalPerformanceMetrics(df)
|
|
|
|
ax1 = op.plot_aht_boxplot_by_skill()
|
|
ax2 = op.plot_resolution_funnel_by_skill()
|
|
|
|
from matplotlib.axes import Axes
|
|
|
|
assert isinstance(ax1, Axes)
|
|
assert isinstance(ax2, Axes)
|