Python

Visión General

El enriquecimiento Python es uno de los operadores de enriquecimiento más potentes y flexibles en mindzieStudio, que te permite escribir código Python personalizado para transformar, analizar y enriquecer tus datos de minería de procesos. Este operador ofrece acceso directo a tus datos del registro de eventos mediante Pandas DataFrames, permitiéndote aprovechar todo el ecosistema Python incluyendo bibliotecas como NumPy, Pandas y lógica de negocio personalizada para crear transformaciones sofisticadas de datos que van más allá de los enriquecimientos estándar.

Con una frecuencia de uso del 95%, Python es uno de los enriquecimientos más comunes en mindzieStudio. Cierra la brecha entre las operaciones estándar de minería de procesos y los flujos de trabajo avanzados de ciencia de datos, permitiendo a científicos de datos y analistas aplicar algoritmos personalizados, reglas comerciales complejas y análisis avanzados directamente dentro de su pipeline de minería de procesos. El operador integra sin problemas la ejecución de código Python con el modelo de datos de mindzieStudio, manejando automáticamente la serialización de datos, conversión de tipos e integración de resultados de vuelta a tu conjunto de datos.

Usos Comunes

  • Calcular KPIs complejos que requieren lógica de negocio personalizada no disponible en calculadoras estándar
  • Aplicar modelos de aprendizaje automático para predicción, clasificación o clustering directamente sobre datos del proceso
  • Realizar procesamiento avanzado de texto y procesamiento de lenguaje natural sobre atributos de eventos
  • Implementar verificaciones de conformidad personalizadas basadas en reglas comerciales complejas
  • Crear atributos derivados utilizando análisis estadístico y operaciones matemáticas avanzadas
  • Integrar fuentes de datos externas llamando a APIs o leyendo archivos externos desde el código Python
  • Construir chequeos personalizados de calidad de datos y reglas de validación específicas para tu dominio de negocio

Configuraciones

Filter: Filtro opcional para limitar qué casos serán procesados por el script Python. Esto permite aplicar transformaciones solo a subconjuntos específicos de tus datos, mejorando el rendimiento y habilitando análisis dirigidos. Cuando no se aplica filtro, el código Python procesa todos los casos del conjunto de datos.

Columns: Selecciona qué columnas existentes de tu conjunto de datos estarán disponibles para el script Python. Estas columnas serán accesibles en los DataFrames case_table y event_table. Solo se pasan a Python las columnas seleccionadas para minimizar el uso de memoria y mejorar el rendimiento. La columna CaseId siempre se incluye automáticamente.

Change Columns: Especifica cuáles de las columnas seleccionadas pueden ser modificadas por tu script Python. Esta configuración te permite actualizar los valores de atributos existentes manteniendo la integridad de los datos. Solo las columnas seleccionadas en Columns pueden marcarse para modificación.

New Columns: Define nuevos atributos que tu script Python creará. Para cada nueva columna, debes especificar:

  • Nombre de la Columna: el nombre interno utilizado en el código Python
  • Nombre para Mostrar: el nombre amigable mostrado en mindzieStudio
  • Tipo de Dato: el tipo de dato (String, Integer, DateTime, Boolean, Double)
  • Tipo de Origen: si el atributo se añade a nivel de Caso o Evento
  • Formato: formato opcional para la visualización del atributo

Python Code: El script Python que se ejecutará sobre tus datos. Tu código tiene acceso a:

  • case_table: DataFrame de Pandas que contiene atributos a nivel de caso
  • event_table: DataFrame de Pandas con datos a nivel de evento, incluyendo columnas para InternalEventIndex, CaseId, ActivityName, ActivityTime y cualquier atributo de evento seleccionado

El script debe modificar estos DataFrames en sitio. Cualquier cambio en columnas existentes (marcadas en Change Columns) o adiciones de nuevas columnas (definidas en New Columns) se integrarán automáticamente de vuelta en tu conjunto de datos.

Python Image: Especifica el entorno de ejecución Python. Las opciones incluyen:

  • LOCAL: Usa la instalación local de Python (si está disponible)
  • Nombre de imagen Docker: Imagen Docker específica con los paquetes Python requeridos
  • Default: entorno Python estándar de mindzie con bibliotecas comunes de ciencia de datos

Ejemplos

Ejemplo 1: Calcular Índice de Eficiencia de Procesamiento de Pedidos

Escenario: En un proceso de pedido a efectivo, necesitas calcular un índice personalizado de eficiencia basado en el valor del pedido, tiempo de procesamiento y número de actividades de retrabajo.

Configuraciones:

  • Filter: Ninguno (procesar todos los casos)
  • Columns: OrderValue, CustomerPriority
  • Change Columns: Ninguno
  • New Columns:
    • Nombre de Columna: EfficiencyScore
    • Nombre para Mostrar: Puntuación de Eficiencia
    • Tipo de Dato: Double
    • Tipo de Origen: Case
  • Python Code:
# Calcular puntuación de eficiencia para cada pedido
import numpy as np

# Contar actividades de retrabajo por caso
rework_activities = ['Order Correction', 'Price Adjustment', 'Approval Retry']
event_table['IsRework'] = event_table['ActivityName'].isin(rework_activities)
rework_counts = event_table.groupby('CaseId')['IsRework'].sum().reset_index()
rework_counts.columns = ['CaseId', 'ReworkCount']

# Calcular duración del caso en días
case_durations = event_table.groupby('CaseId')['ActivityTime'].agg(['min', 'max'])
case_durations['Duration'] = (case_durations['max'] - case_durations['min']).dt.total_seconds() / 86400
case_durations = case_durations.reset_index()[['CaseId', 'Duration']]

# Combinar con tabla de casos
case_table = case_table.merge(rework_counts, on='CaseId', how='left')
case_table = case_table.merge(case_durations, on='CaseId', how='left')

# Calcular puntuación de eficiencia (0-100)
case_table['EfficiencyScore'] = 100 * (
    (case_table['OrderValue'] / case_table['OrderValue'].max()) * 0.4 +
    (1 - case_table['ReworkCount'] / 10) * 0.3 +
    (1 - case_table['Duration'] / 30) * 0.3
)
case_table['EfficiencyScore'] = np.clip(case_table['EfficiencyScore'], 0, 100)

# Eliminar columnas temporales
case_table = case_table.drop(['ReworkCount', 'Duration'], axis=1)
  • Python Image: LOCAL

Resultado: Crea un nuevo atributo de caso "Puntuación de Eficiencia" con valores que van de 0 a 100, donde puntuaciones más altas indican un procesamiento de pedidos más eficiente basado en la combinación de valor del pedido, mínimo retrabajo y tiempo de procesamiento rápido.

Perspectivas: Esta puntuación personalizada ayuda a identificar cuáles pedidos se procesan de forma más eficiente y puede usarse para comparar el desempeño, identificar mejores prácticas y priorizar iniciativas de mejora del proceso.

Ejemplo 2: Detectar Secuencias Anómalas de Eventos

Escenario: En un proceso de tratamiento de pacientes en salud, identifica casos donde la secuencia de procedimientos médicos se desvía de los protocolos estándar.

Configuraciones:

  • Filter: Ninguno
  • Columns: PatientAge, Department
  • Change Columns: Ninguno
  • New Columns:
    • Nombre de Columna: HasAnomalousSequence
    • Nombre para Mostrar: Secuencia Anómala Detectada
    • Tipo de Dato: Boolean
    • Tipo de Origen: Case
    • Nombre de Columna: AnomalyDescription
    • Nombre para Mostrar: Descripción de la Anomalía
    • Tipo de Dato: String
    • Tipo de Origen: Case
  • Python Code:
# Definir patrones esperados de secuencia
normal_sequences = [
    ['Registration', 'Triage', 'Examination', 'Treatment', 'Discharge'],
    ['Registration', 'Triage', 'Examination', 'Lab Test', 'Treatment', 'Discharge'],
    ['Registration', 'Emergency Assessment', 'Treatment', 'Observation', 'Discharge']
]

def check_sequence_anomaly(group):
    activities = group.sort_values('ActivityTime')['ActivityName'].tolist()

    # Verificar actividades repetidas
    if len(activities) != len(set(activities)):
        return True, "Actividades repetidas detectadas"

    # Verificar actividades fuera de orden
    if 'Discharge' in activities and activities.index('Discharge') < len(activities) - 1:
        return True, "Actividades después del alta"

    if 'Registration' in activities and activities.index('Registration') > 0:
        return True, "Registro no es la primera actividad"

    # Verificar si la secuencia coincide con algún patrón normal
    matches_normal = any(
        all(act in activities for act in normal_seq)
        for normal_seq in normal_sequences
    )

    if not matches_normal and len(activities) > 3:
        return True, "Patrón de secuencia no estándar"

    return False, ""

# Aplicar detección de anomalías a cada caso
anomaly_results = event_table.groupby('CaseId').apply(check_sequence_anomaly)

# Añadir resultados a tabla de casos
case_table['HasAnomalousSequence'] = case_table['CaseId'].map(
    lambda x: anomaly_results[x][0] if x in anomaly_results.index else False
)
case_table['AnomalyDescription'] = case_table['CaseId'].map(
    lambda x: anomaly_results[x][1] if x in anomaly_results.index else ""
)
  • Python Image: LOCAL

Resultado: Crea dos nuevos atributos de caso:

  • "Secuencia Anómala Detectada": indicador booleano que señala si el caso tiene una secuencia inusual
  • "Descripción de la Anomalía": texto que explica el tipo de anomalía detectada

Perspectivas: Este enriquecimiento ayuda a identificar casos que se desvían de los protocolos médicos estándar, permitiendo a los equipos de aseguramiento de calidad investigar posibles problemas y garantizar la seguridad del paciente.

Ejemplo 3: Calcular Métricas de Desempeño de Proveedores

Escenario: En un proceso de compras, calcula métricas integrales de desempeño de proveedores basadas en tiempos de entrega, problemas de calidad y completitud de pedidos.

Configuraciones:

  • Filter: ActivityName = "Goods Receipt"
  • Columns: SupplierID, OrderQuantity, ReceivedQuantity
  • Change Columns: Ninguno
  • New Columns:
    • Nombre de Columna: OnTimeDeliveryRate
    • Nombre para Mostrar: Porcentaje de Entrega a Tiempo %
    • Tipo de Dato: Double
    • Tipo de Origen: Case
    • Nombre de Columna: QualityScore
    • Nombre para Mostrar: Puntuación de Calidad del Proveedor
    • Tipo de Dato: Double
    • Tipo de Origen: Case
  • Python Code:
# Calcular desempeño de entrega
def calculate_supplier_metrics(group):
    po_created = group[group['ActivityName'] == 'PO Created']['ActivityTime'].min()
    goods_received = group[group['ActivityName'] == 'Goods Receipt']['ActivityTime'].max()

    # Tiempo esperado de entrega son 5 días hábiles
    expected_days = 5
    actual_days = np.busday_count(po_created.date(), goods_received.date())
    on_time = 1 if actual_days <= expected_days else 0

    # Verificar problemas de calidad
    has_quality_issue = 'Quality Issue' in group['ActivityName'].values
    has_return = 'Return to Supplier' in group['ActivityName'].values

    quality_score = 100
    if has_quality_issue:
        quality_score -= 30
    if has_return:
        quality_score -= 40

    return pd.Series({
        'OnTimeDelivery': on_time,
        'QualityScore': quality_score
    })

# Calcular métricas para cada caso
supplier_metrics = event_table.groupby('CaseId').apply(calculate_supplier_metrics)

# Agregar por proveedor
supplier_performance = case_table.merge(supplier_metrics, left_on='CaseId', right_index=True)
supplier_summary = supplier_performance.groupby('SupplierID').agg({
    'OnTimeDelivery': 'mean',
    'QualityScore': 'mean'
}).reset_index()
supplier_summary.columns = ['SupplierID', 'OnTimeDeliveryRate', 'AvgQualityScore']

# Añadir de nuevo a la tabla de casos
case_table = case_table.merge(
    supplier_summary[['SupplierID', 'OnTimeDeliveryRate', 'AvgQualityScore']],
    on='SupplierID',
    how='left'
)
case_table['OnTimeDeliveryRate'] = case_table['OnTimeDeliveryRate'] * 100
case_table.rename(columns={'AvgQualityScore': 'QualityScore'}, inplace=True)
  • Python Image: LOCAL

Resultado: Crea métricas de desempeño del proveedor a nivel de caso:

  • "Porcentaje de Entrega a Tiempo %": porcentaje de órdenes entregadas a tiempo por este proveedor
  • "Puntuación de Calidad del Proveedor": puntuación de calidad de 0 a 100 basada en problemas de calidad y devoluciones

Perspectivas: Estas métricas permiten a los equipos de compras evaluar objetivamente el desempeño de los proveedores, apoyar decisiones de selección y detectar proveedores que requieran intervenciones para mejorar su desempeño.

Ejemplo 4: Minería de Texto en Comentarios de Proceso

Escenario: En un proceso de gestión de servicios TI, analiza comentarios de tickets para categorizar los problemas y detectar sentimiento.

Configuraciones:

  • Filter: Ninguno
  • Columns: TicketDescription, ResolutionNotes
  • Change Columns: Ninguno
  • New Columns:
    • Nombre de Columna: IssueCategory
    • Nombre para Mostrar: Categoría del Problema
    • Tipo de Dato: String
    • Tipo de Origen: Case
    • Nombre de Columna: CustomerSentiment
    • Nombre para Mostrar: Sentimiento del Cliente
    • Tipo de Dato: String
    • Tipo de Origen: Case
  • Python Code:
import re

# Definir palabras clave para categorización
category_keywords = {
    'Hardware': ['laptop', 'desktop', 'printer', 'mouse', 'keyboard', 'monitor', 'hardware'],
    'Software': ['application', 'software', 'program', 'install', 'update', 'crash', 'error'],
    'Network': ['network', 'internet', 'wifi', 'connection', 'vpn', 'firewall'],
    'Access': ['password', 'login', 'access', 'permission', 'authentication', 'account'],
    'Other': []
}

# Indicadores de sentimiento
negative_words = ['slow', 'broken', 'failed', 'cannot', 'unable', 'frustrated', 'urgent', 'critical']
positive_words = ['resolved', 'working', 'fixed', 'thank', 'great', 'excellent', 'happy']

def categorize_issue(text):
    if pd.isna(text):
        return 'Other'
    text_lower = text.lower()
    for category, keywords in category_keywords.items():
        if any(keyword in text_lower for keyword in keywords):
            return category
    return 'Other'

def analyze_sentiment(text):
    if pd.isna(text):
        return 'Neutral'
    text_lower = text.lower()
    negative_count = sum(1 for word in negative_words if word in text_lower)
    positive_count = sum(1 for word in positive_words if word in text_lower)

    if negative_count > positive_count:
        return 'Negative'
    elif positive_count > negative_count:
        return 'Positive'
    else:
        return 'Neutral'

# Aplicar análisis de texto
case_table['IssueCategory'] = case_table['TicketDescription'].apply(categorize_issue)
case_table['CustomerSentiment'] = case_table['TicketDescription'].apply(analyze_sentiment)
  • Python Image: LOCAL

Resultado: Crea dos atributos derivados de texto:

  • "Categoría del Problema": categorización del problema TI (Hardware, Software, Red, Acceso, Otro)
  • "Sentimiento del Cliente": resultado del análisis de sentimiento (Positivo, Negativo, Neutral)

Perspectivas: Este enriquecimiento permite a los gestores de servicios TI entender la distribución de problemas, priorizar según el sentimiento del cliente e identificar áreas que requieren recursos de soporte o capacitación adicional.

Ejemplo 5: Puntuación de Riesgo de Cumplimiento Financiero

Escenario: En un proceso de aprobación de transacciones financieras, calcula una puntuación de riesgo de cumplimiento basada en múltiples factores de riesgo.

Configuraciones:

  • Filter: TransactionType = "Wire Transfer"
  • Columns: TransactionAmount, CustomerCountry, AccountAge, PreviousFlags
  • Change Columns: Ninguno
  • New Columns:
    • Nombre de Columna: ComplianceRiskScore
    • Nombre para Mostrar: Puntuación de Riesgo de Cumplimiento
    • Tipo de Dato: Integer
    • Tipo de Origen: Case
    • Nombre de Columna: RiskLevel
    • Nombre para Mostrar: Nivel de Riesgo
    • Tipo de Dato: String
    • Tipo de Origen: Case
  • Python Code:
# Definir factores de riesgo y pesos
high_risk_countries = ['Country1', 'Country2', 'Country3']  # Marcador para lista real
suspicious_amount_threshold = 10000
rapid_transaction_threshold = 5  # transacciones por día

def calculate_risk_score(row):
    risk_score = 0

    # Riesgo por monto (0-30 puntos)
    if row['TransactionAmount'] > suspicious_amount_threshold:
        risk_score += min(30, int(row['TransactionAmount'] / suspicious_amount_threshold * 10))

    # Riesgo geográfico (0-25 puntos)
    if row['CustomerCountry'] in high_risk_countries:
        risk_score += 25

    # Riesgo por antigüedad de cuenta (0-20 puntos)
    if pd.notna(row['AccountAge']) and row['AccountAge'] < 30:
        risk_score += 20 - int(row['AccountAge'] / 30 * 20)

    # Riesgo por banderas previas (0-25 puntos)
    if pd.notna(row['PreviousFlags']) and row['PreviousFlags'] > 0:
        risk_score += min(25, row['PreviousFlags'] * 5)

    return risk_score

# Calcular velocidad de transacciones
transaction_counts = event_table[event_table['ActivityName'] == 'Transaction Initiated'].groupby('CaseId').size()
case_table['TransactionVelocity'] = case_table['CaseId'].map(transaction_counts).fillna(0)

# Calcular puntuaciones de riesgo
case_table['ComplianceRiskScore'] = case_table.apply(calculate_risk_score, axis=1)

# Asignar niveles de riesgo
def assign_risk_level(score):
    if score >= 70:
        return 'High'
    elif score >= 40:
        return 'Medium'
    else:
        return 'Low'

case_table['RiskLevel'] = case_table['ComplianceRiskScore'].apply(assign_risk_level)

# Eliminar columnas temporales
case_table = case_table.drop(['TransactionVelocity'], axis=1)
  • Python Image: LOCAL

Resultado: Crea atributos integrales de evaluación de riesgo:

  • "Puntuación de Riesgo de Cumplimiento": puntuación numérica de riesgo de 0 a 100
  • "Nivel de Riesgo": clasificación categórica de riesgo (Alto, Medio, Bajo)

Perspectivas: Esta puntuación de riesgo permite a los equipos de cumplimiento priorizar revisiones de transacciones, automatizar flujos de aprobación basados en niveles de riesgo y garantizar el cumplimiento regulatorio minimizando falsos positivos.

Salida

El operador de enriquecimiento Python produce atributos nuevos o modificados basados en tu código personalizado:

Nuevos Atributos de Caso: Cualquier columna añadida al DataFrame case_table que coincida con los Nuevos Columnas definidos se creará como atributos a nivel de caso en tu conjunto de datos. Estos atributos están disponibles inmediatamente para usar en filtros, calculadoras y otros enriquecimientos.

Nuevos Atributos de Evento: Cualquier columna añadida al DataFrame event_table que coincida con los Nuevos Columnas definidos se creará como atributos a nivel de evento. Estos pueden capturar cálculos o clasificaciones específicas de eventos.

Atributos Modificados: Las columnas existentes especificadas en Change Columns pueden tener sus valores actualizados. El tipo de dato original debe mantenerse, pero los valores pueden transformarse según tu lógica de negocio.

Manejo de Tipos de Datos: El operador maneja automáticamente la conversión de tipos entre Python y mindzieStudio:

  • Cadenas Python → String de mindzieStudio
  • int32/int64 Python → Integer de mindzieStudio
  • float Python → Double de mindzieStudio
  • datetime Python → DateTime de mindzieStudio
  • bool Python → Boolean de mindzieStudio

Eliminación de Casos y Eventos: El uso avanzado permite eliminar casos o eventos filtrándolos fuera de los DataFrames correspondientes. Los casos que no estén presentes en el DataFrame case_table de salida serán eliminados del conjunto de datos.

Los atributos enriquecidos se integran perfectamente con todas las demás funcionalidades de mindzieStudio, permitiéndote aprovechar transformaciones personalizadas de Python a lo largo de tu flujo de análisis de minería de procesos.

Véase También

Esta documentación es parte de la plataforma de minería de procesos mindzie Studio.