数据分析实战：从爬虫到数据挖掘，链家二手房房价预测全流程

import requests
from bs4 import BeautifulSoup
import time
import random
import pandas as pd
from fake_useragent import UserAgent

def get_house_info(url):
    ua = UserAgent()
    headers = {
        'User-Agent': ua.random,
        'Referer': 'https://bj.lianjia.com/',
    }
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        resp.encoding = 'utf-8'
        if resp.status_code != 200:
            return None
        soup = BeautifulSoup(resp.text, 'html.parser')
        # 房源卡片列表
        items = soup.select('ul.sellListContent li.LOGCLICKDATA')
        data = []
        for item in items:
            # 标题
            title = item.select_one('.title a')
            title = title.text.strip() if title else None
            # 地址（区域+板块+小区）
            address = item.select_one('.address .info')
            address = address.text.strip() if address else None
            # 关注度（关注+发布）
            follow = item.select_one('.followInfo .info')
            follow = follow.text.strip() if follow else None
            # 总价（万）
            total_price = item.select_one('.totalPrice .info')
            total_price = total_price.text.strip() if total_price else None
            # 单价（元/平）
            unit_price = item.select_one('.unitPrice .info')
            unit_price = unit_price.text.strip() if unit_price else None
            # 房型、面积、朝向、装修、楼龄等存储在 .info 中，但更可靠的是从 .houseInfo 获取
            house_info = item.select_one('.houseInfo .info')
            house_info = house_info.text.strip() if house_info else None
            
            data.append({
                'title': title,
                'address': address,
                'follow': follow,
                'total_price': total_price,
                'unit_price': unit_price,
                'house_info': house_info
            })
        return data
    except Exception as e:
        print(f"Error: {e}")
        return None

# 爬取前10页（每页30条）
base_url = 'https://bj.lianjia.com/ershoufang/chaoyang/pg{}'
all_data = []
for page in range(1, 11):
    url = base_url.format(page)
    print(f'Crawling {url}')
    page_data = get_house_info(url)
    if page_data:
        all_data.extend(page_data)
    time.sleep(random.uniform(1.5, 3.5))

df_raw = pd.DataFrame(all_data)
df_raw.to_csv('lianjia_raw.csv', index=False, encoding='utf-8-sig')
print(f'Total records: {len(df_raw)}')

import pandas as pd
import re

df = pd.read_csv('lianjia_raw.csv', encoding='utf-8-sig')

def parse_house_info(info):
    if not isinstance(info, str):
        return [None]*5
    parts = [p.strip() for p in info.split('|')]
    # 补全到5个部分
    while len(parts) < 5:
        parts.append(None)
    return parts[:5]

df[['room_type', 'area', 'orientation', 'decoration', 'building_age']] = df['house_info'].apply(
    lambda x: pd.Series(parse_house_info(x))
)

# 清理单价（去掉"元/平米"）
df['unit_price'] = df['unit_price'].str.replace('元/平米', '').str.replace(',', '').astype(float)
# 清理总价（去掉"万"）
df['total_price'] = df['total_price'].str.replace('万', '').astype(float)
# 面积提取数字（可能包含"平米"）
df['area'] = df['area'].str.replace('平米', '').astype(float)
# 关注度解析：如"2人关注 / 3天发布"，提取关注人数
df['follow_num'] = df['follow'].str.extract('(\d+)人关注').astype(float)
# 楼龄处理：如"1998年建"提取年份
df['build_year'] = df['building_age'].str.extract('(\d{4})年').astype(float)
# 户型解析：如"2室1厅" 提取室数
df['rooms'] = df['room_type'].str.extract('(\d+)室').astype(float)

# 丢弃原始冗余列
df_clean = df.drop(['title', 'address', 'follow', 'house_info', 'building_age', 'room_type'], axis=1)
# 删除缺失关键字段的记录
df_clean = df_clean.dropna(subset=['unit_price', 'area', 'total_price'])
print(df_clean.info())

import seaborn as sns
import matplotlib.pyplot as plt

# 箱线图观察
sns.boxplot(x=df_clean['unit_price'])
plt.show()

Q1 = df_clean['unit_price'].quantile(0.25)
Q3 = df_clean['unit_price'].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR
df_clean = df_clean[(df_clean['unit_price'] > lower) & (df_clean['unit_price'] < upper)]

# 面积合理范围（20-300 平米）
df_clean = df_clean[(df_clean['area'] >= 20) & (df_clean['area'] <= 300)]

# 单价分布——右偏，考虑对数变换
sns.histplot(df_clean['unit_price'], kde=True)
plt.title('Unit Price Distribution')
plt.show()

# 相关性热图
num_cols = ['unit_price', 'total_price', 'area', 'follow_num', 'build_year', 'rooms']
corr = df_clean[num_cols].corr()
sns.heatmap(corr, annot=True, cmap='coolwarm')
plt.show()

# 朝向：常见东南/西南/南北等，取前5种，其余归为'其他'
top_ori = df_clean['orientation'].value_counts().head(5).index
df_clean['orientation'] = df_clean['orientation'].apply(lambda x: x if x in top_ori else '其他')

# 装修：简装、精装、毛坯等
top_dec = df_clean['decoration'].value_counts().head(4).index
df_clean['decoration'] = df_clean['decoration'].apply(lambda x: x if x in top_dec else '其他')

df_encoded = pd.get_dummies(df_clean, columns=['orientation', 'decoration'], drop_first=True)

df_encoded['house_age'] = 2026 - df_encoded['build_year']
df_encoded['avg_area_per_room'] = df_encoded['area'] / df_encoded['rooms']
# 对数单价（如果模型对偏态敏感）
df_encoded['log_unit_price'] = np.log1p(df_encoded['unit_price'])  # 用于后续

from sklearn.ensemble import RandomForestRegressor
X = df_encoded.drop(['unit_price', 'log_unit_price'], axis=1)
y = df_encoded['unit_price']
# 处理缺失（若有）
X = X.fillna(X.median())
model = RandomForestRegressor(n_estimators=100, random_state=42)
model.fit(X, y)
importances = pd.Series(model.feature_importances_, index=X.columns).sort_values(ascending=False)
print(importances.head(10))

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.linear_model import Ridge, Lasso
from sklearn.ensemble import RandomForestRegressor
import xgboost as xgb
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score

# 特征矩阵（选择数值+编码后的类别）
features = ['area', 'total_price', 'follow_num', 'build_year', 'rooms', 
            'house_age', 'avg_area_per_room'] + [c for c in df_encoded.columns if c.startswith('orientation_') or c.startswith('decoration_')]
X = df_encoded[features]
y = df_encoded['unit_price']

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 数值列（不包括one-hot）
num_cols = ['area', 'total_price', 'follow_num', 'build_year', 'rooms', 'house_age', 'avg_area_per_room']
cat_cols = [c for c in X.columns if c not in num_cols]

preprocessor = ColumnTransformer([
    ('scaler', StandardScaler(), num_cols),
    ('pass', 'passthrough', cat_cols)  # one-hot已是0/1，无需缩放
])

# 定义模型字典
models = {
    'Ridge': Ridge(alpha=1.0),
    'Lasso': Lasso(alpha=0.01),
    'RandomForest': RandomForestRegressor(n_estimators=200, max_depth=10, random_state=42),
    'XGBoost': xgb.XGBRegressor(n_estimators=200, max_depth=6, learning_rate=0.1, random_state=42)
}

results = {}
for name, model in models.items():
    pipe = Pipeline([
        ('prep', preprocessor),
        ('reg', model)
    ])
    pipe.fit(X_train, y_train)
    y_pred = pipe.predict(X_test)
    mae = mean_absolute_error(y_test, y_pred)
    rmse = mean_squared_error(y_test, y_pred, squared=False)
    r2 = r2_score(y_test, y_pred)
    results[name] = {'MAE': mae, 'RMSE': rmse, 'R2': r2}
    print(f"{name}: MAE={mae:.2f}, RMSE={rmse:.2f}, R2={r2:.4f}")

from sklearn.model_selection import GridSearchCV

param_grid = {
    'reg__max_depth': [4, 6, 8],
    'reg__learning_rate': [0.05, 0.1, 0.2],
    'reg__n_estimators': [150, 200]
}
xgb_pipe = Pipeline([('prep', preprocessor), ('reg', xgb.XGBRegressor(random_state=42))])
grid = GridSearchCV(xgb_pipe, param_grid, cv=5, scoring='neg_mean_absolute_error', n_jobs=-1)
grid.fit(X_train, y_train)
print("Best params:", grid.best_params_)
best_pred = grid.best_estimator_.predict(X_test)
print("Optimized MAE:", mean_absolute_error(y_test, best_pred))

best_model = grid.best_estimator_.named_steps['reg']
# 获取特征名（经过预处理后）
feature_names = num_cols + cat_cols
importance = best_model.feature_importances_
imp_df = pd.DataFrame({'feature': feature_names, 'importance': importance}).sort_values('importance', ascending=False)
plt.figure(figsize=(10,6))
sns.barplot(x='importance', y='feature', data=imp_df.head(10))
plt.title('Top 10 Feature Importance (XGBoost)')
plt.show()

import joblib
joblib.dump(grid.best_estimator_, 'house_price_model.pkl')

# Flask 示例
from flask import Flask, request, jsonify
app = Flask(__name__)
model = joblib.load('house_price_model.pkl')

@app.route('/predict', methods=['POST'])
def predict():
    data = request.get_json()
    # 假设输入与X列顺序一致
    input_df = pd.DataFrame([data['features']], columns=features)
    pred = model.predict(input_df)[0]
    return jsonify({'predicted_unit_price': round(pred, 2)})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)