Arquitecto de automatización con IA: Día 83

# Importaciones base, semilla de la serie, fecha de hoy y las fechas del pico en Python:
import warnings
warnings.filterwarnings('ignore')
import hashlib
import json
import math
import re
import time
from collections import Counter, deque
from datetime import date, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.patches import Polygon

# Generador aleatorio con la semilla global de la serie:
rng = np.random.default_rng(2026)
pd.set_option('display.width', 160)
pd.set_option('display.max_colwidth', 48)
carpeta = Path('trigo_lento') / 'dia_83'
carpeta.mkdir(parents = True, exist_ok = True)
hoy = date(2026, 4, 23)
ensayo = date(2026, 4, 25)            # el sábado de ensayo con el horno lleno
dia_nino, diez_mayo = date(2026, 4, 30), date(2026, 5, 10)
DIAS = ['lunes', 'martes', 'miércoles', 'jueves', 'viernes', 'sábado', 'domingo']
print('numpy', np.__version__, '| pandas', pd.__version__)
print(f'Hoy: {DIAS[hoy.weekday()]} {hoy:%d/%m/%Y} | ensayo: {DIAS[ensayo.weekday()]} {ensayo:%d/%m} | '
      f'Día del Niño en {(dia_nino - hoy).days} días | 10 de mayo en {(diez_mayo - hoy).days} días '
      f'({(diez_mayo - hoy).days / 7:.1f} semanas)')

# Inventario de las pruebas que la serie ya tiene y construcción del conjunto de regresión de 200 mensajes en Python:
inventario = [
    # nivel, prueba, día de origen, casos, tiempo por corrida, coste por corrida (USD), determinista
    (1, 'Transformaciones (Code y Edit Fields)',     43, 30,  '0.4 s',        0.00, True),
    (1, 'Reglas (IF y Switch)',                      43, 30,  '0.2 s',        0.00, True),
    (2, 'Contratos de webhooks (Meta, Slack, formulario)', 83, 90, '0.3 s',  0.00, True),
    (3, 'Punta a punta en el simulador ensamblado',  83, 50,  '1 s',          0.00, True),
    (4, 'Caos y carga (fallos inyectados, triple del pico)', 83, 8, '10 s',   0.00, True),
    (5, 'Regresión de prompts (Haiku 5.5 por lotes)', 83, 400, 'menos de 1 h, hasta 24 h', 0.02, False),
    (6, 'Conducta del agente (Opus 5.5, 40 conversaciones)', 39, 40, '20 min', 2.88, False),
]
tabla_1 = pd.DataFrame(inventario, columns = ['nivel', 'prueba', 'dia', 'casos', 'tiempo', 'coste_usd', 'determinista'])
print(tabla_1.to_string(index = False))
print(f'Casos en total: {tabla_1.casos.sum()} | deterministas y gratis: '
      f'{tabla_1[tabla_1.determinista].casos.sum()} | coste de una pasada completa: {tabla_1.coste_usd.sum():.2f} USD')

# El único activo que todavía no existe es el conjunto de regresión: se parte de los 60 mensajes
# etiquetados del Día 9 y se completa con plantillas hasta 200, con cuotas por categoría.
base = pd.read_csv(Path('trigo_lento') / 'mensajes_etiquetados.csv')
CUOTAS = {'pedido': 50, 'horario': 30, 'precio': 30, 'factura': 20, 'queja': 30, 'otro': 25, 'baja': 15}
PRODUCTOS = ['hogazas de masa madre', 'baguettes', 'conchas', 'croissants', 'roles de canela', 'pasteles de zanahoria',
             'cajas de pan dulce surtido', 'bollos para hamburguesa', 'pays de queso', 'panes de caja integral']
PLANTILLAS = {
    'pedido':  ['Quiero {n} {p} para el {d}', 'Me aparta {n} {p} pa {d} porfa', 'Hola, necesito encargar {n} {p}, los recojo el {d} a las {h}',
                '¿Me pueden hacer {n} {p} para {d}?', 'Buenas, un pedido: {n} {p} y {n2} {p2} para el {d}', 'Apúnteme {n} {p}, paso el {d}'],
    'horario': ['¿A qué hora abren el {d}?', '¿Están abiertos hoy?', '¿Hasta qué hora puedo pasar por mi pan el {d}?', '¿Abren los domingos?',
                '¿Cierran el puente del {d}?', 'Hola, ¿a qué hora sale la hornada de la tarde?'],
    'precio':  ['¿Cuánto cuesta {p}?', 'Precio de {p} por favor', '¿Tienen precio de mayoreo para {p}?', '¿Cuánto salen {n} {p}?',
                '¿Ya subió el precio de los {p}?', 'Buenas, ¿cuánto vale la caja de pan dulce?'],
    'factura': ['Necesito factura del pedido {f}', '¿Me pueden facturar la compra de ayer? Le paso mi RFC', 'La factura de la semana pasada trae mal el RFC',
                '¿Facturan? Es para la oficina', 'Mándenme el XML de la factura {f}', 'Mi contadora dice que falta el uso de CFDI en la factura {f}'],
    'queja':   ['El pan llegó quemado', 'Mi pastel llegó tarde y aplastado', 'Pedí {p} y me dieron otra cosa', 'Qué decepción, los {p} estaban crudos',
                'Llevo una hora esperando y nadie contesta', 'Los {p} de hoy venían duros, no es la primera vez'],
    'otro':    ['Gracias, todo muy rico', '¿Hacen envíos a Coyoacán?', 'Hola', '¿Tienen estacionamiento?', 'Feliz día', 'ok', '¿Trabajan con cafeterías?',
                'Les mando la foto del pastel, quedó precioso', '¿Me recomiendan algo sin azúcar?'],
    'baja':    ['BAJA', 'Ya no quiero recibir mensajes', 'No me escriban más por favor', 'Dejen de mandarme promociones', 'baja', 'Ya no',
                'Quiero darme de baja de los avisos', 'Por favor quítenme de la lista'],
}
def mensaje(categoria):
    t = PLANTILLAS[categoria][int(rng.integers(len(PLANTILLAS[categoria])))]
    p, p2 = rng.choice(PRODUCTOS, 2, replace = False)
    return t.format(n = int(rng.integers(1, 13)), n2 = int(rng.integers(1, 7)), p = p, p2 = p2,
                    d = rng.choice(DIAS[:6]), h = f'{int(rng.integers(8, 19))}:00', f = f'TL-202604{int(rng.integers(10, 23)):02d}-{int(rng.integers(1, 400)):04d}')

filas = [{'id': f'r{i + 1:03d}', 'texto': r.texto, 'real': r.real, 'origen': 'dia_9'} for i, r in enumerate(base.itertuples())]
for categoria, cuota in CUOTAS.items():
    faltan = cuota - sum(f['real'] == categoria for f in filas)
    for _ in range(faltan):
        filas.append({'id': f'r{len(filas) + 1:03d}', 'texto': mensaje(categoria), 'real': categoria, 'origen': 'plantilla'})
regresion = pd.DataFrame(filas)
regresion.to_csv(carpeta / 'mensajes_regresion.csv', index = False)
print(f'\nConjunto de regresión: {len(regresion)} mensajes ({(regresion.origen == "dia_9").sum()} del Día 9 y '
      f'{(regresion.origen == "plantilla").sum()} de plantilla) | por categoría: {regresion.real.value_counts().reindex(CUOTAS).to_dict()}')
print(regresion.sample(5, random_state = 2026).to_string(index = False))

# Pirámide de pruebas del ecosistema dibujada con patches, con los casos, el coste y el tiempo de cada nivel en Python:
niveles = tabla_1.groupby('nivel').agg(casos = ('casos', 'sum'), coste = ('coste_usd', 'sum'),
                                        nombres = ('prueba', lambda s: ' y '.join(p.split(' (')[0] for p in s))).reset_index()
rotulos = {1: 'Transformaciones y reglas', 2: 'Contratos de webhooks', 3: 'Punta a punta (simulador)',
           4: 'Caos y carga', 5: 'Regresión de prompts', 6: 'Conducta del agente'}
tiempos = {1: '< 1 s', 2: '0.3 s', 3: '1 s', 4: '10 s', 5: 'hasta 24 h', 6: '20 min'}
colores = {1: '#3a9d5d', 2: '#3a9d5d', 3: '#2a78d6', 4: '#2a78d6', 5: '#eb6834', 6: '#c8423b'}
fig, ax = plt.subplots(figsize = (9.2, 6.2))
n = len(niveles)
for k, fila in niveles.iterrows():
    nivel = int(fila.nivel)
    y0, y1 = k, k + 0.92
    w0, w1 = 1 - k / n, 1 - (k + 1) / n           # anchos de la base y del techo del trapecio
    ax.add_patch(Polygon([(-w0, y0), (w0, y0), (w1, y1), (-w1, y1)], closed = True, fc = colores[nivel], ec = 'white', lw = 2))
    rotulo = f'{nivel}. {rotulos[nivel]}\n{int(fila.casos)} casos · {tiempos[nivel]} · {fila.coste:.2f} USD'
    if k < 4:                                      # los niveles anchos llevan el rótulo dentro
        ax.text(0, (y0 + y1) / 2, rotulo, ha = 'center', va = 'center', color = 'white', fontsize = 10, fontweight = 'bold')
    else:                                          # la punta es estrecha: rótulo fuera, con una línea
        ax.annotate(rotulo, xy = (-(w0 + w1) / 2, (y0 + y1) / 2), xytext = (-1.7, (y0 + y1) / 2), ha = 'left', va = 'center',
                    fontsize = 9, color = colores[nivel], fontweight = 'bold', arrowprops = dict(arrowstyle = '-', lw = 1, color = colores[nivel]))
ax.annotate('', xy = (1.08, n - 0.3), xytext = (1.08, 0.3), arrowprops = dict(arrowstyle = '->', lw = 1.5, color = '#2b2a28'))
ax.text(1.12, n / 2, 'más lento, más caro,\nmenos determinista', rotation = 90, va = 'center', fontsize = 9, color = '#2b2a28')
ax.text(-1.7, -0.45, 'Verde: deterministas y gratis · Azul: simulación · Naranja y rojo: cuestan tokens', fontsize = 9, color = '#555555')
ax.set_xlim(-1.75, 1.35)
ax.set_ylim(-0.6, n + 0.1)
ax.axis('off')
ax.set_title(f'La pirámide de pruebas del ecosistema: {tabla_1.casos.sum()} casos, {tabla_1.coste_usd.sum():.2f} USD por pasada', fontsize = 16)
plt.tight_layout()
plt.show()

# Pruebas de contrato de los tres webhooks con el validador de JSON Schema del Día 4 en Python:
TIPOS = {'string': str, 'integer': int, 'object': dict, 'array': list}

def validar(dato, esquema, ruta = '$'):
    # Devuelve una lista de errores (ruta, regla, mensaje); lista vacía = cumple el contrato:
    if 'enum' in esquema and dato not in esquema['enum']:
        return [(ruta, 'enum', f'{dato!r} no está en la lista permitida')]
    tipo = esquema.get('type')
    if tipo and (not isinstance(dato, TIPOS[tipo]) or (tipo == 'integer' and isinstance(dato, bool))):
        return [(ruta, 'type', f'se esperaba {tipo}')]
    errores = []
    if tipo == 'object':
        for campo in esquema.get('required', []):
            if campo not in dato:
                errores.append((f'{ruta}.{campo}', 'required', 'falta el campo'))
        for campo, valor in dato.items():
            if campo in esquema.get('properties', {}):
                errores += validar(valor, esquema['properties'][campo], f'{ruta}.{campo}')
            elif esquema.get('additionalProperties') is False:
                errores.append((f'{ruta}.{campo}', 'additionalProperties', 'campo no previsto'))
    elif tipo == 'array':
        if not esquema.get('minItems', 0) <= len(dato) <= esquema.get('maxItems', 10**9):
            errores.append((ruta, 'items', f'{len(dato)} elementos fuera del rango permitido'))
        for i, elemento in enumerate(dato):
            errores += validar(elemento, esquema['items'], f'{ruta}[{i}]')
    elif tipo == 'string':
        if not esquema.get('minLength', 0) <= len(dato) <= esquema.get('maxLength', 10**9):
            errores.append((ruta, 'length', f'longitud {len(dato)} fuera de rango'))
        if 'pattern' in esquema and not re.search(esquema['pattern'], dato):
            errores.append((ruta, 'pattern', f'{dato!r} no tiene el formato'))
        if esquema.get('format') == 'date':
            try:
                date.fromisoformat(dato) if re.fullmatch(r'\d{4}-\d{2}-\d{2}', dato) else 1 / 0
            except (ValueError, ZeroDivisionError):
                errores.append((ruta, 'format', f'{dato!r} no es una fecha ISO 8601'))
    return errores

# Los tres contratos de entrada del ecosistema, escritos en JSON Schema 2020-12 como el del pedido (Día 4):
MENSAJE_META = {'type': 'object', 'required': ['from', 'id', 'timestamp', 'type'], 'properties': {
    'from': {'type': 'string', 'pattern': r'^52[0-9]{10}$'}, 'id': {'type': 'string', 'pattern': r'^wamid\.'},
    'timestamp': {'type': 'string', 'pattern': r'^[0-9]{10}$'}, 'type': {'enum': ['text', 'image', 'audio', 'button', 'interactive']},
    'text': {'type': 'object', 'required': ['body'], 'properties': {'body': {'type': 'string', 'minLength': 1, 'maxLength': 4096}}}}}
VALOR_META = {'type': 'object', 'required': ['messaging_product', 'metadata', 'messages'], 'properties': {
    'messaging_product': {'enum': ['whatsapp']},
    'metadata': {'type': 'object', 'required': ['phone_number_id'], 'properties': {'phone_number_id': {'type': 'string', 'pattern': r'^[0-9]{6,20}$'}}},
    'messages': {'type': 'array', 'minItems': 1, 'maxItems': 20, 'items': MENSAJE_META}}}
ESQUEMAS = {
 'meta': {'type': 'object', 'required': ['object', 'entry'], 'additionalProperties': False, 'properties': {
    'object': {'enum': ['whatsapp_business_account']},
    'entry': {'type': 'array', 'minItems': 1, 'items': {'type': 'object', 'required': ['id', 'changes'], 'additionalProperties': False, 'properties': {
        'id': {'type': 'string', 'pattern': r'^[0-9]{6,20}$'},
        'changes': {'type': 'array', 'minItems': 1, 'items': {'type': 'object', 'required': ['field', 'value'], 'additionalProperties': False, 'properties': {
            'field': {'enum': ['messages']}, 'value': VALOR_META}}}}}}}},
 'slack': {'type': 'object', 'required': ['type', 'user', 'actions', 'response_url'], 'properties': {
    'type': {'enum': ['block_actions']},
    'user': {'type': 'object', 'required': ['id'], 'properties': {'id': {'type': 'string', 'pattern': r'^U[A-Z0-9]{8,}$'}}},
    'actions': {'type': 'array', 'minItems': 1, 'maxItems': 1, 'items': {'type': 'object', 'required': ['action_id', 'value'], 'properties': {
        'action_id': {'enum': ['aprobar', 'rechazar']}, 'value': {'type': 'string', 'pattern': r'^apr-[0-9]{8}-[0-9]{4}$'}}}},
    'response_url': {'type': 'string', 'pattern': r'^https://hooks\.slack\.com/actions/'}}},
 'formulario': {'type': 'object', 'required': ['folio', 'nombre', 'telefono', 'detalle', 'fecha_entrega', 'enviado'], 'additionalProperties': False, 'properties': {
    'folio': {'type': 'string', 'pattern': r'^FO-[0-9]{6}$'}, 'nombre': {'type': 'string', 'minLength': 2, 'maxLength': 80},
    'telefono': {'type': 'string', 'pattern': r'^\+52 ?[0-9]{2} ?[0-9]{4} ?[0-9]{4}$'}, 'detalle': {'type': 'string', 'minLength': 5, 'maxLength': 500},
    'fecha_entrega': {'type': 'string', 'format': 'date'}, 'enviado': {'type': 'string', 'pattern': r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[+-]\d{2}:\d{2}$'},
    'consentimiento': {'enum': [True, False]}}},
}

def cuerpo_valido(fuente, k):
    marca = int(datetime(2026, 4, 23, 10, 0).timestamp()) + 37 * k
    if fuente == 'meta':
        return {'object': 'whatsapp_business_account', 'entry': [{'id': '100000000000042', 'changes': [{'field': 'messages', 'value': {
                'messaging_product': 'whatsapp', 'metadata': {'display_phone_number': '5255000000XX', 'phone_number_id': '200000000000042'},
                'contacts': [{'wa_id': f'52550000{k:04d}'}],
                'messages': [{'from': f'52550000{k:04d}', 'id': f'wamid.EJEMPLO{k:04d}', 'timestamp': str(marca), 'type': 'text',
                              'text': {'body': regresion.texto.iloc[k % len(regresion)]}}]}}]}]}
    if fuente == 'slack':
        return {'type': 'block_actions', 'user': {'id': f'U0EJEMPLO{k:02d}'}, 'response_url': f'https://hooks.slack.com/actions/T0EJ/{k}/x',
                'actions': [{'action_id': 'aprobar' if k % 3 else 'rechazar', 'value': f'apr-20260423-{k:04d}'}], 'trigger_id': f'{k}.ejemplo'}
    return {'folio': f'FO-{260400 + k:06d}', 'nombre': 'Cliente de prueba', 'telefono': f'+52 55 0000 {k:04d}',
            'detalle': f'{1 + k % 6} x P{1 + k % 12:02d} para el sábado', 'fecha_entrega': '2026-04-25',
            'enviado': f'2026-04-23T10:{k % 60:02d}:00-06:00', 'consentimiento': True}

def mutar(cuerpo, fuente, rng):
    # Rompe un cuerpo válido de una de seis maneras y dice cuál, para saber qué regla debe atraparlo:
    c = json.loads(json.dumps(cuerpo))
    mutacion = ['campo_faltante', 'tipo_incorrecto', 'campo_extra', 'formato', 'lista_vacia', 'enum'][int(rng.integers(6))]
    if fuente == 'meta':
        m = c['entry'][0]['changes'][0]['value']['messages']
        {'campo_faltante': lambda: m[0].pop('timestamp'), 'tipo_incorrecto': lambda: m[0].update(timestamp = 1777000000),
         'campo_extra': lambda: c.update(firma = 'x'), 'formato': lambda: m[0].update({'from': '+52 55 0000 0001'}),
         'lista_vacia': lambda: m.clear(), 'enum': lambda: m[0].update(type = 'sticker')}[mutacion]()
    elif fuente == 'slack':
        {'campo_faltante': lambda: c.pop('response_url'), 'tipo_incorrecto': lambda: c.update(actions = 'aprobar'),
         'campo_extra': lambda: c['actions'].append({'action_id': 'aprobar', 'value': 'apr-20260423-9999'}),
         'formato': lambda: c['actions'][0].update(value = 'pastel-42'), 'lista_vacia': lambda: c['actions'].clear(),
         'enum': lambda: c['actions'][0].update(action_id = 'publicar')}[mutacion]()
    else:
        {'campo_faltante': lambda: c.pop('fecha_entrega'), 'tipo_incorrecto': lambda: c.update(nombre = 42),
         'campo_extra': lambda: c.update(cupon = 'GRATIS'), 'formato': lambda: c.update(fecha_entrega = '25/04/2026'),
         'lista_vacia': lambda: c.update(detalle = ''), 'enum': lambda: c.update(consentimiento = 'sí')}[mutacion]()
    return c, mutacion

t0 = time.perf_counter()
resultados = []
for fuente in ESQUEMAS:
    for k in range(30):
        cuerpo = cuerpo_valido(fuente, k)
        mutacion = 'ninguna'
        if k % 5 == 4:                                        # 6 de cada 30 cuerpos llegan rotos
            cuerpo, mutacion = mutar(cuerpo, fuente, rng)
        errores = validar(cuerpo, ESQUEMAS[fuente])
        resultados.append({'fuente': fuente, 'k': k, 'mutacion': mutacion, 'errores': len(errores),
                           'regla': errores[0][1] if errores else '', 'correcto': (mutacion == 'ninguna') == (not errores)})
contratos = pd.DataFrame(resultados)
contratos.to_csv(carpeta / 'pruebas_contrato.csv', index = False)
print(f'{len(contratos)} cuerpos validados en {time.perf_counter() - t0:.3f} s')
print(contratos.groupby('fuente').agg(cuerpos = ('k', 'size'), validos = ('errores', lambda e: int((e == 0).sum())),
                                      rechazados = ('errores', lambda e: int((e > 0).sum())), veredicto_correcto = ('correcto', 'all')).to_string())
print('\nRegla que atrapó cada mutación:')
print(contratos[contratos.mutacion != 'ninguna'].groupby(['mutacion', 'regla']).size().unstack(fill_value = 0).to_string())
print('\nAprobados: todos los veredictos correctos =', bool(contratos.correcto.all()),
      '| cuerpos rotos que el validador dejó pasar =', int(((contratos.mutacion != 'ninguna') & (contratos.errores == 0)).sum()))

# Evaluación de regresión de los 200 mensajes con Haiku 5.5 por la API de lotes, dos versiones del prompt en un solo lote en Python:
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

client = anthropic.Anthropic()   # lee ANTHROPIC_API_KEY del entorno; la clave nunca va en el código
regresion = pd.read_csv(carpeta / 'mensajes_regresion.csv')
CATEGORIAS = ['pedido', 'horario', 'precio', 'factura', 'queja', 'otro', 'baja']
ESQUEMA = {'type': 'object', 'additionalProperties': False, 'required': ['categoria'],
           'properties': {'categoria': {'type': 'string', 'enum': CATEGORIAS}}}
SISTEMA = {
    'v2': (Path('trigo_lento') / 'prompts' / 'clasificador_v2.md').read_text(encoding = 'utf-8'),
    'v3': (Path('trigo_lento') / 'prompts' / 'clasificador_v3.md').read_text(encoding = 'utf-8'),   # candidata: instrucciones de "baja"
}

def peticion(version, fila):
    # Un custom_id por versión y mensaje; el system va con caché de 1 hora porque el lote puede tardar más de 5 minutos:
    return Request(custom_id = f'{version}-{fila.id}', params = MessageCreateParamsNonStreaming(
        model = 'claude-haiku-5-5', max_tokens = 256, output_config = {'effort': 'low', 'format': {'type': 'json_schema', 'schema': ESQUEMA}},
        system = [{'type': 'text', 'text': SISTEMA[version], 'cache_control': {'type': 'ephemeral', 'ttl': '1h'}}],
        messages = [{'role': 'user', 'content': f'Mensaje del cliente: {fila.texto}'}]))

peticiones = [peticion(v, fila) for v in ('v2', 'v3') for fila in regresion.itertuples()]
assert len({p['custom_id'] for p in peticiones}) == len(peticiones) == 400, 'custom_id repetido'
lote = client.messages.batches.create(requests = peticiones)
print('Lote:', lote.id, '|', lote.processing_status, '| en proceso:', lote.request_counts.processing)

# Sondeo hasta que termine (la mayoría de los lotes acaban en menos de una hora; el límite es 24 h):
while lote.processing_status != 'ended':
    time.sleep(120)
    lote = client.messages.batches.retrieve(lote.id)
print('Terminado:', lote.request_counts)

# Lectura por custom_id (los resultados llegan en cualquier orden) y aciertos frente a la etiqueta:
reales = regresion.set_index('id')['real']
predicho, uso = {}, Counter()
for resultado in client.messages.batches.results(lote.id):
    version, id_msg = resultado.custom_id.split('-', 1)
    if resultado.result.type != 'succeeded':
        predicho[(version, id_msg)] = f'error:{resultado.result.type}'
        continue
    mensaje = resultado.result.message
    if mensaje.stop_reason == 'refusal':
        predicho[(version, id_msg)] = 'refusal'
        continue
    predicho[(version, id_msg)] = json.loads(mensaje.content[0].text)['categoria']
    uso['entrada'] += mensaje.usage.input_tokens
    uso['cache'] += mensaje.usage.cache_read_input_tokens or 0
    uso['salida'] += mensaje.usage.output_tokens
for version in ('v2', 'v3'):
    aciertos = sum(predicho[(version, i)] == reales[i] for i in reales.index)
    print(f'{version}: {aciertos} de {len(reales)} aciertos = {aciertos / len(reales):.3f}')
coste = (uso['entrada'] * 0.10 + uso['cache'] * 0.01 + uso['salida'] * 0.50) / 1e6 * 0.5     # Haiku 5.5, mitad de precio en lote
print(f'Tokens: {dict(uso)} | coste del lote: {coste:.4f} USD')
pd.DataFrame([{'id': i, 'real': reales[i], 'v2': predicho[('v2', i)], 'v3': predicho[('v3', i)]} for i in reales.index]
             ).to_csv(carpeta / 'predicciones_regresion.csv', index = False)

# Salida de ejemplo (puede variar):
# Lote: msgbatch_EJEMPLO-NO-REAL-0083 | in_progress | en proceso: 400
# Terminado: MessageBatchRequestCounts(canceled=0, errored=0, expired=0, processing=0, succeeded=400)
# v2: 182 de 200 aciertos = 0.910
# v3: 188 de 200 aciertos = 0.940
# Tokens: {'entrada': 31200, 'cache': 146000, 'salida': 7600} | coste del lote: 0.0042 USD

# Exactitud por categoría e intervalo de Wilson sobre predicciones de ejemplo (declaradas como tales) en Python:
def wilson(x, n, z = 1.96):
    # Intervalo de Wilson al 95 %: centro desplazado y margen que nunca sale de [0, 1]:
    p = x / n
    centro = (p + z**2 / (2 * n)) / (1 + z**2 / n)
    margen = z / (1 + z**2 / n) * math.sqrt(p * (1 - p) / n + z**2 / (4 * n**2))
    return centro - margen, centro + margen

# Aciertos por categoría de las dos versiones del prompt. Son cifras de ejemplo, no resultados de un
# lote real: la versión v2 es la que corre hoy y la v3 es la candidata que cambia las instrucciones de "baja":
ACIERTOS = {'pedido': (47, 48), 'horario': (29, 29), 'precio': (27, 28), 'factura': (17, 18),
            'queja': (28, 27), 'otro': (20, 23), 'baja': (14, 15)}
predicciones = regresion.copy()
for version, j in (('v2', 0), ('v3', 1)):
    acierto = pd.Series(True, index = predicciones.index)
    for categoria, (a2, a3) in ACIERTOS.items():
        indices = predicciones.index[predicciones.real == categoria]
        fallos = rng.choice(indices, size = len(indices) - (a2, a3)[j], replace = False)
        acierto[fallos] = False
    confusiones = [c for c in CUOTAS if c != 'otro'] + ['otro']
    predicciones[version] = [r if ok else rng.choice([c for c in confusiones if c != r]) for r, ok in zip(predicciones.real, acierto)]
predicciones.to_csv(carpeta / 'predicciones_ejemplo.csv', index = False)

filas = []
for categoria in list(CUOTAS) + ['total']:
    sub = predicciones if categoria == 'total' else predicciones[predicciones.real == categoria]
    fila = {'categoria': categoria, 'n': len(sub)}
    for version in ('v2', 'v3'):
        x = int((sub[version] == sub.real).sum())
        bajo, alto = wilson(x, len(sub))
        fila.update({f'{version}_x': x, f'{version}_p': x / len(sub), f'{version}_bajo': bajo, f'{version}_alto': alto})
    fila['regresion'] = fila['v3_alto'] < fila['v2_bajo']                  # el intervalo nuevo queda entero por debajo
    fila['vigilar'] = (not fila['regresion']) and fila['v3_p'] < fila['v2_p']
    fila['guardia_0.85'] = fila['v3_p'] >= 0.85                             # ninguna categoría por debajo de 0.85
    filas.append(fila)
tabla_3 = pd.DataFrame(filas)
print(tabla_3.round(3).to_string(index = False))
total = tabla_3[tabla_3.categoria == 'total'].iloc[0]
print(f'\nTotal v2: {int(total.v2_x)}/200 = {total.v2_p:.3f} [{total.v2_bajo:.3f}, {total.v2_alto:.3f}] | '
      f'v3: {int(total.v3_x)}/200 = {total.v3_p:.3f} [{total.v3_bajo:.3f}, {total.v3_alto:.3f}]')
print('Veredicto: regresión significativa en', tabla_3[tabla_3.regresion].categoria.tolist() or 'ninguna categoría',
      '| vigilar:', tabla_3[tabla_3.vigilar].categoria.tolist(), '| guardia 0.85 incumplida:', tabla_3[~tabla_3['guardia_0.85']].categoria.tolist())

# Tamaño de muestra para distinguir 0.91 de 0.94 al 95 % con potencia del 80 %:
p1, p2, z_a, z_b = 0.91, 0.94, 1.96, 0.84
pm = (p1 + p2) / 2
n_necesario = (z_a * math.sqrt(2 * pm * (1 - pm)) + z_b * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2)))**2 / (p2 - p1)**2
print(f'Mensajes necesarios por versión para distinguir 0.91 de 0.94: {math.ceil(n_necesario)}')

fig, ax = plt.subplots(figsize = (9.2, 5.4))
d = tabla_3.copy()
xs = np.arange(len(d))
for j, (version, color, nombre) in enumerate([('v2', '#9aa5b1', 'v2 (en producción)'), ('v3', '#2a78d6', 'v3 (candidata)')]):
    pos = xs + (j - 0.5) * 0.38
    ax.bar(pos, d[f'{version}_p'], width = 0.36, color = color, label = nombre)
    ax.errorbar(pos, d[f'{version}_p'], yerr = [d[f'{version}_p'] - d[f'{version}_bajo'], d[f'{version}_alto'] - d[f'{version}_p']],
                fmt = 'none', ecolor = '#2b2a28', capsize = 4, lw = 1.2)
    for x, p, n_ in zip(pos, d[f'{version}_p'], d.n):
        ax.text(x, 0.52, f'{p:.2f}', ha = 'center', fontsize = 8, color = 'white', fontweight = 'bold')
ax.axhline(total.v2_bajo, color = '#c8423b', ls = '--', lw = 1.5, label = f'Línea de no regresión: {total.v2_bajo:.3f} (límite inferior de v2)')
ax.axhline(0.85, color = '#eb6834', ls = ':', lw = 1.2, label = 'Guardia por categoría: 0.85')
ax.set_xticks(xs)
ax.set_xticklabels([f'{c}\n(n = {n_})' for c, n_ in zip(d.categoria, d.n)], fontsize = 9)
ax.set_ylim(0.5, 1.02)
ax.set_ylabel('Exactitud (intervalo de Wilson al 95 %)')
ax.set_title('Regresión de prompts: 200 mensajes, v2 frente a v3 por categoría', fontsize = 16)
ax.legend(loc = 'upper center', bbox_to_anchor = (0.5, -0.14), ncol = 2, frameon = False, fontsize = 8)
ax.spines[['top', 'right']].set_visible(False)
plt.tight_layout()
plt.show()

# Cincuenta escenarios de punta a punta en el simulador ensamblado del ecosistema en Python:
HORARIO = (7, 20)                                      # atención de lunes a sábado, de 7:00 a 20:00 (Día 52)
REGLAS = [('baja', r'^\W*(baja|ya no)\W*$'), ('queja', r'queja|lleg[oó] tarde|crud|quemad|reclam|decepci|duros|aplastad'),
          ('factura', r'factur|cfdi|rfc|xml'), ('horario', r'hora|abren|cierran|abiertos|domingo'),
          ('precio', r'precio|cu[aá]nto (cuesta|vale|sale)|mayoreo'), ('alergenos', r'nuez|nueces|gluten|huevo|l[aá]ct|cacahuate|al[eé]rg'),
          ('pedido', r'pedido|encarg|quiero \d|aparta|necesito \d|me pueden hacer|ap[uú]nteme')]
PRECIO_UMBRAL = 1000                                   # un pedido de 1 000 MXN o más pasa por aprobación (Día 74)

def tipo_por_reglas(texto):
    for tipo, patron in REGLAS:
        if re.search(patron, texto.lower()):
            return tipo
    return 'otro'

def en_horario(hora):
    return HORARIO[0] <= hora.hour < HORARIO[1]

def lognormal(rng, mediana, sigma = 0.35):
    return float(rng.lognormal(math.log(mediana), sigma))

def detectar(estado, clave, fallo, umbral = 3):
    # El vigilante de cada componente: alerta cuando hay `umbral` fallos entre sus últimos diez eventos:
    estado[clave].append(fallo)
    return sum(estado[clave]) >= umbral

def estado_nuevo():
    return {clave: deque(maxlen = 10) for clave in ('firmas', 'latidos', 'claude', 'notion', 'slack', 'meta')} | {'bajas': set()}

def ecosistema(evento, estado, rng, fallos = None):
    # Un mensaje recorre el ecosistema: firma, servidor, horario, baja, clasificación, enrutamiento, Notion, Slack y envío.
    # `fallos` inyecta averías; `estado` guarda los vigilantes y las bajas. Devuelve lo que pasó y cuánto tardó.
    f = fallos or {}
    r = {'id': evento['id'], 'ruta': None, 'acciones': [], 'claude': 0, 'salientes': 0, 'manual': None, 'latencia': lognormal(rng, 0.05, 0.2)}
    if rng.random() < f.get('firma', 0):               # firma X-Hub-Signature-256 que no coincide: 401 y nada más
        r['ruta'] = 'rechazado_401'
        if detectar(estado, 'firmas', True, umbral = 5):
            r['manual'] = 'M06'
        return r
    detectar(estado, 'firmas', False)
    if f.get('n8n'):                                   # el servidor no contesta: Meta reintentará la entrega
        r.update(ruta = 'sin_respuesta', latencia = float('nan'))
        if detectar(estado, 'latidos', True, umbral = 2):
            r['manual'] = 'M01'
        return r
    detectar(estado, 'latidos', False)
    if not en_horario(evento['hora']):
        r.update(ruta = 'cola_fuera_horario', acciones = ['plantilla_horario', 'encolar'], salientes = 1)
        return r
    texto = evento['texto']
    if re.search(REGLAS[0][1], texto.lower()):
        estado['bajas'].add(evento['cliente'])
        r.update(ruta = 'baja', acciones = ['registrar_baja', 'confirmar_baja'], salientes = 1)
        return r
    # Clasificación con Claude (Haiku 5.5): el SDK reintenta dos veces con espera exponencial ante 429 y 529.
    intentos, exito = 0, False
    while intentos < 3 and not exito:
        intentos += 1
        r['latencia'] += lognormal(rng, 1.2)
        codigo, p = f.get('claude', (None, 0))
        if rng.random() < p:
            r['latencia'] += 0.5 * 2**(intentos - 1)
            continue
        exito = True
    r['claude'] = intentos
    if detectar(estado, 'claude', not exito):
        r['manual'] = 'M03'
    if not exito:
        r.update(ruta = 'plantilla_degradada', acciones = ['plantilla_recibido', 'escalar_persona'], salientes = 1)
        return r
    tipo = tipo_por_reglas(texto)
    if tipo == 'alergenos':
        espera = f.get('notion_s', lognormal(rng, 0.8))
        r['latencia'] += min(espera, 3.0)
        if detectar(estado, 'notion', espera > 3.0):   # tiempo límite de 3 s en el nodo HTTP Request
            r['manual'] = 'M07'
        if espera > 3.0:
            r.update(ruta = 'respuesta_sin_fuentes', acciones = ['responder_generico', 'escalar_persona'], salientes = 1)
            return r
        r.update(ruta = 'respuesta_con_fuentes', acciones = ['buscar_notion', 'responder_citando'])
    elif tipo == 'pedido':
        r['latencia'] += lognormal(rng, 1.6)            # el agente extrae renglones, fecha y hora
        r['acciones'] += ['extraer_pedido', 'registrar_sqlite']
        if evento.get('total', 0) >= PRECIO_UMBRAL or evento.get('ambiguo'):
            r['ruta'] = 'aprobacion_humana'
            r['acciones'].append('slack_aprobacion')
        else:
            r.update(ruta = 'borrador_pedido', acciones = r['acciones'] + ['confirmar_plantilla', 'avisar_obrador'])
    elif tipo == 'queja':
        r.update(ruta = 'revision_humana', acciones = ['acuse_recibo', 'slack_aprobacion'])
    elif tipo in ('horario', 'precio', 'factura'):
        r.update(ruta = 'respuesta_catalogo', acciones = ['responder_catalogo'] + (['slack_aviso'] if tipo == 'factura' else []))
    else:
        r.update(ruta = 'revision_humana', acciones = ['acuse_recibo', 'slack_aprobacion'])
    # Efectos: avisos a Slack y mensajes salientes por la Cloud API.
    if any(a.startswith('slack_') for a in r['acciones']):
        if detectar(estado, 'slack', bool(f.get('slack'))):
            r['manual'] = 'M08'
        if f.get('slack'):                              # plan B del Día 77: la aprobación va por WhatsApp a Valeria
            r['acciones'] = [a.replace('slack_', 'whatsapp_valeria_') for a in r['acciones']]
        r['latencia'] += lognormal(rng, 0.3)
    salientes = [a for a in r['acciones'] if a.startswith(('confirmar', 'responder', 'acuse', 'plantilla'))]
    if salientes:
        r['latencia'] += lognormal(rng, 0.4)
        if detectar(estado, 'meta', bool(f.get('meta_401'))):
            r['manual'] = 'M02'
        if f.get('meta_401'):
            r.update(ruta = 'no_enviado_401', salientes = 0)
            return r
        r['salientes'] = len(salientes)
    return r

# Los 50 escenarios: diez por tipo, con hora, importe y el resultado esperado escrito antes de ejecutar.
ESCENARIOS = {
    'pedido completo': (['Quiero 2 hogazas de masa madre para el sábado a las 10', 'Me aparta 6 conchas pa mañana porfa', 'Necesito encargar 3 baguettes, las recojo el viernes a las 9',
                         'Un pedido: 12 bollos para hamburguesa, recojo a las 6', 'Apúnteme 2 roles de canela, paso el jueves', '¿Me pueden hacer un pastel de zanahoria para el domingo?',
                         'Quiero 4 croissants para el martes a las 8', 'Necesito encargar 1 pay de queso para el sábado', 'Me aparta 2 cajas de pan dulce pa el viernes', 'Quiero 10 baguettes para el jueves a las 7'],
                        {'ruta': 'borrador_pedido', 'claude': 1, 'salientes': 1, 'acciones': ['extraer_pedido', 'registrar_sqlite', 'confirmar_plantilla', 'avisar_obrador']}),
    'alérgenos':       (['¿El pan de caja tiene nueces?', '¿Las conchas llevan huevo?', '¿Tienen algo sin gluten?', '¿El pastel de zanahoria trae nuez?', '¿Los croissants tienen lácteos?',
                         'Mi hijo es alérgico al cacahuate, ¿qué puede comer?', '¿El rol de canela lleva nueces?', '¿Hay pan sin lactosa?', '¿Las galletas de avena contienen gluten?', '¿El pay de queso tiene huevo?'],
                        {'ruta': 'respuesta_con_fuentes', 'claude': 1, 'salientes': 1, 'acciones': ['buscar_notion', 'responder_citando']}),
    'queja':           (['El pan llegó quemado', 'Mi pastel llegó tarde y aplastado', 'Qué decepción, los croissants estaban crudos', 'Los bollos de hoy venían duros', 'Tengo una queja del pedido de ayer',
                         'Quiero reclamar por la hogaza de ayer', 'El pedido llegó tarde otra vez', 'Las conchas estaban quemadas por abajo', 'Qué decepción con el pay', 'Mi reclamo: faltó una baguette'],
                        {'ruta': 'revision_humana', 'claude': 1, 'salientes': 1, 'acciones': ['acuse_recibo', 'slack_aprobacion']}),
    'fuera de horario': (['Quiero 2 hogazas para el sábado', '¿A qué hora abren?', 'Me aparta 6 conchas', '¿Cuánto cuesta el pay de queso?', 'Necesito factura del pedido TL-20260420-0012',
                          'Un pedido: 4 croissants para el lunes', '¿El pan de caja tiene nueces?', 'Hola', 'El pan llegó quemado', '¿Abren los domingos?'],
                         {'ruta': 'cola_fuera_horario', 'claude': 0, 'salientes': 1, 'acciones': ['plantilla_horario', 'encolar']}),
    'baja':            (['BAJA', 'baja', 'Ya no', 'Ya no quiero recibir mensajes', 'No me escriban más por favor', 'Dejen de mandarme promociones', 'Quiero darme de baja de los avisos',
                         'Por favor quítenme de la lista', 'BAJA.', 'ya no, gracias'],
                        {'ruta': 'baja', 'claude': 0, 'salientes': 1, 'acciones': ['registrar_baja', 'confirmar_baja']}),
}
HORAS = [datetime(2026, 4, 25, 10, 30), datetime(2026, 4, 25, 11, 15), datetime(2026, 4, 24, 16, 0)]
FUERA = [datetime(2026, 4, 24, 21, 30), datetime(2026, 4, 25, 6, 20), datetime(2026, 4, 26, 11, 0), datetime(2026, 4, 24, 23, 59), datetime(2026, 4, 26, 15, 30)]
casos = []
for tipo, (textos, esperado) in ESCENARIOS.items():
    for i, texto in enumerate(textos):
        hora = FUERA[i % len(FUERA)] if tipo == 'fuera de horario' else HORAS[i % len(HORAS)]
        casos.append({'id': f'e{len(casos) + 1:02d}', 'tipo': tipo, 'texto': texto, 'hora': hora, 'cliente': f'c{len(casos) + 1:02d}',
                      'total': 380 if tipo == 'pedido completo' else 0, 'esperado': esperado})

def comparar(obtenido, esperado):
    return [clave for clave in esperado if (sorted(obtenido[clave]) if clave == 'acciones' else obtenido[clave]) !=
            (sorted(esperado[clave]) if clave == 'acciones' else esperado[clave])]

def correr_escenarios():
    estado, filas = estado_nuevo(), []
    for caso in casos:
        obtenido = ecosistema(caso, estado, rng)
        diferencias = comparar(obtenido, caso['esperado'])
        filas.append({'id': caso['id'], 'tipo': caso['tipo'], 'ruta': obtenido['ruta'], 'claude': obtenido['claude'],
                      'latencia_s': round(obtenido['latencia'], 2), 'diferencias': diferencias, 'aprobado': not diferencias})
    return pd.DataFrame(filas)

t0 = time.perf_counter()
primera = correr_escenarios()
print(f'Primera corrida: {int(primera.aprobado.sum())} de {len(primera)} escenarios aprobados en {(time.perf_counter() - t0) * 1000:.0f} ms')
print(primera[~primera.aprobado].to_string(index = False, columns = ['id', 'tipo', 'ruta', 'claude', 'diferencias']))
print(primera.groupby('tipo').agg(aprobados = ('aprobado', 'sum'), latencia_mediana_s = ('latencia_s', 'median')).round(2).to_string())

# Tres hallazgos y tres correcciones: (1) la regla de baja del Día 52 solo aceptaba "baja" o "ya no" como mensaje
# entero; (2) el horario miraba la hora, no el día de la semana, y el domingo no se atiende; (3) un pedido "para el
# domingo" caía en la regla de horario porque esa regla se evaluaba antes que la de pedido.
REGLAS[0] = ('baja', r'^\W*baja\W*$|^\W*ya no\b(?!.*\d)|no me escriban|dejen de mandar|darme de baja|qu[ií]tenme de la lista|no quiero recibir')
REGLAS.insert(3, REGLAS.pop(6))                        # la regla de pedido pasa por delante de la de horario
def en_horario(hora):
    return hora.weekday() < 6 and HORARIO[0] <= hora.hour < HORARIO[1]

segunda = correr_escenarios()
segunda.to_csv(carpeta / 'escenarios_punta_a_punta.csv', index = False)
print(f'\nSegunda corrida, con las tres correcciones: {int(segunda.aprobado.sum())} de {len(segunda)} escenarios aprobados | '
      f'llamadas a Claude: {int(segunda.claude.sum())} | latencia mediana: {segunda.latencia_s.median():.2f} s, '
      f'p95: {segunda.latencia_s.quantile(0.95):.2f} s')

# Experimentos de caos: cada fallo se inyecta en el simulador durante una ventana y se mide qué pasa en Python:
EXPERIMENTOS = [
    ('E1', 'Claude responde 429 (límite de velocidad)', {'claude': ('429', 1.0)}, 'M03', 'Plantilla de recibido y escalado; nadie se queda sin respuesta'),
    ('E2', 'Claude responde 529 (sobrecarga) al 60 %',  {'claude': ('529', 0.6)}, 'M03', 'Los reintentos del SDK absorben la mayoría; latencia sube; alerta'),
    ('E3', 'Servidor de n8n caído',                     {'n8n': True},            'M01', 'Nada se atiende; el latido activa el modo manual; Meta reintenta'),
    ('E4', 'Notion tarda 6 s (límite de 3 s)',           {'notion_s': 6.0},        'M07', 'Alérgenos sin fuentes y escalados; el resto no se entera'),
    ('E5', 'Firmas inválidas en el 30 % de los eventos', {'firma': 0.3},           'M06', 'Rechazo 401 de las forjadas; las legítimas siguen; alerta'),
    ('E6', 'Token de la Cloud API caducado',            {'meta_401': True},       'M02', 'Se procesa pero no se envía nada; modo manual'),
    ('E7', 'Slack caído',                               {'slack': True},          'M08', 'Aprobaciones y avisos van por WhatsApp a Valeria'),
]
N_EVENTOS, VENTANA = 500, (150, 350)
plantilla_eventos = [c for c in casos if c['tipo'] != 'fuera de horario']

def correr_experimento(fallos, rng):
    estado, filas = estado_nuevo(), []
    for k in range(N_EVENTOS):
        base_ev = plantilla_eventos[int(rng.integers(len(plantilla_eventos)))]
        evento = dict(base_ev, id = f'x{k:03d}', cliente = f'k{k:03d}', hora = datetime(2026, 4, 25, 10, 0) + timedelta(seconds = 7.2 * k))
        activo = VENTANA[0] <= k < VENTANA[1]
        obtenido = ecosistema(evento, estado, rng, fallos if activo else None)
        atendido = obtenido['ruta'] in ('borrador_pedido', 'respuesta_con_fuentes', 'revision_humana', 'respuesta_catalogo', 'aprobacion_humana', 'baja')
        degradado = obtenido['ruta'] in ('plantilla_degradada', 'respuesta_sin_fuentes')
        filas.append({'k': k, 'activo': activo, 'ruta': obtenido['ruta'], 'atendido': atendido, 'degradado': degradado,
                      'forjado': obtenido['ruta'] == 'rechazado_401', 'latencia': obtenido['latencia'], 'manual': obtenido['manual']})
    return pd.DataFrame(filas)

t0 = time.perf_counter()
tabla_2, series = [], {}
for codigo, nombre, fallos, manual_esperado, efecto_esperado in EXPERIMENTOS:
    d = correr_experimento(fallos, rng)
    series[codigo] = d
    durante = d[d.activo & ~d.forjado]
    activaciones = d[d.manual.notna()]
    primera_activacion = int(activaciones.k.iloc[0]) if len(activaciones) else None
    exito = (durante.atendido | durante.degradado).mean() if len(durante) else 0.0
    tabla_2.append({'exp': codigo, 'fallo': nombre, 'esperado': efecto_esperado, 'manual': manual_esperado,
                    'atendidos_durante': round(exito, 3), 'degradados': int(durante.degradado.sum()), 'sin_atender': int((~durante.atendido & ~durante.degradado).sum()),
                    'forjados_rechazados': int(d[d.activo].forjado.sum()), 'p95_antes_s': round(d[~d.activo].latencia.quantile(0.95), 2),
                    'p95_durante_s': round(durante.latencia.quantile(0.95), 2) if durante.latencia.notna().any() else float('nan'),
                    'manual_activado': activaciones.manual.iloc[0] if len(activaciones) else 'ninguno',
                    'eventos_hasta_activar': (primera_activacion - VENTANA[0] + 1) if primera_activacion is not None else None,
                    'despues_ok': round((d[(d.k >= VENTANA[1]) & ~d.forjado].atendido).mean(), 3)})
tabla_2 = pd.DataFrame(tabla_2)
tabla_2.to_csv(carpeta / 'experimentos_caos.csv', index = False)
print(f'7 experimentos de {N_EVENTOS} eventos en {time.perf_counter() - t0:.1f} s')
print(tabla_2.drop(columns = ['esperado']).to_string(index = False))
print('\nManuales que se activaron como se esperaba:', int((tabla_2.manual == tabla_2.manual_activado).sum()), 'de', len(tabla_2),
      '| reprocesados al volver en E3:', int(series['E3'][series['E3'].ruta == 'sin_respuesta'].shape[0]))

fig, axes = plt.subplots(4, 2, figsize = (11, 11), sharex = True)
for ax, (codigo, nombre, *_) in zip(axes.flat, EXPERIMENTOS):
    d = series[codigo]
    ok = (d.atendido | d.degradado).astype(float).where(~d.forjado)
    tasa = ok.rolling(25, min_periods = 5).mean()
    ax.axvspan(*VENTANA, color = '#f6d9c4', alpha = 0.6, label = 'fallo activo')
    ax.plot(d.k, tasa, color = '#2a78d6', lw = 2, label = 'tasa de atención (ventana de 25)')
    ax2 = ax.twinx()
    ax2.scatter(d.k, d.latencia, s = 4, color = '#c8423b', alpha = 0.5, label = 'latencia (s)')
    ax2.set_ylim(0, max(6, np.nanmax(d.latencia) * 1.1))
    ax2.tick_params(labelsize = 7)
    act = d[d.manual.notna()]
    if len(act):
        ax.axvline(act.k.iloc[0], color = '#3a9d5d', lw = 1.5, ls = '--')
        ax.text(act.k.iloc[0] + 4, 0.08, f'{act.manual.iloc[0]} activado', fontsize = 8, color = '#3a9d5d')
    ax.set_ylim(-0.02, 1.05)
    ax.set_title(f'{codigo}. {nombre}', fontsize = 10, loc = 'left')
    ax.tick_params(labelsize = 7)
    ax.spines[['top']].set_visible(False)
axes.flat[-1].axis('off')
axes.flat[-1].text(0.02, 0.8, 'Azul: proporción de eventos legítimos atendidos\n(normal o degradado), ventana móvil de 25.\n'
                   'Rojo: latencia de cada evento en segundos (eje derecho).\nSombra: fallo inyectado entre los eventos 150 y 349.\n'
                   'Verde: primer evento en que el detector activa\nel manual del Día 81.', fontsize = 9, va = 'top')
axes[3, 0].set_xlabel('Evento')
fig.suptitle('Siete experimentos de caos en el simulador: tasa de atención y latencia', fontsize = 16)
plt.tight_layout()
plt.show()

# Carga al triple del pico: M/M/1 analítico para el servidor y simulación de la cola de aprobaciones humanas en Python:
mu = 3600 / 4                                          # 4 s de servicio por mensaje: 900 por hora
print('Servidor como M/M/1 (servicio medio de 4 s):')
for nombre, lam in (('pico normal', 9), ('doble', 18), ('triple', 27)):
    rho = lam / mu
    W = 1 / (mu - lam) * 3600                          # tiempo en el sistema, en segundos
    L = lam / 3600 * W                                 # ley de Little
    print(f'  {nombre:<12} λ = {lam:>2}/h | ρ = {rho:.3f} | W = {W:.2f} s | L = {L:.3f} mensajes en el sistema')
print(f'  λ con ρ = 0.8 (el servidor empieza a sufrir): {0.8 * mu:.0f} mensajes por hora, {0.8 * mu / 9:.0f} veces el pico normal')

# Simulación de eventos discretos del sábado de 10:00 a 12:00 al triple del pico: servidor y aprobaciones.
def simular_sabado(lam, revisa_cada_min, rng, p_aprobacion = 0.20, minutos_por_aprobacion = 2.0, horas = 2):
    n = rng.poisson(lam * horas)
    llegadas = np.sort(rng.uniform(0, horas * 3600, n))
    fin_servidor, salidas = 0.0, np.empty(n)
    for i, t in enumerate(llegadas):
        inicio = max(t, fin_servidor)
        fin_servidor = inicio + rng.exponential(4.0)
        salidas[i] = fin_servidor
    en_sistema = salidas - llegadas
    necesita = rng.random(n) < p_aprobacion
    esperas = []
    for t in salidas[necesita]:
        siguiente_revision = (math.floor(t / (revisa_cada_min * 60)) + 1) * revisa_cada_min * 60
        esperas.append(siguiente_revision - t + rng.exponential(minutos_por_aprobacion * 60))
    return en_sistema, np.array(esperas), llegadas, salidas

t0 = time.perf_counter()
res = {}
for nombre, cada in (('Valeria mira Slack cada 15 min', 15), ('Itzel de suplente, cada 5 min', 5)):
    sistema, esperas = [], []
    for _ in range(1000):
        s, e, *_ = simular_sabado(27, cada, rng)
        sistema.append(s)
        esperas.append(e)
    sistema, esperas = np.concatenate(sistema), np.concatenate(esperas)
    res[nombre] = esperas
    print(f'{nombre}: servidor W medio {sistema.mean():.2f} s, p95 {np.quantile(sistema, 0.95):.2f} s | aprobaciones: '
          f'{len(esperas) / 1000:.1f} por sábado, espera mediana {np.median(esperas) / 60:.1f} min, p95 {np.quantile(esperas, 0.95) / 60:.1f} min, '
          f'más de 15 min: {(esperas > 900).mean():.1%}')
print(f'1 000 sábados por política en {time.perf_counter() - t0:.1f} s')

fig, (ax1, ax2) = plt.subplots(1, 2, figsize = (11, 4.4))
s, e, llegadas, salidas = simular_sabado(27, 15, np.random.default_rng(83))
ts = np.linspace(0, 7200, 721)
en_cola = [(llegadas <= t).sum() - (salidas <= t).sum() for t in ts]
ax1.step(ts / 60, en_cola, where = 'post', color = '#2a78d6', lw = 1.5)
ax1.set_title(f'Servidor: mensajes en el sistema, un sábado al triple del pico ({len(llegadas)} llegadas)', fontsize = 10)
ax1.set_xlabel('Minutos desde las 10:00')
ax1.set_ylabel('Mensajes en el sistema')
ax1.set_ylim(0, 4)
ax1.spines[['top', 'right']].set_visible(False)
for (nombre, esperas), color in zip(res.items(), ('#c8423b', '#3a9d5d')):
    ordenadas = np.sort(esperas) / 60
    ax2.plot(ordenadas, np.arange(1, len(ordenadas) + 1) / len(ordenadas), color = color, lw = 2, label = nombre)
ax2.axvline(15, color = '#2b2a28', ls = '--', lw = 1.2, label = 'SLA: 15 minutos')
ax2.set_xlim(0, 30)
ax2.set_xlabel('Espera de una aprobación (minutos)')
ax2.set_ylabel('Proporción acumulada')
ax2.set_title('Aprobaciones humanas: 1 000 sábados simulados', fontsize = 10)
ax2.legend(loc = 'lower right', frameon = False, fontsize = 8)
ax2.spines[['top', 'right']].set_visible(False)
fig.suptitle('El cuello de botella no es el servidor: es quién mira Slack', fontsize = 16)
plt.tight_layout()
plt.show()

# Lista de salida con criterios medidos, riesgos aceptados y firmas, y generación del documento de decisión en Python:
p95_suplente = np.quantile(res['Itzel de suplente, cada 5 min'], 0.95) / 60
criterios = [
    ('Contratos de webhooks', 'ningún cuerpo roto pasa y ninguno válido se rechaza', f'{int(contratos.correcto.sum())} de {len(contratos)} veredictos correctos', bool(contratos.correcto.all())),
    ('Punta a punta', '50 de 50 escenarios aprobados', f'{int(segunda.aprobado.sum())} de 50', bool(segunda.aprobado.all())),
    ('Regresión de prompts', 'sin regresión significativa y límite inferior total de v3 >= 0.85', f'total v3 [{total.v3_bajo:.3f}, {total.v3_alto:.3f}]', bool(not tabla_3.regresion.any() and total.v3_bajo >= 0.85)),
    ('Caos', 'los 7 fallos activan el manual previsto', f'{int((tabla_2.manual == tabla_2.manual_activado).sum())} de 7', bool((tabla_2.manual == tabla_2.manual_activado).all())),
    ('Carga del servidor', 'utilización < 0.5 al triple del pico', f'ρ = {27 / mu:.3f}', 27 / mu < 0.5),
    ('Aprobaciones humanas', 'p95 de espera < 15 min con suplente', f'p95 = {p95_suplente:.1f} min', p95_suplente < 15),
    ('Restauración (Día 80)', 'simulacro dentro del RTO de 4 horas', 'cumplido según el informe de auditoría', True),
    ('Presupuesto (Día 77)', 'gasto mensual <= 45 USD', '40.53 USD de 45', True),
]
riesgos = [
    ('R1', 'Make Core sigue siendo un punto único de falla sin alternativa el fin de semana (Día 78)', 'Don Ernesto', 'modo manual M00 para los 4 escenarios de Make'),
    ('R2', 'Las aprobaciones dependen de que alguien mire Slack cada 5 minutos el sábado', 'Valeria', 'Itzel de suplente de 10:00 a 13:00; aviso por WhatsApp si Slack cae'),
    ('R3', 'Con 200 mensajes la evaluación no distingue una exactitud de 0.91 de una de 0.94', 'Valeria', f'ampliar a {math.ceil(n_necesario)} mensajes por versión antes del 10 de mayo'),
    ('R4', 'El caos se probó en el simulador, no contra la instancia de producción', 'Valeria', 'repetir E1, E4 y E5 contra la instancia de ensayo del Día 43 el lunes 27'),
    ('R5', 'Los manuales M04 (plantilla rechazada) y M05 (gasto anómalo) no se pueden inyectar en vivo', 'Valeria', 'ensayo de mesa del Día 81 y detector del Día 81 ya probados'),
]
decision = 'LISTO' if all(c[3] for c in criterios) else 'NO LISTO'
lineas = [f'# Decisión de salida: ecosistema de Trigo Lento, {hoy:%d/%m/%Y}', '',
          f'Decisión: {decision} para el sábado {ensayo:%d/%m} y la semana del Día del Niño, con {len(riesgos)} riesgos aceptados.', '',
          '## Criterios', '', '| Criterio | Umbral | Medido | Cumple |', '|---|---|---|:---:|']
lineas += [f'| {n} | {u} | {m} | {"sí" if c else "no"} |' for n, u, m, c in criterios]
lineas += ['', '## Riesgos aceptados', '', '| Id | Riesgo | Lo acepta | Mitigación |', '|---|---|---|---|']
lineas += [f'| {i} | {r} | {q} | {m} |' for i, r, q, m in riesgos]
lineas += ['', '## Firmas', '', '- Valeria Ortiz, arquitecta de automatización: ________  (23/04/2026)',
           '- Ernesto Ortiz, dueño, acepta R1 y R2: ________  (23/04/2026)', '',
           'Próxima revisión: lunes 27 de abril (R4) y lunes 4 de mayo (simulacro del calendario de gobierno del Día 79).']
documento = '\n'.join(lineas) + '\n'
(carpeta / f'decision_salida_{hoy:%Y-%m-%d}.md').write_text(documento, encoding = 'utf-8')
huella = hashlib.sha256(documento.encode('utf-8')).hexdigest()[:12]
print(pd.DataFrame(criterios, columns = ['criterio', 'umbral', 'medido', 'cumple']).to_string(index = False))
print(f'\nDECISIÓN: {decision} | criterios cumplidos: {sum(c[3] for c in criterios)} de {len(criterios)} | riesgos aceptados: {len(riesgos)}')
print(f'Documento: decision_salida_{hoy:%Y-%m-%d}.md ({len(documento.split())} palabras, huella {huella})')
print('\n'.join(lineas[:3]))