价格变动场景下产品收入计算方法及数据集统计请求

我们使用Python的Pandas库完成数据处理，步骤如下：

1. 数据预处理 #

读取并格式化两个数据集，重点处理时间字段以确保时间比较的准确性：

import pandas as pd

# 读取样本数据（实际场景可替换为文件读取逻辑）
prices_data = [
    [64, 270000, 239000, '9/10/2018  4:37:00 PM'],
    [3954203, 60000, 64000, '9/11/2018  10:59:00 PM'],
    [3954203, 64000, 60500, '9/17/2018  11:54:00 AM'],
    [3998909, 19000, 17000, '9/10/2018  4:35:00 PM'],
    [3998909, 17000, 15500, '9/16/2018  5:09:00 AM'],
    [4085861, 67000, 62500, '9/11/2018  8:51:00 AM'],
    [4085861, 62500, 58000, '9/17/2018  3:35:00 AM']
]
sales_data = [
    [64, 1, '9/09/2018  4:37:00 PM'],
    [64, 6, '9/11/2018  10:59:00 PM'],
    [3954203, 6, '9/10/2018  11:54:00 AM'],
    [3954203, 1, '9/12/2018  4:35:00 PM'],
    [3954203, 1, '9/18/2018  5:09:00 AM'],
    [4085861, 2, '9/10/2018  8:51:00 AM'],
    [4085861, 1, '9/19/2018  3:35:00 AM']
]

# 转换为DataFrame并统一时间格式
prices = pd.DataFrame(prices_data, columns=['product_id', 'old_price', 'new_price', 'updated_at'])
sales = pd.DataFrame(sales_data, columns=['product_id', 'quantity_ordered', 'ordered_at'])

prices['updated_at'] = pd.to_datetime(prices['updated_at'])
sales['ordered_at'] = pd.to_datetime(sales['ordered_at'])

2. 生成价格生效区间 #

为每个产品补充9月初始价格区间，并明确每段价格的生效时间段：

# 按产品和时间排序价格变动记录
prices = prices.sort_values(['product_id', 'updated_at'])

# 为每个价格变动设置结束时间（下一次变动时间，或9月30日）
prices['price_end'] = prices.groupby('product_id')['updated_at'].shift(-1)
prices['price_end'] = prices['price_end'].fillna(pd.to_datetime('2018-09-30 23:59:59'))

# 生成9月初始价格区间（从9月1日到第一次价格变动前）
initial_prices = []
for product in prices['product_id'].unique():
    first_update = prices[prices['product_id'] == product]['updated_at'].min()
    initial_price = prices[prices['product_id'] == product].iloc[0]['old_price']
    initial_prices.append({
        'product_id': product,
        'price': initial_price,
        'price_start': pd.to_datetime('2018-09-01 00:00:00'),
        'price_end': first_update
    })
initial_prices_df = pd.DataFrame(initial_prices)

# 整理价格变动后的生效区间
price_ranges = prices.rename(columns={'new_price': 'price', 'updated_at': 'price_start'})[['product_id', 'price', 'price_start', 'price_end']]

# 合并初始区间与变动区间
full_price_ranges = pd.concat([initial_prices_df, price_ranges], ignore_index=True)

3. 匹配销售记录与对应价格 #

将每条销售记录匹配到对应的价格区间，获取销售时的实际价格：

# 按时间顺序合并销售数据与价格区间
sales_with_price = pd.merge_asof(
    sales.sort_values('ordered_at'),
    full_price_ranges.sort_values('price_start'),
    left_on='ordered_at',
    right_on='price_start',
    by='product_id',
    direction='backward'
)

# 筛选出销售时间在价格区间内的有效记录
sales_with_price = sales_with_price[sales_with_price['ordered_at'] < sales_with_price['price_end']]

# 计算单条销售记录的收入
sales_with_price['revenue'] = sales_with_price['quantity_ordered'] * sales_with_price['price']

4. 汇总最终结果 #

计算每个产品的总收入，以及各价格对应的收入：

# 按产品+价格维度汇总收入
price_level_revenue = sales_with_price.groupby(['product_id', 'price'])['revenue'].sum().reset_index()
price_level_revenue.rename(columns={'revenue': 'price_total_revenue'}, inplace=True)

# 按产品维度汇总总收入
product_total_revenue = sales_with_price.groupby('product_id')['revenue'].sum().reset_index()
product_total_revenue.rename(columns={'revenue': 'product_total_revenue'}, inplace=True)

# 合并两个维度的结果
final_result = pd.merge(price_level_revenue, product_total_revenue, on='product_id')

样本数据结果示例 #

运行上述代码后，最终结果如下：

内容的提问来源于stack exchange，提问作者Thảo Nguyễn