|
|
@@ -0,0 +1,398 @@
|
|
|
+"""
|
|
|
+体重数据分析 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()
|
