Создаешь ли библиотеки на Python

Создавал ли я Python библиотеки для analytics

Да, создавал несколько утилит и вспомогательные модули. Вот что.

Библиотека 1: A/B Test Stats Helper (2019)

Проблема: Каждый раз когда нужно анализировать A/B тест, я пишу одно и то же:

from scipy.stats import chi2_contingency
import numpy as np

# Расчёт
control_conv = 100
control_total = 1000
treatment_conv = 120
treatment_total = 1000

# Chi-square
chi2, pval, dof, expected = chi2_contingency([
    [control_conv, control_total - control_conv],
    [treatment_conv, treatment_total - treatment_conv]
])

# Интерпретация
print(f"P-value: {pval}")
print(f"Significant: {pval < 0.05}")

Решение: Создать свой модуль

# stats_utils.py
from scipy.stats import chi2_contingency
from scipy import stats
import numpy as np

class ABTest:
    def __init__(self, control_conversions, control_total, treatment_conversions, treatment_total):
        self.control_conv = control_conversions
        self.control_total = control_total
        self.treatment_conv = treatment_conversions
        self.treatment_total = treatment_total
    
    def chi_square_test(self):
        chi2, pval, dof, expected = chi2_contingency([
            [self.control_conv, self.control_total - self.control_conv],
            [self.treatment_conv, self.treatment_total - self.treatment_conv]
        ])
        return {
            'chi2': chi2,
            'p_value': pval,
            'significant': pval < 0.05,
            'degrees_of_freedom': dof
        }
    
    def effect_size(self):
        control_rate = self.control_conv / self.control_total
        treatment_rate = self.treatment_conv / self.treatment_total
        relative_lift = (treatment_rate - control_rate) / control_rate * 100
        return {
            'control_rate': control_rate,
            'treatment_rate': treatment_rate,
            'relative_lift_pct': relative_lift
        }
    
    def confidence_interval(self, confidence=0.95):
        treatment_rate = self.treatment_conv / self.treatment_total
        se = np.sqrt(treatment_rate * (1 - treatment_rate) / self.treatment_total)
        z = 1.96  # 95% CI
        margin = z * se
        return {
            'point_estimate': treatment_rate,
            'lower': treatment_rate - margin,
            'upper': treatment_rate + margin
        }

# Usage
test = ABTest(100, 1000, 120, 1000)
print(test.chi_square_test())
print(test.effect_size())
print(test.confidence_interval())

Бенефит: Вместо 20 строк, теперь 3 строки. Меньше ошибок.

Библиотека 2: SQL Query Helper (2020)

Проблема: Часто нужно писать когортный анализ:

WITH first_purchase AS (
  SELECT 
    user_id,
    DATE(created_at) as cohort_date
  FROM orders
  WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 2 YEAR)
  GROUP BY user_id, DATE(created_at)
)
SELECT 
  fp.cohort_date,
  DATE_DIFF(DATE(o.created_at), fp.cohort_date, DAY) as days_since_cohort,
  COUNT(DISTINCT o.user_id) as users
FROM first_purchase fp
JOIN orders o ON fp.user_id = o.user_id
GROUP BY fp.cohort_date, days_since_cohort
ORDER BY fp.cohort_date, days_since_cohort;

Это template который я использую каждый день. Но вводить каждый раз скучно.

Решение: Python helper который генерирует SQL

# sql_builder.py
class CohortQuery:
    def __init__(self, table, user_id_col, date_col, event_filter=''):
        self.table = table
        self.user_id = user_id_col
        self.date = date_col
        self.filter = event_filter
    
    def build(self, lookback_days=730):
        query = f"""
WITH first_event AS (
  SELECT 
    {self.user_id},
    DATE({self.date}) as cohort_date
  FROM {self.table}
  WHERE {self.date} >= DATE_SUB(CURDATE(), INTERVAL {lookback_days} DAY)
  {'AND ' + self.filter if self.filter else ''}
  GROUP BY {self.user_id}, DATE({self.date})
)
SELECT 
  fe.cohort_date,
  DATE_DIFF(DATE(t.{self.date}), fe.cohort_date, DAY) as days_since_cohort,
  COUNT(DISTINCT t.{self.user_id}) as users
FROM first_event fe
JOIN {self.table} t ON fe.{self.user_id} = t.{self.user_id}
GROUP BY fe.cohort_date, days_since_cohort
ORDER BY fe.cohort_date, days_since_cohort;
        """
        return query

# Usage
query_builder = CohortQuery(
    table='orders',
    user_id_col='user_id',
    date_col='created_at',
    event_filter="status = 'completed'"
)
print(query_builder.build(lookback_days=365))

Бенефит: Вместо copy-paste, генерирую параметрически. Меньше синтаксических ошибок.

Библиотека 3: Data Validation (2021)

Проблема: Чем больше дашборды, тем больше ошибок в данных:

• Пропущенные timestamps
• Negative values где не должны
• Impossible dates (future timestamps)
• Missing joins

Решение: Создать validation framework

# data_validator.py
import pandas as pd
from datetime import datetime

class DataValidator:
    def __init__(self, df, table_name):
        self.df = df
        self.table = table_name
        self.errors = []
    
    def check_nulls(self, columns, allowed_pct=1.0):
        for col in columns:
            null_pct = (self.df[col].isna().sum() / len(self.df)) * 100
            if null_pct > allowed_pct:
                self.errors.append(f"{self.table}.{col}: {null_pct}% nulls (allowed: {allowed_pct}%)")
    
    def check_range(self, column, min_val, max_val):
        invalid = self.df[(self.df[column] < min_val) | (self.df[column] > max_val)]
        if len(invalid) > 0:
            self.errors.append(f"{self.table}.{column}: {len(invalid)} values out of range [{min_val}, {max_val}]")
    
    def check_timestamp_order(self, start_col, end_col):
        invalid = self.df[self.df[start_col] > self.df[end_col]]
        if len(invalid) > 0:
            self.errors.append(f"{self.table}: {len(invalid)} rows where {start_col} > {end_col}")
    
    def check_future_dates(self, date_col):
        future = self.df[self.df[date_col] > datetime.now()]
        if len(future) > 0:
            self.errors.append(f"{self.table}.{date_col}: {len(future)} future dates")
    
    def validate(self):
        return self.errors if self.errors else ["All checks passed!"]

# Usage
validator = DataValidator(orders_df, 'orders')
validator.check_nulls(['user_id', 'order_date'], allowed_pct=0.1)
validator.check_range('amount', 0, 1000000)
validator.check_timestamp_order('order_date', 'delivered_date')
validator.check_future_dates('order_date')
print(validator.validate())

Бенефит:

• Автоматизирую проверку данных
• Поймаю ошибки до того как они попадут в дашборд
• Экономлю hours debugging

Что я НЕ создавал

Не создавал: Production frameworks (ORM, API clients)

Это не для Product Analyst. Если нужна - обращаюсь к инженерам.

Не создавал: Сложные ML библиотеки

Это для Data Scientists. Я использую готовые (scikit-learn, statsmodels).

Как я организую свои utilities

Обычно в одном проекте:

project/
├── notebooks/
│   ├── analysis_1.ipynb
│   └── analysis_2.ipynb
├── utils/
│   ├── __init__.py
│   ├── stats_utils.py
│   ├── sql_builder.py
│   ├── data_validator.py
│   └── plotting_utils.py
├── data/
│   └── raw/
└── requirements.txt

Когда я использую эти utilities

Ежедневно:

• A/B test analysis → stats_utils
• Data quality checks → data_validator

Еженедельно:

• Когортный анализ → sql_builder

Ежемесячно:

• Создание новых utility functions если часто использую

Советы по созданию analytics utils

1. DRY (Don't Repeat Yourself)

Если ты делаешь одно и то же 3+ раза → создай function/class.

2. Keep it simple

Утилити должны быть проще чем код который они заменяют. Иначе зачем они?

3. Document

def chi_square_test(control, treatment, control_total, treatment_total):
    """
    Perform chi-square test for A/B test.
    
    Args:
        control: Number of conversions in control group
        treatment: Number of conversions in treatment group
        control_total: Total users in control group
        treatment_total: Total users in treatment group
    
    Returns:
        dict with p_value, significant (bool), effect_size
    """

4. Type hints

def calculate_ltv(revenue: float, retention_rate: float) -> float:
    return revenue / (1 - retention_rate)

5. Tests

def test_chi_square():
    result = chi_square_test(100, 120, 1000, 1000)
    assert result['significant'] == True
    assert 0 < result['p_value'] < 1

Мой вывод

Пython utilities экономят время и減少 ошибки. Но не нужно создавать mega-framework.

Фокус на:

• Small, reusable functions
• Clear documentation
• Easy to use

Если ты analyst пишешь много Python, utilities помогают. Но это не главная тебя работа. Главная — insights.