用Python构建智能睡眠分析系统：从小米手环数据到可视化看板

睡眠质量直接影响我们的工作效率和生活幸福感。作为数据从业者，我发现自己长期依赖小米手环记录睡眠，却从未深入挖掘这些数据的价值。本文将分享如何用Python构建一个完整的睡眠数据分析系统，从数据提取到自动化可视化，帮助你真正理解自己的睡眠模式。

1. 数据获取与预处理

1.1 小米运动数据导出方案优化

原始方法需要手动备份和转换数据格式，这个过程可以进一步简化。我推荐使用 adb 工具直接从Android设备提取数据库文件：

adb backup -f sleepdata.ab -noapk com.xiaomi.hm.health

这个命令会生成一个 .ab 备份文件，可以用Python的 android_backup 库直接解析：

from android_backup import AndroidBackup

ab = AndroidBackup("sleepdata.ab")
ab.extract_all("extracted_data")

提取后的数据库文件通常位于 apps/com.xiaomi.hm.health/db/ 目录下。相比手动操作，这种方法更适合自动化流程。

1.2 数据库结构解析与高效查询

小米运动的睡眠数据主要存储在 DATE_DATA 表的 SUMMARY 字段中，这个JSON格式的字符串包含丰富的信息。我们可以优化数据提取过程：

import sqlite3
import json
from datetime import datetime

def parse_sleep_data(db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 只查询包含睡眠数据的记录
    cursor.execute("""
        SELECT date, summary 
        FROM DATE_DATA 
        WHERE summary LIKE '%"slp"%'
        ORDER BY date
    """)
    
    sleep_records = []
    for date, summary in cursor.fetchall():
        try:
            data = json.loads(summary)
            slp = data.get("slp", {})
            if not slp:
                continue
                
            record = {
                "date": datetime.fromtimestamp(int(date)//1000).date(),
                "start_time": datetime.fromtimestamp(slp["st"]),
                "end_time": datetime.fromtimestamp(slp["ed"]),
                "deep_sleep": slp.get("dp", 0),
                "light_sleep": slp.get("lt", 0),
                "awake": slp.get("wk", 0),
                "rem": slp.get("dt", 0)
            }
            sleep_records.append(record)
        except json.JSONDecodeError:
            continue
            
    return pd.DataFrame(sleep_records)

这种方法直接过滤无效记录，并使用更健壮的JSON解析方式，避免了原始方案中 ast.literal_eval 的安全隐患。

2. 数据分析与特征工程

2.1 睡眠质量指标计算

原始数据只提供了基础的时间指标，我们可以计算更多有意义的衍生特征：

def calculate_sleep_metrics(df):
    # 计算总睡眠时间(分钟)
    df['total_sleep'] = df['deep_sleep'] + df['light_sleep'] + df['rem']
    
    # 计算各阶段占比
    df['deep_ratio'] = df['deep_sleep'] / df['total_sleep']
    df['light_ratio'] = df['light_sleep'] / df['total_sleep']
    df['rem_ratio'] = df['rem'] / df['total_sleep']
    
    # 计算睡眠效率(实际睡眠时间/床上时间)
    df['sleep_efficiency'] = df['total_sleep'] / (
        (df['end_time'] - df['start_time']).dt.total_seconds() / 60
    )
    
    # 计算入睡时间和起床时间(小时)
    df['bedtime'] = df['start_time'].dt.hour + df['start_time'].dt.minute/60
    df['wakeup_time'] = df['end_time'].dt.hour + df['end_time'].dt.minute/60
    
    # 添加星期几信息
    df['weekday'] = df['start_time'].dt.weekday
    
    return df

这些指标可以帮助我们更全面地评估睡眠质量，而不仅仅是看睡眠时长。

2.2 时间序列特征提取

对于长期睡眠数据分析，时间序列特征尤为重要：

from scipy.stats import zscore

def add_temporal_features(df):
    # 7天移动平均
    df['7d_avg_sleep'] = df['total_sleep'].rolling(7).mean()
    
    # 与前一天的差异
    df['sleep_diff'] = df['total_sleep'].diff()
    
    # 标准化睡眠时长
    df['sleep_zscore'] = zscore(df['total_sleep'])
    
    # 周末标志
    df['is_weekend'] = df['weekday'].isin([5, 6]).astype(int)
    
    return df

3. 可视化分析系统构建

3.1 交互式睡眠趋势面板

使用 plotly 可以创建比 matplotlib 更丰富的交互式可视化：

import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def create_sleep_dashboard(df):
    fig = make_subplots(rows=3, cols=1, shared_xaxes=True,
                       subplot_titles=("睡眠时长趋势", "睡眠阶段分布", "入睡和起床时间"))
    
    # 睡眠时长趋势
    fig.add_trace(
        go.Scatter(x=df['date'], y=df['total_sleep']/60, 
                  name="总睡眠时间(小时)", mode="lines+markers"),
        row=1, col=1
    )
    
    # 睡眠阶段堆叠面积图
    fig.add_trace(
        go.Scatter(x=df['date'], y=df['deep_sleep']/60, 
                  name="深睡", stackgroup="one"),
        row=2, col=1
    )
    fig.add_trace(
        go.Scatter(x=df['date'], y=df['light_sleep']/60, 
                  name="浅睡", stackgroup="one"),
        row=2, col=1
    )
    fig.add_trace(
        go.Scatter(x=df['date'], y=df['rem']/60, 
                  name="REM", stackgroup="one"),
        row=2, col=1
    )
    
    # 入睡和起床时间
    fig.add_trace(
        go.Scatter(x=df['date'], y=df['bedtime'], 
                  name="入睡时间", mode="markers"),
        row=3, col=1
    )
    fig.add_trace(
        go.Scatter(x=df['date'], y=df['wakeup_time'], 
                  name="起床时间", mode="markers"),
        row=3, col=1
    )
    
    fig.update_layout(height=900, showlegend=True)
    return fig

3.2 睡眠质量关联分析

我们可以探索不同因素与睡眠质量的关系：

def sleep_correlation_analysis(df):
    # 创建热力图展示各指标相关性
    corr_matrix = df[['total_sleep', 'deep_ratio', 'sleep_efficiency', 
                     'bedtime', 'wakeup_time', 'is_weekend']].corr()
    
    fig = px.imshow(corr_matrix, text_auto=True,
                   labels=dict(x="指标", y="指标", color="相关系数"),
                   x=corr_matrix.columns, y=corr_matrix.columns)
    
    # 添加工作日/周末对比箱线图
    weekend_fig = px.box(df, x="is_weekend", y="total_sleep",
                        labels={"is_weekend": "是否周末", "total_sleep": "总睡眠时间(分钟)"},
                        points="all")
    
    return corr_matrix, fig, weekend_fig

4. 系统自动化与部署

4.1 自动化数据管道

使用 schedule 库可以轻松设置定期任务：

import schedule
import time
from datetime import datetime

def update_sleep_data():
    print(f"{datetime.now()}: 开始更新睡眠数据...")
    # 这里放置数据提取和分析的完整流程
    print(f"{datetime.now()}: 数据更新完成")

# 每天凌晨3点执行
schedule.every().day.at("03:00").do(update_sleep_data)

while True:
    schedule.run_pending()
    time.sleep(60)

4.2 使用FastAPI构建Web服务

将分析结果通过Web服务暴露出来：

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import uvicorn

app = FastAPI()

@app.get("/dashboard", response_class=HTMLResponse)
async def sleep_dashboard():
    df = load_and_process_data()  # 加载并处理数据
    fig = create_sleep_dashboard(df)
    return fig.to_html(full_html=False)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

4.3 数据持久化方案

对于长期数据存储，建议使用更专业的数据库：

from sqlalchemy import create_engine

# 使用PostgreSQL存储历史数据
engine = create_engine("postgresql://user:password@localhost/sleep_data")

def save_to_database(df):
    # 只保存新增或更新的记录
    existing_dates = pd.read_sql("SELECT DISTINCT date FROM sleep_records", engine)['date']
    new_data = df[~df['date'].isin(existing_dates)]
    
    if not new_data.empty:
        new_data.to_sql("sleep_records", engine, if_exists="append", index=False)
        print(f"新增 {len(new_data)} 条记录")

5. 高级分析与机器学习应用

5.1 睡眠模式聚类分析

使用无监督学习识别不同的睡眠模式：

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

def cluster_sleep_patterns(df):
    features = df[['total_sleep', 'deep_ratio', 'bedtime', 'sleep_efficiency']]
    scaler = StandardScaler()
    scaled_features = scaler.fit_transform(features)
    
    kmeans = KMeans(n_clusters=3, random_state=42)
    df['sleep_cluster'] = kmeans.fit_predict(scaled_features)
    
    # 可视化聚类结果
    cluster_fig = px.scatter_matrix(
        df, dimensions=['total_sleep', 'deep_ratio', 'bedtime'],
        color='sleep_cluster', hover_data=['date']
    )
    
    return df, cluster_fig

5.2 睡眠质量预测模型

构建一个简单的预测模型，预测第二天的睡眠质量：

from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error

def build_sleep_prediction_model(df):
    # 创建滞后特征
    for i in range(1, 4):
        df[f'total_sleep_lag_{i}'] = df['total_sleep'].shift(i)
        df[f'bedtime_lag_{i}'] = df['bedtime'].shift(i)
    
    df = df.dropna()
    
    X = df[['total_sleep_lag_1', 'total_sleep_lag_2', 
            'bedtime_lag_1', 'bedtime_lag_2', 'weekday']]
    y = df['total_sleep']
    
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    model = RandomForestRegressor(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)
    
    predictions = model.predict(X_test)
    mae = mean_absolute_error(y_test, predictions)
    
    return model, mae

在实际项目中，我发现睡眠数据的分析价值远超预期。通过这个系统，我成功识别出咖啡因摄入和晚间屏幕时间对我的睡眠效率有显著影响。系统自动化的部分每周为我节省至少2小时的手工分析时间，而可视化看板让数据洞察一目了然。