""" 建议模型服务 高精确率模型 + DiCE反事实解释，生成个性化建议 """ import logging import time import sys import os import pandas as pd import numpy as np from typing import Dict, Any, List, Optional from pathlib import Path # 添加DiCE模块路径 dice_path_I = Path("/Users/ning/Desktop/idea/代码forSarcoAdvisor/4.DICE建模/DICE/I") dice_path_II = Path("/Users/ning/Desktop/idea/代码forSarcoAdvisor/4.DICE建模/DICE/II") sys.path.append(str(dice_path_I)) sys.path.append(str(dice_path_II)) from schemas.user_input import UserInput, AdvisoryResponse, RecommendationItem from utils.model_loader import model_manager from config import DiCEConfig, SystemConfig # 尝试导入DiCE相关模块 try: import dice_ml from dice_ml import Dice DICE_AVAILABLE = True except ImportError: DICE_AVAILABLE = False logging.warning("DiCE module not available, will use rule-based recommendations") logger = logging.getLogger(__name__) class AdvisoryService: """建议服务类""" def __init__(self): self.model_manager = model_manager self.dice_explainers = {} # DiCE初始化将在模型加载后进行 def initialize_dice(self): """在模型加载后初始化DiCE解释器""" self._init_dice_explainers() def _init_dice_explainers(self): """初始化DiCE解释器""" if not DICE_AVAILABLE: logger.warning("DiCE不可用，将使用基于规则的建议") return try: # 加载DiCE数据和解释器 self._setup_sarcoI_dice() self._setup_sarcoII_dice() logger.info("DiCE解释器初始化完成") except Exception as e: logger.error(f"DiCE解释器初始化失败: {str(e)}") self.dice_explainers = {} def _setup_sarcoI_dice(self): """设置SarcoI DiCE解释器""" try: # 动态确定数据路径 app_path = Path(__file__).parent.parent if (app_path / "data").exists(): # 云端部署路径 data_path = app_path / "data" else: # 本地开发路径 data_path = Path('/Users/ning/Desktop/idea/代码forSarcoAdvisor/4.DICE建模/预筛选') # 🚀 根据配置加载SarcoI数据集 config = DiCEConfig.get_current_config() train_data = pd.read_csv(data_path / 'SarcoI_train_final.csv') # 根据配置决定是否合并测试集 if config["combine_train_test"]: try: test_data = pd.read_csv(data_path / 'SarcoI_test_final.csv') all_data = pd.concat([train_data, test_data], ignore_index=True) logger.info(f"SarcoI DiCE合并训练+测试数据: {len(all_data)} 样本") except FileNotFoundError: all_data = train_data logger.info("SarcoI测试集未找到，仅使用训练集") else: all_data = train_data logger.info(f"SarcoI DiCE使用训练数据: {len(all_data)} 样本") # 根据配置决定是否采样 if config["max_samples"] and len(all_data) > config["max_samples"]: all_data = all_data.sample(n=config["max_samples"], random_state=42) logger.info(f"SarcoI DiCE数据采样至: {len(all_data)} 样本") logger.info(f"SarcoI DiCE最终数据集: {len(all_data)} 样本 (模式: {DiCEConfig.PRECISION_MODE})") # 创建DiCE数据对象 dice_data = dice_ml.Data( dataframe=all_data, continuous_features=['body_mass_index', 'race_ethnicity', 'WWI', 'age_years', 'Activity_Sedentary_Ratio', 'Total_Moderate_Minutes_week', 'Vigorous_MET_Ratio'], outcome_name='SarcoI' ) # 包装模型 wrapped_model = self._wrap_model('sarcoI') dice_model = dice_ml.Model(model=wrapped_model, backend='sklearn') # 🚀 创建极致精度解释器 - 使用genetic方法 + 配置文件中的高精度参数 dice_config_params = DiCEConfig.get_current_config() logger.info(f"SarcoI DiCE使用精度模式: {DiCEConfig.PRECISION_MODE}") # genetic方法是DiCE中最精确但最慢的方法，完美符合您的需求 self.dice_explainers['sarcoI'] = { 'explainer': Dice(dice_data, dice_model, method='genetic'), 'mutable_features': ['Activity_Sedentary_Ratio', 'Total_Moderate_Minutes_week', 'Vigorous_MET_Ratio', 'body_mass_index', 'WWI'], # 🎯 使用配置文件中的高精度参数 'precision_config': dice_config_params } except Exception as e: logger.error(f"SarcoI DiCE设置失败: {str(e)}") def _setup_sarcoII_dice(self): """设置SarcoII DiCE解释器""" try: # 动态确定数据路径 app_path = Path(__file__).parent.parent if (app_path / "data").exists(): # 云端部署路径 data_path = app_path / "data" else: # 本地开发路径 data_path = Path('/Users/ning/Desktop/idea/代码forSarcoAdvisor/4.DICE建模/预筛选') # 🚀 根据配置加载SarcoII数据集 config = DiCEConfig.get_current_config() train_data = pd.read_csv(data_path / 'SarcoII_train_final.csv') # 根据配置决定是否合并测试集 if config["combine_train_test"]: try: test_data = pd.read_csv(data_path / 'SarcoII_test_final.csv') all_data = pd.concat([train_data, test_data], ignore_index=True) logger.info(f"SarcoII DiCE合并训练+测试数据: {len(all_data)} 样本") except FileNotFoundError: all_data = train_data logger.info("SarcoII测试集未找到，仅使用训练集") else: all_data = train_data logger.info(f"SarcoII DiCE使用训练数据: {len(all_data)} 样本") # 根据配置决定是否采样 if config["max_samples"] and len(all_data) > config["max_samples"]: all_data = all_data.sample(n=config["max_samples"], random_state=42) logger.info(f"SarcoII DiCE数据采样至: {len(all_data)} 样本") logger.info(f"SarcoII DiCE最终数据集: {len(all_data)} 样本 (模式: {DiCEConfig.PRECISION_MODE})") # 创建DiCE数据对象 dice_data = dice_ml.Data( dataframe=all_data, continuous_features=['body_mass_index', 'race_ethnicity', 'age_years', 'Activity_Sedentary_Ratio', 'Activity_Diversity_Index', 'WWI', 'Vigorous_MET_Ratio', 'sedentary_minutes'], outcome_name='SarcoII' ) # 包装模型 wrapped_model = self._wrap_model('sarcoII') dice_model = dice_ml.Model(model=wrapped_model, backend='sklearn') # 🚀 创建极致精度解释器 - 使用genetic方法 + 配置文件中的高精度参数 dice_config_params = DiCEConfig.get_current_config() logger.info(f"SarcoII DiCE使用精度模式: {DiCEConfig.PRECISION_MODE}") # genetic方法是DiCE中最精确但最慢的方法，完美符合您的需求 self.dice_explainers['sarcoII'] = { 'explainer': Dice(dice_data, dice_model, method='genetic'), 'mutable_features': ['Activity_Sedentary_Ratio', 'Activity_Diversity_Index', 'WWI', 'Vigorous_MET_Ratio', 'sedentary_minutes', 'body_mass_index'], # 🎯 使用配置文件中的高精度参数 'precision_config': dice_config_params } except Exception as e: logger.error(f"SarcoII DiCE设置失败: {str(e)}") def _wrap_model(self, model_type: str): """包装模型以适配DiCE""" class ModelWrapper: def __init__(self, model, threshold, model_manager_ref, model_type): self.model = model self.threshold = threshold self.model_manager = model_manager_ref self.model_type = model_type def predict(self, X): X_processed = self._process_input(X) proba = self.model.predict_proba(X_processed)[:, 1] return (proba >= self.threshold).astype(int) def predict_proba(self, X): X_processed = self._process_input(X) return self.model.predict_proba(X_processed) def _process_input(self, X): if isinstance(X, pd.DataFrame): X_processed = X.copy() for col in X_processed.columns: if X_processed[col].dtype.name == 'category': X_processed[col] = X_processed[col].astype('int64') return X_processed else: return X # 检查模型是否存在 if model_type not in self.model_manager.advisory_models: raise KeyError(f"建议模型 '{model_type}' 未找到") model = self.model_manager.advisory_models[model_type] threshold = self.model_manager.thresholds[model_type]['advisory'] return ModelWrapper(model, threshold, self.model_manager, model_type) async def generate_recommendations(self, user_data: UserInput, risk_types: List[str] = None, num_recommendations: int = 3, language: str = "zh") -> AdvisoryResponse: """ 生成个性化建议 Args: user_data: 用户输入数据 risk_types: 需要生成建议的风险类型 num_recommendations: 每种类型的建议数量 Returns: AdvisoryResponse: 建议结果 """ start_time = time.time() fallback_used = False try: # 转换用户数据 user_dict = user_data.model_dump() # 默认为高风险的类型生成建议 if risk_types is None: risk_types = self._determine_risk_types(user_dict) # 生成建议 sarcoI_recommendations = [] sarcoII_recommendations = [] if 'sarcoI' in risk_types: sarcoI_recommendations = await self._generate_sarcoI_recommendations( user_dict, num_recommendations, language ) if 'sarcoII' in risk_types: sarcoII_recommendations = await self._generate_sarcoII_recommendations( user_dict, num_recommendations, language ) # 如果没有生成任何建议，使用降级方案 if not sarcoI_recommendations and not sarcoII_recommendations: logger.warning("DiCE建议生成失败，使用降级方案") fallback_recs = self._generate_fallback_recommendations(user_dict, risk_types, language) if 'sarcoI' in risk_types: sarcoI_recommendations = fallback_recs[:2] # 前2个给SarcoI if 'sarcoII' in risk_types: sarcoII_recommendations = fallback_recs[2:4] if len(fallback_recs) > 2 else fallback_recs # 后2个给SarcoII fallback_used = True # 合并所有建议用于API返回 all_recommendations = sarcoI_recommendations + sarcoII_recommendations # 生成优先级行动和目标指标 priority_actions = self._generate_priority_actions(sarcoI_recommendations, sarcoII_recommendations) target_metrics = self._generate_target_metrics(user_dict, sarcoI_recommendations, sarcoII_recommendations) processing_time = time.time() - start_time response = AdvisoryResponse( sarcoI_recommendations=sarcoI_recommendations, sarcoII_recommendations=sarcoII_recommendations, recommendations=all_recommendations, # 添加合并的建议列表 priority_actions=priority_actions, target_metrics=target_metrics, processing_time=processing_time, success=True, fallback_used=fallback_used ) logger.info(f"建议生成完成: SarcoI={len(sarcoI_recommendations)}, SarcoII={len(sarcoII_recommendations)}, " f"fallback={fallback_used}, 耗时={processing_time:.2f}s") return response except Exception as e: logger.error(f"建议生成失败: {str(e)}") # 返回基础建议 basic_recommendations = self._generate_basic_recommendations(language) processing_time = time.time() - start_time if language == 'en': priority_actions = ["Consult Professional Doctor", "Increase Physical Activity", "Improve Diet"] target_metrics = {"Recommendation": "Seek professional medical guidance"} else: priority_actions = ["咨询专业医生", "增加体力活动", "改善饮食习惯"] target_metrics = {"建议": "寻求专业医疗指导"} return AdvisoryResponse( sarcoI_recommendations=basic_recommendations, sarcoII_recommendations=[], recommendations=basic_recommendations, # 添加合并的建议列表 priority_actions=priority_actions, target_metrics=target_metrics, processing_time=processing_time, success=False, fallback_used=True ) def _determine_risk_types(self, user_dict: Dict) -> List[str]: """确定需要生成建议的风险类型""" risk_types = [] try: # 使用建议模型进行高精确率预测 sarcoI_result = self.model_manager.predict_advisory(user_dict, 'sarcoI') sarcoII_result = self.model_manager.predict_advisory(user_dict, 'sarcoII') if sarcoI_result['risk_level'] in ['medium', 'high']: risk_types.append('sarcoI') if sarcoII_result['risk_level'] in ['medium', 'high']: risk_types.append('sarcoII') except Exception as e: logger.error(f"风险类型判断失败: {str(e)}") # 默认返回所有类型 risk_types = ['sarcoI', 'sarcoII'] return risk_types if risk_types else ['sarcoI', 'sarcoII'] async def _generate_sarcoI_recommendations(self, user_dict: Dict, num_recs: int, language: str = "zh") -> List[RecommendationItem]: """生成SarcoI建议""" recommendations = [] try: if 'sarcoI' in self.dice_explainers: logger.info("SarcoI 尝试使用DiCE生成建议...") # 使用DiCE生成建议，添加超时保护 import signal def dice_timeout_handler(signum, frame): raise TimeoutError("DiCE生成超时") # 设置2分钟超时 signal.signal(signal.SIGALRM, dice_timeout_handler) signal.alarm(120) try: dice_recs = self._generate_dice_recommendations(user_dict, 'sarcoI', num_recs, language) recommendations.extend(dice_recs) logger.info(f"SarcoI DiCE生成成功，获得{len(dice_recs)}个建议") finally: signal.alarm(0) # 如果DiCE建议不足，补充规则建议 if len(recommendations) < num_recs: logger.info("SarcoI DiCE建议不足，补充规则建议...") rule_recs = self._generate_sarcoI_rule_recommendations(user_dict, language) recommendations.extend(rule_recs[:num_recs - len(recommendations)]) except Exception as e: logger.error(f"SarcoI建议生成失败: {str(e)}") logger.info("SarcoI 使用规则建议作为降级方案") recommendations = self._generate_sarcoI_rule_recommendations(user_dict, language)[:num_recs] return recommendations[:num_recs] async def _generate_sarcoII_recommendations(self, user_dict: Dict, num_recs: int, language: str = "zh") -> List[RecommendationItem]: """生成SarcoII建议""" recommendations = [] try: if 'sarcoII' in self.dice_explainers: # 使用DiCE生成建议 dice_recs = self._generate_dice_recommendations(user_dict, 'sarcoII', num_recs, language) recommendations.extend(dice_recs) # 如果DiCE建议不足，补充规则建议 if len(recommendations) < num_recs: rule_recs = self._generate_sarcoII_rule_recommendations(user_dict, language) recommendations.extend(rule_recs[:num_recs - len(recommendations)]) except Exception as e: logger.error(f"SarcoII建议生成失败: {str(e)}") recommendations = self._generate_sarcoII_rule_recommendations(user_dict, language)[:num_recs] return recommendations[:num_recs] def _generate_dice_recommendations(self, user_dict: Dict, model_type: str, num_recs: int, language: str = "zh") -> List[RecommendationItem]: """使用DiCE生成反事实建议""" recommendations = [] import time try: dice_start_time = time.time() logger.info(f"开始{model_type} DiCE分析...") dice_config = self.dice_explainers[model_type] explainer = dice_config['explainer'] mutable_features = dice_config['mutable_features'] # 准备查询数据 query_df = self._prepare_dice_query(user_dict, model_type) logger.info(f"{model_type} DiCE查询数据准备完成: {query_df.shape}") logger.info(f"{model_type} 查询数据内容: {query_df.iloc[0].to_dict()}") # 检查用户当前预测结果 wrapped_model = self._wrap_model(model_type) current_prediction = wrapped_model.predict_proba(query_df)[0] current_class = wrapped_model.predict(query_df)[0] logger.info(f"{model_type} 当前预测: 类别={current_class}, 概率={current_prediction}") # 如果当前预测已经是低风险，生成维持性建议 if current_class == 0: logger.info(f"{model_type} 当前预测为低风险类别(0)，生成维持性建议") maintenance_recommendations = self._generate_maintenance_recommendations(user_dict, model_type, language) recommendations.extend(maintenance_recommendations) return recommendations # 生成反事实 - 保持原始参数确保精确度 logger.info(f"{model_type} 开始生成反事实，目标类别=0，可变特征={mutable_features}") # 🚀 使用配置文件中的极致精度参数 dice_config_params = DiCEConfig.get_current_config() logger.info(f"{model_type} 使用DiCE精度模式: {DiCEConfig.PRECISION_MODE}") logger.info(f"{model_type} DiCE配置参数: {dice_config_params}") dice_params_list = [ # 🚀 极致精度配置：追求最高质量 { "total_CFs": dice_config_params.get('max_counterfactuals', 500), "desired_class": 0, "features_to_vary": mutable_features, "proximity_weight": dice_config_params.get('proximity_weight', 0.01), "diversity_weight": dice_config_params.get('diversity_weight', 15.0) }, # 🔥 高精度备用配置 { "total_CFs": dice_config_params.get('max_counterfactuals', 500) // 2, "desired_class": 0, "features_to_vary": mutable_features, "proximity_weight": dice_config_params.get('proximity_weight', 0.01) * 2, "diversity_weight": dice_config_params.get('diversity_weight', 15.0) * 0.8 }, # 🛡️ 保守配置（如果前两个失败） { "total_CFs": 100, "desired_class": 0, "features_to_vary": mutable_features, "proximity_weight": 0.1, "diversity_weight": 5.0 } ] # 🚀 设置DiCE总体超时 - 极致精度模式无时间限制 dice_timeout = dice_config_params.get('timeout_seconds', None) if dice_timeout: logger.info(f"{model_type} DiCE设置超时: {dice_timeout}秒") else: logger.info(f"{model_type} DiCE无超时限制 (极致精度模式)") counterfactuals = None for i, params in enumerate(dice_params_list): try: logger.info(f"{model_type} 尝试DiCE参数组合 {i+1}: {params}") logger.info(f"{model_type} 查询数据形状: {query_df.shape}") logger.info(f"{model_type} 可变特征: {params.get('features_to_vary', [])}") # 🚀 生成反事实，添加详细日志和计时 dice_gen_start = time.time() logger.info(f"{model_type} 开始DiCE生成 (极致精度模式)...") logger.info(f"{model_type} 预期生成 {params['total_CFs']} 个反事实样本") logger.info(f"{model_type} 多样性权重: {params['diversity_weight']}, 接近度权重: {params['proximity_weight']}") counterfactuals = explainer.generate_counterfactuals(query_df, **params) dice_gen_time = time.time() - dice_gen_start logger.info(f"{model_type} DiCE生成完成，耗时: {dice_gen_time:.2f}秒") # 检查是否生成了有效的反事实 if (hasattr(counterfactuals, 'cf_examples_list') and len(counterfactuals.cf_examples_list) > 0 and counterfactuals.cf_examples_list[0].final_cfs_df is not None and len(counterfactuals.cf_examples_list[0].final_cfs_df) > 0): logger.info(f"{model_type} DiCE参数组合 {i+1} 成功生成反事实") break else: logger.warning(f"{model_type} DiCE参数组合 {i+1} 未生成有效反事实") counterfactuals = None except Exception as param_e: logger.warning(f"{model_type} DiCE参数组合 {i+1} 失败: {str(param_e)}") counterfactuals = None if counterfactuals is None: logger.error(f"{model_type} 所有DiCE参数组合均失败") dice_time = time.time() - dice_start_time logger.info(f"{model_type} DiCE分析完成，耗时: {dice_time:.2f}秒") # 解析反事实结果 logger.info(f"{model_type} 反事实对象类型: {type(counterfactuals)}") logger.info(f"{model_type} 反事实对象属性: {dir(counterfactuals)}") if hasattr(counterfactuals, 'cf_examples_list'): logger.info(f"{model_type} cf_examples_list长度: {len(counterfactuals.cf_examples_list)}") if len(counterfactuals.cf_examples_list) > 0: cf_example = counterfactuals.cf_examples_list[0] logger.info(f"{model_type} cf_example类型: {type(cf_example)}") logger.info(f"{model_type} cf_example属性: {dir(cf_example)}") if hasattr(cf_example, 'final_cfs_df'): cf_df = cf_example.final_cfs_df logger.info(f"{model_type} final_cfs_df: {cf_df is not None}") if cf_df is not None: logger.info(f"{model_type} 反事实DataFrame形状: {cf_df.shape}") logger.info(f"{model_type} 反事实内容:\n{cf_df}") # 临时简化：直接使用前N个反事实，避免Series比较问题 logger.info(f"{model_type} 使用简化筛选，直接选择前{num_recs}个反事实") best_counterfactuals = cf_df.head(num_recs) logger.info(f"{model_type} 筛选出 {len(best_counterfactuals)} 个最佳反事实") # 处理筛选后的反事实 for i, (_, cf_row) in enumerate(best_counterfactuals.iterrows()): try: logger.info(f"{model_type} 处理第{i+1}个最佳反事实: {cf_row.to_dict()}") # 安全的特征变化分析 try: changes = self._analyze_feature_changes(query_df.iloc[0], cf_row, mutable_features) logger.info(f"{model_type} 特征变化: {changes}") except Exception as e: logger.error(f"{model_type} 特征变化分析失败: {str(e)}") continue # 安全的建议生成，避免重复 try: rec = self._changes_to_recommendation(changes, model_type, i + 1, language) if rec: # 检查是否与已有建议重复 is_duplicate = False for existing_rec in recommendations: if existing_rec.title == rec.title: is_duplicate = True break if not is_duplicate: recommendations.append(rec) logger.info(f"{model_type} 生成建议: {rec.title}") else: logger.info(f"{model_type} 跳过重复建议: {rec.title}") else: logger.warning(f"{model_type} 第{i+1}个反事实未生成有效建议") except Exception as e: logger.error(f"{model_type} 建议生成失败: {str(e)}") continue except Exception as e: logger.error(f"{model_type} 处理第{i+1}个反事实时出错: {str(e)}") import traceback logger.error(f"{model_type} 错误详情: {traceback.format_exc()}") continue else: logger.error(f"{model_type} final_cfs_df 为 None") else: logger.error(f"{model_type} cf_example 没有 final_cfs_df 属性") else: logger.error(f"{model_type} cf_examples_list 为空") else: logger.error(f"{model_type} counterfactuals 没有 cf_examples_list 属性") except Exception as e: dice_time = time.time() - dice_start_time if 'dice_start_time' in locals() else 0 logger.error(f"{model_type} DiCE建议生成失败 (耗时: {dice_time:.2f}秒): {str(e)}") # 🚀 追求极致精度模式：移除所有超时限制，无论耗时多久都要获得最佳结果 # 不再有任何时间限制，DiCE将运行到完成为止 logger.info(f"{model_type} DiCE分析耗时: {dice_time:.2f}秒 - 极致精度模式，无时间限制") return recommendations def _select_best_counterfactuals(self, original_row, cf_df, mutable_features, max_count): """🚀 极致精度反事实筛选 - 多维度质量评估，确保最优建议""" import pandas as pd import numpy as np if len(cf_df) <= max_count: return cf_df # 🎯 扩展特征分组，更细致的分类确保建议全面性 feature_groups = { 'weight_management': ['body_mass_index', 'WWI'], 'aerobic_exercise': ['Total_Moderate_Minutes_week', 'Vigorous_MET_Ratio'], 'sedentary_behavior': ['Activity_Sedentary_Ratio', 'sedentary_minutes'], 'exercise_diversity': ['Activity_Diversity_Index'], 'lifestyle_balance': [] # 综合性改变 } # 🏆 多维度质量评分系统 scored_cfs = [] for idx, cf_row in cf_df.iterrows(): # 1. 计算变化幅度合理性 (0-1分) change_reasonableness = self._calculate_change_reasonableness(original_row, cf_row, mutable_features) # 2. 计算特征变化多样性 (0-1分) change_diversity = self._calculate_change_diversity(original_row, cf_row, mutable_features) # 3. 计算实际可操作性 (0-1分) actionability = self._calculate_actionability(original_row, cf_row, mutable_features) # 4. 计算健康改善潜力 (0-1分) health_impact = self._calculate_health_impact(original_row, cf_row, mutable_features) # 🎯 综合质量评分 (加权平均) quality_score = ( change_reasonableness * 0.25 + # 变化合理性 change_diversity * 0.20 + # 多样性 actionability * 0.30 + # 可操作性 (最重要) health_impact * 0.25 # 健康影响 ) # 获取主要变化特征类别 primary_feature = self._get_primary_change_feature(original_row, cf_row, mutable_features) feature_category = self._get_feature_category(primary_feature, feature_groups) scored_cfs.append({ 'cf_row': cf_row, 'quality_score': quality_score, 'feature_category': feature_category, 'primary_feature': primary_feature, 'change_reasonableness': change_reasonableness, 'change_diversity': change_diversity, 'actionability': actionability, 'health_impact': health_impact }) # 🏅 按质量评分排序 scored_cfs.sort(key=lambda x: x['quality_score'], reverse=True) # 🎪 智能选择策略：优先质量，兼顾多样性 selected_cfs = [] used_categories = set() # 第一轮：每个类别选择最高质量的建议 for cf_info in scored_cfs: if cf_info['feature_category'] not in used_categories: selected_cfs.append(cf_info['cf_row']) used_categories.add(cf_info['feature_category']) if len(selected_cfs) >= max_count: break # 第二轮：如果还需要更多建议，选择剩余的高质量建议 if len(selected_cfs) < max_count: for cf_info in scored_cfs: if len(selected_cfs) >= max_count: break # 检查是否已经选择过这个建议 already_selected = False for selected_cf in selected_cfs: if self._are_cfs_similar(selected_cf, cf_info['cf_row'], mutable_features): already_selected = True break if not already_selected: selected_cfs.append(cf_info['cf_row']) return pd.DataFrame(selected_cfs) def _calculate_change_reasonableness(self, original_row, cf_row, mutable_features): """计算变化幅度的合理性 (0-1分)""" reasonableness_scores = [] for feature in mutable_features: if feature in original_row and feature in cf_row: original_val = original_row[feature] cf_val = cf_row[feature] # 计算相对变化幅度 if original_val != 0: relative_change = abs(cf_val - original_val) / abs(original_val) else: relative_change = abs(cf_val) # 根据特征类型设定合理变化范围 if 'bmi' in feature.lower() or 'weight' in feature.lower(): # 体重相关：10-30%变化较合理 if 0.1 <= relative_change <= 0.3: score = 1.0 elif relative_change < 0.1: score = relative_change / 0.1 # 变化太小 else: score = max(0.3, 1.0 - (relative_change - 0.3) / 0.5) # 变化太大 elif 'activity' in feature.lower() or 'minutes' in feature.lower(): # 运动相关：20-100%变化较合理 if 0.2 <= relative_change <= 1.0: score = 1.0 elif relative_change < 0.2: score = relative_change / 0.2 else: score = max(0.2, 1.0 - (relative_change - 1.0) / 1.0) else: # 其他特征：通用评分 if 0.1 <= relative_change <= 0.5: score = 1.0 elif relative_change < 0.1: score = relative_change / 0.1 else: score = max(0.2, 1.0 - (relative_change - 0.5) / 0.5) reasonableness_scores.append(score) return np.mean(reasonableness_scores) if reasonableness_scores else 0.5 def _calculate_change_diversity(self, original_row, cf_row, mutable_features): """计算特征变化的多样性 (0-1分)""" changed_features = 0 total_features = len(mutable_features) for feature in mutable_features: if feature in original_row and feature in cf_row: if abs(original_row[feature] - cf_row[feature]) > 1e-6: # 有显著变化 changed_features += 1 # 多样性评分：变化特征数量 / 总特征数量 diversity_ratio = changed_features / total_features if total_features > 0 else 0 # 理想情况是改变2-4个特征，给予最高分 if 2 <= changed_features <= 4: return 1.0 elif changed_features == 1: return 0.7 # 单一变化，分数较低 elif changed_features > 4: return max(0.5, 1.0 - (changed_features - 4) * 0.1) # 变化太多，逐渐降分 else: return 0.3 # 没有变化，最低分 def _calculate_actionability(self, original_row, cf_row, mutable_features): """计算实际可操作性 (0-1分)""" actionability_scores = [] for feature in mutable_features: if feature in original_row and feature in cf_row: original_val = original_row[feature] cf_val = cf_row[feature] change = abs(cf_val - original_val) # 根据特征类型评估可操作性 if 'minutes' in feature.lower(): # 运动时间：每周增加30-150分钟较容易实现 if 30 <= change <= 150: score = 1.0 elif change < 30: score = 0.8 # 变化太小，效果有限 else: score = max(0.4, 1.0 - (change - 150) / 200) # 变化太大，难以实现 elif 'bmi' in feature.lower(): # BMI：减少1-5较容易实现 if 1 <= change <= 5: score = 1.0 elif change < 1: score = 0.7 else: score = max(0.3, 1.0 - (change - 5) / 10) elif 'ratio' in feature.lower(): # 比例类特征：0.1-0.3的变化较合理 if 0.1 <= change <= 0.3: score = 1.0 elif change < 0.1: score = 0.8 else: score = max(0.4, 1.0 - (change - 0.3) / 0.5) else: # 默认评分 score = 0.8 actionability_scores.append(score) return np.mean(actionability_scores) if actionability_scores else 0.5 def _calculate_health_impact(self, original_row, cf_row, mutable_features): """计算健康改善潜力 (0-1分)""" impact_scores = [] for feature in mutable_features: if feature in original_row and feature in cf_row: original_val = original_row[feature] cf_val = cf_row[feature] # 根据特征类型和变化方向评估健康影响 if 'bmi' in feature.lower(): # BMI降低有益健康 if cf_val < original_val: # BMI降低 reduction = (original_val - cf_val) / original_val score = min(1.0, reduction * 5) # BMI每降低20%得满分 else: score = 0.3 # BMI增加不利健康 elif 'minutes' in feature.lower() and 'moderate' in feature.lower(): # 中等强度运动增加有益健康 if cf_val > original_val: # 运动时间增加 increase = (cf_val - original_val) / max(original_val, 1) score = min(1.0, increase * 2) # 运动时间每增加50%得满分 else: score = 0.4 # 运动减少不利健康 elif 'sedentary' in feature.lower(): # 久坐时间减少有益健康 if cf_val < original_val: # 久坐时间减少 reduction = (original_val - cf_val) / original_val score = min(1.0, reduction * 3) # 久坐时间每减少33%得满分 else: score = 0.2 # 久坐增加不利健康 elif 'vigorous' in feature.lower(): # 高强度运动比例增加有益健康 if cf_val > original_val: increase = (cf_val - original_val) / max(original_val, 0.1) score = min(1.0, increase * 1.5) else: score = 0.5 else: # 默认中等影响 score = 0.6 impact_scores.append(score) return np.mean(impact_scores) if impact_scores else 0.5 def _are_cfs_similar(self, cf1, cf2, mutable_features, threshold=0.1): """判断两个反事实是否相似""" similarity_count = 0 total_features = 0 for feature in mutable_features: if feature in cf1 and feature in cf2: total_features += 1 val1, val2 = cf1[feature], cf2[feature] # 计算相对差异 if abs(val1) > 1e-6 or abs(val2) > 1e-6: relative_diff = abs(val1 - val2) / max(abs(val1), abs(val2), 1e-6) if relative_diff < threshold: similarity_count += 1 else: similarity_count += 1 # 都接近0，认为相似 similarity_ratio = similarity_count / total_features if total_features > 0 else 0 return similarity_ratio > 0.8 # 80%以上特征相似则认为是相似的反事实 def _calculate_counterfactual_quality(self, original_row, cf_row, mutable_features): """计算反事实的质量分数""" score = 0 for feature in mutable_features: if feature in original_row.index and feature in cf_row.index: # 确保获取标量值 original_val = float(original_row[feature]) cf_val = float(cf_row[feature]) change = abs(cf_val - original_val) # 根据特征类型评分 if feature == 'body_mass_index': # BMI变化：合理范围内的变化得分更高 if 18.5 <= cf_val <= 24.9: # 正常BMI范围 score += 10 elif 25 <= cf_val <= 29.9: # 超重范围 score += 5 else: # 过低或过高 score -= 5 # 变化幅度合理性 if change > 50: # BMI变化超过50不合理 score -= 20 elif change > 20: # 变化过大 score -= 10 elif feature in ['Total_Moderate_Minutes_week', 'Vigorous_MET_Ratio', 'Activity_Sedentary_Ratio']: # 运动相关特征：有改善但不过度 if change > 0 and change <= original_val * 2: # 合理增长 score += 8 elif change > original_val * 2: # 过度增长 score -= 5 elif feature == 'WWI': # WWI变化：适度减少 if cf_val < original_val and change <= 3: # 适度减少 score += 6 elif change > 5: # 变化过大 score -= 8 return score def _get_primary_change_feature(self, original_row, cf_row, mutable_features): """找到变化最大的特征""" max_change = 0 primary_feature = None for feature in mutable_features: if feature in original_row.index and feature in cf_row.index: # 确保获取标量值 original_val = float(original_row[feature]) cf_val = float(cf_row[feature]) # 计算相对变化 if abs(original_val) > 1e-10: # 避免除零错误 relative_change = abs((cf_val - original_val) / original_val) else: relative_change = abs(cf_val) if relative_change > max_change: max_change = relative_change primary_feature = feature return primary_feature or mutable_features[0] def _get_feature_category(self, feature, feature_groups): """获取特征所属的类别""" for category, features in feature_groups.items(): if feature in features: return category return 'other' def _prepare_dice_query(self, user_dict: Dict, model_type: str) -> pd.DataFrame: """准备DiCE查询数据""" if model_type == 'sarcoI': features = ['body_mass_index', 'race_ethnicity', 'WWI', 'age_years', 'Activity_Sedentary_Ratio', 'Total_Moderate_Minutes_week', 'Vigorous_MET_Ratio'] else: # sarcoII features = ['body_mass_index', 'race_ethnicity', 'age_years', 'Activity_Sedentary_Ratio', 'Activity_Diversity_Index', 'WWI', 'Vigorous_MET_Ratio', 'sedentary_minutes'] feature_values = [] missing_features = [] for feature in features: if feature in user_dict and user_dict[feature] is not None: value = user_dict[feature] feature_values.append(value) logger.debug(f"{model_type} 特征 {feature}: {value} (来自用户数据)") else: # 默认值 defaults = { 'Total_Moderate_Minutes_week': 150, 'Activity_Diversity_Index': 2, 'sedentary_minutes': 480, 'Activity_Sedentary_Ratio': 0.5, 'Vigorous_MET_Ratio': 0.2 } default_val = defaults.get(feature, 0) feature_values.append(default_val) missing_features.append(feature) logger.warning(f"{model_type} 特征 {feature}: {default_val} (默认值)") if missing_features: logger.info(f"{model_type} DiCE缺失特征使用默认值: {missing_features}") query_df = pd.DataFrame([feature_values], columns=features) # 验证数据范围 self._validate_dice_query_data(query_df, model_type) return query_df def _validate_dice_query_data(self, query_df: pd.DataFrame, model_type: str): """验证DiCE查询数据的合理性""" data = query_df.iloc[0] # 检查关键特征的合理性 if data['body_mass_index'] < 10 or data['body_mass_index'] > 50: logger.warning(f"{model_type} BMI异常: {data['body_mass_index']}") if data['WWI'] < 5 or data['WWI'] > 20: logger.warning(f"{model_type} WWI异常: {data['WWI']}") if data['age_years'] < 18 or data['age_years'] > 80: logger.warning(f"{model_type} 年龄异常: {data['age_years']}") # 检查活动相关特征 if 'Activity_Sedentary_Ratio' in data: if data['Activity_Sedentary_Ratio'] < 0 or data['Activity_Sedentary_Ratio'] > 1: logger.warning(f"{model_type} 久坐比例异常: {data['Activity_Sedentary_Ratio']}") logger.info(f"{model_type} DiCE数据验证完成") def _analyze_feature_changes(self, original, counterfactual, mutable_features) -> List[Dict]: """分析特征变化""" changes = [] for feature in mutable_features: if feature in original.index and feature in counterfactual.index: # 确保获取标量值，避免Series比较问题 original_val = float(original[feature]) cf_val = float(counterfactual[feature]) if abs(cf_val - original_val) > 1e-6: # 安全的百分比计算 if abs(original_val) > 1e-10: change_pct = ((cf_val - original_val) / original_val) * 100 else: change_pct = 0.0 changes.append({ 'feature': feature, 'original': original_val, 'target': cf_val, 'change': cf_val - original_val, 'change_pct': change_pct }) return changes def _changes_to_recommendation(self, changes: List[Dict], model_type: str, rec_id: int, language: str = "zh") -> Optional[RecommendationItem]: """将特征变化转换为多样化的精确建议""" if not changes: return None # 按特征类型优先级选择，确保多样性 feature_priority = { 'Total_Moderate_Minutes_week': 1, # 运动优先 'Vigorous_MET_Ratio': 2, 'Activity_Diversity_Index': 3, 'body_mass_index': 4, # 体重其次 'WWI': 5, 'Activity_Sedentary_Ratio': 6 } # 根据建议ID选择不同类型的特征，确保5个建议的多样性 if rec_id == 1: # 第一个建议：中等强度运动 preferred_features = ['Total_Moderate_Minutes_week', 'moderate_minutes'] elif rec_id == 2: # 第二个建议：高强度运动 preferred_features = ['Vigorous_MET_Ratio', 'vigorous_minutes'] elif rec_id == 3: # 第三个建议：运动多样性 preferred_features = ['Activity_Diversity_Index', 'Activity_Sedentary_Ratio'] elif rec_id == 4: # 第四个建议：体重管理 preferred_features = ['body_mass_index', 'WWI'] else: # 第五个建议：久坐行为改善 preferred_features = ['sedentary_minutes', 'Activity_Sedentary_Ratio'] # 寻找首选特征的变化 selected_change = None for feature in preferred_features: for change in changes: if change['feature'] == feature: selected_change = change break if selected_change: break # 如果没找到首选特征，选择变化最大的 if not selected_change: changes.sort(key=lambda x: abs(x['change']), reverse=True) selected_change = changes[0] feature = selected_change['feature'] original_val = selected_change['original'] target_val = selected_change['target'] change_val = selected_change['change'] # 生成精确数值建议 title, description, target_change, expected_impact = self._get_precise_feature_recommendation( feature, original_val, target_val, change_val, model_type, language ) # 基于变化幅度计算优先级 change_magnitude = abs(change_val / original_val) if original_val != 0 else abs(change_val) if change_magnitude > 0.3: # 变化超过30% priority = "High" if language == 'en' else "高" elif change_magnitude > 0.15: # 变化超过15% priority = "Medium" if language == 'en' else "中" else: priority = "Low" if language == 'en' else "低" return RecommendationItem( title=title, description=description, priority=priority, target_change=target_change, expected_impact=expected_impact ) def _get_precise_feature_recommendation(self, feature: str, original_val: float, target_val: float, change_val: float, model_type: str, language: str = "zh") -> tuple: """获取基于精确数值的特征建议""" if feature == 'body_mass_index': if language == 'en': title = "Exercise-Based Weight Management Plan" description = f"Current BMI: {original_val:.1f} → Target: {target_val:.1f}" if original_val > target_val: weight_loss = (original_val - target_val) * 2.5 # Assuming height 1.7m, 1 BMI ≈ 2.5kg description += f"\n• Weight Loss Goal: {weight_loss:.1f} kg" description += f"\n• Timeline: {weight_loss/0.75:.0f} weeks (0.5-1kg/week)" description += f"\n• Exercise Plan:" description += f"\n - Aerobic: 300+ min/week" description += f"\n - Strength training: 3x/week" description += f"\n - HIIT: 2x/week" target_change = f"BMI: {original_val:.1f} → {target_val:.1f} (change: {change_val:+.1f})" expected_impact = f"Every 1-point BMI reduction decreases sarcopenia risk by 8-12% through exercise" else: title = "运动减重精确管理计划" description = f"当前BMI: {original_val:.1f} → 目标: {target_val:.1f}" if original_val > target_val: weight_loss = (original_val - target_val) * 2.5 # 假设身高1.7m，每1BMI约2.5kg description += f"\n• 减重目标: {weight_loss:.1f} 公斤" description += f"\n• 时间计划: {weight_loss/0.75:.0f} 周 (每周0.5-1公斤)" description += f"\n• 运动方案:" description += f"\n - 有氧运动: 每周300+分钟" description += f"\n - 力量训练: 每周3次" description += f"\n - HIIT训练: 每周2次" target_change = f"BMI: {original_val:.1f} → {target_val:.1f} (变化: {change_val:+.1f})" expected_impact = f"BMI每降低1点，通过运动肌少症风险降低约8-12%" elif feature == 'Total_Moderate_Minutes_week': if language == 'en': title = "Moderate-Intensity Exercise Precision Plan" description = f"Current weekly: {original_val:.0f}min → Target: {target_val:.0f}min" if target_val > original_val: increase = target_val - original_val daily_increase = increase / 7 description += f"\n• Increase: {increase:.0f} min/week ({daily_increase:.0f} min/day)" # Specific exercise recommendations if daily_increase <= 15: description += f"\n• Start: Daily brisk walk {daily_increase:.0f}min" elif daily_increase <= 30: description += f"\n• Plan: Walk 20min + swim/cycle {daily_increase-20:.0f}min" else: description += f"\n• Plan: Walk 30min + other exercises {daily_increase-30:.0f}min" description += f"\n• Activity Mix:" description += f"\n - Walking/jogging: {increase*0.4:.0f}min/week" description += f"\n - Swimming: {increase*0.3:.0f}min/week" description += f"\n - Cycling: {increase*0.2:.0f}min/week" description += f"\n - Tai Chi/Yoga: {increase*0.1:.0f}min/week" # Progressive plan weeks_to_target = max(4, int(increase / 30)) description += f"\n\nProgressive plan ({weeks_to_target}-week target):" for week in range(1, min(weeks_to_target + 1, 5)): weekly_target = original_val + (increase * week / weeks_to_target) description += f"\nWeek {week}: {weekly_target:.0f}min/week" target_change = f"Moderate exercise: {original_val:.0f}min/week → {target_val:.0f}min/week" expected_impact = f"Every 30min/week increase reduces sarcopenia risk by 5-8%, muscle strength improves 10-15%" else: title = "中等强度运动精确计划" description = f"当前每周: {original_val:.0f}分钟 → 目标: {target_val:.0f}分钟" if target_val > original_val: increase = target_val - original_val daily_increase = increase / 7 description += f"\n• 增加: {increase:.0f} 分钟/周 (每天{daily_increase:.0f}分钟)" # 具体的运动建议 if daily_increase <= 15: description += f"\n• 开始: 每天快走{daily_increase:.0f}分钟" elif daily_increase <= 30: description += f"\n• 方案: 快走20分钟 + 游泳/骑车{daily_increase-20:.0f}分钟" else: description += f"\n• 方案: 快走30分钟 + 其他运动{daily_increase-30:.0f}分钟" description += f"\n• 活动组合:" description += f"\n - 快走/慢跑: {increase*0.4:.0f}分钟/周" description += f"\n - 游泳: {increase*0.3:.0f}分钟/周" description += f"\n - 骑车: {increase*0.2:.0f}分钟/周" description += f"\n - 太极/瑜伽: {increase*0.1:.0f}分钟/周" # 渐进式计划 weeks_to_target = max(4, int(increase / 30)) description += f"\n\n渐进式计划（{weeks_to_target}周达标）：" for week in range(1, min(weeks_to_target + 1, 5)): weekly_target = original_val + (increase * week / weeks_to_target) description += f"\n第{week}周：{weekly_target:.0f}分钟/周" target_change = f"中等强度运动: {original_val:.0f}分钟/周 → {target_val:.0f}分钟/周" expected_impact = f"每增加30分钟/周，肌少症风险降低约5-8%，肌肉力量提升10-15%" elif feature == 'Vigorous_MET_Ratio': if language == 'en': title = "High-Intensity Exercise Precision Enhancement Plan" description = f"Current ratio: {original_val:.3f} → Target ratio: {target_val:.3f}" if target_val > original_val: increase = target_val - original_val description += f"\n📈 Need to increase high-intensity exercise ratio by {increase:.3f}" # Convert to actual time recommendations weekly_minutes = increase * 150 # Based on 150min/week baseline daily_minutes = weekly_minutes / 7 description += f"\n⏱️ Equivalent to adding {weekly_minutes:.0f} min/week high-intensity exercise" description += f"\n (about {daily_minutes:.0f} min/day)" # Specific high-intensity exercise plans description += f"\n🔥 Recommended high-intensity exercise combination:" if weekly_minutes <= 60: description += f"\n • Strength training: 3x/week, {weekly_minutes/3:.0f}min each" description += f"\n • HIIT training: 2x/week, {weekly_minutes/4:.0f}min each" elif weekly_minutes <= 120: description += f"\n • Strength training: 3x/week, 25min each" description += f"\n • Running/sprints: 2x/week, {(weekly_minutes-75)/2:.0f}min each" description += f"\n • HIIT training: 2x/week, 15min each" else: description += f"\n • Strength training: 4x/week, 30min each" description += f"\n • Running: 3x/week, {(weekly_minutes-120)/3:.0f}min each" description += f"\n • HIIT training: 2x/week, 20min each" # Safety reminders and progressive plan description += f"\n⚠️ Safety reminders:" description += f"\n • Warm up thoroughly for 5-10 minutes before exercise" description += f"\n • Cool down and stretch for 10 minutes after exercise" description += f"\n • Progress gradually to avoid exercise injuries" # Progressive intensity enhancement weeks_to_target = max(6, int(weekly_minutes / 20)) description += f"\n📅 {weeks_to_target}-week progressive plan:" for week in [1, 2, 4, weeks_to_target]: if week <= weeks_to_target: weekly_target = weekly_minutes * week / weeks_to_target description += f"\n Week {week}: {weekly_target:.0f}min/week high-intensity exercise" target_change = f"High-intensity exercise ratio: {original_val:.3f} → {target_val:.3f}" expected_impact = f"Every 0.1 increase in high-intensity ratio reduces sarcopenia risk by 10-15%, muscle strength improves 20-30%" else: title = "高强度运动精确提升计划" description = f"当前比例: {original_val:.3f} → 目标比例: {target_val:.3f}" if target_val > original_val: increase = target_val - original_val description += f"\n📈 需要提升 {increase:.3f} 的高强度运动比例" # 转换为实际时间建议 weekly_minutes = increase * 150 # 假设基于150分钟/周的基础 daily_minutes = weekly_minutes / 7 description += f"\n⏱️ 相当于每周增加 {weekly_minutes:.0f} 分钟高强度运动" description += f"\n (每天约 {daily_minutes:.0f} 分钟)" # 具体的高强度运动计划 description += f"\n🔥 推荐高强度运动组合：" if weekly_minutes <= 60: description += f"\n • 力量训练：3次/周，每次{weekly_minutes/3:.0f}分钟" description += f"\n • HIIT训练：2次/周，每次{weekly_minutes/4:.0f}分钟" elif weekly_minutes <= 120: description += f"\n • 力量训练：3次/周，每次25分钟" description += f"\n • 跑步/冲刺：2次/周，每次{(weekly_minutes-75)/2:.0f}分钟" description += f"\n • HIIT训练：2次/周，每次15分钟" else: description += f"\n • 力量训练：4次/周，每次30分钟" description += f"\n • 跑步：3次/周，每次{(weekly_minutes-120)/3:.0f}分钟" description += f"\n • HIIT训练：2次/周，每次20分钟" # 安全提醒和渐进计划 description += f"\n⚠️ 安全提醒：" description += f"\n • 运动前充分热身5-10分钟" description += f"\n • 运动后拉伸放松10分钟" description += f"\n • 循序渐进，避免运动损伤" # 渐进式强度提升 weeks_to_target = max(6, int(weekly_minutes / 20)) description += f"\n📅 {weeks_to_target}周渐进计划：" for week in [1, 2, 4, weeks_to_target]: if week <= weeks_to_target: weekly_target = weekly_minutes * week / weeks_to_target description += f"\n 第{week}周：{weekly_target:.0f}分钟/周高强度运动" target_change = f"高强度运动比例: {original_val:.3f} → {target_val:.3f}" expected_impact = f"高强度运动比例每提升0.1，肌少症风险降低10-15%，肌肉力量提升20-30%" elif feature == 'Activity_Sedentary_Ratio': if language == 'en': title = "Sedentary Behavior Precision Improvement Plan" description = f"Current activity/sedentary ratio: {original_val:.3f} → Target ratio: {target_val:.3f}" if target_val > original_val: increase = target_val - original_val # Assuming 16 hours of waking time daily_activity_increase = increase * 16 * 60 # Convert to minutes description += f"\n📈 Need to improve activity/sedentary ratio by {increase:.3f}" description += f"\n⏱️ Equivalent to increasing daily activity by {daily_activity_increase:.0f} minutes" # Specific activity increase strategies description += f"\n🎯 Activity increase strategies:" if daily_activity_increase <= 30: description += f"\n • Stand and move for 3-5 minutes every hour" description += f"\n • Take stairs instead of elevators" description += f"\n • Stand during phone calls/meetings" elif daily_activity_increase <= 60: description += f"\n • Stand and move for 5 minutes every 45 minutes" description += f"\n • Take a 15-minute walk during lunch break" description += f"\n • Use standing desk for 2-3 hours" description += f"\n • Park farther away/get off transit early to walk" else: description += f"\n • Stand and move for 5 minutes every 30 minutes" description += f"\n • Walk/bike to work for 30 minutes" description += f"\n • Use standing desk for 4-6 hours" description += f"\n • Actively engage in household activities" # Different strategies for weekdays and weekends description += f"\n📅 Time-specific strategies:" description += f"\n Weekdays:" description += f"\n • Set 30-minute activity reminders" description += f"\n • Do stretching exercises between meetings" description += f"\n • Take 10-15 minute walks after lunch" description += f"\n Weekends:" description += f"\n • Increase outdoor activities to {daily_activity_increase*1.5:.0f} minutes" description += f"\n • Engage in housework, gardening, walking shopping" description += f"\n • Reduce TV/phone screen time" # Practical tools recommendations description += f"\n🛠️ Practical tools:" description += f"\n • Set sedentary reminders on phone" description += f"\n • Use pedometer/fitness tracker" description += f"\n • Adjustable height desk" description += f"\n • Exercise ball instead of office chair" target_change = f"Activity/sedentary ratio: {original_val:.3f} → {target_val:.3f}" expected_impact = f"Every 0.1 increase in activity ratio reduces sarcopenia risk by 6-10%, metabolism improves 15-20%" else: title = "久坐行为精确改善计划" description = f"当前活动/久坐比例: {original_val:.3f} → 目标比例: {target_val:.3f}" if target_val > original_val: increase = target_val - original_val # 假设16小时清醒时间 daily_activity_increase = increase * 16 * 60 # 转换为分钟 description += f"\n📈 需要提升 {increase:.3f} 的活动/久坐比例" description += f"\n⏱️ 相当于每天增加活动 {daily_activity_increase:.0f} 分钟" # 具体的活动增加策略 description += f"\n🎯 活动增加策略：" if daily_activity_increase <= 30: description += f"\n • 每小时起身活动3-5分钟" description += f"\n • 走楼梯代替电梯" description += f"\n • 站立接电话/开会" elif daily_activity_increase <= 60: description += f"\n • 每45分钟起身活动5分钟" description += f"\n • 午休时间散步15分钟" description += f"\n • 使用站立办公桌2-3小时" description += f"\n • 远距离停车/提前下车步行" else: description += f"\n • 每30分钟起身活动5分钟" description += f"\n • 上下班步行/骑车30分钟" description += f"\n • 站立办公4-6小时" description += f"\n • 主动做家务活动" # 工作日和周末的不同策略 description += f"\n📅 分时段策略：" description += f"\n 工作日：" description += f"\n • 设置每30分钟活动提醒" description += f"\n • 会议间隙做伸展运动" description += f"\n • 午餐后散步10-15分钟" description += f"\n 周末：" description += f"\n • 户外活动增加到{daily_activity_increase*1.5:.0f}分钟" description += f"\n • 家务活动、园艺、购物步行" description += f"\n • 减少看电视/手机时间" # 实用工具推荐 description += f"\n🛠️ 实用工具：" description += f"\n • 手机设置久坐提醒" description += f"\n • 使用计步器/智能手环" description += f"\n • 可调节高度办公桌" description += f"\n • 瑜伽球代替办公椅" target_change = f"活动/久坐比例: {original_val:.3f} → {target_val:.3f}" expected_impact = f"活动比例每提升0.1，肌少症风险降低6-10%，新陈代谢提升15-20%" elif feature == 'WWI': if language == 'en': title = "Core-Focused Waist Management Plan" description = f"Current WWI: {original_val:.2f} → Target WWI: {target_val:.2f}" if original_val > target_val: decrease = original_val - target_val # WWI = waist/height^0.5, assuming height 1.7m waist_reduction = decrease * (1.7 ** 0.5) description += f"\n📉 Need to reduce WWI by {decrease:.2f}" description += f"\n📏 Equivalent to waist reduction of approximately {waist_reduction:.1f} cm" description += f"\n💪 Exercise recommendations:" description += f"\n • Core strengthening: Planks, crunches, Russian twists" description += f"\n • Aerobic exercise: Focus on fat burning" description += f"\n • HIIT training: Target abdominal fat reduction" target_change = f"WWI: {original_val:.2f} → {target_val:.2f}" expected_impact = f"Every 0.5 WWI reduction decreases sarcopenia risk by 7-12%" else: title = "核心运动腰围精确管理计划" description = f"当前WWI: {original_val:.2f} → 目标WWI: {target_val:.2f}" if original_val > target_val: decrease = original_val - target_val # WWI = 腰围/身高^0.5，假设身高1.7m waist_reduction = decrease * (1.7 ** 0.5) description += f"\n📉 需要减少WWI {decrease:.2f}" description += f"\n📏 相当于腰围减少约 {waist_reduction:.1f} 厘米" description += f"\n💪 运动建议：" description += f"\n • 核心训练：平板支撑、卷腹、俄式转体" description += f"\n • 有氧运动：专注脂肪燃烧" description += f"\n • HIIT训练：针对腹部脂肪减少" target_change = f"WWI: {original_val:.2f} → {target_val:.2f}" expected_impact = f"WWI每降低0.5，肌少症风险降低约7-12%" elif feature == 'Activity_Diversity_Index': if language == 'en': title = "Exercise Diversity Precision Enhancement Plan" description = f"Current diversity index: {original_val:.1f} → Target: {target_val:.1f}" if target_val > original_val: increase = target_val - original_val target_types = int(target_val) + 2 description += f"\n📈 Need to increase activity diversity by {increase:.1f}" description += f"\n🎯 Goal: Engage in {target_types} different types of exercise weekly" # Exercise type classification and recommendations description += f"\n🏃‍♂️ Recommended exercise type combinations:" description += f"\n Aerobic exercises (2-3 types):" description += f"\n • Brisk walking/jogging - 3-4x/week, 30min each" description += f"\n • Swimming/water exercise - 2x/week, 45min each" description += f"\n • Cycling - 2-3x/week, 40min each" description += f"\n 💪 Strength training (1-2 types):" description += f"\n • Weight training - 2-3x/week, 45min each" description += f"\n • Bodyweight training - 3x/week, 30min each" description += f"\n • Resistance band training - 2x/week, 25min each" description += f"\n 🧘‍♀️ Flexibility & balance (1-2 types):" description += f"\n • Yoga - 2-3x/week, 60min each" description += f"\n • Tai Chi - 2x/week, 45min each" description += f"\n • Pilates - 2x/week, 50min each" description += f"\n 🏐 Recreational sports (1 type):" description += f"\n • Ball sports - 1-2x/week" description += f"\n • Dancing - 1-2x/week" description += f"\n • Hiking/mountain climbing - 1x/week" # Weekly plan example description += f"\n📅 Diverse weekly plan example:" description += f"\n Monday: Strength training (45min)" description += f"\n Tuesday: Brisk walking (30min) + Yoga (30min)" description += f"\n Wednesday: Swimming (45min)" description += f"\n Thursday: Bodyweight training (30min) + Tai Chi (30min)" description += f"\n Friday: Cycling (40min)" description += f"\n Saturday: Ball sports (60min)" description += f"\n Sunday: Hiking/climbing (90min)" # Progressive increase recommendations description += f"\n🔄 Progressive increase strategy:" current_types = max(1, int(original_val)) for week in range(1, 5): week_types = min(current_types + week, target_types) description += f"\n Week {week}: {week_types} exercise types" target_change = f"Activity diversity: {original_val:.1f} → {target_val:.1f}" expected_impact = f"Every 1-point increase in exercise diversity reduces sarcopenia risk by 4-8%, whole-body coordination improves 25%" else: title = "运动多样性精确提升计划" description = f"当前多样性指数: {original_val:.1f} → 目标: {target_val:.1f}" if target_val > original_val: increase = target_val - original_val target_types = int(target_val) + 2 description += f"\n📈 需要增加 {increase:.1f} 的活动多样性" description += f"\n🎯 目标：每周进行 {target_types} 种不同类型的运动" # 运动类型分类和建议 description += f"\n🏃‍♂️ 推荐运动类型组合：" description += f"\n 有氧运动类 (2-3种)：" description += f"\n • 快走/慢跑 - 每周3-4次，每次30分钟" description += f"\n • 游泳/水中运动 - 每周2次，每次45分钟" description += f"\n • 骑自行车 - 每周2-3次，每次40分钟" description += f"\n 💪 力量训练类 (1-2种)：" description += f"\n • 器械训练 - 每周2-3次，每次45分钟" description += f"\n • 自重训练 - 每周3次，每次30分钟" description += f"\n • 弹力带训练 - 每周2次，每次25分钟" description += f"\n 🧘‍♀️ 柔韧平衡类 (1-2种)：" description += f"\n • 瑜伽 - 每周2-3次，每次60分钟" description += f"\n • 太极 - 每周2次，每次45分钟" description += f"\n • 普拉提 - 每周2次，每次50分钟" description += f"\n 🏐 趣味运动类 (1种)：" description += f"\n • 球类运动 - 每周1-2次" description += f"\n • 舞蹈 - 每周1-2次" description += f"\n • 爬山/徒步 - 每周1次" # 周计划示例 description += f"\n📅 多样性周计划示例：" description += f"\n 周一：力量训练 (45分钟)" description += f"\n 周二：快走 (30分钟) + 瑜伽 (30分钟)" description += f"\n 周三：游泳 (45分钟)" description += f"\n 周四：自重训练 (30分钟) + 太极 (30分钟)" description += f"\n 周五：骑车 (40分钟)" description += f"\n 周六：球类运动 (60分钟)" description += f"\n 周日：徒步/爬山 (90分钟)" # 渐进式增加建议 description += f"\n🔄 渐进式增加策略：" current_types = max(1, int(original_val)) for week in range(1, 5): week_types = min(current_types + week, target_types) description += f"\n 第{week}周：{week_types}种运动类型" target_change = f"活动多样性: {original_val:.1f} → {target_val:.1f}" expected_impact = f"运动多样性每提升1点，肌少症风险降低4-8%，全身协调性提升25%" elif feature == 'sedentary_minutes': if language == 'en': title = "Daily Sedentary Time Precision Control Plan" description = f"Current daily sedentary: {original_val:.0f}min → Target: {target_val:.0f}min" if original_val > target_val: decrease = original_val - target_val hours_decrease = decrease / 60 description += f"\n📉 Need to reduce daily sedentary time by {decrease:.0f} minutes ({hours_decrease:.1f} hours)" # Time-segmented reduction strategies description += f"\n🕐 Time-segmented reduction strategies:" if decrease <= 60: description += f"\n • Stand and move for 5 minutes every hour" description += f"\n • Stand or walk during phone meetings" description += f"\n • Stand for 10 minutes after meals" elif decrease <= 120: description += f"\n • Stand and move for 5-8 minutes every 45 minutes" description += f"\n • Take 15-20 minute walks during lunch break" description += f"\n • Do simple exercises while watching TV" description += f"\n • Use standing desk for 1-2 hours" else: description += f"\n • Stand and move for 5-10 minutes every 30 minutes" description += f"\n • Walk/bike to work instead of driving" description += f"\n • Use standing desk for 3-4 hours" description += f"\n • Actively take on tasks requiring movement" # Specific alternative activities description += f"\n🚶‍♂️ Sedentary replacement activities:" description += f"\n Work hours:" description += f"\n • Stand during meetings/phone calls" description += f"\n • Walk to colleagues' desks for communication" description += f"\n • Take stairs instead of elevators" description += f"\n • Walk farther for water/restroom breaks" description += f"\n Leisure time:" description += f"\n • Stretch while watching TV" description += f"\n • Do housework while listening to music" description += f"\n • Stand while reading/studying" description += f"\n • Take walks instead of scrolling phone" # Practical tools and reminders description += f"\n🛠️ Practical tools and reminders:" description += f"\n • Set {30 if decrease > 120 else 45 if decrease > 60 else 60}-minute sedentary reminders on phone" description += f"\n • Use standing desk or adjustable desk" description += f"\n • Wear fitness tracker to monitor sedentary time" description += f"\n • Place reminder notes on desk" # Progressive reduction plan description += f"\n📅 4-week progressive reduction plan:" for week in range(1, 5): week_target = original_val - (decrease * week / 4) daily_reduction = decrease * week / 4 description += f"\n Week {week}: Daily sedentary {week_target:.0f}min (reduce {daily_reduction:.0f}min)" else: title = "每日久坐时间精确控制计划" description = f"当前每日久坐: {original_val:.0f}分钟 → 目标: {target_val:.0f}分钟" if original_val > target_val: decrease = original_val - target_val hours_decrease = decrease / 60 description += f"\n📉 需要每天减少久坐 {decrease:.0f} 分钟 ({hours_decrease:.1f}小时)" # 分时段减少策略 description += f"\n🕐 分时段减少策略：" if decrease <= 60: description += f"\n • 每小时起身活动5分钟" description += f"\n • 电话会议时站立或走动" description += f"\n • 用餐后站立10分钟" elif decrease <= 120: description += f"\n • 每45分钟起身活动5-8分钟" description += f"\n • 午休时间散步15-20分钟" description += f"\n • 看电视时做简单运动" description += f"\n • 站立办公1-2小时" else: description += f"\n • 每30分钟起身活动5-10分钟" description += f"\n • 上下班步行/骑车代替开车" description += f"\n • 站立办公3-4小时" description += f"\n • 主动承担需要走动的任务" # 具体的替代活动 description += f"\n🚶‍♂️ 久坐替代活动：" description += f"\n 工作时间：" description += f"\n • 站立开会/打电话" description += f"\n • 走到同事桌前交流" description += f"\n • 爬楼梯代替电梯" description += f"\n • 远距离取水/上厕所" description += f"\n 休闲时间：" description += f"\n • 看电视时做拉伸" description += f"\n • 边听音乐边做家务" description += f"\n • 站立阅读/学习" description += f"\n • 散步代替刷手机" # 实用工具和提醒 description += f"\n🛠️ 实用工具和提醒：" description += f"\n • 手机设置每{30 if decrease > 120 else 45 if decrease > 60 else 60}分钟久坐提醒" description += f"\n • 使用站立办公桌或升降桌" description += f"\n • 佩戴智能手环监测久坐" description += f"\n • 在办公桌放置提醒标语" # 渐进式减少计划 description += f"\n📅 4周渐进减少计划：" for week in range(1, 5): week_target = original_val - (decrease * week / 4) daily_reduction = decrease * week / 4 description += f"\n 第{week}周：每日久坐{week_target:.0f}分钟 (减少{daily_reduction:.0f}分钟)" target_change = f"每日久坐时间: {original_val:.0f}分钟 → {target_val:.0f}分钟" expected_impact = f"每日久坐时间每减少60分钟，肌少症风险降低约5-9%" else: title = f"{feature}优化计划" description = f"当前值: {original_val:.3f} → 目标值: {target_val:.3f}" target_change = f"{feature}: {original_val:.3f} → {target_val:.3f}" expected_impact = "遵循专业指导，预期改善肌少症风险" return title, description, target_change, expected_impact def _get_recommendation_texts(self, language: str = "zh") -> Dict: """获取双语建议文本""" texts = { "zh": { "increase_vigorous": { "title": "增加高强度运动", "description": "每周至少进行75分钟高强度运动，如跑步、游泳或力量训练\n\n具体建议：\n• 跑步：每次20-30分钟，每周3次\n• 力量训练：每次45分钟，每周2次\n• 游泳：每次30分钟，每周2次", "priority": "高", "target_change": "高强度运动比例提升至0.3以上", "expected_impact": "显著改善肌肉力量和心肺功能" }, "increase_moderate": { "title": "增加中等强度运动", "description": "每周进行至少150分钟中等强度运动\n\n推荐活动：\n• 快走：每次30分钟，每周5次\n• 骑车：每次40分钟，每周3次\n• 太极：每次45分钟，每周2次", "priority": "中", "target_change": "中等强度运动增至200分钟/周", "expected_impact": "提升整体体能和肌肉耐力" }, "reduce_sedentary": { "title": "减少久坐时间", "description": "控制每日久坐时间，增加活动频率\n\n实施方案：\n• 每30分钟起身活动5分钟\n• 站立办公2-3小时\n• 步行会议和电话\n• 使用提醒工具", "priority": "高", "target_change": "久坐比例降至0.4以下", "expected_impact": "改善肌肉活性，降低肌少症风险" }, "weight_management": { "title": "体重管理", "description": "通过运动控制体重在健康范围\n\n目标设定：\n• BMI控制在18.5-24.9\n• 每周减重0.5-1公斤\n• 有氧运动300分钟/周\n• 力量训练3次/周", "priority": "中", "target_change": "BMI控制在18.5-24.9范围内", "expected_impact": "减轻关节负担，改善身体成分" }, "reduce_sedentary": { "title": "减少久坐时间", "description": "每小时起身活动5-10分钟，减少连续久坐时间", "priority": "高", "target_change": "久坐比例降至0.4以下", "expected_impact": "改善肌肉活性，降低肌少症风险" }, "increase_moderate": { "title": "增加中等强度运动", "description": "每周进行至少150分钟中等强度运动，如快走、骑车", "priority": "中", "target_change": "中等强度运动增至200分钟/周", "expected_impact": "提升整体体能和肌肉耐力" }, "weight_management": { "title": "体重管理", "description": "通过合理饮食和运动，维持健康体重", "priority": "中", "target_change": "BMI控制在18.5-24.9范围内", "expected_impact": "减轻关节负担，改善身体成分" }, "protein_intake": { "title": "增加蛋白质摄入", "description": "每日摄入1.2-1.6g/kg体重的优质蛋白质", "priority": "高", "target_change": "蛋白质摄入达到推荐标准", "expected_impact": "促进肌肉合成，维持肌肉质量" }, "regular_monitoring": { "title": "定期健康监测", "description": "每6个月进行一次肌肉质量和功能评估", "priority": "中", "target_change": "建立定期监测计划", "expected_impact": "早期发现肌肉质量变化，及时调整干预策略" }, "increase_diversity": { "title": "增加运动多样性", "description": "尝试多种不同类型的体力活动，如有氧运动、力量训练、柔韧性练习", "priority": "高", "target_change": "活动类型增加至4种以上", "expected_impact": "全面改善身体功能，降低SarcoII风险" }, "strength_training": { "title": "加强力量训练", "description": "每周进行2-3次力量训练，重点锻炼大肌群", "priority": "高", "target_change": "建立规律的力量训练计划", "expected_impact": "增强肌肉力量，改善身体功能" }, "increase_activity": { "title": "增加体力活动", "description": "每周至少进行150分钟中等强度或75分钟高强度运动", "priority": "高", "target_change": "达到WHO推荐的运动量标准", "expected_impact": "全面改善身体健康状况" }, "balanced_nutrition": { "title": "均衡营养摄入", "description": "确保充足的蛋白质、维生素D和钙的摄入", "priority": "中", "target_change": "建立健康的饮食习惯", "expected_impact": "支持肌肉健康和骨骼强度" } }, "en": { "increase_vigorous": { "title": "Increase High-Intensity Exercise", "description": "Engage in at least 75 minutes of vigorous exercise weekly\n\nSpecific recommendations:\n• Running: 20-30 minutes per session, 3 times weekly\n• Strength training: 45 minutes per session, 2 times weekly\n• Swimming: 30 minutes per session, 2 times weekly", "priority": "High", "target_change": "Increase vigorous exercise ratio to above 0.3", "expected_impact": "Significantly improve muscle strength and cardiopulmonary function" }, "reduce_sedentary": { "title": "Reduce Sedentary Time", "description": "Control daily sedentary time and increase activity frequency\n\nImplementation plan:\n• Stand and move for 5 minutes every 30 minutes\n• Use standing desk for 2-3 hours\n• Walking meetings and phone calls\n• Use reminder tools", "priority": "High", "target_change": "Reduce sedentary ratio to below 0.4", "expected_impact": "Improve muscle activity and reduce sarcopenia risk" }, "increase_moderate": { "title": "Increase Moderate-Intensity Exercise", "description": "Engage in at least 150 minutes of moderate exercise weekly\n\nRecommended activities:\n• Brisk walking: 30 minutes per session, 5 times weekly\n• Cycling: 40 minutes per session, 3 times weekly\n• Tai Chi: 45 minutes per session, 2 times weekly", "priority": "Medium", "target_change": "Increase moderate exercise to 200 minutes/week", "expected_impact": "Enhance overall fitness and muscle endurance" }, "weight_management": { "title": "Weight Management", "description": "Maintain healthy weight through balanced diet and exercise", "priority": "Medium", "target_change": "Keep BMI within 18.5-24.9 range", "expected_impact": "Reduce joint burden and improve body composition" }, "protein_intake": { "title": "Increase Protein Intake", "description": "Consume 1.2-1.6g/kg body weight of high-quality protein daily", "priority": "High", "target_change": "Meet recommended protein intake standards", "expected_impact": "Promote muscle synthesis and maintain muscle mass" }, "regular_monitoring": { "title": "Regular Health Monitoring", "description": "Conduct muscle mass and function assessment every 6 months", "priority": "Medium", "target_change": "Establish regular monitoring plan", "expected_impact": "Early detection of muscle mass changes, timely intervention adjustment" }, "increase_diversity": { "title": "Increase Exercise Diversity", "description": "Try various types of physical activities, such as aerobic exercise, strength training, and flexibility exercises", "priority": "High", "target_change": "Increase activity types to 4 or more", "expected_impact": "Comprehensively improve body function and reduce SarcoII risk" }, "strength_training": { "title": "Strengthen Resistance Training", "description": "Perform strength training 2-3 times per week, focusing on major muscle groups", "priority": "High", "target_change": "Establish regular strength training plan", "expected_impact": "Enhance muscle strength and improve body function" }, "increase_activity": { "title": "Increase Physical Activity", "description": "Engage in at least 150 minutes of moderate-intensity or 75 minutes of vigorous-intensity exercise weekly", "priority": "High", "target_change": "Meet WHO recommended physical activity standards", "expected_impact": "Comprehensively improve overall health" }, "balanced_nutrition": { "title": "Balanced Nutrition Intake", "description": "Ensure adequate intake of protein, vitamin D, and calcium", "priority": "Medium", "target_change": "Establish healthy eating habits", "expected_impact": "Support muscle health and bone strength" } } } return texts def _generate_sarcoI_rule_recommendations(self, user_dict: Dict, language: str = "zh") -> List[RecommendationItem]: """生成SarcoI基于规则的建议""" recommendations = [] texts = self._get_recommendation_texts(language) # 基于用户数据生成规则建议 bmi = user_dict.get('body_mass_index', 25) sedentary_ratio = user_dict.get('Activity_Sedentary_Ratio', 0.5) moderate_minutes = user_dict.get('Total_Moderate_Minutes_week', 150) vigorous_ratio = user_dict.get('Vigorous_MET_Ratio', 0.2) if vigorous_ratio < 0.2: rec = texts[language]["increase_vigorous"] recommendations.append(RecommendationItem( title=rec["title"], description=rec["description"], priority=rec["priority"], target_change=rec["target_change"], expected_impact=rec["expected_impact"] )) if moderate_minutes < 150: rec = texts[language]["increase_moderate"] recommendations.append(RecommendationItem( title=rec["title"], description=rec["description"], priority=rec["priority"], target_change=rec["target_change"], expected_impact=rec["expected_impact"] )) if bmi > 25: rec = texts[language]["weight_management"] recommendations.append(RecommendationItem( title=rec["title"], description=rec["description"], priority=rec["priority"], target_change=rec["target_change"], expected_impact=rec["expected_impact"] )) return recommendations def _generate_maintenance_recommendations(self, user_dict: Dict, model_type: str, language: str = "zh") -> List[RecommendationItem]: """ 为低风险用户生成维持性建议 Args: user_dict: 用户数据字典 model_type: 模型类型 ('sarcoI' 或 'sarcoII') Returns: 维持性建议列表 """ recommendations = [] # 基础维持建议 if language == 'en': recommendations.append(RecommendationItem( title="Maintain Current Health Status", description="Your current sarcopenia risk is low, continue maintaining your existing healthy lifestyle", priority="Medium", target_change="Maintain current exercise and dietary habits", expected_impact="Continue reducing sarcopenia risk, maintain muscle health" )) else: recommendations.append(RecommendationItem( title="保持当前健康状态", description="您当前的肌少症风险较低，建议继续保持现有的健康生活方式", priority="中", target_change="维持当前的运动和饮食习惯", expected_impact="持续降低肌少症风险，保持肌肉健康" )) # 基于当前活动水平的维持建议 if model_type == 'sarcoI': current_moderate = user_dict.get('Total_Moderate_Minutes_week', 150) current_vigorous = user_dict.get('Vigorous_MET_Ratio', 0.2) if current_moderate >= 150: if language == 'en': recommendations.append(RecommendationItem( title="Continue Moderate-Intensity Exercise", description=f"You currently engage in {current_moderate:.0f} minutes of moderate exercise weekly, continue maintaining this", priority="Medium", target_change="Maintain at least 150 minutes of moderate exercise weekly", expected_impact="Maintain cardiopulmonary function and muscle endurance" )) else: recommendations.append(RecommendationItem( title="继续中等强度运动", description=f"您当前每周进行{current_moderate:.0f}分钟中等强度运动，建议继续保持", priority="中", target_change="维持每周至少150分钟中等强度运动", expected_impact="保持心肺功能和肌肉耐力" )) if current_vigorous >= 0.2: if language == 'en': recommendations.append(RecommendationItem( title="Maintain High-Intensity Exercise", description="Your high-intensity exercise ratio is good, continue maintaining strength training and high-intensity exercise", priority="Medium", target_change="Maintain current high-intensity exercise level", expected_impact="Maintain muscle mass and strength" )) else: recommendations.append(RecommendationItem( title="保持高强度运动", description="您的高强度运动比例良好，建议继续保持力量训练和高强度运动", priority="中", target_change="维持当前的高强度运动水平", expected_impact="保持肌肉质量和力量" )) # 预防性建议 age = user_dict.get('age_years', 45) if age > 50: if language == 'en': recommendations.append(RecommendationItem( title="Age-Related Prevention", description="As you age, it's recommended to regularly monitor muscle mass and appropriately increase protein intake", priority="Medium", target_change="Conduct muscle mass assessment every 6 months", expected_impact="Early detection of muscle mass changes, timely intervention adjustment" )) else: recommendations.append(RecommendationItem( title="年龄相关预防", description="随着年龄增长，建议定期监测肌肉质量，适当增加蛋白质摄入", priority="中", target_change="每6个月进行一次肌肉质量评估", expected_impact="早期发现肌肉质量变化，及时调整干预策略" )) return recommendations def _generate_sarcoII_rule_recommendations(self, user_dict: Dict, language: str = "zh") -> List[RecommendationItem]: """生成SarcoII基于规则的建议""" recommendations = [] texts = self._get_recommendation_texts(language) diversity_index = user_dict.get('Activity_Diversity_Index', 2) sedentary_minutes = user_dict.get('sedentary_minutes', 480) sedentary_ratio = user_dict.get('Activity_Sedentary_Ratio', 0.5) if diversity_index < 3: rec = texts[language]["increase_diversity"] recommendations.append(RecommendationItem( title=rec["title"], description=rec["description"], priority=rec["priority"], target_change=rec["target_change"], expected_impact=rec["expected_impact"] )) if sedentary_minutes > 480: rec = texts[language]["reduce_sedentary"] recommendations.append(RecommendationItem( title=rec["title"], description=rec["description"], priority=rec["priority"], target_change=rec["target_change"], expected_impact=rec["expected_impact"] )) if sedentary_ratio > 0.6: if language == 'en': recommendations.append(RecommendationItem( title="Reduce Sedentary Activity Ratio", description="Increase standing and light activity time, reduce sedentary activity proportion", priority="Medium", target_change="Reduce sedentary activity ratio to below 50%", expected_impact="Improve daily activity level, promote muscle health" )) else: recommendations.append(RecommendationItem( title="降低久坐活动比例", description="增加站立和轻度活动时间，减少久坐活动占比", priority="中", target_change="久坐活动比例降至50%以下", expected_impact="提高日常活动水平，促进肌肉健康" )) return recommendations def _generate_fallback_recommendations(self, user_dict: Dict, risk_types: List[str] = None, language: str = "zh") -> List[RecommendationItem]: """生成降级建议""" recommendations = [] texts = self._get_recommendation_texts(language) # 基础建议 rec = texts[language]["increase_activity"] recommendations.append(RecommendationItem( title=rec["title"], description=rec["description"], priority=rec["priority"], target_change=rec["target_change"], expected_impact=rec["expected_impact"] )) rec = texts[language]["increase_diversity"] recommendations.append(RecommendationItem( title=rec["title"], description=rec["description"], priority=rec["priority"], target_change=rec["target_change"], expected_impact=rec["expected_impact"] )) # 基于用户数据的个性化建议 bmi = user_dict.get('body_mass_index', 25) if bmi > 25: if language == 'en': recommendations.append(RecommendationItem( title="Weight Management", description="Control weight through balanced diet and exercise, target BMI between 18.5-24.9", priority="Medium", target_change=f"Reduce weight to BMI below {bmi-2:.1f}", expected_impact="Reduce sarcopenia risk, improve overall health" )) else: recommendations.append(RecommendationItem( title="体重管理", description="通过合理饮食和运动控制体重，目标BMI在18.5-24.9之间", priority="中", target_change=f"减重至BMI {bmi-2:.1f}以下", expected_impact="降低肌少症风险，改善整体健康" )) sedentary_minutes = user_dict.get('sedentary_minutes', 480) if sedentary_minutes > 480: # 超过8小时 if language == 'en': recommendations.append(RecommendationItem( title="Reduce Sedentary Time", description="Stand and move for 5-10 minutes every hour, reduce continuous sitting time", priority="Medium", target_change="Control daily sedentary time within 8 hours", expected_impact="Improve blood circulation, maintain muscle vitality" )) else: recommendations.append(RecommendationItem( title="减少久坐时间", description="每小时起身活动5-10分钟，减少连续久坐时间", priority="中", target_change="每日久坐时间控制在8小时以内", expected_impact="改善血液循环，维持肌肉活力" )) return recommendations def _generate_basic_recommendations(self, language: str = "zh") -> List[RecommendationItem]: """生成基础建议""" if language == 'en': return [ RecommendationItem( title="Consult Professional Doctor", description="Seek professional medical guidance and conduct detailed health assessment", priority="High", target_change="Obtain personalized medical advice", expected_impact="Develop scientific health management plan" ), RecommendationItem( title="Regular Exercise", description="Maintain regular physical activity, including aerobic exercise and strength training", priority="Medium", target_change="Establish long-term exercise habits", expected_impact="Maintain and improve physical health" ) ] else: return [ RecommendationItem( title="咨询专业医生", description="建议寻求专业医疗指导，进行详细的健康评估", priority="高", target_change="获得个性化医疗建议", expected_impact="制定科学的健康管理方案" ), RecommendationItem( title="规律运动", description="保持规律的体力活动，包括有氧运动和力量训练", priority="中", target_change="建立长期运动习惯", expected_impact="维持和改善身体健康" ) ] def _generate_priority_actions(self, sarcoI_recs: List[RecommendationItem], sarcoII_recs: List[RecommendationItem]) -> List[str]: """生成优先级行动列表""" actions = [] # 收集高优先级建议 for rec in sarcoI_recs + sarcoII_recs: if rec.priority == "高": actions.append(rec.title) # 如果没有高优先级建议，添加基础建议 if not actions: actions = ["增加体力活动", "改善生活方式", "定期健康检查"] return actions[:3] # 最多返回3个优先行动 def _generate_target_metrics(self, user_dict: Dict, sarcoI_recs: List[RecommendationItem], sarcoII_recs: List[RecommendationItem]) -> Dict[str, Any]: """生成目标指标""" metrics = {} # 基于用户当前状态和建议生成目标 current_bmi = user_dict.get('body_mass_index', 25) current_activity = user_dict.get('Total_Moderate_Minutes_week', 150) if current_bmi > 25: metrics['目标BMI'] = f"{current_bmi - 2:.1f} - {current_bmi - 1:.1f}" if current_activity < 150: metrics['每周运动时间'] = "150分钟以上" metrics['评估周期'] = "4-6周" metrics['下次检查'] = "3个月后" return metrics # 创建全局建议服务实例 advisory_service = AdvisoryService()