SQL ও pandas দিয়ে ML-এর উপযোগী ডেটা তৈরি

          ফিচার: কাটঅফের আগের ইতিহাস                  টার্গেট: এরপর কী ঘটে
  ───────────────────────────────────────────────────|─────────────────────────|──▶ সময়
                                                  কাটঅফ                  কাটঅফ + ৯০ দিন
                                                (2026-04-01)              (2026-06-30)
  যে ফিচার কাটঅফের ডান দিকের কিছু পড়ে, সেটাই লিক (LEAK)।

WITH order_totals AS (
    SELECT o.order_id, o.customer_id, o.order_date, o.status,
           SUM(oi.quantity * oi.unit_price) AS total
    FROM orders AS o
    JOIN order_items AS oi ON oi.order_id = o.order_id
    GROUP BY o.order_id, o.customer_id, o.order_date, o.status
)
SELECT c.customer_id,
       c.city,
       CAST(julianday('2026-04-01') - julianday(c.joined_on) AS INTEGER) AS tenure_days,
       COUNT(t.order_id)          AS frequency,
       COALESCE(SUM(t.total), 0)  AS monetary,
       CAST(julianday('2026-04-01') - julianday(MAX(t.order_date)) AS INTEGER) AS recency_days,
       CASE WHEN EXISTS (
                SELECT 1 FROM orders AS f
                WHERE f.customer_id = c.customer_id
                  AND f.status <> 'cancelled'
                  AND f.order_date >= '2026-04-01' AND f.order_date < '2026-07-01'
            ) THEN 0 ELSE 1 END AS churned
FROM customers AS c
LEFT JOIN order_totals AS t
       ON t.customer_id = c.customer_id
      AND t.order_date < '2026-04-01'          -- পয়েন্ট-ইন-টাইম ফিল্টার
      AND t.status <> 'cancelled'
WHERE c.joined_on < '2026-04-01'               -- কাটঅফের দিনের পপুলেশন
GROUP BY c.customer_id, c.city, c.joined_on
ORDER BY c.customer_id;

+-------------+------------+-------------+-----------+----------+--------------+---------+
| customer_id | city       | tenure_days | frequency | monetary | recency_days | churned |
+-------------+------------+-------------+-----------+----------+--------------+---------+
|           1 | Dhaka      |         149 |         2 |     5100 |           57 |       0 |
|           2 | Chattogram |         108 |         2 |    19500 |           24 |       0 |
|           3 | Dhaka      |          82 |         2 |     6400 |            4 |       0 |
|           4 | Sylhet     |          69 |         0 |        0 | NULL         |       1 |
|           5 | NULL       |          55 |         1 |     4000 |           17 |       1 |
|           6 | Khulna     |          42 |         0 |        0 | NULL         |       0 |
|           7 | Dhaka      |           2 |         0 |        0 | NULL         |       0 |
+-------------+------------+-------------+-----------+----------+--------------+---------+

import numpy as np
import pandas as pd
from sqlhelp import con, show

rng = np.random.default_rng(42)          # একটাই seed: প্রতিবার একই ডেটাসেট
N = 2000
END = pd.Timestamp("2026-06-30")         # ডেটার শেষ দিন

signup = pd.Timestamp("2025-01-01") + pd.to_timedelta(rng.integers(0, 450, N), unit="D")
plan = rng.choice(["free", "basic", "pro"], N, p=[0.5, 0.35, 0.15])
city = rng.choice(["Dhaka", "Chattogram", "Sylhet", "Khulna", "Rajshahi", None], N,
                  p=[0.42, 0.2, 0.12, 0.11, 0.1, 0.05])
age = rng.integers(18, 61, N).astype(float)
age[rng.random(N) < 0.07] = np.nan                       # কখনো পূরণ করা হয়নি
typo = rng.random(N) < 0.015
age[typo] = rng.choice([0, 7, 230], typo.sum())          # ডেটা এন্ট্রির ভুল

rate = rng.gamma(2.0, 1.0, N) * np.select([plan == "pro", plan == "basic"], [1.6, 1.2], 1.0)
life = rng.exponential(np.select([plan == "pro", plan == "basic"], [500, 360], 220))
churn_at = signup + pd.to_timedelta(life.round(), unit="D")

customers, orders, tickets = [], [], []
oid = tid = 0
for i in range(N):
    stop = min(churn_at[i], END)
    days = (stop - signup[i]).days
    closed = None
    if churn_at[i] <= END and rng.random() < 0.6:          # চলে যাওয়াদের কেউ কেউ অ্যাকাউন্ট বন্ধ করেন
        c = churn_at[i] + pd.Timedelta(days=int(rng.integers(0, 45)))
        closed = c.date().isoformat() if c <= END else None
    customers.append((i + 1, signup[i].date().isoformat(), plan[i], city[i],
                      None if np.isnan(age[i]) else int(age[i]), closed))
    if days <= 0:
        continue
    n = rng.poisson(rate[i] * days / 30)
    for d in np.sort(rng.integers(0, days, n)):
        left = days - d if churn_at[i] <= END else 999
        if rng.random() > min(1.0, left / 75):          # চলে যাওয়ার আগে আগ্রহ কমে আসে
            continue
        oid += 1
        amount = round(float(rng.lognormal(6.6 + 0.3 * (plan[i] == "pro"), 0.6)))
        orders.append((oid, i + 1, (signup[i] + pd.Timedelta(days=int(d))).date().isoformat(), amount))
    m = rng.poisson(0.15 * days / 30 + (2.0 if churn_at[i] <= END else 0))
    for d in rng.integers(max(0, days - 60) if churn_at[i] <= END else 0, days, m):
        tid += 1                                          # চলে যাওয়ার আগে অভিযোগ বাড়ে
        tickets.append((tid, i + 1, (signup[i] + pd.Timedelta(days=int(d))).date().isoformat()))

con.executescript("""
DROP TABLE IF EXISTS sim_tickets;
DROP TABLE IF EXISTS sim_orders;
DROP TABLE IF EXISTS sim_customers;
CREATE TABLE sim_customers (customer_id INTEGER PRIMARY KEY, signup_date TEXT NOT NULL,
    plan TEXT NOT NULL, city TEXT, age INTEGER, closed_on TEXT);
CREATE TABLE sim_orders (order_id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL,
    order_date TEXT NOT NULL, amount INTEGER NOT NULL,
    FOREIGN KEY (customer_id) REFERENCES sim_customers (customer_id));
CREATE TABLE sim_tickets (ticket_id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL,
    opened_on TEXT NOT NULL,
    FOREIGN KEY (customer_id) REFERENCES sim_customers (customer_id));
""")
with con:
    con.executemany("INSERT INTO sim_customers VALUES (?, ?, ?, ?, ?, ?)", customers)
    con.executemany("INSERT INTO sim_orders VALUES (?, ?, ?, ?)", orders)
    con.executemany("INSERT INTO sim_tickets VALUES (?, ?, ?)", tickets)

show("""SELECT (SELECT COUNT(*) FROM sim_customers) AS customers,
               (SELECT COUNT(*) FROM sim_orders) AS orders,
               (SELECT COUNT(*) FROM sim_tickets) AS tickets,
               (SELECT MAX(order_date) FROM sim_orders) AS last_order""")

+-----------+--------+---------+------------+
| customers | orders | tickets | last_order |
+-----------+--------+---------+------------+
|      2000 |  26281 |    4331 | 2026-06-29 |
+-----------+--------+---------+------------+

FEATURES_SQL = """
WITH active AS (
    SELECT DISTINCT customer_id
    FROM sim_orders
    WHERE order_date >= date(:cutoff, '-90 days') AND order_date < :cutoff
),
history AS (                                   -- sales: everything BEFORE the cutoff
    SELECT customer_id,
           COUNT(*) AS orders_all,
           SUM(CASE WHEN order_date >= date(:cutoff, '-30 days') THEN 1 ELSE 0 END) AS orders_30d,
           SUM(CASE WHEN order_date >= date(:cutoff, '-90 days') THEN 1 ELSE 0 END) AS orders_90d,
           SUM(amount) AS spend_all,
           CAST(julianday(:cutoff) - julianday(MAX(order_date)) AS INTEGER) AS recency_days
    FROM sim_orders
    WHERE order_date < :cutoff
    GROUP BY customer_id
),
support AS (                                   -- support: tickets in the last 90 days
    SELECT customer_id, COUNT(*) AS tickets_90d
    FROM sim_tickets
    WHERE opened_on >= date(:cutoff, '-90 days') AND opened_on < :cutoff
    GROUP BY customer_id
),
future AS (                                    -- the label window: AFTER the cutoff
    SELECT DISTINCT customer_id
    FROM sim_orders
    WHERE order_date >= :cutoff AND order_date < date(:cutoff, '+90 days')
)
SELECT :cutoff AS cutoff, c.customer_id, c.plan, c.city, c.age,
       CAST(julianday(:cutoff) - julianday(c.signup_date) AS INTEGER) AS tenure_days,
       h.orders_all, h.orders_30d, h.orders_90d, h.spend_all, h.recency_days,
       COALESCE(s.tickets_90d, 0) AS tickets_90d,
       CASE WHEN f.customer_id IS NULL THEN 1 ELSE 0 END AS churned
FROM sim_customers AS c
JOIN active AS a ON a.customer_id = c.customer_id
JOIN history AS h ON h.customer_id = c.customer_id
LEFT JOIN support AS s ON s.customer_id = c.customer_id
LEFT JOIN future AS f ON f.customer_id = c.customer_id
WHERE c.closed_on IS NULL OR c.closed_on >= :cutoff
ORDER BY c.customer_id
"""


def snapshot(cutoff):
    return pd.read_sql(FEATURES_SQL, con, params={"cutoff": cutoff})


snaps = {cut: snapshot(cut) for cut in ["2025-10-01", "2026-01-01", "2026-04-01"]}
for cut, df in snaps.items():
    print(cut, f"rows={len(df)}", f"churn rate={df['churned'].mean():.1%}")
print(snaps["2026-04-01"].drop(columns="cutoff").head(3).to_string(index=False))

2025-10-01 rows=744 churn rate=21.2%
2026-01-01 rows=898 churn rate=20.9%
2026-04-01 rows=992 churn rate=22.3%
 customer_id plan       city  age  tenure_days  orders_all  orders_30d  orders_90d  spend_all  recency_days  tickets_90d  churned
           2 free Chattogram 27.0          107           4           0           2       2909            51            2        1
           3 free      Dhaka 40.0          161           8           1           5       5718            24            0        0
           6 free     Khulna 50.0           69           1           0           1        389            59            1        1

def prepare(df):
    df = df.copy()
    df["age"] = df["age"].where(df["age"].between(13, 100))     # 0, 7, 230 -> NaN
    df["city"] = df["city"].fillna("unknown")                    # না থাকাটাও একটা তথ্য
    df["log_spend"] = np.log1p(df["spend_all"])                  # লম্বা লেজটা সামলানো
    df["trend"] = df["orders_30d"] / (df["orders_90d"] / 3)      # শেষ মাস বনাম ৩ মাসের গড়
    return df


raw = snaps["2026-04-01"]
print("invalid ages:", int((~raw["age"].between(13, 100) & raw["age"].notna()).sum()))
print("missing ages:", int(raw["age"].isna().sum()), "| missing cities:", int(raw["city"].isna().sum()))
print("spend_all: median", raw["spend_all"].median(), "max", raw["spend_all"].max())
print(prepare(raw)[["log_spend", "trend"]].describe().round(2).loc[["mean", "min", "max"]].to_string())

invalid ages: 18
missing ages: 73 | missing cities: 46
spend_all: median 9232.0 max 162510
      log_spend  trend
mean       9.01   1.08
min        5.19   0.00
max       12.00   3.00

from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import precision_score, recall_score, roc_auc_score
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

NUMERIC = ["age", "tenure_days", "orders_all", "orders_90d", "log_spend",
           "recency_days", "tickets_90d", "trend"]
CATEGORICAL = ["plan", "city"]


def make_model(numeric=NUMERIC):
    features = ColumnTransformer([
        ("num", make_pipeline(SimpleImputer(strategy="median"), StandardScaler()), numeric),
        ("cat", OneHotEncoder(handle_unknown="ignore"), CATEGORICAL),
    ])
    return make_pipeline(features, LogisticRegression(class_weight="balanced", max_iter=1000))


def evaluate(train, test, numeric=NUMERIC):
    model = make_model(numeric).fit(train[numeric + CATEGORICAL], train["churned"])
    proba = model.predict_proba(test[numeric + CATEGORICAL])[:, 1]
    return model, roc_auc_score(test["churned"], proba), (proba >= 0.5).astype(int)


train = prepare(pd.concat([snaps["2025-10-01"], snaps["2026-01-01"]], ignore_index=True))
test = prepare(snaps["2026-04-01"])
model, auc, pred = evaluate(train, test)

print(f"train rows {len(train)}, test rows {len(test)}")
print(f"always 'stays' accuracy: {1 - test['churned'].mean():.3f}")
print(f"model accuracy:          {(pred == test['churned']).mean():.3f}")
print(f"ROC AUC {auc:.3f} | precision {precision_score(test['churned'], pred):.2f}"
      f" | recall {recall_score(test['churned'], pred):.2f}")
weights = pd.Series(model[-1].coef_[0], index=model[0].get_feature_names_out())
print(weights.sort_values().round(2).iloc[[0, 1, -2, -1]].to_string())

train rows 1642, test rows 992
always 'stays' accuracy: 0.777
model accuracy:          0.806
ROC AUC 0.861 | precision 0.55 | recall 0.72
cat__plan_pro       -0.52
num__log_spend      -0.51
num__recency_days    0.63
num__tickets_90d     1.62

LEAKY_SQL = FEATURES_SQL.replace("    WHERE order_date < :cutoff\n", "")   # কাটঅফ দিতে ভুলে গেছি
leaky = {cut: pd.read_sql(LEAKY_SQL, con, params={"cutoff": cut}) for cut in snaps}
leaky_train = prepare(pd.concat([leaky["2025-10-01"], leaky["2026-01-01"]], ignore_index=True))
leaky_test = prepare(leaky["2026-04-01"])
print("leak 1 recency_days min:", leaky_test["recency_days"].min())
print(f"leak 1 ROC AUC: {evaluate(leaky_train, leaky_test)[1]:.3f}")

closed = pd.read_sql("SELECT customer_id, closed_on IS NOT NULL AS account_closed FROM sim_customers", con)
train2, test2 = train.merge(closed, on="customer_id"), test.merge(closed, on="customer_id")
print(f"leak 2 ROC AUC: {evaluate(train2, test2, NUMERIC + ['account_closed'])[1]:.3f}")
print(f"honest ROC AUC: {auc:.3f}")

leak 1 recency_days min: -89
leak 1 ROC AUC: 0.979
leak 2 ROC AUC: 0.880
honest ROC AUC: 0.861

import hashlib
import json


def check(df, cutoff):
    assert not df.duplicated(["customer_id", "cutoff"]).any(), "one row per customer and cutoff"
    assert set(df["churned"].unique()) <= {0, 1}, "target must be 0/1"
    assert (df["recency_days"] >= 0).all(), "a feature looked past the cutoff"
    assert 0.05 < df["churned"].mean() < 0.5, "churn rate outside the expected range"
    assert df["cutoff"].eq(cutoff).all()


manifest = []
for cut, df in snaps.items():
    check(df, cut)
    df.to_parquet(f"churn_features_{cut}.parquet", index=False)
    digest = hashlib.sha256(df.to_csv(index=False).encode()).hexdigest()[:16]
    manifest.append({"file": f"churn_features_{cut}.parquet", "rows": len(df), "sha256_16": digest})

with open("manifest.json", "w") as f:
    json.dump({"seed": 42, "query": "FEATURES_SQL v1", "snapshots": manifest}, f, indent=2)
print(json.dumps(manifest[-1]))

try:
    check(leaky["2026-04-01"], "2026-04-01")
except AssertionError as e:
    print("leaky snapshot rejected:", e)

{"file": "churn_features_2026-04-01.parquet", "rows": 992, "sha256_16": "ea1a0912c25b37aa"}
leaky snapshot rejected: a feature looked past the cutoff

-- ১. গ্রাহক 7 আর 8 যোগ দিয়েছেন 2026-03-01-এর পরে, তাই তাঁরা পপুলেশনে নেই।
WITH order_totals AS (
    SELECT o.order_id, o.customer_id, o.order_date, o.status,
           SUM(oi.quantity * oi.unit_price) AS total
    FROM orders AS o
    JOIN order_items AS oi ON oi.order_id = o.order_id
    GROUP BY o.order_id, o.customer_id, o.order_date, o.status
)
SELECT c.customer_id,
       COUNT(t.order_id)         AS frequency,
       COALESCE(SUM(t.total), 0) AS monetary,
       CAST(julianday('2026-03-01') - julianday(MAX(t.order_date)) AS INTEGER) AS recency_days,
       CASE WHEN EXISTS (
                SELECT 1 FROM orders AS f
                WHERE f.customer_id = c.customer_id
                  AND f.status <> 'cancelled'
                  AND f.order_date >= '2026-03-01' AND f.order_date < '2026-06-01'
            ) THEN 0 ELSE 1 END AS churned
FROM customers AS c
LEFT JOIN order_totals AS t
       ON t.customer_id = c.customer_id
      AND t.order_date < '2026-03-01'
      AND t.status <> 'cancelled'
WHERE c.joined_on < '2026-03-01'
GROUP BY c.customer_id
ORDER BY c.customer_id;

# ২. SQL-এ একটা কলাম বেশি, ফিচার লিস্টে একটা নাম বেশি
AOV_SQL = FEATURES_SQL.replace(
    "h.spend_all, h.recency_days,",
    "h.spend_all, h.recency_days, ROUND(1.0 * h.spend_all / h.orders_all, 2) AS avg_order_value,")
aov = {cut: pd.read_sql(AOV_SQL, con, params={"cutoff": cut}) for cut in snaps}
aov_train = prepare(pd.concat([aov["2025-10-01"], aov["2026-01-01"]], ignore_index=True))
aov_test = prepare(aov["2026-04-01"])
print(f"ROC AUC with avg_order_value: {evaluate(aov_train, aov_test, NUMERIC + ['avg_order_value'])[1]:.3f}")


# ৩. প্রতিটা ট্রেনিং জবের আগে চালানোর মতো একটা পয়েন্ট-ইন-টাইম টেস্ট
def assert_point_in_time(cutoff):
    df = snapshot(cutoff)
    truth = pd.read_sql(
        "SELECT customer_id, COUNT(*) AS n FROM sim_orders WHERE order_date < :cutoff GROUP BY customer_id",
        con, params={"cutoff": cutoff})
    merged = df.merge(truth, on="customer_id")
    assert len(merged) == len(df), "every customer in the snapshot has history"
    assert (merged["orders_all"] == merged["n"]).all(), "orders_all counted orders after the cutoff"
    assert (df["recency_days"] >= 0).all(), "a last order after the cutoff"
    print(cutoff, "is point-in-time correct")


assert_point_in_time("2026-04-01")