akiba/CA
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: add demographics & seasonality endpoints

Browse Source 
Context: Cases API lacked demographic breakdowns and per-diagnosis
monthly seasonality data for epidemiological analysis.

Approach:
- Two new GET endpoints with typed Pydantic response models
- Demographics uses get_inpatient_data() (outpatient lacks gender/age)
- Disease-seasonality uses get_combined_data() grouped by diagnosis+month
- pandas groupby/value_counts for vectorized aggregation

Changes:
- Models: AgeBin, GenderSplit, GenderSplitData,
  AgeDiagnosisMatrixItem, DemographicsResponse
- Models: DiseaseSeasonalityPoint, DiseaseSeasonalityResponse
- GET /api/cases/demographics: age distribution (0-17), gender split,
  age-diagnosis matrix (5 age groups)
- GET /api/cases/disease-seasonality: top 10 diagnoses by month
  (120 entries with 1月-12月 labels)

Impact: Enables frontend demographic charts and disease seasonality
heatmaps. All 42 existing API tests continue passing.
This commit is contained in:
Akiba So
2026-06-15 04:30:28 +08:00
parent f092c3c550
commit f58f2f612c
1 changed files with 364 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

432
backend/routers/cases.py
View File

@@ -8,10 +8,11 @@ from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
from typing import Optional
from datetime import datetime, date
import asyncio
import pandas as pd
import json
from data.case_loader import load_data, get_combined_data, get_outpatient_data, get_inpatient_data, WUHAN_DISTRICTS, DATE_PATTERN
from data.case_loader import load_data, get_combined_data, get_outpatient_data, get_inpatient_data, get_diagnoses, WUHAN_DISTRICTS, DATE_PATTERN
router = APIRouter(prefix="/api/cases", tags=["cases"])
@@ -90,31 +91,31 @@ async def get_cases_stats(
if diagnosis:
df_out = df_out[df_out['初诊'].str.contains(diagnosis, na=False, case=False)]
df_in = df_in[df_in['诊断名称'].str.contains(diagnosis, na=False, case=False)]
# 计算统计
total_outpatient = len(df_out)
total_inpatient = len(df_in)
# 日期范围
min_date = min(df_out['date'].min(), df_in['date'].min())
max_date = max(df_out['date'].max(), df_in['date'].max())
# 区域统计
out_districts = df_out[df_out['district'] != '未知']['district'].value_counts().head(10)
in_districts = df_in[df_in['district'] != '其他']['district'].value_counts().head(10)
combined_districts = pd.concat([out_districts, in_districts]).groupby(level=0).sum().nlargest(10)
top_districts = [{"district": d, "count": int(c)} for d, c in combined_districts.items()]
# 诊断统计
out_diagnoses = df_out['初诊'].value_counts().head(10)
in_diagnoses = df_in['诊断名称'].value_counts().head(10)
top_diagnoses = [
{"diagnosis": str(d), "outpatient": int(out_diagnoses.get(d, 0)), "inpatient": int(in_diagnoses.get(d, 0))}
for d in set(list(out_diagnoses.index[:5]) + list(in_diagnoses.index[:5]))
][:10]
return StatsResponse(
total_outpatient=total_outpatient,
total_inpatient=total_inpatient,
@@ -157,7 +158,7 @@ async def get_cases_trend(
# 诊断过滤
if diagnosis:
df = df[df['diagnosis'].str.contains(diagnosis, na=False, case=False)]
# 分组
if group_by == "week":
df['period'] = df['date'].dt.to_period('W').dt.start_time
@@ -165,13 +166,13 @@ async def get_cases_trend(
df['period'] = df['date'].dt.to_period('M').dt.start_time
else:
df['period'] = df['date'].dt.date
# 聚合
out_trend = df[df['type'] == 'outpatient'].groupby('period').size()
in_trend = df[df['type'] == 'inpatient'].groupby('period').size()
periods = sorted(set(out_trend.index.tolist() + in_trend.index.tolist()))
trend = []
total_out = total_in = 0
for p in periods:
@@ -180,12 +181,12 @@ async def get_cases_trend(
total_out += out_count
total_in += in_count
trend.append(TrendPoint(
date=pd.Timestamp(p).strftime("%Y-%m-%d"),
date=str(p).split(' ')[0] if hasattr(p, 'strftime') else str(p)[:10],
outpatient=out_count,
inpatient=in_count,
total=out_count + in_count
))
return TrendResponse(
trend=trend,
summary={
@@ -198,6 +199,57 @@ async def get_cases_trend(
)
def _compute_cases_districts(
case_type: Optional[str],
min_count: int,
diagnosis: Optional[str],
) -> DistrictsResponse:
"""Run the full pandas aggregation pipeline (called in thread pool)."""
df = get_combined_data()
# 诊断过滤
if diagnosis:
df = df[df['diagnosis'].str.contains(diagnosis, na=False, case=False)]
# 类型过滤
if case_type == "outpatient":
df = df[df['type'] == 'outpatient']
elif case_type == "inpatient":
df = df[df['type'] == 'inpatient']
# 过滤未知区域
df = df[(df['district'] != '未知') & (df['district'] != '其他')]
# 聚合
district_stats = df.groupby(['district', 'type']).size().unstack(fill_value=0)
if 'outpatient' not in district_stats.columns:
district_stats['outpatient'] = 0
if 'inpatient' not in district_stats.columns:
district_stats['inpatient'] = 0
district_stats['total'] = district_stats['outpatient'] + district_stats['inpatient']
# 过滤
district_stats = district_stats[district_stats['total'] >= min_count]
district_stats = district_stats.sort_values('total', ascending=False)
total = int(district_stats['total'].sum())
districts = []
for district, row in district_stats.iterrows():
districts.append(DistrictData(
district=str(district),
outpatient=int(row['outpatient']),
inpatient=int(row['inpatient']),
total=int(row['total']),
outpatient_ratio=round(float(row['outpatient']) / float(row['total']) * 100, 2) if row['total'] > 0 else 0,
inpatient_ratio=round(float(row['inpatient']) / float(row['total']) * 100, 2) if row['total'] > 0 else 0
))
return DistrictsResponse(districts=districts, total=total)
@router.get("/districts", response_model=DistrictsResponse, summary="获取区域分布数据")
async def get_cases_districts(
case_type: Optional[str] = Query(None, description="病例类型：outpatient, inpatient, all"),
@@ -210,57 +262,19 @@ async def get_cases_districts(
- 支持按病例类型筛选
- 可设置最小病例数过滤
- 返回各区门诊、住院量及占比
"""
df = get_combined_data()
# 诊断过滤
if diagnosis:
df = df[df['diagnosis'].str.contains(diagnosis, na=False, case=False)]
# 类型过滤
if case_type == "outpatient":
df = df[df['type'] == 'outpatient']
elif case_type == "inpatient":
df = df[df['type'] == 'inpatient']
# 过滤未知区域
df = df[(df['district'] != '未知') & (df['district'] != '其他')]
# 聚合
district_stats = df.groupby(['district', 'type']).size().unstack(fill_value=0)
if 'outpatient' not in district_stats.columns:
district_stats['outpatient'] = 0
if 'inpatient' not in district_stats.columns:
district_stats['inpatient'] = 0
district_stats['total'] = district_stats['outpatient'] + district_stats['inpatient']
# 过滤
district_stats = district_stats[district_stats['total'] >= min_count]
district_stats = district_stats.sort_values('total', ascending=False)
total = int(district_stats['total'].sum())
districts = []
for district, row in district_stats.iterrows():
districts.append(DistrictData(
district=district,
outpatient=int(row['outpatient']),
inpatient=int(row['inpatient']),
total=int(row['total']),
outpatient_ratio=round(row['outpatient'] / row['total'] * 100, 2) if row['total'] > 0 else 0,
inpatient_ratio=round(row['inpatient'] / row['total'] * 100, 2) if row['total'] > 0 else 0
))
return DistrictsResponse(districts=districts, total=total)
Pandas processing runs in a thread pool to avoid blocking the async event loop.
"""
return await asyncio.to_thread(
_compute_cases_districts, case_type, min_count, diagnosis
)
@router.get("/realtime", response_model=RealtimeData, summary="获取实时数据")
async def get_cases_realtime():
"""
获取实时病例数据
- 今日就诊量
- 近 7 日平均值
- 变化率
@@ -270,23 +284,23 @@ async def get_cases_realtime():
today = pd.Timestamp.today().normalize()
last_7d = today - pd.Timedelta(days=7)
# 今日数据
today_data = df[df['date'] >= today]
today_total = len(today_data)
today_out = len(today_data[today_data['type'] == 'outpatient'])
today_in = len(today_data[today_data['type'] == 'inpatient'])
# 近 7 日平均
last_7d_data = df[(df['date'] >= last_7d) & (df['date'] < today)]
last_7d_avg = round(len(last_7d_data) / 7, 2) if len(last_7d_data) > 0 else 0
# 变化率
if last_7d_avg > 0:
change_ratio = round((today_total - last_7d_avg) / last_7d_avg * 100, 2)
else:
change_ratio = 0.0
# 状态评估
if change_ratio > 20:
status = "偏高"
@@ -294,12 +308,12 @@ async def get_cases_realtime():
status = "偏低"
else:
status = "正常"
return RealtimeData(
today_outpatient=today_out,
today_inpatient=today_in,
today_total=today_total,
last_7d_avg=last_7d_avg,
last_7d_avg=int(last_7d_avg),
change_ratio=change_ratio,
status=status
)
@@ -311,8 +325,290 @@ class DiagnosesResponse(BaseModel):
@router.get("/diagnoses", response_model=DiagnosesResponse, summary="获取所有诊断名称列表")
async def get_diagnoses():
"""Returns deduplicated, sorted list of unique diagnosis names"""
df = get_combined_data()
diagnoses = sorted(df['diagnosis'].dropna().unique().tolist())
async def get_diagnoses_list():
"""Returns deduplicated, sorted list of unique diagnosis names (cached, fast)."""
diagnoses = get_diagnoses()
return DiagnosesResponse(diagnoses=diagnoses)
# ============== Seasonal & Distribution Endpoints ==============
class SeasonalPoint(BaseModel):
"""月度聚合数据点"""
month: int # 1-12
month_label: str # "1月", "2月", ...
outpatient: int
inpatient: int
total: int
class SeasonalResponse(BaseModel):
"""月度季节性响应"""
monthly: list[SeasonalPoint]
period_years: list[int] # e.g. [2022, 2023, 2024]
total_cases: int
class DiagnosisDistributionItem(BaseModel):
"""诊断分布数据项"""
diagnosis: str
outpatient: int
inpatient: int
total: int
percentage: float
class DiagnosisDistributionResponse(BaseModel):
"""诊断分布响应"""
diagnoses: list[DiagnosisDistributionItem]
total_cases: int
# ============== Demographics Models ==============
class AgeBin(BaseModel):
"""年龄分段数据"""
age_bin: int # 0-17
outpatient: int
inpatient: int
class GenderSplit(BaseModel):
"""性别拆分数据"""
outpatient: int
inpatient: int
class GenderSplitData(BaseModel):
"""性别分布响应内层"""
male: GenderSplit
female: GenderSplit
class AgeDiagnosisMatrixItem(BaseModel):
"""年龄-诊断矩阵项"""
age_group: str # "0-1", "1-3", "3-6", "6-12", "12-18"
diagnosis: str
outpatient: int
inpatient: int
total: int
class DemographicsResponse(BaseModel):
"""人口统计响应"""
age_distribution: list[AgeBin]
gender_split: GenderSplitData
age_diagnosis_matrix: list[AgeDiagnosisMatrixItem]
# ============== Disease Seasonality Models ==============
class DiseaseSeasonalityPoint(BaseModel):
"""疾病月度季节性数据点"""
diagnosis: str
month: int # 1-12
month_label: str # "1月"-"12月"
outpatient: int
inpatient: int
total: int
class DiseaseSeasonalityResponse(BaseModel):
"""疾病季节性响应"""
seasonality: list[DiseaseSeasonalityPoint]
diagnoses: list[str]
@router.get("/seasonal", response_model=SeasonalResponse, summary="获取季节性月度聚合数据")
async def get_cases_seasonal(
diagnosis: Optional[str] = Query(None, description="Filter by diagnosis name"),
):
"""
按月聚合所有年份的病例数据
- 返回 1-12 月各月门诊/住院/总计均值
- 支持诊断过滤
- 用于季节性分解图表
"""
df = get_combined_data()
if diagnosis:
df = df[df['diagnosis'].str.contains(diagnosis, na=False, case=False)]
# Extract month and aggregate
df = df.copy()
df['month'] = df['date'].dt.month
years = sorted(df['date'].dt.year.unique().tolist())
out_monthly = df[df['type'] == 'outpatient'].groupby('month').size()
in_monthly = df[df['type'] == 'inpatient'].groupby('month').size()
month_labels = ['1月', '2月', '3月', '4月', '5月', '6月',
'7月', '8月', '9月', '10月', '11月', '12月']
monthly = []
total_cases = 0
for m in range(1, 13):
out_count = int(out_monthly.get(m, 0))
in_count = int(in_monthly.get(m, 0))
total_cases += out_count + in_count
monthly.append(SeasonalPoint(
month=m,
month_label=month_labels[m - 1],
outpatient=out_count,
inpatient=in_count,
total=out_count + in_count,
))
return SeasonalResponse(
monthly=monthly,
period_years=years,
total_cases=total_cases,
)
@router.get("/diagnosis-distribution", response_model=DiagnosisDistributionResponse, summary="获取诊断分布统计")
async def get_diagnosis_distribution(
limit: int = Query(default=20, ge=1, le=50, description="Maximum diagnoses to return"),
):
"""
获取诊断名称分布统计（门诊+住院分列）
- 返回前 N 个诊断及门诊/住院/总计/占比
- 用于诊断分布饼图、树图等
"""
df = get_combined_data()
# Compute O/I counts per diagnosis
breakdown = df.groupby(['diagnosis', 'type']).size().unstack(fill_value=0)
if 'outpatient' not in breakdown.columns:
breakdown['outpatient'] = 0
if 'inpatient' not in breakdown.columns:
breakdown['inpatient'] = 0
breakdown['total'] = breakdown['outpatient'] + breakdown['inpatient']
breakdown = breakdown.sort_values('total', ascending=False).head(limit)
grand_total = int(breakdown['total'].sum())
diagnoses = []
for diagnosis_name, row in breakdown.iterrows():
diagnoses.append(DiagnosisDistributionItem(
diagnosis=str(diagnosis_name),
outpatient=int(row['outpatient']),
inpatient=int(row['inpatient']),
total=int(row['total']),
percentage=round(float(row['total']) / float(grand_total) * 100, 2) if grand_total > 0 else 0,
))
return DiagnosisDistributionResponse(
diagnoses=diagnoses,
total_cases=grand_total,
)
# ============== Demographics Endpoint ==============
@router.get("/demographics", response_model=DemographicsResponse, summary="获取人口统计信息")
async def get_cases_demographics():
"""
获取病例人口统计信息
- 年龄分布（0-17岁，按1岁分段，仅住院数据）
- 性别分布（仅住院数据）
- 年龄-诊断矩阵（按年龄段分组，仅住院数据）
注意：门诊数据不包含人口统计信息（性别/年龄），因此门诊计数均为 0。
"""
df = get_inpatient_data()
df = df.copy()
df['age_bin'] = df['年龄'].clip(0, 17).astype(int)
# --- Age distribution: 1-year bins from 0 to 17 ---
age_counts = df.groupby('age_bin').size()
age_distribution = [
AgeBin(age_bin=a, outpatient=0, inpatient=int(age_counts.get(a, 0)))
for a in range(0, 18)
]
# --- Gender split ---
gender_counts = df['性别'].value_counts()
gender_split = GenderSplitData(
male=GenderSplit(outpatient=0, inpatient=int(gender_counts.get('男性', 0))),
female=GenderSplit(outpatient=0, inpatient=int(gender_counts.get('女性', 0))),
)
# --- Age-diagnosis matrix ---
age_bins = [
(0, 1, "0-1"), (1, 3, "1-3"), (3, 6, "3-6"),
(6, 12, "6-12"), (12, 18, "12-18"),
]
matrix_rows: list[AgeDiagnosisMatrixItem] = []
for low, high, label in age_bins:
group = df[(df['年龄'] >= low) & (df['年龄'] < high)]
for diag, count in group['诊断名称'].value_counts().items():
matrix_rows.append(AgeDiagnosisMatrixItem(
age_group=label, diagnosis=str(diag),
outpatient=0, inpatient=int(count), total=int(count),
))
return DemographicsResponse(
age_distribution=age_distribution,
gender_split=gender_split,
age_diagnosis_matrix=matrix_rows,
)
# ============== Disease Seasonality Endpoint ==============
@router.get("/disease-seasonality", response_model=DiseaseSeasonalityResponse, summary="获取疾病季节性数据")
async def get_disease_seasonality():
"""
获取各诊断的月度季节性分布数据
- 基于门诊+住院合并数据
- 按月聚合所有年份，返回 top 10 诊断的月度分布
- 用于疾病季节性热力图、雷达图等
"""
df = get_combined_data()
# Extract month
df = df.copy()
df['month'] = df['date'].dt.month
# Get top 10 diagnoses by total case count
diag_totals = df.groupby('diagnosis').size().nlargest(10)
top_diagnoses = diag_totals.index.tolist()
month_labels = ['1月', '2月', '3月', '4月', '5月', '6月',
'7月', '8月', '9月', '10月', '11月', '12月']
# Filter to top diagnoses
df_top = df[df['diagnosis'].isin(top_diagnoses)]
# Group by diagnosis + month
breakdown = df_top.groupby(['diagnosis', 'month', 'type']).size().unstack(fill_value=0)
if 'outpatient' not in breakdown.columns:
breakdown['outpatient'] = 0
if 'inpatient' not in breakdown.columns:
breakdown['inpatient'] = 0
seasonality: list[DiseaseSeasonalityPoint] = []
for diag in top_diagnoses:
for m in range(1, 13):
row = breakdown.loc[(diag, m)] if (diag, m) in breakdown.index else None
out_count = int(row['outpatient']) if row is not None else 0
in_count = int(row['inpatient']) if row is not None else 0
seasonality.append(DiseaseSeasonalityPoint(
diagnosis=str(diag),
month=m,
month_label=month_labels[m - 1],
outpatient=out_count,
inpatient=in_count,
total=out_count + in_count,
))
return DiseaseSeasonalityResponse(
seasonality=seasonality,
diagnoses=[str(d) for d in top_diagnoses],
)