Поиск в CSV Я.Метрики топ-100 сайтов-источников по лидогенерации

Задача

Найти сайты (основные домены), которые генерируют трафик приводящий лиды с учётом количества лидов, количества сессий, продолжительность лидогенерации (чтобы отсечь разовые всплески), показатель конверсий.

Начальные условия

Из Я.Метрики выкачены данные в CSV-файлы по сессиям (визитам), достижениям цели со сторонних сайтов "Referrer". Колонки датафрейма:

• "External referrer domain" — тип object, домен с которого был переход;
• "Date of visit" — тип datetime64[ns], дата;
• "Sessions" — тип int64, количество сессий (визитов) за дату;
• Цели в Я.Метрике:
• "Goals_reached_id_1" — тип int64, количество лидов по цели с идентификатором id_1 за дату;
• "Goals_reached_id_2" — тип int64, количество лидов по цели с идентификатором id_2 за дату;
  ...
• "Goals_reached_id_n" — количество лидов по цели с идентификатором id_n за дату.

Открываем Jupyter Notebook, пишем код...

1. Начальные настройки

# Рекомендуется установить
pip install tldextract --quiet

import glob
import locale
import numpy as np
import os
import pandas as pd
import re

# Отключаем предупреждения в консоль
import warnings
warnings.filterwarnings('ignore')

# Для визаулизации загрузки CSV-файлов
from tqdm import tqdm

# Настройка локали для форматирования чисел с пробелами
locale.setlocale(locale.LC_ALL, 'ru_RU.utf8')

# Цель из Я.Метрики
DEMO_COL = 'Goals_reached_1'

2. Инициация. Загрузка данных

folder_path = 'path_to_csv_folder/'

# Если CSV лежат в папке 'path_to_csv_folder' в папках по годам, то...
# Укажите список папок, из которых нужно выгружать файлы
target_folders = ['2026']

# Собираем все CSV-файлы только из указанных папок
csv_files = []
for target_folder in target_folders:
····folder_pattern = os.path.join(folder_path, target_folder, '**', '*.csv')
····csv_files.extend(glob.glob(folder_pattern, recursive=True))

# Сортируем и удаляем дубликаты (на случай, если папки повторяются)
csv_files = sorted(set(csv_files))
if not csv_files:
····raise FileNotFoundError(f"❌ Не найдено CSV-файлов в папках: {', '.join(target_folders)}")
print(f"📁 Найдено {len(csv_files):,} CSV-файлов для загрузки из папок: {', '.join(target_folders)}\n")
dfs = []
failed_files = []

for file in tqdm(csv_files, desc="Загрузка", unit="файл", ncols=80, colour='green'):
····try:
········df_temp = pd.read_csv(file, sep=None, engine='python', encoding='utf-8-sig')
········dfs.append(df_temp)
····except Exception as e:
········failed_files.append((file, str(e)))
········continue

# Итоги загрузки
if not dfs:
····raise ValueError("❌ Не удалось загрузить ни одного файла")
df = pd.concat(dfs, ignore_index=True)
print(f"\n✅ Загрузка завершена:")
print(f" • Успешно: {len(dfs):>5} файлов → {len(df)} строк")
print(f" • Ошибок: {len(failed_files):>5} файлов")

if failed_files:
····print(f"\n⚠️ Примеры ошибок (первые 3):")
····for i, (file, error) in enumerate(failed_files[:3], 1):
········print(f" {i}. {os.path.basename(file)[:50]:<50} → {error[:60]}...")

Загрузка файлов CSV

3. Обработка загруженных данных

# Удаление ненужных столбцов
df.drop(labels=['name_col1', 'name_col2', ... name_colN], axis=1, inplace=True, errors='raise')
print(f"✅ Удалены лишние колонки")

# Если данные загружать из API, то них есть строка с 'Totals and averages'.
# Удаляем эту строку из таблицы, она не нужна.
df = df[df['Device type'] != 'Totals and averages'].copy()

# Переводим 'Date of visit' → формат 'datetime64[ns]'
df['Date of visit'] = pd.to_datetime(df['Date of visit'], errors='coerce')
print(f"✅ Перевели 'Date of visit' в формат datetime64[ns]")

# Выводим в консоль Период данных
print(f"✅ Период данных: {df['Date of visit'].min().strftime("%Y-%m-%d")} - {df['Date of visit'].max().strftime("%Y-%m-%d")}.")

# Смотрим что получилось
df.info()

4. Вспомогательные функции

Функция extract_registered_domain извлекает основной домен, превращает любые вариации ссылок в единый формат domain.suffix, отбрасывая поддомены, пути и параметры.

Пример:

• https://aiml.ru/inbox → aiml.ru
• https://sub.egpe.ru/page?utm=1 → egpe.ru
• NaN / None → None

def extract_registered_domain(series):
····# Если установлен tldextract
····try:
········import tldextract
········def _extract(url):
············if pd.isna(url):
················return None
············ext = tldextract.extract(str(url).lower().strip())
············if ext.domain and ext.suffix:
················return f"{ext.domain}.{ext.suffix}"
············return None
········return series.apply(_extract)

····# Если не установлен tldextract то через re
····except ImportError:
········def _extract_simple(url):
············if pd.isna(url):
················return None
············s = str(url).lower().strip()
············s = re.sub(r'^https?://', '', s)
············s = re.sub(r'^www\.', '', s)
············s = re.sub(r':\d+.*$', '', s)
············s = re.sub(r'[/?#].*$', '', s)
············parts = s.rstrip('.').split('.')
············if len(parts) >= 2:
················if parts[-1] in ['ru', 'com', 'org', 'net', 'biz', 'info']:
····················return '.'.join(parts[-2:])
················if len(parts) >= 3 and parts[-2] in ['co', 'com', 'org', 'gov', 'ac'] and parts[-1] in ['uk', 'au', 'jp']:
····················return '.'.join(parts[-3:])
················return '.'.join(parts[-2:])
············return None
········return series.apply(_extract_simple)

Функция is_valid_domain_vectorized возвращает булеву маску (True/False) для фильтрации валидных доменов в векторизованном режиме. Проверяет каждую строку в колонке и говорит, является ли она «похожей на настоящий домен».

def is_valid_domain_vectorized(series):
····# Приводит к строке, убирает пробелы в начале и в конце, делает нижний регистр
····s = series.astype(str).str.strip().str.lower()

····# Отсекает пустые, служебные и технические значения: '', 'nan', 'localhost', '127.0.0.1'
····invalid_values = ['', 'nan', 'none', 'null', 'not specified', 'localhost', '127.0.0.1', '0.0.0.0', '::1', 'local']
····mask_valid = ~s.isin(invalid_values)

····# Отбрасывает чистые IPv4 (с портом или без) '192.168.1.1', '10.0.0.5:8080'
····ip_pattern = r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(:\d+)?$'
····mask_not_ip = ~s.str.match(ip_pattern, na=False)

····# Домен должен содержать хотя бы одну точку
····mask_has_dot = s.str.contains(r'\.', na=False)

····# Отсекает «псевдо-домены» типа 123.456.789
····mask_not_only_digits = ~s.str.match(r'^[\d\.]+$', na=False)

····# Отсекает слишком короткие строки 'a', 'ab' → ❌, 'a.ru' → ✅
····mask_min_len = s.str.len() >= 3

····# Все условия выше должны выполниться одновременно
····return mask_valid & mask_not_ip & mask_has_dot & mask_not_only_digits & mask_min_len

5. Извлечение доменов

print("🔍 Извлекаем домены...")
df['registered_domain'] = extract_registered_domain(df['External referrer domain'])
valid_mask = is_valid_domain_vectorized(df['registered_domain'])
df_valid = df[valid_mask & df['registered_domain'].notna()].copy()
print(f"✅ Валидных строк с доменами: {len(df_valid):,}")

6. Агрегация по году и неделям

df_agg = df_valid.groupby(['registered_domain', 'Date of visit'], as_index=False).agg({'Sessions': 'sum', DEMO_COL: 'sum'})

# Сортировка по датам
df_agg.sort_values(by='Date of visit', inplace=True)

# Колонка 'Год-Неделя'
df_agg['year_week'] = df_agg['Date of visit'].dt.strftime('%Y-%W')
df_agg_week_site = df_agg.groupby(['registered_domain', 'year_week'], as_index=False).agg({'Sessions': 'sum', DEMO_COL: 'sum'})
df_agg_week_site.info()

7. Настройки критериев отбора лучших сайтов

TOP_LIST = 100 # Сколько сайтов попадёт в топ
MIN_WEEKS = 4 # Мин. недель наблюдения
MIN_PCT_ABOVE_Q75 = 0.50 # Мин. доля недель в топ-25% по лидам
MIN_AVG_LEADS = 3 # Мин. среднее лидов в неделю
MIN_TOTAL_SESSIONS = 50 # Мин. сессий за период

# Раасчёт 75-го перцентиля по неделям
weekly_q75 = df_agg_week_site.groupby('year_week')[DEMO_COL].quantile(0.75).reset_index()
weekly_q75.columns = ['year_week', 'demo_q75_week']

df_with_q75 = df_agg_week_site.merge(weekly_q75, on='year_week', how='left')
df_with_q75['above_q75'] = np.where(
····df_with_q75['demo_q75_week'] == 0,
····df_with_q75[DEMO_COL] >= 1,
····df_with_q75[DEMO_COL] >= df_with_q75['demo_q75_week'])

8. Агрегация по доменам

domain_stats = df_with_q75.groupby('registered_domain').agg(
····weeks_observed=('year_week', 'count'),
····weeks_above_q75=('above_q75', 'sum'),
····avg_demo_leads=(DEMO_COL, 'mean'),
····total_sessions=('Sessions', 'sum'),
····total_demo_leads=(DEMO_COL, 'sum')).reset_index()

domain_stats['pct_above_q75'] = domain_stats['weeks_above_q75'] / domain_stats['weeks_observed']

9. Добавляем колоноки

# 9.1. Фильтр стабильности
domain_stats['is_stable_leader'] = (
····(domain_stats['weeks_observed'] >= MIN_WEEKS) &
····(domain_stats['pct_above_q75'] >= MIN_PCT_ABOVE_Q75) &
····(domain_stats['avg_demo_leads'] >= MIN_AVG_LEADS) &
····(domain_stats['total_sessions'] >= MIN_TOTAL_SESSIONS))

stable_count = domain_stats['is_stable_leader'].sum()

# 9.2. Приоритетный скор
def calc_hybrid_score(row):
····freq_score = row['pct_above_q75'] * 40
····volume_score = min(row['avg_demo_leads'] * 3, 30)
····scale_score = np.log1p(row['total_sessions']) * 2
····return round(freq_score + volume_score + scale_score, 2)

domain_stats['hybrid_score'] = domain_stats.apply(calc_hybrid_score, axis=1)

# 9.3. Конверсии
domain_stats['cr_demo_per_session'] = np.where(
····domain_stats['total_sessions'] > 0,
····(domain_stats['total_demo_leads'] / domain_stats['total_sessions']).round(6), 0.0)

Скоринг (минимальные требования)

Сайт считается «полезным» (is_stable_leader = True) только если выполняются все условия:

weeks_observed >= MIN_WEEKS

• ≥ 4 недели
  Отсекает сайты с 1-2 появлениями (случайность)

pct_above_q75 >= MIN_PCT_ABOVE_Q75

• ≥ 50%
  Сайт должен быть в топ-25% чаще, чем нет

avg_demo_leads >= MIN_AVG_LEADS

• ≥ 3 лида/неделю
  Отсекает сайты с «символическими» 1-2 лидами

total_sessions >= MIN_TOTAL_SESSIONS

• ≥ 50 сессий
  Фильтр «мусорного» трафика

10. TOP-N = TOP_LIST

if stable_count >= TOP_LIST:
top_N = domain_stats[domain_stats['is_stable_leader']].nlargest(TOP_LIST, 'hybrid_score').reset_index(drop=True)
else:
print(f"\nℹ️ Стабильных лидеров только {stable_count}. Дополняем топ по скорингу...")
stable_part = domain_stats[domain_stats['is_stable_leader']]
remaining_part = domain_stats[~domain_stats['is_stable_leader']].nlargest(TOP_LIST - len(stable_part), 'hybrid_score')
top_N = pd.concat([stable_part, remaining_part], ignore_index=True).nlargest(TOP_LIST, 'hybrid_score').reset_index(drop=True)

11. Подготовка к экспорту

export_cols = [
····'registered_domain','weeks_observed',
····'pct_above_q75', 'avg_demo_leads',
····'total_demo_leads', 'total_sessions',
····'cr_demo_per_session', 'hybrid_score']

top_N_export = top_N[export_cols].copy()

# Переименование для Excel
top_N_export = top_N_export.rename(columns={
····'pct_above_q75': 'pct_weeks_in_top25',
····'avg_demo_leads': 'avg_demo_leads_per_week',
····'total_demo_leads': 'total_demo_leads_period'})

# Округление
top_N_export['pct_weeks_in_top25'] = (top_N_export['pct_weeks_in_top25'] * 100).round(1)
top_N_export['avg_demo_leads_per_week'] = top_N_export['avg_demo_leads_per_week'].round(2)
top_N_export['cr_demo_per_session'] = top_N_export['cr_demo_per_session'].round(6)

# Сортировка и сохранение 💾
top_N_export = top_N_export.sort_values('hybrid_score', ascending=False).reset_index(drop=True)
OUTPUT_FILE = f'top{TOP_LIST}_stable_domains_with_cr.xlsx'
top_N_export.to_excel(OUTPUT_FILE, index=False)
print(f"✅ Готово! Топ-{TOP_LIST} с конверсией сохранён в {OUTPUT_FILE}")
print(f"📊 Диапазон hybrid_score: {top_N_export['hybrid_score'].min():.2f} — {top_N_export['hybrid_score'].max():.2f}")

Весь код на GitHub.