Bitirme projesi: Anadolu Market müşteri analitiği paketi

Proje: müşteri analitiği paketi

Anadolu Market'in pazarlama ekibi üç soru soruyor: "Hangi müşterilere kampanya gönderelim?", "Yeni gelen müşteriler ikinci ay geri dönüyor mu?", "Hangi ürün ve kategori ciroyu taşıyor?" Bu projede üç soruyu cevaplayan, komut satırından çalışan ve testli bir paket kuracaksın. Paket, önceki derslerin hepsini birleştirir: pandas (groupby, qcut, pivot), asyncio (sayfalı veri çekme), argparse (CLI), unittest (doğrulama).

1. Veri katmanı

Veri ders içinde küçük bir sözlük listesi; gerçek hayatta aynı yapı CSV ya da API'den gelir. İlk iş tarihi gerçek tarih türüne çevirmek ve ay sütununu çıkarmak:

import pandas as pd

KAYITLAR = [
    {"musteri": "Ayşe", "tarih": "2024-01-05", "urun": "Kalem", "kategori": "Kırtasiye", "tutar": 120.0},
    {"musteri": "Mehmet", "tarih": "2024-01-12", "urun": "Kulaklık", "kategori": "Elektronik", "tutar": 850.0},
    {"musteri": "Ayşe", "tarih": "2024-02-03", "urun": "Defter", "kategori": "Kırtasiye", "tutar": 60.0},
    {"musteri": "Zeynep", "tarih": "2024-02-10", "urun": "Kulaklık", "kategori": "Elektronik", "tutar": 850.0},
    {"musteri": "Ayşe", "tarih": "2024-03-15", "urun": "Kulaklık", "kategori": "Elektronik", "tutar": 850.0},
    {"musteri": "Mehmet", "tarih": "2024-03-20", "urun": "Kalem", "kategori": "Kırtasiye", "tutar": 120.0},
    {"musteri": "Ali", "tarih": "2024-03-22", "urun": "Çanta", "kategori": "Aksesuar", "tutar": 400.0},
    {"musteri": "Zeynep", "tarih": "2024-04-01", "urun": "Defter", "kategori": "Kırtasiye", "tutar": 60.0},
    {"musteri": "Ali", "tarih": "2024-04-18", "urun": "Kalem", "kategori": "Kırtasiye", "tutar": 120.0},
    {"musteri": "Elif", "tarih": "2024-04-25", "urun": "Çanta", "kategori": "Aksesuar", "tutar": 400.0},
]

def siparis_df(kayitlar):
    df = pd.DataFrame(kayitlar)
    df["tarih"] = pd.to_datetime(df["tarih"])
    df["ay"] = df["tarih"].dt.to_period("M").astype(str)
    return df

df = siparis_df(KAYITLAR)
print(df[["musteri", "ay", "tutar"]].head())
print("satır:", len(df), "| müşteri:", df["musteri"].nunique())

to_period("M") tarihi aya yuvarlar; astype(str) ile "2024-01" gibi karşılaştırması kolay bir metne çevirmek iki pandas sürümünde de aynı davranır.

2. RFM segmentasyonu

RFM üç ölçüt: Recency (son siparişten bu yana geçen gün; az olması iyi), Frequency (sipariş sayısı), Monetary (toplam harcama). Her müşteri için üçünü hesaplayıp her birini 1-4 arası skora çeviririz; skorların toplamı segment belirler.

import pandas as pd

df = pd.DataFrame({
    "musteri": ["Ayşe", "Mehmet", "Ayşe", "Zeynep", "Ayşe", "Mehmet", "Ali", "Zeynep", "Ali", "Elif"],
    "tarih": pd.to_datetime(["2024-01-05", "2024-01-12", "2024-02-03", "2024-02-10", "2024-03-15",
                             "2024-03-20", "2024-03-22", "2024-04-01", "2024-04-18", "2024-04-25"]),
    "urun": ["Kalem", "Kulaklık", "Defter", "Kulaklık", "Kulaklık", "Kalem", "Çanta", "Defter", "Kalem", "Çanta"],
    "kategori": ["Kırtasiye", "Elektronik", "Kırtasiye", "Elektronik", "Elektronik", "Kırtasiye", "Aksesuar", "Kırtasiye", "Kırtasiye", "Aksesuar"],
    "tutar": [120.0, 850.0, 60.0, 850.0, 850.0, 120.0, 400.0, 60.0, 120.0, 400.0],
})

def rfm_hesapla(df, bugun):
    bugun = pd.Timestamp(bugun)
    g = df.groupby("musteri", observed=True)
    rfm = pd.DataFrame({
        "recency": (bugun - g["tarih"].max()).dt.days,
        "frequency": g["tarih"].count(),
        "monetary": g["tutar"].sum(),
    })
    return rfm.reset_index()

rfm = rfm_hesapla(df, "2024-05-01")
print(rfm)

Skorlama için quantile (çeyreklik) kullanırız: müşterileri ölçüte göre dört eşit gruba böl, en iyi gruba 4 ver. pd.qcut bunu yapar; ancak aynı değere sahip çok müşteri varsa "kenarlar tekrarlı" hatası verir. Çare, önce rank(method="first") ile her müşteriye benzersiz bir sıra vermek. Recency'de küçük değer iyi olduğu için etiketleri ters çeviririz.

import pandas as pd

df = pd.DataFrame({
    "musteri": ["Ayşe", "Mehmet", "Ayşe", "Zeynep", "Ayşe", "Mehmet", "Ali", "Zeynep", "Ali", "Elif"],
    "tarih": pd.to_datetime(["2024-01-05", "2024-01-12", "2024-02-03", "2024-02-10", "2024-03-15",
                             "2024-03-20", "2024-03-22", "2024-04-01", "2024-04-18", "2024-04-25"]),
    "urun": ["Kalem", "Kulaklık", "Defter", "Kulaklık", "Kulaklık", "Kalem", "Çanta", "Defter", "Kalem", "Çanta"],
    "kategori": ["Kırtasiye", "Elektronik", "Kırtasiye", "Elektronik", "Elektronik", "Kırtasiye", "Aksesuar", "Kırtasiye", "Kırtasiye", "Aksesuar"],
    "tutar": [120.0, 850.0, 60.0, 850.0, 850.0, 120.0, 400.0, 60.0, 120.0, 400.0],
})

def rfm_hesapla(df, bugun):
    bugun = pd.Timestamp(bugun)
    g = df.groupby("musteri", observed=True)
    rfm = pd.DataFrame({
        "recency": (bugun - g["tarih"].max()).dt.days,
        "frequency": g["tarih"].count(),
        "monetary": g["tutar"].sum(),
    })
    return rfm.reset_index()

def rfm_skor(rfm, q=4):
    rfm = rfm.copy()
    etiket = list(range(1, q + 1))
    rfm["R"] = pd.qcut(rfm["recency"].rank(method="first"), q, labels=etiket[::-1]).astype(int)
    rfm["F"] = pd.qcut(rfm["frequency"].rank(method="first"), q, labels=etiket).astype(int)
    rfm["M"] = pd.qcut(rfm["monetary"].rank(method="first"), q, labels=etiket).astype(int)
    rfm["skor"] = rfm["R"] + rfm["F"] + rfm["M"]
    return rfm

def segment(skor, q=4):
    ust = 3 * q
    if skor >= ust - 2:
        return "Şampiyon"
    if skor >= ust - 5:
        return "Sadık"
    if skor >= 6:
        return "Potansiyel"
    return "Riskli"

rfm = rfm_hesapla(df, "2024-05-01")
skorlu = rfm_skor(rfm)
skorlu["segment"] = skorlu["skor"].apply(segment)
print(skorlu[["musteri", "R", "F", "M", "skor", "segment"]])

3. Kohort analizi

Kohort (cohort), ilk siparişini aynı ayda veren müşteri grubudur. Soru: Ocak kohortundan kaç müşteri şubatta (1. ay), martta (2. ay) yine alışveriş yaptı? Hesap üç adım: her müşterinin ilk ayı, her siparişin ilk aya göre kaçıncı ayda olduğu, kohort × ay_no tablosunda benzersiz müşteri sayısı. Ay farkını sürümden bağımsız hesaplamak için yıl*12+ay tamsayısını kullanıyoruz.

import pandas as pd

df = pd.DataFrame({
    "musteri": ["Ayşe", "Mehmet", "Ayşe", "Zeynep", "Ayşe", "Mehmet", "Ali", "Zeynep", "Ali", "Elif"],
    "tarih": pd.to_datetime(["2024-01-05", "2024-01-12", "2024-02-03", "2024-02-10", "2024-03-15",
                             "2024-03-20", "2024-03-22", "2024-04-01", "2024-04-18", "2024-04-25"]),
    "urun": ["Kalem", "Kulaklık", "Defter", "Kulaklık", "Kulaklık", "Kalem", "Çanta", "Defter", "Kalem", "Çanta"],
    "kategori": ["Kırtasiye", "Elektronik", "Kırtasiye", "Elektronik", "Elektronik", "Kırtasiye", "Aksesuar", "Kırtasiye", "Kırtasiye", "Aksesuar"],
    "tutar": [120.0, 850.0, 60.0, 850.0, 850.0, 120.0, 400.0, 60.0, 120.0, 400.0],
})

def ay_index(seri):
    return seri.dt.year * 12 + seri.dt.month

def kohort_tablosu(df):
    df = df.copy()
    ilk = df.groupby("musteri", observed=True)["tarih"].min()
    df["ilk"] = df["musteri"].map(ilk)
    df["kohort"] = df["ilk"].dt.to_period("M").astype(str)
    df["ay_no"] = ay_index(df["tarih"]) - ay_index(df["ilk"])
    tablo = (df.groupby(["kohort", "ay_no"], observed=True)["musteri"]
               .nunique().unstack(fill_value=0))
    return tablo

tablo = kohort_tablosu(df)
print(tablo)
print("Ocak kohortu 2. ayda:", int(tablo.loc["2024-01", 2]))

Tabloyu okuma: satır kohort, sütun ilk aydan itibaren geçen ay. 0. sütun kohort büyüklüğüdür; sonraki sütunları ona bölersen tutundurma oranını (retention rate) elde edersin. Üretimde bu tablo ısı haritasına dönüşür.

4. En çok satan ürün ve kategori

import pandas as pd

df = pd.DataFrame({
    "musteri": ["Ayşe", "Mehmet", "Ayşe", "Zeynep", "Ayşe", "Mehmet", "Ali", "Zeynep", "Ali", "Elif"],
    "tarih": pd.to_datetime(["2024-01-05", "2024-01-12", "2024-02-03", "2024-02-10", "2024-03-15",
                             "2024-03-20", "2024-03-22", "2024-04-01", "2024-04-18", "2024-04-25"]),
    "urun": ["Kalem", "Kulaklık", "Defter", "Kulaklık", "Kulaklık", "Kalem", "Çanta", "Defter", "Kalem", "Çanta"],
    "kategori": ["Kırtasiye", "Elektronik", "Kırtasiye", "Elektronik", "Elektronik", "Kırtasiye", "Aksesuar", "Kırtasiye", "Kırtasiye", "Aksesuar"],
    "tutar": [120.0, 850.0, 60.0, 850.0, 850.0, 120.0, 400.0, 60.0, 120.0, 400.0],
})

def en_cok_satan(df, sutun="urun", n=3):
    g = df.groupby(sutun, observed=True)["tutar"].agg(["count", "sum"]).reset_index()
    g.columns = [sutun, "adet", "ciro"]
    return (g.sort_values(["ciro", sutun], ascending=[False, True])
             .head(n).reset_index(drop=True))

print(en_cok_satan(df))
print(en_cok_satan(df, "kategori", 2))

Sıralamada ikinci anahtar olarak ürün adını eklemek, eşit ciroda sonucu deterministik yapar; testler bunu sever.

5. Sahte API'den sayfalı veri çekme

Gerçek sistemde siparişler bir API'den sayfa sayfa gelir: ?sayfa=0&boyut=100. Ağ erişimimiz yok, bu yüzden API'yi asyncio.sleep ile bekleyen bir coroutine olarak simüle ediyoruz. Yanıt, veriyi ve "son sayfa mı" bilgisini taşır; istemci son sayfaya kadar döner.

import asyncio

KAYITLAR = [{"id": i, "tutar": 100 + i * 10} for i in range(10)]

async def api_sayfa(kayitlar, sayfa, boyut):
    await asyncio.sleep(0.01)            # ağ gecikmesi simülasyonu
    bas = sayfa * boyut
    return {"veri": kayitlar[bas:bas + boyut], "son": bas + boyut >= len(kayitlar)}

async def hepsini_cek(kayitlar, boyut):
    hepsi, sayfa = [], 0
    while True:
        yanit = await api_sayfa(kayitlar, sayfa, boyut)
        hepsi.extend(yanit["veri"])
        if yanit["son"]:
            break
        sayfa += 1
    return hepsi, sayfa + 1

loop = asyncio.new_event_loop()
veri, sayfa_sayisi = loop.run_until_complete(hepsini_cek(KAYITLAR, 4))
loop.close()
print(len(veri), "kayıt,", sayfa_sayisi, "sayfa")

Sayfalar birbirine bağlı (bir sonrakini çekmek için öncekinin "son" bilgisi gerekir), bu yüzden sıralı await doğru. Toplam sayfa sayısı önceden biliniyorsa gather ile paralel çekmek mümkündür; alıştırmada iki yaklaşımı da göreceksin.

6. CLI: argparse ile alt komutlar

Pazarlama ekibi kod yazmadan python -m analitik rfm --bugun 2024-05-01 diyebilmeli. parse_args fonksiyonuna liste verebilmek test için kritik: sys.argv yerine parse_args(["rfm", "--q", "3"]) ile her komutu testte deneyebilirsin.

import argparse

def parse_args(argv=None):
    p = argparse.ArgumentParser(prog="analitik", description="Anadolu Market müşteri analitiği")
    p.add_argument("komut", choices=["rfm", "kohort", "top"])
    p.add_argument("--bugun", default="2024-05-01")
    p.add_argument("--q", type=int, default=4)
    p.add_argument("--n", type=int, default=3)
    p.add_argument("--sutun", choices=["urun", "kategori"], default="urun")
    return p.parse_args(argv)

a = parse_args(["top", "--sutun", "kategori", "--n", "2"])
print(a.komut, a.sutun, a.n)
try:
    parse_args(["sil"])
except SystemExit as e:
    print("geçersiz komut, çıkış kodu:", e.code)

main(argv) fonksiyonu argümana göre ilgili fonksiyonu çağırıp sonucu yazdırır; iş mantığı CLI'dan ayrı kaldığı için testler main'e dokunmadan hesap fonksiyonlarını doğrular.

7. unittest ile doğrulama

import unittest

class TestRfm(unittest.TestCase):
    def setUp(self):
        self.df = siparis_df(KAYITLAR)

    def test_recency_ayse(self):
        rfm = rfm_hesapla(self.df, "2024-05-01").set_index("musteri")
        self.assertEqual(int(rfm.loc["Ayşe", "recency"]), 47)

    def test_skor_araligi(self):
        sk = rfm_skor(rfm_hesapla(self.df, "2024-05-01"))
        self.assertTrue(sk["skor"].between(3, 12).all())

if __name__ == "__main__":
    unittest.main(argv=["x"], exit=False)

setUp her testten önce taze veri kurar. Tarayıcıda unittest.main(argv=["x"], exit=False) kalıbı gerekir; aksi halde unittest tarayıcının argümanlarını okumaya çalışır ve çıkış yapar.

anadolu-analitik/
├── pyproject.toml          # bağımlılıklar, [project.scripts] analitik = "analitik.cli:main"
├── README.md               # ne yapar, nasıl kurulur, örnek komut, örnek çıktı
├── analitik.toml           # varsayılan ayarlar (gizli bilgi YOK)
├── .env.example            # gereken ortam değişkenlerinin listesi
├── src/analitik/
│   ├── __init__.py  __main__.py  cli.py
│   ├── ayar.py             # Ayar dataclass + yükleyici
│   ├── veri.py  dogrula.py # okuma ve doğrulama
│   ├── rfm.py  kohort.py  urun.py
│   ├── api.py              # yeniden deneme + sayfalama
│   └── rapor.py            # Markdown/CSV çıktı
├── tests/                  # pytest; her modüle bir test dosyası
└── .github/workflows/ci.yml

import tomllib
from dataclasses import dataclass, fields

@dataclass(frozen=True)
class Ayar:
    bugun: str = "2024-05-01"
    q: int = 4
    cikti: str = "rapor.md"
    sayfa_boyutu: int = 100

def ayar_yukle(toml_metin, ortam):
    degerler = {}
    tablo = tomllib.loads(toml_metin).get("analitik", {})
    gecerli = {f.name: f.type for f in fields(Ayar)}
    for k, v in tablo.items():
        if k not in gecerli:
            raise ValueError(f"bilinmeyen ayar: {k}")
        degerler[k] = v
    for k, tur in gecerli.items():             # ANALITIK_Q=5 gibi
        anahtar = "ANALITIK_" + k.upper()
        if anahtar in ortam:
            degerler[k] = int(ortam[anahtar]) if tur is int else ortam[anahtar]
    return Ayar(**degerler)

dosya = '[analitik]\nq = 5\ncikti = "aylik.md"\n'
print(ayar_yukle(dosya, {}))
print(ayar_yukle(dosya, {"ANALITIK_Q": "3", "PATH": "/usr/bin"}))

import logging, sys, random

def log_kur(seviye="INFO"):
    log = logging.getLogger("analitik")
    log.handlers.clear()
    h = logging.StreamHandler(sys.stdout)
    h.setFormatter(logging.Formatter("%(levelname)-7s %(name)s: %(message)s"))
    log.addHandler(h)
    log.setLevel(seviye)
    log.propagate = False
    return log

log = log_kur()

def yeniden_dene(fonk, deneme=4, taban=0.5, tavan=8.0, uyu=lambda s: None):
    for i in range(deneme):
        try:
            return fonk()
        except ConnectionError as e:
            if i == deneme - 1:
                log.error("vazgeçildi: %s", e)
                raise
            bekle = min(tavan, taban * 2 ** i) * random.uniform(0.8, 1.2)   # jitter
            log.warning("deneme %d başarısız (%s), %.2f sn sonra", i + 1, e, bekle)
            uyu(bekle)

random.seed(1)
cagri = {"n": 0}
def kararsiz_api():
    cagri["n"] += 1
    if cagri["n"] < 3:
        raise ConnectionError("503 Service Unavailable")
    return {"veri": [1, 2, 3], "son": True}

print(yeniden_dene(kararsiz_api))
log.debug("bu satır görünmez: seviye INFO")

import os, tempfile

def markdown_tablo(satirlar, sutunlar):
    bas = "| " + " | ".join(sutunlar) + " |"
    ayr = "|" + "|".join("---:" if isinstance(satirlar[0][s], (int, float)) else "---" for s in sutunlar) + "|"
    govde = ["| " + " | ".join(f"{r[s]:,.0f}" if isinstance(r[s], float) else str(r[s]) for s in sutunlar) + " |" for r in satirlar]
    return "\n".join([bas, ayr, *govde])

def atomik_yaz(yol, metin):
    klasor = os.path.dirname(os.path.abspath(yol))
    fd, gecici = tempfile.mkstemp(dir=klasor, suffix=".tmp")
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        f.write(metin)
    os.replace(gecici, yol)

satirlar = [{"urun": "Kulaklık", "adet": 3, "ciro": 2550.0}, {"urun": "Çanta", "adet": 2, "ciro": 800.0}]
rapor = "# Aylık satış raporu\n\n" + markdown_tablo(satirlar, ["urun", "adet", "ciro"]) + "\n"
for _ in range(2):                         # iki kez çalıştır: sonuç değişmez
    atomik_yaz("rapor.md", rapor)
print(open("rapor.md", encoding="utf-8").read())

import veri
import pandas as pd

def siparis_tablosu():
    o = veri.oku("anadolu_market", "orders", parse_dates=["order_date"])
    oi = veri.oku("anadolu_market", "order_items")
    oi["tutar"] = oi["quantity"] * oi["unit_price"] * (1 - oi["discount"])
    return o.join(oi.groupby("order_id")["tutar"].sum(), on="order_id")

s = siparis_tablosu()
print(len(s), "sipariş, toplam", round(s["tutar"].sum(), 2), "TL")
print(s.groupby("status")["tutar"].sum().round(0).to_dict())

import veri

print(veri.oku("kesit_bi", "RFMSegmentleri")[["Sira", "Segment", "Eylem"]].to_string(index=False))

import veri
import pandas as pd

def siparis_tablosu():
    o = veri.oku("anadolu_market", "orders", parse_dates=["order_date"])
    oi = veri.oku("anadolu_market", "order_items")
    oi["tutar"] = oi["quantity"] * oi["unit_price"] * (1 - oi["discount"])
    return o.join(oi.groupby("order_id")["tutar"].sum(), on="order_id")

def rfm_hesapla(s, bugun="2025-01-01"):
    d = s[s["status"] == "delivered"]
    r = d.groupby("customer_id").agg(son=("order_date", "max"), F=("order_id", "size"), M=("tutar", "sum"))
    r["R"] = (pd.Timestamp(bugun) - r["son"]).dt.days
    return r[["R", "F", "M"]]

def _puan(seri, q=5, ters=False):
    p = pd.qcut(seri.rank(method="first"), q, labels=False) + 1
    return (q + 1 - p) if ters else p

def rfm_skorla(r, q=5):
    r = r.copy()
    r["r"] = _puan(r["R"], q, ters=True)
    r["f"] = _puan(r["F"], q)
    r["m"] = _puan(r["M"], q)
    return r

def segment_ata(r, f):
    if r >= 4 and f >= 4:
        return "Şampiyon"
    if r == 3 and f >= 4:
        return "Sadık"
    if r >= 4 and f >= 2:
        return "Potansiyel sadık"
    if r >= 4:
        return "Yeni"
    if r == 3:
        return "Dikkat gerektiren"
    if f >= 3:
        return "Risk altında"
    return "Kaybedilmiş"

r = rfm_skorla(rfm_hesapla(siparis_tablosu()))
r["Segment"] = [segment_ata(a, b) for a, b in zip(r["r"], r["f"])]
ozet = r.groupby("Segment").agg(musteri=("M", "size"), ciro=("M", "sum")).reset_index()
tablo = veri.oku("kesit_bi", "RFMSegmentleri").merge(ozet, on="Segment", how="left").sort_values("Sira")
tablo["pay"] = (tablo["ciro"] / tablo["ciro"].sum() * 100).round(1)
print(tablo[["Segment", "musteri", "pay"]].to_string(index=False))

import veri

o = veri.oku("anadolu_market", "orders", parse_dates=["order_date"])
bi = veri.oku("kesit_bi", "Musteriler").set_index("MusteriNo")["IlkSiparis"].dropna()
for ad, kume in [("tüm siparişler", o), ("iptal hariç", o[o["status"] != "cancelled"])]:
    ilk = kume.groupby("customer_id")["order_date"].min().dt.strftime("%Y-%m-%d")
    print(ad, "eşleşen:", int((bi == ilk.reindex(bi.index)).sum()), "/", len(bi))