""" 体重数据分析 CLI ================= 两个子命令： python analyze.py generate 生成 180 天合成数据（synthetic-user-001）写入 MySQL python analyze.py report 从库里读数据，输出终端报告 + 一张趋势 PNG 设计原则： - 合成数据有真实的减重轨迹 + 测量噪声 + 标签事件 + 若干异常点， 方便直接演示"数据洞察"，不需要真实用户数据。 - 所有 SQL 都写出来，方便面试官 review。 """ from __future__ import annotations import sys from dataclasses import dataclass from datetime import date, datetime, timedelta from pathlib import Path from uuid import uuid4 import matplotlib matplotlib.use("Agg") # 无 GUI 环境也能出图 import matplotlib.pyplot as plt # Windows 中文字体，避免图里中文变方框 plt.rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei", "DejaVu Sans"] plt.rcParams["axes.unicode_minus"] = False import numpy as np import pandas as pd import pymysql import typer from rich.console import Console from rich.table import Table app = typer.Typer(help="体重数据合成与分析 CLI", no_args_is_help=True) console = Console() # --------------------------------------------------------------------------- # 数据库连接配置（与 backend/.env 保持一致） # --------------------------------------------------------------------------- import os DB_CONFIG = dict( host=os.getenv("DB_HOST", "localhost"), port=int(os.getenv("DB_PORT", "3306")), user=os.getenv("DB_USERNAME", "analytics"), password=os.getenv("DB_PASSWORD", "123456"), database=os.getenv("DB_DATABASE", "campus_health"), charset="utf8mb4", autocommit=False, ) SYNTH_USER_ID = "synthetic-user-001" SYNTH_USER_PHONE = "13900000001" SYNTH_USER_NICKNAME = "合成用户·小A" OUTPUT_DIR = Path(__file__).parent / "output" def get_conn(password: str | None = None) -> pymysql.Connection: cfg = dict(DB_CONFIG) if password: cfg["password"] = password try: return pymysql.connect(**cfg) except pymysql.err.OperationalError as e: if e.args[0] == 1045 and not password: # 密码错误或未提供，交互式问一次 pwd = typer.prompt("MySQL 密码", hide_input=True) cfg["password"] = pwd return pymysql.connect(**cfg) console.print(f"[red]数据库连接失败：{e}[/red]") console.print( "请确认：1) MySQL 服务已启动 2) 已执行 backend/database/init.sql 和 schema.sql" ) sys.exit(1) # --------------------------------------------------------------------------- # 数据生成器 # --------------------------------------------------------------------------- @app.command() def generate( days: int = typer.Option(180, help="生成多少天的数据"), start_weight: float = typer.Option(75.0, help="初始体重 kg"), target_weight: float = typer.Option(66.0, help="目标体重 kg"), seed: int = typer.Option(42, help="随机种子，保证可复现"), ): """生成 180 天合成体重数据并写入 campus_health.weight_records""" rng = np.random.default_rng(seed) end_date = date.today() start_date = end_date - timedelta(days=days - 1) conn = get_conn() with conn.cursor() as cur: # 幂等：先清掉这个用户的旧数据 cur.execute("DELETE FROM weight_record_tags WHERE weight_record_id IN (SELECT id FROM weight_records WHERE user_id=%s)", (SYNTH_USER_ID,)) cur.execute("DELETE FROM weight_records WHERE user_id=%s", (SYNTH_USER_ID,)) cur.execute("DELETE FROM weight_goals WHERE user_id=%s", (SYNTH_USER_ID,)) cur.execute("DELETE FROM anomaly_logs WHERE user_id=%s", (SYNTH_USER_ID,)) cur.execute( "INSERT INTO users (id, phone, nickname, grade) VALUES (%s,%s,%s,%s) " "ON DUPLICATE KEY UPDATE nickname=VALUES(nickname)", (SYNTH_USER_ID, SYNTH_USER_PHONE, SYNTH_USER_NICKNAME, "大三"), ) # ---- 目标 ---- cur.execute( "INSERT INTO weight_goals (id,user_id,target_weight,target_body_fat,target_date," "start_weight,start_body_fat,start_date,weekly_target,status) " "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'active')", ( str(uuid4()), SYNTH_USER_ID, target_weight, 21.0, end_date + timedelta(days=30), start_weight, 28.0, start_date, 0.5, ), ) # ---- 标签事件（相对天数偏移）---- tag_events = { 0: "开始运动", 30: "饮食改变", 60: "生病", 90: "假期", 120: "目标调整", 150: "压力期", } # 取系统标签 ID cur.execute("SELECT id, tag_name FROM tags WHERE user_id IS NULL") tag_id_by_name = {name: tid for tid, name in cur.fetchall()} # ---- 逐日生成 ---- records = [] # (record_id, date, weight, body_fat, muscle, condition, notes) total_drop = start_weight - target_weight for i in range(days): d = start_date + timedelta(days=i) # 周末 20% 概率不称 if d.weekday() >= 5 and rng.random() < 0.35: continue # 基线减重曲线：线性，但在标签事件点有折点 progress = i / days baseline = start_weight - total_drop * progress # 事件扰动 if 60 <= i < 67: # 生病：急性脱水，体重骤降但体脂反升 baseline -= 1.8 elif 90 <= i < 98: # 假期：饮食放纵，反弹 baseline += 1.5 elif 120 <= i < 140: # 压力期：皮质醇高，平台 baseline += 0.4 # 测量噪声 noise = rng.normal(0, 0.25) # 测量条件：80% 早晨空腹，20% 餐后 if rng.random() < 0.8: condition = "fasting" time_str = "07:45:00" condition_offset = 0.0 else: condition = "after_meal" time_str = "12:30:00" condition_offset = rng.uniform(0.8, 1.4) # 餐后称重偏高 weight = round(baseline + noise + condition_offset, 2) # 体脂：跟随体重下降，但生病周会反向 bf_base = 28.0 - 7.0 * progress if 60 <= i < 67: bf_base += 1.5 body_fat = round(np.clip(bf_base + rng.normal(0, 0.2), 5, 60), 2) # 肌肉量：缓慢上升 muscle = round(26.0 + 1.2 * progress + rng.normal(0, 0.15), 2) notes = None if i in tag_events: notes = f"事件：{tag_events[i]}" records.append((str(uuid4()), d, time_str, weight, body_fat, muscle, condition, notes)) cur.executemany( "INSERT INTO weight_records (id,user_id,record_date,measurement_time,weight,body_fat,muscle_mass,measurement_condition,notes) " "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", [(r[0], SYNTH_USER_ID, r[1], r[2], r[3], r[4], r[5], r[6], r[7]) for r in records], ) # 关联标签到事件当天的记录 for offset, tag_name in tag_events.items(): d = start_date + timedelta(days=offset) cur.execute( "SELECT id FROM weight_records WHERE user_id=%s AND record_date=%s LIMIT 1", (SYNTH_USER_ID, d), ) row = cur.fetchone() if row and tag_name in tag_id_by_name: cur.execute( "INSERT INTO weight_record_tags (id,weight_record_id,tag_id) VALUES (%s,%s,%s) " "ON DUPLICATE KEY UPDATE id=id", (str(uuid4()), row[0], tag_id_by_name[tag_name]), ) # 埋一个 anomaly_log 示例 cur.execute( "INSERT INTO anomaly_logs (id,user_id,anomaly_type,severity,message,detected_date) " "VALUES (%s,%s,'rapid_change','warning','近7天体重骤降2.1kg（脱水/生病期），请结合体脂与症状判断',%s)", (str(uuid4()), SYNTH_USER_ID, start_date + timedelta(days=63)), ) conn.commit() conn.close() console.print(f"[green]✓ 已写入 {len(records)} 条记录，{len(tag_events)} 个标签事件[/green]") console.print(f" 日期范围：{start_date} → {end_date}") console.print(f" 体重区间：{start_weight}kg → {target_weight}kg（线性减重曲线 + 事件扰动）") # --------------------------------------------------------------------------- # 分析报告 # --------------------------------------------------------------------------- def load_data() -> pd.DataFrame: sql = """ SELECT wr.record_date, wr.weight, wr.body_fat, wr.muscle_mass, wr.measurement_condition, wr.notes, (SELECT GROUP_CONCAT(t.tag_name) FROM weight_record_tags wrt JOIN tags t ON t.id = wrt.tag_id WHERE wrt.weight_record_id = wr.id) AS tags FROM weight_records wr WHERE wr.user_id = %s AND wr.deleted_at IS NULL ORDER BY wr.record_date """ conn = get_conn() df = pd.read_sql(sql, conn, params=[SYNTH_USER_ID]) conn.close() df["record_date"] = pd.to_datetime(df["record_date"]) return df def report_overview(df: pd.DataFrame) -> None: table = Table(title="总体进度", show_header=True, header_style="bold cyan") table.add_column("指标") table.add_column("数值", justify="right") current = df.iloc[-1] initial = df.iloc[0] total_change = current["weight"] - initial["weight"] days = (df["record_date"].iloc[-1] - df["record_date"].iloc[0]).days weekly_avg = total_change / days * 7 # 最近 4 周趋势 recent = df[df["record_date"] >= df["record_date"].max() - pd.Timedelta(days=28)] recent_weekly = (recent["weight"].iloc[-1] - recent["weight"].iloc[0]) / 4 table.add_row("记录天数", str(len(df))) table.add_row("起始体重", f"{initial['weight']:.2f} kg") table.add_row("当前体重", f"{current['weight']:.2f} kg") table.add_row("累计变化", f"{total_change:+.2f} kg") table.add_row("周均变化（全程）", f"{weekly_avg:+.3f} kg/周") table.add_row("周均变化（近4周）", f"{recent_weekly:+.3f} kg/周") table.add_row("体脂变化", f"{current['body_fat'] - initial['body_fat']:+.2f} %") console.print(table) def report_condition_bias(df: pd.DataFrame) -> None: """空腹 vs 餐后的称重偏差——这是个真实且有价值的洞察""" fasting = df[df["measurement_condition"] == "fasting"]["weight"] after = df[df["measurement_condition"] == "after_meal"]["weight"] if len(fasting) < 3 or len(after) < 3: console.print("[yellow]样本不足，跳过测量偏差分析[/yellow]") return table = Table(title="测量条件偏差（餐后 vs 空腹）", show_header=True, header_style="bold magenta") table.add_column("条件") table.add_column("样本数", justify="right") table.add_column("均值 (kg)", justify="right") table.add_column("标准差", justify="right") table.add_row("空腹", str(len(fasting)), f"{fasting.mean():.2f}", f"{fasting.std():.2f}") table.add_row("餐后", str(len(after)), f"{after.mean():.2f}", f"{after.std():.2f}") table.add_row("差值", "—", f"{after.mean() - fasting.mean():+.2f}", "—") console.print(table) console.print( f"[dim]→ 餐后称重比空腹平均高 {after.mean()-fasting.mean():.2f} kg，" "跨条件比较时必须先按条件分层，否则会把'刚吃完饭'误读成'长胖了'。[/dim]" ) def report_tag_impact(df: pd.DataFrame) -> None: """每个标签事件前后 7 天的周减重速度对比""" events = df[df["tags"].notna() & (df["tags"] != "")] if events.empty: return table = Table(title="标签事件影响（事件日 ±7 天的周减重速度）", show_header=True, header_style="bold green") table.add_column("标签") table.add_column("事件日") table.add_column("前7天周速 (kg/周)") table.add_column("后7天周速 (kg/周)") table.add_column("变化") for _, row in events.iterrows(): tags = str(row["tags"]).split(",") for tag in tags: event_date = row["record_date"] pre = df[(df["record_date"] < event_date) & (df["record_date"] >= event_date - pd.Timedelta(days=7))] post = df[(df["record_date"] > event_date) & (df["record_date"] <= event_date + pd.Timedelta(days=7))] if len(pre) < 2 or len(post) < 2: continue pre_speed = (pre["weight"].iloc[-1] - pre["weight"].iloc[0]) / max(len(pre) - 1, 1) * 7 post_speed = (post["weight"].iloc[-1] - post["weight"].iloc[0]) / max(len(post) - 1, 1) * 7 delta = post_speed - pre_speed table.add_row( tag, event_date.strftime("%Y-%m-%d"), f"{pre_speed:+.2f}", f"{post_speed:+.2f}", f"{delta:+.2f}", ) console.print(table) def report_anomaly(df: pd.DataFrame) -> None: """简单的异常检测：周变化 > 2kg 或 体重降但体脂升""" df = df.sort_values("record_date").reset_index(drop=True) table = Table(title="自动异常扫描", show_header=True, header_style="bold red") table.add_column("日期") table.add_column("类型") table.add_column("说明") for i in range(7, len(df)): week_ago = df.iloc[i - 7] now = df.iloc[i] dw = now["weight"] - week_ago["weight"] dbf = now["body_fat"] - week_ago["body_fat"] if abs(dw) > 2.0: table.add_row( now["record_date"].strftime("%Y-%m-%d"), "rapid_change", f"7天体重变化 {dw:+.2f} kg（阈值 ±2.0）", ) if dw < -0.5 and dbf > 0.5: table.add_row( now["record_date"].strftime("%Y-%m-%d"), "body_fat_anomaly", f"体重 {dw:+.2f} kg 但体脂 {dbf:+.2f}%——疑似脱水而非减脂", ) if table.row_count == 0: console.print("[green]未发现异常[/green]") else: console.print(table) def draw_chart(df: pd.DataFrame) -> Path: OUTPUT_DIR.mkdir(exist_ok=True) fig, ax = plt.subplots(figsize=(11, 5.5)) fasting = df[df["measurement_condition"] == "fasting"] after = df[df["measurement_condition"] == "after_meal"] ax.scatter(after["record_date"], after["weight"], s=18, alpha=0.5, color="#f59e0b", label="餐后") ax.plot(fasting["record_date"], fasting["weight"], color="#2563eb", linewidth=1.5, label="空腹趋势") # 标签事件竖线 events = df[df["tags"].notna() & (df["tags"] != "")] for _, row in events.iterrows(): ax.axvline(row["record_date"], color="#9ca3af", linestyle="--", alpha=0.6) ax.text(row["record_date"], ax.get_ylim()[1], row["tags"].split(",")[0], rotation=90, fontsize=8, va="top", ha="right", color="#374151") ax.set_title("Synthetic User — Weight Trajectory") ax.set_xlabel("Date") ax.set_ylabel("Weight (kg)") ax.legend(loc="upper right") ax.grid(alpha=0.3) fig.tight_layout() out = OUTPUT_DIR / "weight_trend.png" fig.savefig(out, dpi=130) plt.close(fig) return out @app.command() def report(): """从库里读数据，输出分析报告 + 趋势图""" df = load_data() if df.empty: console.print("[red]没有数据，请先运行：python analyze.py generate[/red]") raise typer.Exit(1) console.print(f"\n[bold]数据窗口：[/bold]{df['record_date'].iloc[0].date()} → {df['record_date'].iloc[-1].date()}，共 {len(df)} 条\n") report_overview(df) console.print() report_condition_bias(df) console.print() report_tag_impact(df) console.print() report_anomaly(df) chart_path = draw_chart(df) console.print(f"\n[green]✓ 趋势图已保存：{chart_path}[/green]") if __name__ == "__main__": app()