本文介绍如何基于带状态标记的房产数据，计算每日单位面积房价均值变化，并分解其驱动因素（价格调整、新增房源、成交退出）对整体变动的贡献。
本文介绍如何基于带状态标记的房产数据，计算每日单位面积房价均值变化，并分解其驱动因素（价格调整、新增房源、成交退出）对整体变动的贡献。
在房地产数据分析中，仅观察均价涨跌不足以支撑业务决策；关键在于归因——明确“为什么涨/跌”。本教程以结构化方式实现均价变动的三因子归因分析：by_price_change（存量在售房源价格调整）、by_new_premises（新增挂牌房源拉高/压低均价）、by_sale（已售房源退出导致样本结构变化）。所有计算严格遵循业务规则：仅 cur（当前在售）和 new（新挂牌）状态参与均价计算；sold（已售）与 nf_sale（非在售）被排除。
✅ 步骤一：数据预处理与有效样本筛选
首先确保日期为 datetime 类型，并按业务逻辑过滤有效记录：
import pandas as pd import numpy as np df['date'] = pd.to_datetime(df['date']) # 仅保留参与均价计算的状态：'cur' 和 'new' df_valid = df[df['status'].isin(['cur', 'new'])].copy()
✅ 步骤二：计算每日均价及基准值
按日聚合总面积与总价格，得出单位面积均价（avg），并关联前一日均价（avg_yest）用于变化率计算：
# 按日汇总
daily = df_valid.groupby('date').agg(
area=('area', 'sum'),
price=('price', 'sum')
).reset_index()
daily['avg'] = daily['price'] / daily['area']
# 关联前一日均价（使用 shift(1) 实现时间对齐）
daily = daily.sort_values('date').reset_index(drop=True)
daily['avg_yest'] = daily['avg'].shift(1)
daily['avg_yest'] = daily['avg_yest'].fillna(daily['avg'].iloc[0]) # 首日无昨日值，设为自身
daily['avg_change_ratio'] = (daily['avg'] - daily['avg_yest']) / daily['avg_yest']✅ 步骤三：三因子归因分解（核心逻辑）
归因需满足：avg_change_ratio ≈ by_price_change + by_new_premises + by_sale，且各因子有明确业务含义：
by_price_change：由 cur 状态房源价格变动引起的均价变化
→ 计算当日 cur 房源的加权平均价格变动率（以昨日同 num 的 cur 房源为基准）by_new_premises：由新增 new 房源引入的结构性影响
→ 计算 new 房源均价与昨日整体均价的差值，按其面积占比加权by_sale：由 sold 房源退出导致的样本权重再分配效应
→ 计算昨日 sold 房源（若存在）的均价与昨日整体均价的偏离，按其原面积占比反向修正
# 构建归因所需中间表：含 num, date, status, price, area, price_yest
df_full = df.copy()
df_full['date'] = pd.to_datetime(df_full['date'])
# 1. Price change effect: 仅对比连续两日均为 'cur' 的房源
cur_today = df_full[(df_full['status'] == 'cur') & (df_full['date'] == daily['date'].max())]
cur_yest = df_full[(df_full['status'] == 'cur') & (df_full['date'] == daily['date'].max() - pd.Timedelta(days=1))]
merged_cur = pd.merge(cur_today[['num', 'price']],
cur_yest[['num', 'price']],
on='num', suffixes=('_t', '_y'))
if not merged_cur.empty:
weight = merged_cur['price_y'].sum() / daily[daily['date'] == merged_cur['_y'].name]['price'].iloc[0] if 'price' in daily.columns else 1
price_effect = (merged_cur['price_t'] - merged_cur['price_y']).sum() / merged_cur['price_y'].sum()
else:
price_effect = 0.0
# 2. New premises effect: new 房源均价 vs 昨日整体均价
new_today = df_full[(df_full['status'] == 'new') & (df_full['date'] == daily['date'].max())]
if not new_today.empty:
new_avg = new_today['price'].sum() / new_today['area'].sum()
new_effect = (new_avg - daily[daily['date'] == daily['date'].max() - pd.Timedelta(days=1)]['avg'].iloc[0]) * (new_today['area'].sum() / daily[daily['date'] == daily['date'].max() - pd.Timedelta(days=1)]['area'].iloc[0])
else:
new_effect = 0.0
# 3. Sale effect: sold 房源退出导致的权重偏移（简化版：用 sold 房源均价偏离度 × 其面积占比）
sold_yest = df_full[(df_full['status'] == 'sold') & (df_full['date'] == daily['date'].max() - pd.Timedelta(days=1))]
if not sold_yest.empty:
sold_avg = sold_yest['price'].sum() / sold_yest['area'].sum()
yest_avg = daily[daily['date'] == daily['date'].max() - pd.Timedelta(days=1)]['avg'].iloc[0]
sale_effect = (yest_avg - sold_avg) * (sold_yest['area'].sum() / daily[daily['date'] == daily['date'].max() - pd.Timedelta(days=1)]['area'].iloc[0])
else:
sale_effect = 0.0
# 归一化至变化率维度（除以昨日均价）
daily['by_price_change'] = price_effect / daily['avg_yest'].iloc[-1] if len(daily) > 1 else 0.0
daily['by_new_premises'] = new_effect / daily['avg_yest'].iloc[-1] if len(daily) > 1 else 0.0
daily['by_sale'] = sale_effect / daily['avg_yest'].iloc[-1] if len(daily) > 1 else 0.0⚠️ 注意事项：
- 归因结果需满足数值可加性验证：np.isclose(daily['avg_change_ratio'], daily['by_price_change'] + daily['by_new_premises'] + daily['by_sale'], atol=1e-4)
- 若数据跨度仅两天（如示例），by_price_change 可能受限于匹配房源数，建议补充历史数据提升鲁棒性
- 实际业务中，“nf_sale → new” 转化应纳入新增逻辑，本例中 new 已隐含该行为
最终输出结构清晰、可解释性强的归因报表，直接支撑运营复盘与策略调优。
