一键导入
cohort-evolution
Lens 3 -- Track a single cohort's behavior over time (activity, frequency, value decay)
用 Codex 或 Claude 帮你安装 复制这段 Prompt,粘贴到 Codex、Claude 或其他助手里,让它检查 Skill 页面并帮你完成安装。
菜单
Lens 3 -- Track a single cohort's behavior over time (activity, frequency, value decay)
用 Codex 或 Claude 帮你安装 复制这段 Prompt,粘贴到 Codex、Claude 或其他助手里,让它检查 Skill 页面并帮你完成安装。
基于 SOC 职业分类
Turn completed audit outputs into an executive-ready document (Word/PDF) organized by insight, not by lens -- Pyramid Principle, SCQA, action titles, embedded exhibits
Complete customer-base audit orchestrator -- runs all lenses with parallel sub-agents and review
Lens 4 -- Compare two acquisition cohorts side-by-side using left-aligned analysis
Lens 5 -- Assess overall customer base health via C3 chart, acquisition flow, and repeat rates
Lens 1 -- Analyze customer heterogeneity via profit decomposition, distributions, and deciles
Load, validate, and aggregate transaction data for customer-base audit
| name | cohort-evolution |
| description | Lens 3 -- Track a single cohort's behavior over time (activity, frequency, value decay) |
Lens 3 of the customer-base audit. Track how a single acquisition cohort evolves over multiple periods. Focuses on retention/activity decay, buying patterns, and time-to-nth-purchase.
Orders DataFrame with columns: customer_id, date, period, spend, profit (optional), cohort.
import polars as pl
cohort_label = "TARGET_COHORT" # e.g., "2020-Q1"
cohort_data = orders.filter(pl.col("cohort") == cohort_label)
cohort_size = cohort_data["customer_id"].n_unique()
activity = cohort_data.group_by("period").agg(
pl.col("customer_id").n_unique().alias("n_active"),
pl.col("spend").sum().alias("total_revenue"),
pl.col("spend").mean().alias("avg_spend"),
pl.len().alias("n_transactions"),
).sort("period").with_columns(
(pl.col("n_active") / cohort_size).alias("pct_active"),
(pl.col("n_transactions") / pl.col("n_active")).alias("aof"),
(pl.col("total_revenue") / pl.col("n_transactions")).alias("aov"),
)
# Plotly line chart
import plotly.graph_objects as go
fig = go.Figure(go.Scatter(
x=activity["period"].to_list(), y=activity["pct_active"].to_list(),
mode="lines+markers", name="% Active"
))
fig.update_layout(template="plotly_white", title=f"Cohort {cohort_label} Activity Decay",
xaxis_title="Period", yaxis_title="% Active", yaxis=dict(tickformat=".0%"))
For cohorts observed over multiple years, create binary purchase patterns:
# Create customer x period matrix of Y/N
cust_periods = cohort_data.group_by("customer_id", "period").agg(
pl.len().alias("n_orders")
)
periods = sorted(cust_periods["period"].unique().to_list())
# Pivot to wide format
wide = cust_periods.pivot(on="period", index="customer_id", values="n_orders").fill_null(0)
# Convert to binary patterns (Y/N strings)
for p in periods:
if p in wide.columns:
wide = wide.with_columns(
pl.when(pl.col(p) > 0).then(pl.lit("Y")).otherwise(pl.lit("N")).alias(p)
)
# Count pattern frequencies
pattern_col = pl.concat_str([pl.col(p) for p in periods if p in wide.columns], separator="")
patterns = wide.with_columns(pattern_col.alias("pattern")).group_by("pattern").agg(
pl.len().alias("count")
).sort("count", descending=True).with_columns(
(pl.col("count") / pl.col("count").sum()).alias("pct")
)
# Rank each customer's orders chronologically
ranked = cohort_data.sort("customer_id", "date").with_columns(
pl.col("date").rank("ordinal").over("customer_id").alias("purchase_num")
)
first = ranked.filter(pl.col("purchase_num") == 1).select("customer_id", pl.col("date").alias("first_date"))
for n in [2, 3, 5]:
nth = ranked.filter(pl.col("purchase_num") == n).select("customer_id", pl.col("date").alias("nth_date"))
time_to_nth = first.join(nth, on="customer_id").with_columns(
(pl.col("nth_date") - pl.col("first_date")).dt.total_days().alias("days_to_nth")
)
median_days = time_to_nth["days_to_nth"].median()
pct_reached = len(time_to_nth) / cohort_size
print(f"Purchase #{n}: {pct_reached:.1%} reached, median {median_days:.0f} days")
second = ranked.filter(pl.col("purchase_num") == 2).select("customer_id", pl.col("date").alias("second_date"))
days_to_second = first.join(second, on="customer_id").with_columns(
(pl.col("second_date") - pl.col("first_date")).dt.total_days().alias("days")
)
# Build CDF
max_days = int(days_to_second["days"].max())
cdf = []
for d in range(0, max_days + 1, max(1, max_days // 50)):
pct = days_to_second.filter(pl.col("days") <= d).height / cohort_size
cdf.append({"days": d, "pct_reached_2nd": pct})
cdf_df = pl.DataFrame(cdf)
${CLAUDE_PLUGIN_ROOT}/references/methodology.md -- Cohort analysis framework${CLAUDE_PLUGIN_ROOT}/references/expected_patterns.md -- Typical decay curves${CLAUDE_PLUGIN_ROOT}/references/common_pitfalls.md -- Pitfall #8: ignoring one-and-done buyers