Search⌘ K
AI Features

Common DS Coding Question Patterns

Explore common patterns in data science coding questions involving event metrics, data reshaping, statistical calculations, and simulations. Understand how to define problem grain, validate assumptions, and implement correct solutions using Python and pandas. This lesson helps you approach coding prompts methodically, ensuring accurate interpretation and reliable results.

Coding problems involving D7 retention, calculating a median without pandas, and estimating probabilities through simulation may appear unrelated. However, these problems follow recognizable problem patterns that help identify the required intermediate results, appropriate calculations, and potential sources of error.

Consider three representative prompts:

  • Compute D7 retention for each signup cohort from user_id, event_time, and event_name.

  • Convert a wide monthly sales table into region-level monthly totals.

  • Estimate the probability that at least one of ten trials succeeds when each trial has probability 0.1.

The first prompt is an event metric. We need a user-level signup table before we can count a cohort. The second is a transformation pipeline. We need to preserve the table’s grain while reshaping and aggregating. The third is a small simulation. We need one random draw per trial and one boolean result per simulated run.

A fourth common shape asks us to implement a familiar statistic from primitives. A prompt such as “write a median function without a library” tests whether we make edge-case policies explicit, especially for sorted inputs, even-length inputs, and missing values.

The fastest reliable start is to label the shape, state the output grain, and name one invariant to inspect. For the retention prompt, the first intermediate table is one row per user_id with a signup_date. For the reshaping prompt, the first check is whether the output still has one row per region and month. For the simulation prompt, the first check is whether each simulated run contains exactly ten independent Bernoulli trials.

Retention from raw events

For an event metric, staged tables make the denominator visible before aggregation hides it. Suppose an interviewer asks us to compute D7 retention for users who signed up on each calendar date.

A defensible cohort definition is narrower. A user belongs to the cohort determined by the date of that user’s signup event. An eligible user has reached the seventh day after signup within the observed event window. A retained user generated at least one qualifying event on day seven. The result has one row per cohort_date, not one row per event.

The code builds those stages from a small events table and prints the cohort summary. Check that retained_users never exceeds eligible_users, and check that a user with multiple day-seven events still contributes only one retained user.

Files
main.py
README.md
Python 3.14.0
import pandas as pd
def build_events_df() -> pd.DataFrame:
# Inline event log: one row per event.
data = [
{"user_id": 1, "event_time": "2024-01-01 09:15:00", "event_name": "signup"},
{"user_id": 1, "event_time": "2024-01-08 08:00:00", "event_name": "open_app"},
{"user_id": 1, "event_time": "2024-01-08 12:00:00", "event_name": "purchase"},
{"user_id": 2, "event_time": "2024-01-01 10:00:00", "event_name": "signup"},
{"user_id": 2, "event_time": "2024-01-05 18:00:00", "event_name": "open_app"},
{"user_id": 3, "event_time": "2024-01-02 11:30:00", "event_name": "signup"},
{"user_id": 3, "event_time": "2024-01-09 13:00:00", "event_name": "open_app"},
{"user_id": 4, "event_time": "2024-01-04 07:45:00", "event_name": "signup"},
{"user_id": 4, "event_time": "2024-01-11 09:10:00", "event_name": "open_app"},
{"user_id": 5, "event_time": "2024-01-06 16:20:00", "event_name": "signup"},
{"user_id": 5, "event_time": "2024-01-13 09:00:00", "event_name": "open_app"},
{"user_id": 6, "event_time": "2024-01-10 12:00:00", "event_name": "signup"},
{"user_id": 6, "event_time": "2024-01-17 07:00:00", "event_name": "open_app"},
# Same cohort as users 1 and 2, active on day 8 (not day 7).
{"user_id": 7, "event_time": "2024-01-01 14:00:00", "event_name": "signup"},
{"user_id": 7, "event_time": "2024-01-09 10:00:00", "event_name": "open_app"},
# Signed up too recently: day 7 (2024-01-19) is after the last observed date.
{"user_id": 8, "event_time": "2024-01-12 10:00:00", "event_name": "signup"},
{"user_id": 8, "event_time": "2024-01-14 09:00:00", "event_name": "open_app"},
]
df = pd.DataFrame(data)
df["event_time"] = pd.to_datetime(df["event_time"])
return df
def derive_signup_dates(events: pd.DataFrame) -> pd.DataFrame:
signup_events = events.loc[events["event_name"].eq("signup"), ["user_id", "event_time"]].copy()
# Cohort is determined by the user's first observed signup event.
signup_dates = (
signup_events.sort_values(["user_id", "event_time"])
.groupby("user_id", as_index=False)
.agg(signup_time=("event_time", "min"))
)
signup_dates["signup_date"] = signup_dates["signup_time"].dt.normalize()
return signup_dates[["user_id", "signup_date"]]
def compute_d7_retention(events: pd.DataFrame, signups: pd.DataFrame, as_of: pd.Timestamp) -> pd.DataFrame:
eligible_signups = signups.loc[signups["signup_date"].le(as_of - pd.Timedelta(days=7))].copy()
# Eligible denominator is unique users whose D7 has occurred within the observed window.
eligible_users_by_cohort = (
eligible_signups.groupby("signup_date", as_index=False)
.agg(eligible_users=("user_id", "nunique"))
.rename(columns={"signup_date": "cohort_date"})
)
activity = events.loc[~events["event_name"].eq("signup"), ["user_id", "event_time"]].copy()
activity["event_date"] = activity["event_time"].dt.normalize()
# Attach each event to a cohort; validate prevents accidental fan-out.
activity_with_cohort = activity.merge(eligible_signups, on="user_id", how="inner", validate="many_to_one")
activity_with_cohort["d7_date"] = activity_with_cohort["signup_date"] + pd.Timedelta(days=7)
# Exact D7 definition: activity on signup_date + 7 calendar days.
d7_users = activity_with_cohort.loc[activity_with_cohort["event_date"].eq(activity_with_cohort["d7_date"]), [
"user_id",
"signup_date",
]].drop_duplicates(subset=["user_id", "signup_date"])
retained_users_by_cohort = (
d7_users.groupby("signup_date", as_index=False)
.agg(retained_users=("user_id", "nunique"))
.rename(columns={"signup_date": "cohort_date"})
)
cohort_summary = eligible_users_by_cohort.merge(
retained_users_by_cohort,
on="cohort_date",
how="left",
validate="one_to_one",
)
cohort_summary["retained_users"] = cohort_summary["retained_users"].fillna(0).astype(int)
cohort_summary["retention_rate"] = cohort_summary["retained_users"] / cohort_summary["eligible_users"]
cohort_summary = cohort_summary.sort_values("cohort_date").reset_index(drop=True)
# Denominator safety: a retained user must also be eligible.
assert (cohort_summary["retained_users"] <= cohort_summary["eligible_users"]).all()
# Extra guard: no user should be counted retained unless they were eligible.
eligible_user_set = set(eligible_signups["user_id"].unique())
assert set(d7_users["user_id"].unique()).issubset(eligible_user_set)
return cohort_summary
def main() -> None:
events = build_events_df()
signups = derive_signup_dates(events)
# Use the max observed event date as the "as_of" boundary for eligibility.
as_of = events["event_time"].max().normalize()
cohort_summary = compute_d7_retention(events=events, signups=signups, as_of=as_of)
print(cohort_summary.to_string(index=False))
if __name__ == "__main__":
main()

In the main.py ...