Проблема

Уже достаточно давно я разрабатываю и поддерживаю собственный сервис валидации Email адресов Cleancontact. Сервис позволяет проверить email адрес на предмет реального существования и таким образом повысить качество базы данных пользователей и снизить процент недоставок Email, что очень важно для репутации рассыльщика. Но что если к нам на сайт приходят злоумышленники, которые выполняют массовые регистрации с использованием реальных почт, на которые доходят письма. Тема антифрода достаточно богатая для обсуждения, и с фродом сталкивались так или иначе все владельцы сайтов, где из регистрации можно извлечь хоть какую-то выгоду. И даже без выгоды, фродеры могу выполнять массовые регистрации чисто из спортивного интереса. Капчи так или иначе обходятся, пулы IP используются для массовых регистраций с разных IP, и в итоге при разработке системы антифрода часто приходишь к скорингу: когда нельзя однозначно сказать это злоумышленник или нет, но можно посчитать определенную вероятность этого и отсеивать пользователей по некоторому порогу.

Злоумышленники и спам-боты регистрируются на сервисах клиентов с адресами вроде:

xnjjeka.2.e.1.1@gmail.com
dhduwbsb.wu.166@gmail.com
a.l.eli.c.sal.los.2.71@gmail.com
xjd82jd91@gmail.com

Такие адреса генерируются скриптами. Самый массовый приём на Gmail: взять один реальный ящик alicsallos271@gmail.com и расставить в нём точки. Gmail игнорирует точки, поэтому письма приходят на один и тот же ящик, а для сервиса каждый вариант выглядит как новый пользователь. SMTP проверка на все варианты честно отвечает “ящик существует”, и в целом фродеры часто используют именно валидные email адреса, на которые доходят сообщения. Да, можно было ввести логическую проверку на количество точек, и отсекать адреса вида xnjjeka.2.e.1.1@gmail.com, но это было бы локальное решение одной проблемы, а мы хотим сделать что-то более общее.

Визуально мы легко видим разницу между ivan.ivanov.88@gmail.com и xnjjeka.2.e.1.1@gmail.com. И хочется, чтобы ее видел и сервис, и выдавал числовой скоринг: насколько адрес визуально “человеческий”.

Важное ограничение, которое обозначим сразу - скоринг не превращает адрес в не-валидный. Странные адреса бывают настоящими. Это дополнительный сигнал риска рядом с SMTP проверкой, MX проверкой и чёрными DEA списками.

Решение и стек

Обучать модель удобно на Python, а работать она у нас будет внутри Node.js бэкенда без отдельного Python-процесса. Поэтому разделим реализацию:

Python модель

• берем CSV файл по результатам прошлых проверок

• используем CatBoost

• экспортируем полученную модель в ONNX

Node.js / TypeScript бэкенд

• подключаем файл модели

• выделям признаки идентично python

• используем onnxruntime-node

• подключаем скоринг в антифроде

Стек тренировочной части:

• Python 3.12, зависимости через uv.

• CatBoost https://catboost.ai/ - open source библиотека разработанная Яндексом. Использует усовершенствованный алгоритм градиентного бустинг. Выбран за нативный экспорт в ONNX и устойчивость к дефолтным гиперпараметрам. CatBoost часто показывает отличные результаты на стандартных настройках, то есть работает из коробки.

• pandas и pyarrow для датасета, scikit-learn для метрик.

• onnx и onnxruntime для экспорта и проверки паритета.

• используем старый добрый ML без каких либо LLM, все гораздо проще, быстрее, модель тренируется очень быстро, и API отвечает максимально быстро

• признаки должны быть только числовыми и воспроизводимыми один в один на TypeScript

Реализация на Python по шагам

Шаг 1. Нормализация признаков

Всё, что модель видит, это первая часть адреса до собачки в нижнем регистре. Домен специально не используем: иначе модель быстро выучит, что например gmail.com хороший, а некий условный mail.ru плохой, а мы хотим оценивать только написание.

def split_email(email: str) -> tuple[str, str]:
  email = email.strip().lower()
  if "@" not in email:
    return email, ""
  local, domain = email.rsplit("@", 1)
  return local, domain


def tokenize(local: str) -> list[str]:
  return [t for t in re.split(r"[._-]", local) if t]

Буквы это [a-z], гласные aeiou, токены делятся по . _ -, энтропия считается по log2. Эти же правила реализует TypeScript экстрактор.

Шаг 2. Лексические признаки

Опишем первый набор признаков: длина строки, доля цифр, количество разделителей, самая длинная серия цифр и букв, энтропия, повторы символов, наличие года.

def entropy(text: str) -> float:
  if not text:
    return 0.0
  n = len(text)
  return -sum((c / n) * math.log2(c / n) for c in Counter(text).values())


def longest_run(text: str, charset: str) -> int:
  best = cur = 0
  for ch in text:
    cur = cur + 1 if ch in charset else 0
    best = max(best, cur)
  return best


def lexical_features(local: str) -> dict[str, float]:
  n = len(local)
  denom = max(n, 1)
  letters = sum(ch in LETTERS for ch in local)
  digits = sum(ch in DIGITS for ch in local)
  dots, dashes, underscores = local.count("."), local.count("-"), local.count("_")
  digit_runs = _DIGIT_RUN.findall(local)
  first, last = local[:1], local[-1:]

  return {
    "length": n,
    "letter_count": letters,
    "digit_count": digits,
    "digit_ratio": digits / denom,
    "dot_count": dots,
    "dash_count": dashes,
    "underscore_count": underscores,
    "separator_ratio": (dots + dashes + underscores) / denom,
    "starts_with_digit": int(first in DIGITS and first != ""),
    "ends_with_digit": int(last in DIGITS and last != ""),
    "starts_with_letter": int(first in LETTERS and first != ""),
    "ends_with_letter": int(last in LETTERS and last != ""),
    "longest_digit_run": longest_run(local, DIGITS),
    "longest_letter_run": longest_run(local, LETTERS),
    "unique_char_count": len(set(local)),
    "unique_char_ratio": len(set(local)) / denom,
    "entropy": entropy(local),
    "repeated_char_count": sum(local[i] == local[i - 1] for i in range(1, n)),
    "has_year": int(any(len(r) == 4 and YEAR_MIN <= int(r) <= YEAR_MAX for r in digit_runs)),
    "has_4_digit_number": int(any(len(r) >= 4 for r in digit_runs)),
  }

Шаг 3. Первая проблема - результат SMTP проверки не годятся как разметка

Поскольку я разрабатываю сервис валидации, то первой идеей было посмотреть корреляцию, связаны ли реальность (доставляемость сообщений) на этот адрес и его внешний вид. Результаты показали AUC=0.51 на тестовой выборке. То есть модель была не лучше случайного выбора.

Разбор данных показал почему. Несуществующие адреса выглядят по-человечески: pollinakrasnova@gmail.com, ivanbekov@icloud.com. Часто встречаются опечатки домена вроде @mai.ru. А адреса с тремя и более точками на Gmail существуют в 95% случаев, потому что это как раз варианты реальных ящиков. Модель, обученная на существовании, давала xnjjeka.2.e.1.1@gmail.com скоринг 0.94. Что в какой-то степени было правдой, ведь gmail dot trick используют с реальными ящиками, но для тренировки модели - бесполезно.

Поэтому адрес существует и адрес выглядит человеческим это разные задачи. И остановимся только на второй.

Шаг 4. Разметка для “человечности”: реальные и синтетические негативы

Для позитивов берем подтверждённые SMTP адреса, из которых убраны все dot-trick варианты.

def _is_dot_variant(df: pd.DataFrame) -> pd.Series:
  gmail = df["domain"].isin(("gmail.com", "googlemail.com"))
  alias_local = df["aliasEmail"].fillna("").map(lambda e: split_email(e)[0])
  flagged = (df["suspicious"] == "True") | (df["gmailDotTrick"] == "True")
  aliased = (alias_local != "") & (alias_local == df["local"].str.replace(".", ""))
  dotted = df["local"].str.count(r"\.") >= 2
  return gmail & (flagged | (aliased & dotted))

Реальные негативы: эти самые dot-trick варианты. Их в моей тестовой выборке было около 7 тысяч, но это одно семейство фейков. Если учить только на нём, модель запомнит “много точек = плохо” и ничего больше.

Поэтому остальные семейства синтезируются. C одной стороны может показаться плохо, что нужно синтезировать данные для тренировки. Но ведь по сути это то что и происходит в реальности - когда злоумышленники генерируют адреса для регистраций - они тоже сгенерированные. На каждый позитив сделаем один негатив из шести семейств с заданными весами. Здесь:

• random_letters - kqzmwpdhx

• random_alnum - xjd82jd92

• shuffled - перемешиваем буквы адреса

• dot_trick - реальный адрес с 2–5 случайными точками

• consonant_mash - qmxjwp82

• keyboard_walk - qwerty123, asdfgh

FAMILIES = {
  "random_letters": 0.20,   # kqzmwpdhx
  "random_alnum": 0.20,     # xjd82jd92, dhduwbsb.wu.177
  "shuffled": 0.25,         # буквы реального адреса перемешаны
  "dot_trick": 0.15,        # реальный адрес с 2–5 случайными точками
  "consonant_mash": 0.10,   # qmxjwp82
  "keyboard_walk": 0.10,    # qwerty123, asdfgh
}

def random_letters(rng: random.Random, pool: list[str]) -> str:
  return _rand_str(rng, LETTERS, rng.randint(5, 14))


def random_alnum(rng: random.Random, pool: list[str]) -> str:
  parts = []
  for _ in range(rng.randint(1, 3)):
    alphabet = LETTERS + DIGITS if rng.random() < 0.7 else DIGITS
    parts.append(_rand_str(rng, alphabet, rng.randint(1, 8)))
  return rng.choice([".", "_", "-", ""]).join(parts)


def shuffled(rng: random.Random, pool: list[str]) -> str:
  for _ in range(20):
    chars = list(rng.choice(pool))
    if sum(c in LETTERS for c in chars) >= 6:
      rng.shuffle(chars)
      return "".join(chars)
  return random_letters(rng, pool)


def dot_trick(rng: random.Random, pool: list[str]) -> str:
  base = rng.choice(pool).replace(".", "")
  if len(base) < 4:
    base = random_letters(rng, pool)
  slots = sorted(rng.sample(range(1, len(base)), min(rng.randint(2, 5), len(base) - 1)))
  out, prev = [], 0
  for s in slots:
    out.append(base[prev:s])
    prev = s
  out.append(base[prev:])
  return ".".join(out)


def consonant_mash(rng: random.Random, pool: list[str]) -> str:
  n = rng.randint(5, 12)
  alphabet = CONSONANTS if rng.random() < 0.6 else CONSONANTS + VOWELS[:2]
  s = _rand_str(rng, alphabet, n)
  return s + (_rand_str(rng, DIGITS, rng.randint(1, 3)) if rng.random() < 0.5 else "")


def keyboard_walk(rng: random.Random, pool: list[str]) -> str:
  parts = []
  for _ in range(rng.randint(1, 3)):
    row = rng.choice(KEYBOARD_ROWS)
    start = rng.randint(0, len(row) - 3)
    parts.append(row[start : start + rng.randint(3, len(row) - start)])
  return "".join(parts)

Семейство shuffled особенно важно. Перемешанный ivanivanov сохраняет длину, долю цифр и энтропию оригинала, но ломает последовательности букв. Оно заставляет модель опираться на n-граммы, а не на статистику символов.

Синтетические негативы генерируются для каждого split отдельно и только из позитивов этого split.

Шаг 5. N-граммы

N-граммы используются, чтобы определить, насколько последовательности букв похожи на те, которые обычно встречаются в человеческих именах и email-адресах. Частоты буквенных биграмм и триграмм считаются по позитивам train выборки и сохраняются в JSON. Для адреса считается средняя и минимальная log-частота его n-грамм и доля редких.

def build_vocab(train_locals: list[str], rare_quantile: float = 0.1) -> NgramVocab:
  bi, tri = Counter(), Counter()
  for local in train_locals:
    text = letters_only(local)
    bi.update(ngrams(text, 2))
    tri.update(ngrams(text, 3))
  bi_table, bi_unknown = _logfreq_table(bi, sum(bi.values()))
  tri_table, tri_unknown = _logfreq_table(tri, sum(tri.values()))
  return NgramVocab(
    bigram_logfreq=bi_table,
    trigram_logfreq=tri_table,
    rare_bigram_threshold=_quantile(list(bi_table.values()), rare_quantile),
    rare_trigram_threshold=_quantile(list(tri_table.values()), rare_quantile),
    unknown_bigram_logfreq=bi_unknown,
    unknown_trigram_logfreq=tri_unknown,
  )


def _stats(grams, table, unknown, rare):
  values = [table.get(g, unknown) for g in grams]
  return (
    sum(values) / len(values),
    min(values),
    sum(v < rare for v in values) / len(values),
  )

В итоге rare_trigram_ratio оказался самым важным признаком модели, его доля около 30%. Для ivanov триграммы iva, van, ano, nov частые. Русскоязычные фамилии вообще достаточно однообразные внешне, и их легче опознавать. Для dhduwbsb почти все редкие.

Шаг 6. Проблема с иностранными именами

Наши данные на 80% состоят из русскоязычного транслита. Словарь n-грамм унаследовал это. hans.mueller@gmx.de получал 0.41, а kenji.tanaka@yahoo.co.jp 0.005. То есть японское имя для модели выглядело как случайная строка.

Это хочется поправить, ведь сервис не обязательно проверяет только российские адреса и имена и могут быть зарубежные пользователи.

Сначала я добавил словари имён. Использовал два открытых источника: базу имён с разбивкой по 55 регионам, включая Россию, Китай, Индию и арабские страны, и базу фамилий из 20 стран. После нормализации в [a-z] получилось 44 тысячи имён и 86 тысяч фамилий. По ним считаются признаки: есть ли в адресе имя или фамилия, является ли первый токен именем, какую долю букв покрывает самое длинное найденное имя.

def name_features(local: str, names: NameDict) -> dict[str, float]:
  segments = letter_segments(local)
  letters = sum(len(s) for s in segments)
  tokens = [t.strip("0123456789") for t in tokenize(local)]
  is_name = lambda t: t in names.first or t in names.last
  longest_first = _longest_match(segments, names.first)
  longest_last = _longest_match(segments, names.last)
  return {
    "contains_first_name": int(longest_first > 0),
    "contains_last_name": int(longest_last > 0),
    "first_token_is_name": int(bool(tokens) and is_name(tokens[0])),
    "name_token_count": sum(is_name(t) for t in tokens),
    "name_char_ratio": max(longest_first, longest_last) / letters if letters else 0.0,
  }

При этом эти же словари можно использовать на вход расчета n-граммы. Триграммы mue, ell, ken, nji перестали быть редкими. Только с name-признаками kenji.tanaka оставался на 0.005, а с триграммами поднялся до 0.99.

Заодно обнаружилась странная вещь. В списке фамилий были qwert, erty, yuiop. Из-за них qwerty123 получал 0.98 как “фамилия”. Это нужно фильтровать заранее, при подготовке данных словаря. Такие записи теперь фильтруются, а в модель добавлен признак longest_keyboard_run: самая длинная подстрока, совпадающая с рядом клавиатуры.

Шаг 7. Обучение модели

После подготовки признаков обучение через CatBoost очень компактное и занимает двадцать строк.

model = CatBoostClassifier(
  iterations=1500,
  depth=6,
  learning_rate=0.05,
  loss_function="Logloss",
  eval_metric="AUC",
  random_seed=42,
)
model.fit(
  X["train"], y["train"],
  eval_set=(X["valid"], y["valid"]),
  early_stopping_rounds=300,
)

Также был забавный момент с early stopping (когда качество долго не улучшается CatBoost может решить что дальше пробовать бессмысленно и останавливается). Из-за разбиения по времени реальные dot-trick негативы почти целиком попали в train, и валидационная кривая получилась немонотонной. Пик на 25-й итерации, провал, затем медленный рост. Увеличение early_stopping_rounds до 300 позволило CatBoost переждать этот провал и продолжить обучение. В итоге AUC оказался примерно на 0.02 выше.

При этом общего AUC здесь недостаточно. Стоит отдельно смотреть AUC на реальных негативах. Синтетические данные всё-таки синтетические (масло масляное), поэтому хорошая метрика на них ещё не означает, что модель нормально работает на реальном трафике. Также смотрим калибровку по децилям (а точнее по бинам 0-0.1 и 0.9-1.0) и проверяем важность признаков.

Итог для финальной модели на тестовой выборке:

Требования к CSV-файлам

Тренировка работает на выгрузке из сервиса валидации email адресов:

Значения в кавычках, NULL как строка, булевы как True и False. Несколько файлов можно подавать вместе, в том числе из разных баз: дедупликация идёт по адресу, а createdAt приводится к UTC, поэтому разные таймзоны не мешают.

Опционально подаётся второй тип CSV с колонкой fakeReason: из него берутся строки, помеченные как alias-группы или сгенерированные, как дополнительные реальные негативы.

Запуск тренировки

uv sync

# подключаем словари имен
uv run python -m email_ml.dictionaries

# 2. готовим датасет. Разметка, дедупликация, split по времени, синтетические негативы
uv run python -m email_ml.dataset \
  --input data/raw/emails-*.csv \
  --fake-csv data/raw/data-with-fake-reason.csv \
  --output data/processed/naturalness.parquet

# 3. обучение, набор признаков v4, словари в корпусе n-грамм
uv run python -m email_ml.train \
  --dataset data/processed/naturalness.parquet \
  --feature-set v4 --ngram-names \
  --out models/nat-v4n

train печатает AUC по каждому типу негативов, калибровку, важность признаков и скоринг пробного списка из 23 адресов, а рядом с моделью сохраняет nat-v4n.train.json с теми же цифрами и nat-v4n.ngrams.json со словарём n-грамм.

Наборы признаков версионированы: v1 лексика, v2 плюс структура, v3 плюс n-граммы, v4 плюс имена и клавиатурные ряды. Это позволяет повторить всю историю экспериментов одной сменой флага.

Проверка на новых адресах

Теперь, когда модель готова, хочется проверить а как она вообще работает. Получилось ли хоть что-то полезное? Для этого вызовем ее через uv run python -m email_ml.predict

uv run python -m email_ml.predict ivan.ivanov@gmail.com xnjjeka.2.e.1.1@gmail.com dmitry@bask.ws stanislav85@mail.ru vmifvhbgti@rambler.ru satoshi.nakamoto@mail.com geohot@icloud.me mafiaboy@yahoo.com 
0.994  ivan.ivanov@gmail.com
0.000  xnjjeka.2.e.1.1@gmail.com
0.943  dmitry@bask.ws
1.000  stanislav85@mail.ru
0.000  vmifvhbgti@rambler.ru
0.998  satoshi.nakamoto@mail.com
0.206  geohot@icloud.me
0.812  mafiaboy@yahoo.com

Экспорт в ONNX

Далее мы хотим подключить модель в Typescript, а для этого нам нужно выгрузить ее в ONNX

uv run python -m email_ml.export --model models/nat-v4n.cbm --out models/email-pattern-v1
uv run pytest tests/test_onnx_parity.py
uv run python -m email_ml.predict --model models/email-pattern-v1.onnx ivan.ivanov@gmail.com

CatBoost экспортирует в ONNX одной строкой, но с подвохом. В граф добавляется оператор ZipMap, и выход probabilities получает тип seq(map(int64, float)). JavaScript API onnxruntime-node работает только с тензорами. Поэтому экспорт вырезает ZipMap и выставляет наружу сырой тензор вероятностей [N, 2]:

def strip_zipmap(model: onnx.ModelProto) -> onnx.ModelProto:
  graph = model.graph
  zipmap = next((n for n in graph.node if n.op_type == "ZipMap"), None)
  if zipmap is None:
    return model
  tree = next(n for n in graph.node if zipmap.input[0] in n.output)
  graph.node.remove(zipmap)
  for out in list(graph.output):
    if out.name == zipmap.output[0]:
      graph.output.remove(out)
  tree.output[list(tree.output).index(zipmap.input[0])] = "probabilities"
  graph.output.append(
    helper.make_tensor_value_info("probabilities", TensorProto.FLOAT, ["N", 2])
  )
  onnx.checker.check_model(model)
  return model

Экспорт создаёт четыре артефакта:

• email-pattern-v1.onnx

• email-pattern-v1.meta.json - порядок признаков, имена входа и выхода, пути к словарям, метрики. TypeScript читает порядок признаков отсюда и не хардкодит его.

• email-pattern-v1.ngrams.json - словарь n-грамм.

• tests/fixtures/parity.json - мы хотим убедиться, что наш будущий TS код коррекно воспроизводит векторы. Поэтому сюда кладем адреса с векторами признаков и скорингом. TypeScript-экстрактор обязан воспроизвести векторы точно, а скоринг с точностью до пятого знака.

Экспорт сам сверяет onnxruntime с CatBoost и падает при расхождении больше 1e-5.

Для одной статьи уже достаточно много информации, поэтому TS код напишу отдельно в следующей.

Исходный код проекта https://github.com/baskakov/email-human-ml/