Spaces:
Sleeping
Sleeping
| """ | |
| 建议模型服务 | |
| 高精确率模型 + 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() |