使用 Pandas 计算按月续费订阅的销售额
结论
不必逐日展开合同，也不建议为每条记录调用 pd.date_range()。更合适的做法是先用向量化计算每份合同的付款月数，再展开实际发生付款的客户和月份记录。
最终可以直接得到粒度明确的明细表：
CustomerID billing_month sales
1111 2015-07-01 10
1111 2015-08-01 10
...
有了这张明细表，就可以通过 groupby() 汇总月度、季度和年度销售额。表中仍然保留 CustomerID，也方便后续建模。
下面的代码基于一个明确的前提：付款日是合同开始日的月度周年日。如果某个月没有对应日期，就使用该月最后一天。例如，合同从 1 月 31 日开始，2 月的付款日就是 2 月最后一天。如果合同恰好在付款日当天结束，该期费用仍然计入。
处理思路
先计算每份合同的 StartTime 和 EndTime 相差多少个自然月，再判断结束月份的付款日是否已经到达。
例如：
2015-07-10至2016-03-07：付款月份为 2015 年 7 月至 2016 年 2 月，共 8 次。2016-01-05至2016-01-19：只有首次付款，共 1 次。- 如果结束时间是
2016-02-05，第二次付款日已经到达，共计 2 次。
得到付款次数后，每次付款所在的月份都是从开始月份起连续排列的，因此可以直接展开月份序号，不需要逐日检查合同状态。
向量化生成客户月度明细
import numpy as np
import pandas as pd
MONTHLY_FEE = 10
WINDOW_START = pd.Timestamp("2015-01-01")
WINDOW_END = pd.Timestamp("2020-12-31")
def build_customer_month(contracts):
df = contracts.copy()
df["StartTime"] = pd.to_datetime(df["StartTime"], errors="coerce")
df["EndTime"] = pd.to_datetime(df["EndTime"], errors="coerce")
# 清理无法解析或结束时间早于开始时间的合同
valid = (
df["StartTime"].notna()
& df["EndTime"].notna()
& df["EndTime"].ge(df["StartTime"])
)
df = df.loc[valid].reset_index(drop=True)
if df.empty:
return pd.DataFrame(
{
"CustomerID": pd.Series(dtype=contracts["CustomerID"].dtype),
"billing_month": pd.Series(dtype="datetime64[ns]"),
"sales": pd.Series(dtype="int64"),
}
)
start = df["StartTime"]
end = df["EndTime"]
# StartTime 到 EndTime 的自然月距离
month_gap = (
(end.dt.year - start.dt.year) * 12
+ end.dt.month
- start.dt.month
).to_numpy(dtype=np.int64)
# 计算结束月份中的付款周年日。
# 如果该月没有对应日期，则使用月末。
anniversary_day = np.minimum(
start.dt.day.to_numpy(),
end.dt.days_in_month.to_numpy(),
)
end_month_anniversary = pd.to_datetime(
{
"year": end.dt.year.to_numpy(),
"month": end.dt.month.to_numpy(),
"day": anniversary_day,
}
)
# 首次付款计为第 1 次；结束月份的付款日已到达时再计一期
payment_count = (
month_gap
+ (end_month_anniversary <= end).to_numpy(dtype=np.int64)
)
# 将月份转换成连续整数，便于快速展开
# 例如 2015-01 -> 2015 * 12，2015-02 -> 2015 * 12 + 1
first_month = (
start.dt.year.to_numpy(dtype=np.int64) * 12
+ start.dt.month.to_numpy(dtype=np.int64)
- 1
)
last_month = first_month + payment_count - 1
window_first = WINDOW_START.year * 12 + WINDOW_START.month - 1
window_last = WINDOW_END.year * 12 + WINDOW_END.month - 1
# 只保留 2015-01 至 2020-12 范围内的付款月份
selected_first = np.maximum(first_month, window_first)
selected_last = np.minimum(last_month, window_last)
counts = np.maximum(selected_last - selected_first + 1, 0)
total_rows = int(counts.sum())
if total_rows == 0:
return pd.DataFrame(
{
"CustomerID": pd.Series(dtype=df["CustomerID"].dtype),
"billing_month": pd.Series(dtype="datetime64[ns]"),
"sales": pd.Series(dtype="int64"),
}
)
# 每个合同在结果中重复相应的付款次数
source_rows = np.repeat(np.arange(len(df)), counts)
# 生成每个合同内部的 0、1、2... 月偏移量
group_starts = np.cumsum(counts) - counts
offsets = (
np.arange(total_rows, dtype=np.int64)
- np.repeat(group_starts, counts)
)
billing_number = np.repeat(selected_first, counts) + offsets
billing_year = billing_number // 12
billing_month_number = billing_number % 12 + 1
billing_month = pd.to_datetime(
{
"year": billing_year,
"month": billing_month_number,
"day": np.ones(total_rows, dtype=np.int8),
}
)
detail = pd.DataFrame(
{
"CustomerID": df["CustomerID"].to_numpy()[source_rows],
"billing_month": billing_month,
"sales": np.full(total_rows, MONTHLY_FEE, dtype=np.int64),
}
)
# 同一客户可能有多份合同，因此按客户和月份合并
return (
detail.groupby(
["CustomerID", "billing_month"],
as_index=False,
sort=False,
)["sales"]
.sum()
)调用方式：
contracts = pd.DataFrame(
{
"CustomerID": [1111, 1112, 1113],
"StartTime": ["2015-07-10", "2016-01-05", "2015-10-18"],
"EndTime": ["2016-03-07", "2016-01-19", "2020-09-01"],
}
)
customer_month = build_customer_month(contracts)
print(customer_month.head())结果中每一行都表示某位客户在某个月产生的订阅销售额。如果同一客户同时持有多份合同，当月销售额可能是 $20、$30 等，不一定只有 $10。
计算月度、季度和年度销售额
月度总销售额
monthly_sales = (
customer_month.groupby("billing_month", as_index=False)["sales"]
.sum()
.sort_values("billing_month")
)如果还要显示销售额为 0 的月份，可以补齐完整的月份索引：
all_months = pd.date_range(
WINDOW_START,
WINDOW_END,
freq="MS",
)
monthly_sales = (
monthly_sales.set_index("billing_month")
.reindex(all_months, fill_value=0)
.rename_axis("billing_month")
.reset_index()
)季度总销售额
quarterly_sales = (
customer_month.assign(
quarter=customer_month["billing_month"].dt.to_period("Q")
)
.groupby("quarter", as_index=False)["sales"]
.sum()
.sort_values("quarter")
)quarter 的类型是 Period。如果导出系统只接受字符串，可以进行转换：
quarterly_sales["quarter"] = quarterly_sales["quarter"].astype(str)年度总销售额
annual_sales = (
customer_month.assign(
year=customer_month["billing_month"].dt.year
)
.groupby("year", as_index=False)["sales"]
.sum()
.sort_values("year")
)按客户生成季度和年度明细
customer_month 已经保存了客户的月度明细。在此基础上，可以继续汇总季度和年度客户明细：
customer_quarter = (
customer_month.assign(
quarter=customer_month["billing_month"].dt.to_period("Q")
)
.groupby(["CustomerID", "quarter"], as_index=False)["sales"]
.sum()
)
customer_year = (
customer_month.assign(
year=customer_month["billing_month"].dt.year
)
.groupby(["CustomerID", "year"], as_index=False)["sales"]
.sum()
)如果机器学习模型要求使用宽表，可以把各个月份转换为特征列：
customer_month_features = (
customer_month.pivot_table(
index="CustomerID",
columns="billing_month",
values="sales",
aggfunc="sum",
fill_value=0,
)
.sort_index(axis=1)
)客户和月份较多时，长表通常更节省内存，也更适合保存为 Parquet，再交给其他数据处理工具使用。
数百万条合同的处理方式
向量化计算付款次数通常不会成为主要瓶颈。更占内存的是展开后的客户月度明细。假设每份合同平均产生 36 次付款，100 万份合同最多可能展开为约 3600 万行。
处理这类数据时，应采用分块方式：
monthly_parts = []
for chunk in pd.read_csv(
"contracts.csv",
usecols=["CustomerID", "StartTime", "EndTime"],
chunksize=200_000,
):
part = build_customer_month(chunk)
# 如果只需要总销售额，可立即按月压缩
part_monthly = (
part.groupby("billing_month", as_index=False)["sales"]
.sum()
)
monthly_parts.append(part_monthly)
monthly_sales = (
pd.concat(monthly_parts, ignore_index=True)
.groupby("billing_month", as_index=False)["sales"]
.sum()
.sort_values("billing_month")
)如果需要保留客户级明细，不宜长期把所有分块存放在列表中。可以逐块写入 Parquet：
for chunk_number, chunk in enumerate(
pd.read_csv(
"contracts.csv",
usecols=["CustomerID", "StartTime", "EndTime"],
chunksize=200_000,
)
):
part = build_customer_month(chunk)
part.to_parquet(
f"customer_month_part_{chunk_number:05d}.parquet",
index=False,
)读取这些分块后，还要再次按 CustomerID 和 billing_month 汇总，因为同一客户的多份合同可能位于不同的输入分块中。
注意事项
明确月底付款规则
对于 1 月 29 日、1 月 30 日、1 月 31 日 等日期，“一个月后”可能有不同的业务定义。上面的代码采用以下规则：以原始开始日作为月度周年日，如果当月没有对应日期，就取月末。
如果实际系统按”以上一次付款日期为基础连续加一个月”计算，月底日期可能逐步漂移。此时不能直接使用这里的付款次数公式，需要根据真实计费系统的日期规则调整。
合同结束日是否收费
当前代码使用：
end_month_anniversary <= EndTime因此，结束日期恰好等于付款日期时，该期费用仍会收取。这符合”在下一付款日期之前结束才不付款”的描述。
如果业务规则规定结束日当天不收费，应改为严格小于：
end_month_anniversary < EndTime时间字段可能包含时分秒
如果 StartTime 和 EndTime 含有具体时间，例如付款发生在上午 10 点，而合同在当天上午 9 点结束，仅比较日期就不够。此时需要保留时间部分，并明确付款发生的具体时刻。
重复合同与重复记录
同一客户在同一个月有多份有效合同时，代码会累加每份合同的费用。如果输入数据中可能存在重复合同，应先根据合同编号或业务唯一键去重，不能只按 CustomerID 去重。
金额类型
固定的整数美元费用可以用整数保存。如果费用含有小数，不建议直接使用二进制浮点数进行财务汇总。可以先把金额换算成”分”并用整数保存，也可以使用符合业务精度要求的十进制类型。
备注：内容仅供参考。