Python for Analysts

# A type is a promise about what you can do with a value
users   = 1200              # int   — whole number
conv    = 0.048             # float — fractional
metric  = "checkout_rate"   # str   — text
is_sig  = True              # bool  — yes/no
missing = None              # None  — there is NO value (not zero!)

print(type(users), type(conv))

# The classic trap: numbers that are secretly strings
a, b = "120", "80"
print(a + b)            # 12080  — glued, not added
print(int(a) + int(b))  # 200    — as intended

# None is not zero
never_bought = None
refunded_all = 0
print(never_bought is None, refunded_all == 0)

# List — order matters, access by position
revenue = [120.5, 80.0, 340.2, 55.0]
print(revenue[0], revenue[-1], revenue[1:3])

# Dict — lookup by key
user_plan = {101: "pro", 102: "free", 103: "pro"}
print(user_plan[102])                 # free
print(user_plan.get(999, "unknown"))  # safe default, no error

# Set — uniqueness and fast membership tests
paid_users = {101, 103, 107}
print(107 in paid_users)

# Why the choice matters for speed
import time
big_list = list(range(1_000_000))
big_set  = set(big_list)

t0 = time.perf_counter(); 999_999 in big_list; t_list = time.perf_counter() - t0
t0 = time.perf_counter(); 999_999 in big_set;  t_set  = time.perf_counter() - t0
print(f"list: {t_list*1000:.2f} ms | set: {t_set*1000:.4f} ms")

# if / elif / else: Python checks conditions top to bottom
# and runs the FIRST branch whose condition is True.
def segment(revenue):
    if revenue >= 1000:
        return "VIP"
    elif revenue >= 100:        # reached only if the line above was False
        return "regular"
    elif revenue > 0:
        return "small"
    else:
        return "inactive"

for r in [2500, 340, 15, 0]:
    print(r, "->", segment(r))

# Comparisons combine with and / or / not
is_paid, is_new = True, False
print(is_paid and not is_new)          # True

# "Truthiness": empty things count as False
for value in [0, "", [], None, 42, "hi", [1]]:
    print(repr(value), "->", bool(value))

# One-line if (the "ternary" form)
p_value = 0.03
print("significant" if p_value < 0.05 else "not significant")

# for: repeat an action for every element of a collection
sales = [120, 80, 300]
for amount in sales:
    print("sale:", amount)

# range(n): the numbers 0, 1, ..., n-1
for day in range(3):
    print("day", day)

# enumerate: position AND value — instead of range(len(...))
for i, amount in enumerate(sales, start=1):
    print(f"order #{i}: {amount}")

# zip: walk two lists side by side
users = ["ann", "bob", "eve"]
for user, amount in zip(users, sales):
    print(user, amount)

# while: repeat WHILE a condition holds — you must make it stop
balance, months = 1000.0, 0
while balance < 1500:
    balance *= 1.05              # +5% per month
    months += 1
print("months to reach 1500:", months)

# break stops the loop early; continue skips to the next element
for amount in [120, -5, 300, 9999, 40]:
    if amount < 0:
        continue                 # skip invalid values
    if amount > 5000:
        print("suspicious amount, stopping")
        break
    print("ok:", amount)

orders = [
    {"id": 1, "amount": 120, "status": "paid"},
    {"id": 2, "amount": 0,   "status": "cancelled"},
    {"id": 3, "amount": 340, "status": "paid"},
]

amounts  = [o["amount"] for o in orders]                        # transform
paid     = [o["amount"] for o in orders if o["status"] == "paid"]  # + filter
by_id    = {o["id"]: o["amount"] for o in orders}               # dict
statuses = {o["status"] for o in orders}                        # set

print(amounts)
print(paid)
print(by_id)
print(sorted(statuses))

def lift(control, variant):
    """Relative uplift in %. The docstring is what your editor
    shows as a hint later — it is not decoration."""
    if control == 0:
        return None                 # honestly return "unknown"
    return (variant - control) / control * 100

print(round(lift(0.098, 0.1136), 1))
print(lift(0, 0.05))                # None — no division by zero

def summarize(values, digits=2, with_median=True):
    out = {"n": len(values), "mean": round(sum(values)/len(values), digits)}
    if with_median:
        srt = sorted(values); mid = len(srt)//2
        out["median"] = round(srt[mid] if len(srt) % 2 else (srt[mid-1]+srt[mid])/2, digits)
    return out

print(summarize([120, 80, 340, 55]))

# 1. Positional vs keyword arguments, defaults
def conversion(conversions, visitors, as_percent=True, digits=1):
    rate = conversions / visitors
    return round(rate * 100, digits) if as_percent else rate

print(conversion(45, 900))                         # positional
print(conversion(visitors=900, conversions=45))    # keywords: order no longer matters
print(conversion(45, 900, as_percent=False))

# 2. THE TRAP: a mutable default is created ONCE and shared by every call
def add_event_bad(event, log=[]):
    log.append(event)
    return log

print(add_event_bad("open"))
print(add_event_bad("click"))     # surprise: "open" is still there!

def add_event_good(event, log=None):
    if log is None:               # a fresh list on every call
        log = []
    log.append(event)
    return log

print(add_event_good("open"))
print(add_event_good("click"))

# 3. Returning several values = returning a tuple, then unpacking it
def min_max_mean(values):
    return min(values), max(values), sum(values) / len(values)

lo, hi, avg = min_max_mean([120, 80, 340, 55])
print(lo, hi, avg)

# 4. *args collects extra positional args, **kwargs extra keyword args
def log_metric(name, *values, **tags):
    print(name, values, tags)

log_metric("dau", 120, 135, 128, platform="ios", country="RU")

# 5. Scope: a variable created inside a function lives only there
total = 0
def add(x):
    total = x          # a NEW local variable, the global one is untouched
    return total
add(5)
print("global total is still", total)

# 6. lambda: a tiny unnamed function, most often used as a sort key
orders = [("ann", 300), ("bob", 120), ("eve", 540)]
print(sorted(orders, key=lambda o: o[1], reverse=True))
print(max(orders, key=lambda o: o[1]))

# Strings are everywhere: messy CSV columns, logs, interview questions
raw = "  Ivan PETROV ;[email protected]; Moscow  "

name, email, city = [part.strip() for part in raw.split(";")]
print(name.title(), "|", email.lower(), "|", city)

print(", ".join(["RU", "KZ", "BY"]))          # list -> one string
print("checkout_page_v2".replace("_", " "))
print("2026-03-15".startswith("2026"))

# f-strings: the cleanest way to format numbers for a report
conv, revenue, users = 0.04837, 1234567.891, 42
print(f"conversion: {conv:.1%}")             # percent, 1 decimal
print(f"revenue:    {revenue:,.0f}")         # thousands separator
print(f"users:      {users:>6}")             # right-align in 6 chars
print(f"{'metric':<10}|{'value':>8}")         # simple aligned table

# Strings are immutable: methods return a NEW string
s = "hello"
s.upper()
print(s)             # still "hello"
s = s.upper()
print(s)

# A real export: some values are broken. The script must not die on them.
raw_amounts = ["120.5", "80", "n/a", "", "340,2", "55"]

parsed, failed = [], []
for raw in raw_amounts:
    try:
        value = float(raw.replace(",", "."))   # may raise ValueError
    except ValueError:
        failed.append(raw)                      # catch ONLY what we expect
    else:
        parsed.append(value)                    # runs only if no error

print("parsed:", parsed)
print("failed:", failed)

# raise: stop with a clear message when an input makes no sense
def safe_rate(part, whole):
    if whole <= 0:
        raise ValueError(f"whole must be positive, got {whole}")
    return part / whole

try:
    safe_rate(5, 0)
except ValueError as e:
    print("caught:", e)

# finally runs no matter what — use it to clean up
try:
    result = 10 / 2
finally:
    print("done, result =", result)

import time

def has_duplicate_slow(items):      # O(n^2) — every pair compared
    for i in range(len(items)):
        for j in range(i+1, len(items)):
            if items[i] == items[j]:
                return True
    return False

def has_duplicate_fast(items):      # O(n) — one pass, remembering
    seen = set()
    for x in items:
        if x in seen:
            return True
        seen.add(x)
    return False

data = list(range(5000))            # worst case: no duplicates at all

t0 = time.perf_counter(); has_duplicate_slow(data); slow = time.perf_counter()-t0
t0 = time.perf_counter(); has_duplicate_fast(data); fast = time.perf_counter()-t0
print(f"O(n^2): {slow*1000:.1f} ms")
print(f"O(n):   {fast*1000:.2f} ms")
print(f"faster: {slow/fast:.0f}x")

# Pattern: a dict remembers what we have already seen.
# Analytical meaning: find two orders that add up to a target sum.
def two_sum(nums, target):
    seen = {}                       # value -> index
    for i, n in enumerate(nums):
        if target - n in seen:      # has the complement appeared before?
            return [seen[target - n], i]
        seen[n] = i
    return []

print(two_sum([120, 45, 300, 80, 55], 135))   # 80 + 55

# Same pattern: match two tables by key without a nested loop
users  = {101: "RU", 102: "KZ", 103: "RU"}
orders = [(101, 120), (103, 340), (999, 50)]
joined = [(uid, amt, users.get(uid, "unknown")) for uid, amt in orders]
print(joined)

from collections import Counter

pages = ["/home","/pricing","/home","/docs","/home","/pricing","/blog"]

counts = Counter(pages)
print(counts)                       # how many times each page appeared
print(counts.most_common(2))        # ranking, ready-made

# Share of traffic, not just raw counts
total = sum(counts.values())
for page, n in counts.most_common(3):
    print(f"{page:<10} {n}  {n/total*100:.0f}%")

# Best 3 consecutive days of sales.
# Naive: recompute the sum for every position — O(n*k)
# Sliding window: add the entering value, subtract the leaving one — O(n)
def max_window(values, k):
    current = sum(values[:k])
    best = current
    for i in range(k, len(values)):
        current += values[i] - values[i - k]
        best = max(best, current)
    return best

sales = [120, 80, 300, 55, 400, 90, 20]
print(max_window(sales, 3))         # 300 + 55 + 400

# pandas does exactly this with rolling()
import pandas as pd
print(pd.Series(sales).rolling(3).sum().max())

from collections import defaultdict

# Pattern: group by a key you compute yourself.
def group_by_key(items, key_fn):
    groups = defaultdict(list)
    for x in items:
        groups[key_fn(x)].append(x)
    return dict(groups)

# Anagrams: the key is the sorted letters
words = ["listen", "silent", "enlist", "google", "gogole"]
print(group_by_key(words, lambda w: "".join(sorted(w))))

# Same mechanism, real task: merge duplicate customers
names = ["Ivan Petrov", "ivan  petrov", "Anna Sidorova"]
normalise = lambda s: " ".join(s.lower().split())
print(group_by_key(names, normalise))

import random, time
from collections import Counter, defaultdict
import pandas as pd

random.seed(1)
pages = [random.choice(["/home", "/pricing", "/docs", "/blog", "/cart"]) for _ in range(1_000_000)]

def with_get(items):                 # 1. a plain dict and .get()
    counts = {}
    for x in items:
        counts[x] = counts.get(x, 0) + 1
    return counts

def with_defaultdict(items):         # 2. defaultdict: a missing key starts at 0
    counts = defaultdict(int)
    for x in items:
        counts[x] += 1
    return dict(counts)

def with_counter(items):             # 3. Counter: counting is its only job
    return dict(Counter(items))

def with_pandas(items):              # 4. pandas, the way it looks in a notebook
    return pd.Series(items).value_counts().to_dict()

def with_list_count(items):          # 5. list.count() for every distinct value
    return {x: items.count(x) for x in set(items)}

reference = with_get(pages)
print("1,000,000 page views, 5 distinct pages:")
for fn in [with_get, with_defaultdict, with_counter, with_pandas, with_list_count]:
    t0 = time.perf_counter()
    same = fn(pages) == reference
    print(f"  {fn.__name__:<17} {(time.perf_counter() - t0) * 1000:7.1f} ms   same answer: {same}")

# The trap: list.count() rescans the list once per distinct value.
# With 5 pages that is 5 scans; with thousands of user IDs it is thousands.
user_ids = [random.randint(1, 5_000) for _ in range(50_000)]
print("50,000 events, ~5,000 distinct users:")
for fn in [with_counter, with_list_count]:
    t0 = time.perf_counter()
    fn(user_ids)
    print(f"  {fn.__name__:<17} {(time.perf_counter() - t0) * 1000:7.1f} ms")

import heapq, random, time
from collections import Counter
import pandas as pd

random.seed(2)
emails = [f"user{random.randint(1, 10_000)}@mail.com" for _ in range(50_000)]

def with_seen_set(items):            # 1. the classic loop: a set remembers what we took
    seen, result = set(), []
    for x in items:
        if x not in seen:
            seen.add(x)
            result.append(x)
    return result

def with_dict_fromkeys(items):       # 2. dict keys are unique and keep insertion order
    return list(dict.fromkeys(items))

def with_pandas(items):              # 3. pandas: keeps the first occurrence too
    return pd.Series(items).drop_duplicates().tolist()

def with_list_check(items):          # 4. "x not in result" scans a LIST every time
    result = []
    for x in items:
        if x not in result:
            result.append(x)
    return result

reference = with_seen_set(emails)
print("50,000 sign-ups, ~10,000 distinct e-mails, first occurrence kept:")
for fn in [with_seen_set, with_dict_fromkeys, with_pandas, with_list_check]:
    t0 = time.perf_counter()
    same = fn(emails) == reference
    print(f"  {fn.__name__:<19} {(time.perf_counter() - t0) * 1000:8.1f} ms   same answer: {same}")

# Top 3 most active users — three ways to ask the same question
counts = Counter(emails)
top_sorted = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)[:3]
top_heap = heapq.nlargest(3, counts.items(), key=lambda kv: kv[1])
top_counter = counts.most_common(3)
print("top 3:", top_counter)
print("sorted == nlargest == most_common:", top_sorted == top_heap == top_counter)

import numpy as np, pandas as pd, time

s = pd.Series(np.random.rand(100_000))

t0 = time.perf_counter(); loop = [v * 1.2 for v in s];  t_loop = time.perf_counter()-t0
t0 = time.perf_counter(); vec  = s * 1.2;               t_vec  = time.perf_counter()-t0

print(f"loop:   {t_loop*1000:.1f} ms")
print(f"vector: {t_vec*1000:.1f} ms")
print(f"faster: {t_loop/t_vec:.0f}x")

# numpy also gives whole-array maths in one call
a = np.array([120.0, 80.0, 340.0, 55.0])
print(a.mean(), a.std().round(2), np.percentile(a, 90))

import pandas as pd, numpy as np

rng = np.random.default_rng(42)      # fixed seed -> reproducible
n = 500

users = pd.DataFrame({
    "user_id": np.arange(1, n+1),                              # int
    "signup":  pd.to_datetime("2026-01-01")                    # real dates
               + pd.to_timedelta(rng.integers(0, 180, n), unit="D"),
    "country": pd.Categorical(rng.choice(["RU","KZ","BY"], n, p=[.6,.25,.15])),
    "plan":    pd.Categorical(rng.choice(["free","pro"], n, p=[.75,.25]),
                              categories=["free","pro"], ordered=True),
    "revenue": np.round(rng.lognormal(2.5, 1.1, n), 2),
})
users.loc[rng.choice(n, 25, replace=False), "revenue"] = np.nan   # realistic gaps

print(users.dtypes.to_string())

obj_kb = users["country"].astype(object).memory_usage(deep=True)/1024
cat_kb = users["country"].memory_usage(deep=True)/1024
print(f"\nobject: {obj_kb:.1f} KB | category: {cat_kb:.1f} KB "
      f"-> {obj_kb/cat_kb:.0f}x smaller")

import pandas as pd

wide = pd.DataFrame({
    "user_id": [1, 2, 3],
    "jan": [100, 150, 90], "feb": [120, 140, 110], "mar": [130, 160, 95],
})
print("WIDE — easy for a human to read:")
print(wide.to_string(index=False))

# melt: wide -> long. One row = one observation
long = wide.melt(id_vars="user_id", var_name="month", value_name="revenue")
print("\nLONG — easy to analyse:")
print(long.head(4).to_string(index=False))

# pivot: long -> wide again, for a report
print("\nBack via pivot:")
print(long.pivot(index="user_id", columns="month", values="revenue").to_string())

import pandas as pd

orders = pd.DataFrame({
    "order_id": [1, 2, 3, 4, 5, 6],
    "user_id":  [10, 10, 11, 12, 12, 13],
    "country":  ["RU", "RU", "KZ", "KZ", "RU", "KZ"],
    "amount":   [120.0, 0.0, 340.5, 55.0, 0.0, 780.0],
    "status":   ["paid","cancelled","paid","paid","cancelled","paid"],
})

print(orders.shape)                    # rows, columns
print(orders.loc[0, "amount"])         # by label
print(orders.iloc[0, 3])               # by position

by_country = orders.query("status == 'paid'").groupby("country").agg(
    orders=("order_id", "count"),
    buyers=("user_id", "nunique"),     # distinct, not rows
    revenue=("amount", "sum"),
).round(1)
print(by_country)

users = pd.DataFrame({"user_id": [10,11,12,13], "plan": ["free","pro","free","pro"]})
print(orders.merge(users, on="user_id", how="left").groupby("plan")["amount"].sum())

import pandas as pd, numpy as np

df = pd.DataFrame({
    "user_id": [1, 2, 2, 3, 4, 5],
    "signup":  ["2026-01-05","2026-01-07","2026-01-07","2026-02-01",None,"2026-02-14"],
    "revenue": [100.0, np.nan, np.nan, 250.0, 80.0, -15.0],
    "country": ["RU","ru "," KZ","KZ","RU","XX"],
})

print("shape:", df.shape)
print("\nmissing values:\n", df.isna().sum().to_string())
print("\nfull duplicate rows:", df.duplicated().sum())

df["signup"]  = pd.to_datetime(df["signup"], errors="coerce")   # dates as dates
df["country"] = df["country"].str.strip().str.upper()           # unify categories
print("\ncountries after cleaning:", df["country"].unique().tolist())
print("negative revenue:", (df["revenue"] < 0).sum(), "rows")

df = df.drop_duplicates(subset=["user_id"])      # duplicate users
df = df[df["revenue"].fillna(0) >= 0]            # drop impossible values
df["revenue"] = df["revenue"].fillna(0)          # NaN here means "did not pay"

print("\nresult:", df.shape)

import pandas as pd

orders = pd.DataFrame({"country": ["RU","KZ","RU"],
                       "amount":  [1200, 300, 800],
                       "status":  ["paid","paid","cancelled"]})

# BAD: pandas cannot tell if you mean the copy or the original
subset = orders[orders["country"] == "RU"]
# subset["is_big"] = subset["amount"] > 1000    -> SettingWithCopyWarning

# GOOD 1: say explicitly that this is a separate table
subset = orders[orders["country"] == "RU"].copy()
subset["is_big"] = subset["amount"] > 1000

# GOOD 2: modify the original through .loc, no intermediate table
orders.loc[orders["country"] == "RU", "is_big"] = orders["amount"] > 1000

# One readable chain instead of five throwaway variables
top = (orders
       .query("status == 'paid' and amount > 0")
       .groupby("country")["amount"]
       .sum()
       .sort_values(ascending=False)
       .head(5))
print(top.to_string())

import pandas as pd
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

events = pd.DataFrame({
    "user_id": [1,1,1,2,2,3,3,3,3,4,5,5,6,6,6,7,8,8,8,8],
    "step": ["visit","signup_form","submitted","visit","signup_form",
             "visit","signup_form","submitted","activated","visit",
             "visit","signup_form","visit","signup_form","submitted",
             "visit","visit","signup_form","submitted","activated"],
})
order = ["visit", "signup_form", "submitted", "activated"]

# nunique: count DISTINCT users per step, not events
funnel = events.groupby("step")["user_id"].nunique().reindex(order)
conv = (funnel / funnel.iloc[0] * 100).round(1)
print(pd.DataFrame({"users": funnel, "conv_%": conv}))

fig, ax = plt.subplots(figsize=(7, 3.2))
y = range(len(funnel))[::-1]
ax.barh(y, funnel.values, color="#5b50c8", height=0.62)
for yi, n, c in zip(y, funnel.values, conv.values):
    ax.text(n + 0.1, yi, f"{n}  ({c:.0f}%)", va="center", color="#1e2233")
ax.set_yticks(list(y)); ax.set_yticklabels(order)
ax.set_xlabel("Unique users"); ax.set_xlim(0, funnel.max() * 1.25)
ax.grid(axis="y", visible=False)
ax.set_title("Each step keeps fewer users; activation halves", loc="left", fontweight="bold")
plt.show()

import numpy as np, pandas as pd, seaborn as sns
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

rng = np.random.default_rng(7)
revenue = pd.Series(np.round(rng.lognormal(2.6, 1.15, 2000), 2))

print("mean (ARPU):", round(revenue.mean(), 2))
print("median:     ", round(revenue.median(), 2))
top10 = revenue.nlargest(200).sum() / revenue.sum() * 100
print("top 10% share of revenue:", round(top10, 1), "%")
print("users below the mean:", round((revenue < revenue.mean()).mean() * 100, 1), "%")

fig, ax = plt.subplots(figsize=(7, 3.2))
sns.histplot(revenue, bins=60, color="#5b50c8", alpha=0.55, edgecolor="none", ax=ax)
ax.axvline(revenue.mean(), color="#c2410c", lw=2, label=f"mean {revenue.mean():.0f}")
ax.axvline(revenue.median(), color="#1a7a4a", lw=2, ls="--", label=f"median {revenue.median():.0f}")
ax.set_xlim(0, revenue.quantile(0.99))
ax.set_xlabel("Revenue per user"); ax.set_ylabel("Users")
ax.legend(frameon=False)
ax.set_title("The mean sits far right of the typical user", loc="left", fontweight="bold")
plt.show()

import pandas as pd, seaborn as sns
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

# Users still active N months after signup, per signup-month cohort
df = pd.DataFrame({
    "cohort":   ["2026-01"]*4 + ["2026-02"]*3 + ["2026-03"]*2 + ["2026-04"],
    "month":    [0, 1, 2, 3,   0, 1, 2,   0, 1,   0],
    "retained": [1000, 302, 217, 156,   1100, 356, 256,   900, 311,   1250],
})

pivot = df.pivot(index="cohort", columns="month", values="retained")
ret = pivot.div(pivot[0], axis=0) * 100      # % of each cohort's own month 0
print(ret.round(1).to_string())

fig, ax = plt.subplots(figsize=(7, 3.0))
# vmax=50: month 0 is always 100%, so cap the scale to let the rest show colour
sns.heatmap(ret, annot=True, fmt=".0f", cmap="Purples", cbar=False, vmax=50,
            linewidths=3, linecolor="white", ax=ax)
ax.grid(False)
ax.set_yticklabels(ax.get_yticklabels(), rotation=0)
ax.set_xlabel("Months since signup"); ax.set_ylabel("Signup cohort")
ax.set_title("Read down a column: later cohorts retain better", loc="left", fontweight="bold")
plt.show()

import numpy as np, pandas as pd
from scipy import stats
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

data = {"Control": (2500, 245), "Variant A": (2500, 284), "Variant B": (2500, 262)}

rows = []
for name, (n, c) in data.items():
    p = c / n
    se = np.sqrt(p * (1 - p) / n)                 # standard error of a proportion
    rows.append({"group": name, "p": p * 100,
                 "lo": (p - 1.96 * se) * 100, "hi": (p + 1.96 * se) * 100})
df = pd.DataFrame(rows)
print(df.round(2).to_string(index=False))

n1, c1 = data["Control"]; n2, c2 = data["Variant A"]
p1, p2 = c1 / n1, c2 / n2
se_diff = np.sqrt(p1 * (1 - p1) / n1 + p2 * (1 - p2) / n2)
z = (p2 - p1) / se_diff
print(f"\nz = {z:.3f}, p-value = {2 * (1 - stats.norm.cdf(abs(z))):.4f}")

fig, ax = plt.subplots(figsize=(7, 3.0))
y = range(len(df))
ax.errorbar(df["p"], y, xerr=[df["p"] - df["lo"], df["hi"] - df["p"]],
            fmt="o", ms=9, lw=2.2, capsize=6, color="#5b50c8", ecolor="#a59cf7")
ax.axvline(df.loc[0, "p"], color="#9aa0b8", ls="--", lw=1.2)
for yi, r in zip(y, df.itertuples()):
    ax.text(r.hi + 0.12, yi, f"{r.p:.2f}%", va="center", fontweight="bold", color="#1e2233")
ax.set_yticks(list(y)); ax.set_yticklabels(df["group"])
ax.set_xlabel("Conversion, % (bars = 95% CI)")
ax.grid(axis="y", visible=False)
ax.set_title("Intervals overlap the control: no winner yet", loc="left", fontweight="bold")
plt.show()

import numpy as np, pandas as pd
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

rng = np.random.default_rng(5)
df = pd.DataFrame({
    "recency":   rng.exponential(30, 800).clip(1, 180).round(0),
    "frequency": rng.poisson(4, 800) + 1,
    "monetary":  rng.lognormal(3.2, 1.0, 800).round(2),
})

# qcut splits into equal-sized groups (quintiles).
# Recency is reversed: fewer days since purchase = better = score 5
df["R"] = pd.qcut(df["recency"], 5, labels=[5, 4, 3, 2, 1]).astype(int)
df["F"] = pd.qcut(df["frequency"].rank(method="first"), 5, labels=[1, 2, 3, 4, 5]).astype(int)

def segment(r):
    if r.R >= 4 and r.F >= 4: return "Champions"
    if r.R >= 4 and r.F <= 2: return "New"
    if r.R <= 2 and r.F >= 4: return "At risk"
    if r.R <= 2 and r.F <= 2: return "Dormant"
    return "Middle"

df["segment"] = df.apply(segment, axis=1)
print(df["segment"].value_counts().to_string())

colors = {"Champions": "#1a7a4a", "At risk": "#c2410c", "New": "#5b50c8",
          "Dormant": "#9aa0b8", "Middle": "#c4bcf5"}
fig, ax = plt.subplots(figsize=(7, 3.4))
for name, g in df.groupby("segment"):
    ax.scatter(g["recency"], g["frequency"], s=np.sqrt(g["monetary"]) * 3.2,
               alpha=0.55, color=colors[name], label=name, edgecolors="none")
ax.set_xlabel("Recency — days since last purchase"); ax.set_ylabel("Frequency — purchases")
ax.set_xlim(0, 150)
ax.legend(frameon=False, ncol=3, fontsize=8.5, loc="upper right")
ax.set_title("Dot size = money spent. Red = worth winning back", loc="left", fontweight="bold")
plt.show()

import numpy as np, pandas as pd
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

rng = np.random.default_rng(3)
days = pd.date_range("2026-05-01", periods=90)
values = 1000 + np.arange(90) * 4 + rng.normal(0, 45, 90)
values[62:66] -= 380                          # an incident happened here
s = pd.Series(values.round(0), index=days)

roll = s.rolling(7, center=True).mean()       # smooth with a weekly window
resid = s - roll                              # deviation from the trend
thr = 2.5 * resid.std()                       # threshold: 2.5 sigma
anom = s[np.abs(resid) > thr]

print(f"threshold: +/-{thr:.0f}   points found: {len(anom)}")
print(anom.to_string())

fig, ax = plt.subplots(figsize=(7.4, 3.2))
ax.plot(s.index, s.values, color="#a59cf7", lw=1.1, label="daily signups")
ax.plot(roll.index, roll.values, color="#5b50c8", lw=2.4, label="7-day rolling mean")
ax.scatter(anom.index, anom.values, color="#c2410c", s=52, zorder=5,
           edgecolors="white", linewidths=1.4, label="anomaly")
ax.legend(frameon=False, loc="upper left")
ax.set_title("The rolling mean exposes a dip lost in the noise", loc="left", fontweight="bold")
fig.autofmt_xdate()
plt.show()

import numpy as np, pandas as pd
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

rng = np.random.default_rng(21)
segs = ["Mobile iOS", "Mobile Android", "Desktop", "Tablet"]
level = {"Mobile iOS": 62, "Mobile Android": 55, "Desktop": 70, "Tablet": 40}
decay = {"Mobile iOS": 0.9, "Mobile Android": 1.6, "Desktop": 0.5, "Tablet": 2.3}
d = pd.DataFrame([{"segment": s, "week": w,
                   "retention": level[s] - decay[s] * w + rng.normal(0, 1.2)}
                  for s in segs for w in range(12)])

# Weekly drop per segment = slope of a straight-line fit
for s in segs:
    g = d[d.segment == s]
    print(f"{s:<15} {np.polyfit(g.week, g.retention, 1)[0]:+.2f} pp/week")

fig, axes = plt.subplots(2, 2, figsize=(7.4, 4.2), sharex=True, sharey=True)
for ax, seg in zip(axes.ravel(), segs):
    for other in segs:                       # grey context: all the others
        if other != seg:
            g = d[d.segment == other]
            ax.plot(g.week, g.retention, color="#dfe2ee", lw=1.2)
    g = d[d.segment == seg]                  # the highlighted segment
    ax.plot(g.week, g.retention, color="#5b50c8", lw=2.4)
    ax.set_title(seg, loc="left", fontsize=10, fontweight="bold")
fig.supxlabel("Week", color="#4a4f66"); fig.supylabel("Retention, %", color="#4a4f66")
plt.tight_layout(); plt.show()

import numpy as np, pandas as pd, seaborn as sns
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

rng = np.random.default_rng(21)
dd = pd.DataFrame({
    "group": ["Control"] * 300 + ["Variant"] * 300,
    "value": np.r_[rng.lognormal(2.6, 0.9, 300), rng.lognormal(2.78, 0.9, 300)],
})
print(dd.groupby("group")["value"].agg(["mean", "median"]).round(1))

fig, ax = plt.subplots(figsize=(7, 3.1))
sns.boxplot(data=dd, x="value", y="group", hue="group", legend=False, ax=ax,
            width=0.45, showfliers=False, palette=["#9aa0b8", "#5b50c8"], linewidth=1.3)
sns.stripplot(data=dd, x="value", y="group", ax=ax,
              size=2.6, alpha=0.32, color="#1e2233", jitter=0.28)
for i, (name, g) in enumerate(dd.groupby("group")):
    ax.scatter(g["value"].mean(), i, marker="D", s=56, color="#c2410c",
               zorder=6, edgecolors="white")       # the mean as a diamond
ax.set_xlim(0, dd["value"].quantile(0.985))
ax.set_xlabel("Revenue per user"); ax.set_ylabel("")
ax.grid(axis="y", visible=False)
ax.set_title("Box = median & quartiles, diamond = mean, dots = raw data", loc="left", fontweight="bold")
plt.show()

import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

steps  = ["Revenue\nMay", "New\ncustomers", "Win-back", "Upgrades",
          "Downgrades", "Churn", "Revenue\nJune"]
start  = 420
deltas = [85, 22, 48, -19, -63]

fig, ax = plt.subplots(figsize=(7.4, 3.4))
cum, peak = start, start
ax.bar(0, start, color="#5b50c8", width=0.6)
ax.text(0, start + 10, start, ha="center", fontweight="bold")
for i, v in enumerate(deltas, start=1):
    bottom = cum if v > 0 else cum + v       # the key line of a waterfall
    color = "#1a7a4a" if v > 0 else "#c2410c"
    ax.bar(i, abs(v), bottom=bottom, width=0.6, color=color)
    ax.plot([i - 0.7, i - 0.3], [cum, cum], color="#9aa0b8", lw=1, ls=":")
    ax.text(i, bottom + abs(v) + 10, f"{v:+d}", ha="center", fontweight="bold", color=color)
    cum += v
    peak = max(peak, cum)
ax.bar(len(steps) - 1, cum, color="#5b50c8", width=0.6)
ax.text(len(steps) - 1, cum + 10, cum, ha="center", fontweight="bold")
print("June revenue:", cum)
ax.set_xticks(range(len(steps))); ax.set_xticklabels(steps, fontsize=8.5)
ax.set_ylim(0, peak * 1.15); ax.set_ylabel("Revenue, k")   # room above the PEAK, not the end
ax.grid(axis="x", visible=False)
ax.set_title("Where the growth came from, factor by factor", loc="left", fontweight="bold")
plt.show()

import numpy as np, pandas as pd, seaborn as sns
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

rng = np.random.default_rng(11)
n = 600
sessions = rng.poisson(8, n) + 1
df = pd.DataFrame({"sessions": sessions})
df["page_views"]   = sessions * rng.normal(4, 0.8, n)
df["time_on_site"] = df["page_views"] * rng.normal(0.9, 0.25, n)
df["orders"]       = rng.poisson(0.12 * sessions)
df["revenue"]      = df["orders"] * rng.lognormal(3.4, 0.5, n)
df["tickets"]      = rng.poisson(0.4, n) + (df["orders"] > 3)

corr = df.corr().round(2)
print(corr["revenue"].sort_values(ascending=False).to_string())

mask = np.triu(np.ones_like(corr, dtype=bool))      # hide the mirrored half
# drop the first row and last column — after masking they are empty
corr_t, mask_t = corr.iloc[1:, :-1], mask[1:, :-1]
fig, ax = plt.subplots(figsize=(6.4, 4.2))
sns.heatmap(corr_t, mask=mask_t, annot=True, fmt=".2f", cmap="RdBu_r", vmin=-1, vmax=1,
            center=0, square=True, linewidths=2, linecolor="white",
            cbar_kws={"shrink": 0.75}, ax=ax)
ax.grid(False)
ax.set_yticklabels(ax.get_yticklabels(), rotation=0)
ax.set_title("Correlation matrix: lower triangle only", loc="left", fontweight="bold")
plt.show()

import pandas as pd
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

df = pd.DataFrame({
    "channel": ["Organic", "Paid search", "Social", "Email", "Referral", "Display"],
    "before":  [4.1, 3.2, 1.8, 6.5, 5.0, 0.9],
    "after":   [4.6, 3.1, 2.9, 7.1, 4.4, 1.0],
})
df["change"] = df["after"] - df["before"]
df = df.sort_values("change").reset_index(drop=True)   # sort by change
print(df.round(1).to_string(index=False))

fig, ax = plt.subplots(figsize=(7, 3.4))
for i, r in df.iterrows():
    color = "#1a7a4a" if r.change > 0 else "#c2410c"
    ax.plot([r.before, r.after], [i, i], color=color, lw=3, alpha=0.55, zorder=1)
ax.scatter(df["before"], df.index, color="#9aa0b8", s=70, zorder=2, label="before redesign")
ax.scatter(df["after"], df.index, color="#5b50c8", s=70, zorder=3, label="after redesign")
ax.set_yticks(df.index); ax.set_yticklabels(df["channel"])
ax.set_xlabel("Conversion, %")
ax.grid(axis="y", visible=False)
ax.legend(frameon=False, loc="lower right")
ax.set_title("Dumbbell: level AND change at a glance", loc="left", fontweight="bold")
plt.show()

import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

q1 = {"Free": 18, "Starter": 31, "Pro": 46, "Team": 52, "Enterprise": 61}
q2 = {"Free": 20, "Starter": 33, "Pro": 39, "Team": 55, "Enterprise": 63}

fig, ax = plt.subplots(figsize=(5.6, 4.2))
for plan in q1:
    worse = q2[plan] < q1[plan]
    color, lw = ("#c2410c", 2.6) if worse else ("#c9cce0", 1.6)
    ax.plot([0, 1], [q1[plan], q2[plan]], color=color, lw=lw, marker="o")
    ax.text(-0.05, q1[plan], f"{plan}  {q1[plan]}%", ha="right", va="center",
            color="#1e2233" if worse else "#6b6f85")
    ax.text(1.05, q2[plan], f"{q2[plan]}%", ha="left", va="center",
            color="#c2410c" if worse else "#6b6f85", fontweight="bold" if worse else "normal")
    print(f"{plan:<11} {q1[plan]} -> {q2[plan]}  ({q2[plan]-q1[plan]:+d})")

ax.set_xlim(-0.6, 1.3); ax.set_xticks([0, 1]); ax.set_xticklabels(["Q1", "Q2"])
ax.set_yticks([]); ax.grid(False)
for side in ["left", "bottom"]:
    ax.spines[side].set_visible(False)
ax.set_title("Day-30 retention by plan: one line breaks the pattern", loc="left", fontweight="bold")
plt.show()

import numpy as np, pandas as pd
import matplotlib.pyplot as plt
plt.rcParams.update({
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.edgecolor": "#d5d8e6", "axes.labelcolor": "#4a4f66",
    "xtick.color": "#4a4f66", "ytick.color": "#4a4f66",
    "axes.grid": True, "axes.axisbelow": True,
    "grid.color": "#eceef5", "font.size": 10,
})

rng = np.random.default_rng(4)
days = pd.date_range("2026-01-05", periods=26 * 7)          # 26 full weeks
weekday_effect = np.array([1.00, 1.05, 1.07, 1.04, 0.95, 0.62, 0.58])   # Mon..Sun
dau = (1000 * weekday_effect[days.dayofweek]
       * (1 + 0.004 * np.arange(len(days)))                 # slow growth
       * rng.normal(1, 0.04, len(days)))
s = pd.Series(dau.round(0), index=days)

print(s.groupby(s.index.dayofweek).mean().round(0)
       .rename(index=dict(enumerate(["Mon","Tue","Wed","Thu","Fri","Sat","Sun"]))).to_string())

grid = s.values.reshape(26, 7).T                            # rows = weekdays, cols = weeks
fig, ax = plt.subplots(figsize=(7.6, 2.6))
im = ax.imshow(grid, cmap="Purples", aspect="auto")
ax.set_yticks(range(7)); ax.set_yticklabels(["Mon","Tue","Wed","Thu","Fri","Sat","Sun"])
ax.set_xticks(range(0, 26, 4)); ax.set_xticklabels([f"wk {i+1}" for i in range(0, 26, 4)])
ax.grid(False)
for side in ["left", "bottom"]:
    ax.spines[side].set_visible(False)
fig.colorbar(im, ax=ax, shrink=0.8, label="DAU")
ax.set_title("Calendar heatmap: weekends dip, every single week", loc="left", fontweight="bold")
plt.show()