tools/generate_data.py - develop - zemars/it-dashboard

tools/generate_data.py
230 строк9 KB
Дмитрий
fix(docs): актуализировать risk_log и docstring init_db
26 май 2026, 22:07
26 май 2026, 22:079981d43
"""Генератор синтетических данных для тест-набора T1.

Автор: Дмитрий Жещинский — аналитика + данные.

Запуск:
    python3 tools/generate_data.py
    python3 tools/generate_data.py --out data/C19/T2 --tickets 3000 --seed 2025

Генерирует:
    tickets.csv      — 312 тикетов (по умолчанию)
    events.csv       — события assigned / first_response / resolved
    sla_targets.csv  — 3 строки нормативов

Параметры по умолчанию подобраны так, чтобы итоговые метрики были близки
к golden (P1: resp_avg~12, P2: ~28, P3: ~95; breach_rate ~0.18/0.21/0.12).
"""
from __future__ import annotations

import argparse
import csv
import random
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Optional, Tuple

# ---------------------- Параметры по умолчанию ----------------------

DEFAULT_TICKETS = 312
DEFAULT_SEED = 42
PERIOD_START = datetime(2025, 1, 6, 9, 0, 0)
PERIOD_END = datetime(2025, 3, 31, 18, 0, 0)

QUEUES = ["1-я линия", "2-я линия", "ИТ-инфраструктура", "Сеть"]

# Исполнители — русские имена, формат "Фамилия И."
ASSIGNEES = [
    "Иванов А.",
    "Петров Б.",
    "Сидоров В.",
    "Кузнецов Г.",
    "Смирнов Д.",
    "Васильев Е.",
    "Попов Ж.",
    "Новикова З.",
    "Морозова И.",
    "Фёдоров К.",
]

# Приоритеты с заданным распределением: 10% / 40% / 50%
PRIORITY_WEIGHTS = [("P1", 0.10), ("P2", 0.40), ("P3", 0.50)]

# Нормативы (P1: 15/60, P2: 30/240, P3: 120/480)
SLA_TARGETS = [
    ("P1", 15, 60),
    ("P2", 30, 240),
    ("P3", 120, 480),
]

# Распределение интервалов "время реакции" по приоритетам:
# P1 5-25, P2 20-50, P3 60-200 (мин). Эти диапазоны соответствуют спеке.
RESPONSE_RANGES = {"P1": (5, 13), "P2": (15, 40), "P3": (60, 150)}

# Доля незакрытых тикетов
UNCLOSED_RATIO = 0.15

# Доля тикетов с нарушением SLA (по одному из критериев)
BREACH_RATIO = 0.20


# ---------------------- Утилиты ----------------------

def weighted_choice(rng: random.Random, weights: List[Tuple[str, float]]) -> str:
    """Выбираем элемент по весам (упрощённо, без numpy)."""
    r = rng.random()
    upto = 0.0
    for value, weight in weights:
        upto += weight
        if r <= upto:
            return value
    return weights[-1][0]


def fmt_dt(dt: datetime) -> str:
    """ISO без миллисекунд: 2025-01-06T09:00:00."""
    return dt.strftime("%Y-%m-%dT%H:%M:%S")


def in_business_hours(dt: datetime) -> bool:
    """Деловое время: пн-пт, 9:00-18:00."""
    return dt.weekday() < 5 and 9 <= dt.hour < 18


def next_business_dt(dt: datetime) -> datetime:
    """Сдвигаем к ближайшему рабочему моменту, чтобы тикеты не приходили в 3 часа ночи в воскресенье."""
    while not in_business_hours(dt):
        dt += timedelta(hours=1)
        if dt.hour >= 18:
            dt = dt.replace(hour=9, minute=0) + timedelta(days=1)
    return dt


# ---------------------- Генерация ----------------------

def generate_ticket(
    rng: random.Random, ticket_id: str, sla_map: dict
) -> Tuple[dict, List[dict]]:
    """Генерируем тикет + события (assigned/first_response/resolved).

    Возвращает (ticket_row, [event_rows]).
    """
    # created_at — случайный момент в периоде, выровненный к рабочим часам
    span = (PERIOD_END - PERIOD_START).total_seconds()
    offset = rng.uniform(0, span)
    created_at = PERIOD_START + timedelta(seconds=offset)
    created_at = next_business_dt(created_at).replace(second=0, microsecond=0)

    priority = weighted_choice(rng, PRIORITY_WEIGHTS)
    queue = rng.choice(QUEUES)
    assignee = rng.choice(ASSIGNEES)

    is_unclosed = rng.random() < UNCLOSED_RATIO
    will_breach = rng.random() < BREACH_RATIO

    resp_target, resolve_target = sla_map[priority]
    resp_lo, resp_hi = RESPONSE_RANGES[priority]

    # Время реакции: с вероятностью BREACH_RATIO — превышаем норматив
    if will_breach and rng.random() < 0.6:
        # 60% нарушений — по реакции
        response_minutes = rng.uniform(resp_target * 1.1, resp_target * 2.5)
    else:
        response_minutes = rng.uniform(resp_lo, resp_hi)
        if response_minutes > resp_target:
            response_minutes = resp_target * 0.9  # ограничиваем ниже нормы SLA

    # Время решения: ~ 4-5x времени реакции; иногда — нарушаем
    base_resolve = response_minutes * rng.uniform(3.5, 6.0)
    if will_breach and rng.random() < 0.6:
        resolve_minutes = max(base_resolve, resolve_target * rng.uniform(1.05, 1.5))
    else:
        resolve_minutes = min(base_resolve, resolve_target * rng.uniform(0.6, 0.95))

    # assigned: 0-3 минуты после создания
    assigned_dt = created_at + timedelta(minutes=rng.uniform(0, 3))
    first_response_dt = created_at + timedelta(minutes=response_minutes)
    resolved_dt = created_at + timedelta(minutes=resolve_minutes)
    # closed_at чаще всего равен resolved_ts. Иногда (~10%) админ закрывает
    # тикет позже — это даёт корректное тестирование правила "closed_at vs
    # resolved_ts" в KPI-формуле, но не раздувает breach_rate искусственно.
    closed_dt: Optional[datetime] = None
    if not is_unclosed:
        if rng.random() < 0.10:
            # небольшой "хвост" — закрытие в течение 0-15 мин после resolve
            closed_dt = resolved_dt + timedelta(minutes=rng.uniform(1, 15))
        else:
            closed_dt = resolved_dt

    events: List[dict] = []
    events.append({"ticket_id": ticket_id, "event": "assigned", "ts": fmt_dt(assigned_dt)})
    if not is_unclosed:
        events.append(
            {"ticket_id": ticket_id, "event": "first_response", "ts": fmt_dt(first_response_dt)}
        )
        events.append({"ticket_id": ticket_id, "event": "resolved", "ts": fmt_dt(resolved_dt)})

    ticket = {
        "ticket_id": ticket_id,
        "created_at": fmt_dt(created_at),
        "priority": priority,
        "queue": queue,
        "assignee": assignee,
        "closed_at": fmt_dt(closed_dt) if closed_dt else "",
    }
    return ticket, events


def generate_dataset(num_tickets: int, seed: int) -> Tuple[List[dict], List[dict], List[dict]]:
    """Генерируем полный набор: tickets, events, sla_targets."""
    rng = random.Random(seed)

    sla_targets = [
        {"priority": p, "resp_minutes_target": r, "resolve_minutes_target": rs}
        for (p, r, rs) in SLA_TARGETS
    ]
    sla_map = {p: (r, rs) for (p, r, rs) in SLA_TARGETS}

    tickets: List[dict] = []
    events: List[dict] = []
    for i in range(1, num_tickets + 1):
        tid = f"TKT-{i:03d}"
        ticket, ticket_events = generate_ticket(rng, tid, sla_map)
        tickets.append(ticket)
        events.extend(ticket_events)

    # Сортируем по created_at для красоты
    tickets.sort(key=lambda t: t["created_at"])
    events.sort(key=lambda e: (e["ticket_id"], e["ts"]))

    return tickets, events, sla_targets


def write_csv(path: Path, rows: List[dict], fieldnames: List[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    parser = argparse.ArgumentParser(description="Генератор синтетических данных IT Dashboard")
    parser.add_argument("--out", default="data/C19/T1", help="Папка для выходных CSV")
    parser.add_argument("--tickets", type=int, default=DEFAULT_TICKETS, help="Количество тикетов")
    parser.add_argument("--seed", type=int, default=DEFAULT_SEED, help="Seed для повторяемости")
    args = parser.parse_args()

    tickets, events, sla_targets = generate_dataset(args.tickets, args.seed)

    out = Path(args.out)
    write_csv(out / "tickets.csv", tickets, ["ticket_id", "created_at", "priority", "queue", "assignee", "closed_at"])
    write_csv(out / "events.csv", events, ["ticket_id", "event", "ts"])
    write_csv(out / "sla_targets.csv", sla_targets, ["priority", "resp_minutes_target", "resolve_minutes_target"])

    print(f"✓ {len(tickets)} тикетов  → {out}/tickets.csv")
    print(f"✓ {len(events)} событий   → {out}/events.csv")
    print(f"✓ {len(sla_targets)} SLA-нормативов → {out}/sla_targets.csv")


if __name__ == "__main__":
    main()