颠覆“年轻就要拼命拼”，计算健康损耗与收益，颠覆透支身体，输出可持续奋斗模型。

可持续奋斗模型：健康与收益的智能平衡系统

一、实际应用场景描述

场景：互联网/创业/高压行业的工作者

- 28岁程序员连续加班3个月，项目结束后发现严重失眠、脱发、胃炎

- 25岁创业者为了融资，每天工作16小时，3年后公司成功了，但身体垮了

- 30岁产品经理为了晋升，长期熬夜赶项目，体检发现多项指标异常

现状：社会鼓吹"年轻就要拼命拼"，导致年轻人用健康换短期收益，形成"35岁危机"的恶性循环。

二、引入痛点

1. 健康债务隐形化：身体损伤是渐进的，年轻时察觉不到，积累到临界点爆发

2. 收益计算片面化：只算金钱和职位，不算健康成本和机会成本

3. 社会压力绑架："不拼就是不上进"的道德枷锁

4. 缺乏决策框架：不知道什么时候该冲刺，什么时候该休整

三、核心逻辑讲解

可持续奋斗模型三大支柱

┌─────────────────────────────────────────────────────────┐

│ 可持续奋斗模型 │

├─────────────┬─────────────┬─────────────┤

│ 健康损耗器 │ 收益计算器 │ 平衡决策器 │

├─────────────┼─────────────┼─────────────┤

│ 量化健康成本 │ 多维收益评估 │ 动态平衡策略 │

│ 预测健康风险 │ 计算净现值 │ 优化奋斗节奏 │

│ 健康状态追踪 │ 长期价值分析 │ 防透支机制 │

└─────────────┴─────────────┴─────────────┘

核心算法：

- 健康损耗指数 = (工作强度×持续时间) × 个人健康系数 - 恢复能力

- 综合收益 = 经济收益 + 成长收益 + 健康折损成本

- 奋斗可持续性 = 健康储备 / 健康损耗率

四、代码模块化实现

项目结构

sustainable_struggle/

├── core/

│ ├── __init__.py

│ ├── health_depletion.py # 健康损耗器

│ ├── benefit_calculator.py # 收益计算器

│ └── balance_decider.py # 平衡决策器

├── models/

│ ├── __init__.py

│ └── work_life.py # 工作生活数据模型

├── utils/

│ ├── __init__.py

│ └── decision_helpers.py # 决策辅助工具

├── main.py # 主程序入口

├── config.py # 配置文件

└── README.md # 项目说明

1. 配置文件 (config.py)

"""

配置文件：定义可持续奋斗模型的核心参数

"""

from dataclasses import dataclass, field

from typing import Dict, List, Tuple

from enum import Enum

class HealthDomain(Enum):

"""健康领域分类"""

PHYSICAL = "physical" # 身体健康

MENTAL = "mental" # 心理健康

COGNITIVE = "cognitive" # 认知功能

EMOTIONAL = "emotional" # 情绪健康

SOCIAL = "social" # 社交健康

@dataclass

class HealthConfig:

"""健康相关配置"""

# 基础健康参数

BASE_HEALTH_RESERVE: float = 100.0 # 基础健康储备

DAILY_RECOVERY_RATE: float = 2.0 # 每日自然恢复率

WEEKLY_RECOVERY_BONUS: float = 5.0 # 周末恢复加成

# 工作强度影响系数

INTENSITY_TO_PHYSICAL: float = 0.8 # 工作强度对身体健康的影响

INTENSITY_TO_MENTAL: float = 1.0 # 工作强度对心理健康的影响

INTENSITY_TO_COGNITIVE: float = 0.6 # 工作强度对认知功能的影响

INTENSITY_TO_EMOTIONAL: float = 0.9 # 工作强度对情绪健康的影响

INTENSITY_TO_SOCIAL: float = 0.4 # 工作强度对社交健康的影响

# 年龄修正因子（不同年龄的恢复能力和耐受度不同）

AGE_FACTORS: Dict[int, float] = field(default_factory=lambda: {

20: 1.2, # 20岁恢复能力强

25: 1.1,

30: 1.0, # 基准

35: 0.9,

40: 0.8,

45: 0.7,

50: 0.6

})

# 健康风险阈值

CRITICAL_HEALTH_THRESHOLD: float = 30.0 # 健康储备临界值

WARNING_HEALTH_THRESHOLD: float = 50.0 # 健康储备警告值

# 恢复活动效果

RECOVERY_ACTIVITIES: Dict[str, float] = field(default_factory=lambda: {

"quality_sleep": 8.0, # 优质睡眠

"regular_exercise": 6.0, # 规律运动

"healthy_diet": 4.0, # 健康饮食

"meditation": 5.0, # 冥想放松

"social_connection": 3.0, # 社交连接

"hobbies": 4.0, # 兴趣爱好

"vacation": 12.0, # 度假休息

"medical_checkup": 2.0 # 医疗检查

})

@dataclass

class WorkConfig:

"""工作相关配置"""

# 工作时长影响

HOURS_BASELINE: float = 8.0 # 基准工作时长

HOURS_OVERLOAD_PENALTY: float = 1.5 # 超时工作惩罚系数

HOURS_MAX_SUSTAINABLE: float = 10.0 # 最大可持续工作时长

# 工作强度等级

INTENSITY_LEVELS: Dict[str, float] = field(default_factory=lambda: {

"low": 0.3,

"moderate": 0.5,

"high": 0.7,

"extreme": 0.9,

"crunch": 1.0

})

# 项目周期影响

SPRINT_DURATION_WEEKS: int = 2 # 冲刺周期

SPRINT_RECOVERY_WEEKS: int = 1 # 冲刺后恢复周数

# 职业阶段参数

CAREER_STAGES: Dict[str, Dict] = field(default_factory=lambda: {

"entry": {

"growth_multiplier": 1.5,

"health_tolerance": 0.8,

"max_sprint_cycles": 3

},

"mid": {

"growth_multiplier": 1.2,

"health_tolerance": 0.6,

"max_sprint_cycles": 2

},

"senior": {

"growth_multiplier": 1.0,

"health_tolerance": 0.4,

"max_sprint_cycles": 1

}

})

@dataclass

class BenefitConfig:

"""收益相关配置"""

# 收益类型权重

ECONOMIC_WEIGHT: float = 0.4 # 经济收益权重

GROWTH_WEIGHT: float = 0.3 # 成长收益权重

NETWORK_WEIGHT: float = 0.2 # 人脉收益权重

REPUTATION_WEIGHT: float = 0.1 # 声誉收益权重

# 时间价值参数

DISCOUNT_RATE: float = 0.05 # 年折现率

LONG_TERM_BENEFIT_PERIOD: int = 10 # 长期收益计算期（年）

# 健康成本折算

HEALTH_COST_RATIO: float = 0.3 # 健康成本占收益的比例

FUTURE_HEALTH_PENALTY: float = 0.5 # 未来健康问题的折现比例

# 健康状态描述

HEALTH_STATUS_DESCRIPTIONS = {

"excellent": "健康状态极佳，可承受高强度工作",

"good": "健康状态良好，适合持续奋斗",

"fair": "健康状态一般，需要注意恢复",

"poor": "健康状态较差，建议减少工作强度",

"critical": "健康状态危急，必须立即调整"

}

# 奋斗建议

STRUGGLE_ADVICE = {

"sprint": "可以进行短期冲刺，但需安排充分恢复",

"maintain": "保持当前节奏，注重健康管理",

"decelerate": "建议减速，增加恢复时间",

"pause": "建议暂停高强度工作，专注恢复",

"restructure": "需要重新规划工作生活平衡"

}

2. 数据模型 (models/work_life.py)

"""

工作生活数据模型：存储和管理工作与健康数据

"""

from dataclasses import dataclass, field

from datetime import datetime, timedelta

from typing import List, Optional, Dict

from enum import Enum

import uuid

class WorkIntensity(Enum):

"""工作强度枚举"""

LOW = "low"

MODERATE = "moderate"

HIGH = "high"

EXTREME = "extreme"

CRUNCH = "crunch"

class CareerStage(Enum):

"""职业阶段枚举"""

ENTRY = "entry"

MID = "mid"

SENIOR = "senior"

LEADER = "leader"

@dataclass

class DailyWorkRecord:

"""

每日工作记录

用于追踪日常工作对健康的影响

"""

record_id: str = field(default_factory=lambda: str(uuid.uuid4()))

date: datetime = field(default_factory=datetime.now)

hours_worked: float = 8.0 # 工作时长

intensity: WorkIntensity = WorkIntensity.MODERATE # 工作强度

meeting_hours: float = 2.0 # 会议时长

deep_work_hours: float = 4.0 # 深度工作时长

interruptions: int = 10 # 打断次数

commute_time: float = 1.0 # 通勤时间

overtime: bool = False # 是否加班

overtime_hours: float = 0.0 # 加班时长

def calculate_daily_intensity_score(self, config: 'WorkConfig') -> float:

"""

计算当日综合强度分数

Returns:

float: 强度分数(0-1)

"""

base_score = config.INTENSITY_LEVELS.get(self.intensity.value, 0.5)

# 时长加成

if self.hours_worked > config.HOURS_BASELINE:

overtime_factor = (self.hours_worked - config.HOURS_BASELINE) / 2

base_score += overtime_factor * config.HOURS_OVERLOAD_PENALTY

# 加班加成

if self.overtime:

base_score += 0.1 * min(self.overtime_hours, 4)

# 打断惩罚（影响认知负荷）

interruption_penalty = min(self.interruptions / 50, 0.15)

base_score -= interruption_penalty

return min(max(base_score, 0.0), 1.0)

@dataclass

class RecoveryActivity:

"""

恢复活动记录

用于记录和追踪健康恢复行为

"""

activity_id: str = field(default_factory=lambda: str(uuid.uuid4()))

activity_type: str = "" # 活动类型

duration_minutes: int = 30 # 持续时间（分钟）

intensity_level: str = "medium" # 活动强度

performed_at: datetime = field(default_factory=datetime.now)

effectiveness_rating: float = 0.5 # 主观有效性评分(0-1)

notes: str = "" # 备注

def calculate_recovery_points(self, config: 'HealthConfig') -> float:

"""

计算恢复点数

Returns:

float: 恢复点数

"""

base_points = config.RECOVERY_ACTIVITIES.get(self.activity_type, 2.0)

# 时长加成

duration_bonus = min(self.duration_minutes / 60, 2.0)

# 有效性加成

effectiveness_multiplier = 0.5 + (self.effectiveness_rating * 0.5)

total_points = base_points * duration_bonus * effectiveness_multiplier

return round(total_points, 2)

@dataclass

class ProjectCycle:

"""

项目周期记录

用于追踪项目对工作健康的影响

"""

cycle_id: str = field(default_factory=lambda: str(uuid.uuid4()))

project_name: str = ""

start_date: datetime = field(default_factory=datetime.now)

end_date: Optional[datetime] = None

planned_duration_weeks: int = 8

actual_duration_weeks: float = 8.0

peak_intensity: WorkIntensity = WorkIntensity.HIGH

crunch_periods: List[Dict] = field(default_factory=list) # 冲刺期记录

health_before: float = 100.0

health_after: float = 100.0

performance_metrics: Dict = field(default_factory=dict) # 绩效指标

def add_crunch_period(self, start_date: datetime, end_date: datetime, intensity: WorkIntensity):

"""添加冲刺期记录"""

self.crunch_periods.append({

"start": start_date.isoformat(),

"end": end_date.isoformat(),

"intensity": intensity.value,

"duration_days": (end_date - start_date).days

})

@dataclass

class HealthProfile:

"""

健康档案

记录个人的基础健康状态和特性

"""

profile_id: str = field(default_factory=lambda: str(uuid.uuid4()))

age: int = 25

gender: str = "unspecified"

base_health_reserve: float = 100.0

health_conditions: List[str] = field(default_factory=list) # 既有健康状况

genetic_factors: List[str] = field(default_factory=list) # 遗传因素

lifestyle_factors: Dict[str, float] = field(default_factory=dict) # 生活方式因素

stress_tolerance: float = 0.5 # 压力耐受度(0-1)

recovery_ability: float = 0.5 # 恢复能力(0-1)

def get_age_factor(self, config: 'HealthConfig') -> float:

"""获取年龄修正因子"""

age_factors = config.AGE_FACTORS

# 找到最接近的年龄段

closest_age = min(age_factors.keys(), key=lambda x: abs(x - self.age))

return age_factors[closest_age]

def get_adjusted_health_reserve(self, config: 'HealthConfig') -> float:

"""计算调整后的健康储备"""

age_factor = self.get_age_factor(config)

conditions_penalty = len(self.health_conditions) * 5.0

return (self.base_health_reserve * age_factor) - conditions_penalty

@dataclass

class WorkLifeBalance:

"""

工作生活平衡主数据类

整合所有工作、健康、收益数据

"""

user_id: str = field(default_factory=lambda: str(uuid.uuid4()))

name: str = ""

career_stage: CareerStage = CareerStage.ENTRY

health_profile: HealthProfile = field(default_factory=HealthProfile)

# 时间线记录

daily_records: List[DailyWorkRecord] = field(default_factory=list)

recovery_activities: List[RecoveryActivity] = field(default_factory=list)

project_cycles: List[ProjectCycle] = field(default_factory=list)

# 累积指标

cumulative_health_damage: float = 0.0

cumulative_benefits: Dict[str, float] = field(default_factory=lambda: {

"economic": 0.0,

"growth": 0.0,

"network": 0.0,

"reputation": 0.0

})

# 当前状态

current_health_reserve: float = 100.0

current_balance_score: float = 75.0

# 时间戳

created_at: datetime = field(default_factory=datetime.now)

last_updated: datetime = field(default_factory=datetime.now)

def add_daily_record(self, record: DailyWorkRecord):

"""添加每日工作记录"""

self.daily_records.append(record)

self.last_updated = datetime.now()

def add_recovery_activity(self, activity: RecoveryActivity):

"""添加恢复活动记录"""

self.recovery_activities.append(activity)

self.last_updated = datetime.now()

def add_project_cycle(self, cycle: ProjectCycle):

"""添加项目周期记录"""

self.project_cycles.append(cycle)

self.last_updated = datetime.now()

def get_recent_records(self, days: int = 30) -> List[DailyWorkRecord]:

"""获取最近N天的记录"""

cutoff_date = datetime.now() - timedelta(days=days)

return [r for r in self.daily_records if r.date >= cutoff_date]

def get_weekly_summary(self, week_start: datetime) -> Dict:

"""获取一周的汇总数据"""

week_end = week_start + timedelta(days=7)

weekly_records = [

r for r in self.daily_records

if week_start <= r.date < week_end

]

if not weekly_records:

return {"status": "no_data"}

total_hours = sum(r.hours_worked for r in weekly_records)

avg_intensity = statistics.mean(

[r.calculate_daily_intensity_score(WorkConfig()) for r in weekly_records]

)

overtime_days = sum(1 for r in weekly_records if r.overtime)

return {

"week_start": week_start.isoformat(),

"total_hours": round(total_hours, 1),

"average_intensity": round(avg_intensity, 2),

"overtime_days": overtime_days,

"record_count": len(weekly_records)

}

# 导入statistics模块

import statistics

3. 健康损耗器 (core/health_depletion.py)

"""

健康损耗器模块

负责量化工作对健康的影响，预测健康风险

"""

from dataclasses import dataclass, field

from datetime import datetime, timedelta

from typing import List, Dict, Tuple, Optional

from . import config

import statistics

import math

@dataclass

class HealthDepletionResult:

"""

健康损耗计算结果

包含详细的损耗分析和风险评估

"""

total_depletion: float # 总健康损耗

domain_depletion: Dict[str, float] # 各领域损耗

depletion_rate: float # 损耗速率

health_projection: List[float] # 健康预测曲线

risk_level: str # 风险等级

critical_domains: List[str] # 关键风险领域

assessment_time: datetime = field(default_factory=datetime.now)

def to_dict(self) -> Dict:

return {

"total_depletion": round(self.total_depletion, 2),

"domain_depletion": {k: round(v, 2) for k, v in self.domain_depletion.items()},

"depletion_rate": round(self.depletion_rate, 3),

"health_projection": [round(p, 2) for p in self.health_projection],

"risk_level": self.risk_level,

"critical_domains": self.critical_domains,

"assessment_time": self.assessment_time.isoformat()

}

class HealthDepletionCalculator:

"""

健康损耗计算器主类

核心功能：

1. 计算工作对健康各领域的损耗

2. 预测未来健康趋势

3. 识别高风险领域

4. 评估健康储备消耗速度

"""

def __init__(self, health_config: config.HealthConfig = None):

self.config = health_config or config.HealthConfig()

self.calculation_history: List[HealthDepletionResult] = []

def calculate_daily_depletion(

self,

work_record: 'models.DailyWorkRecord',

health_profile: 'models.HealthProfile'

) -> Dict[str, float]:

"""

计算单日工作对健康各领域的损耗

Args:

work_record: 当日工作记录

health_profile: 健康档案

Returns:

Dict: 各领域损耗值

"""

intensity_score = work_record.calculate_daily_intensity_score(WorkConfig())

# 获取个人健康修正因子

age_factor = health_profile.get_age_factor(self.config)

stress_tolerance = health_profile.stress_tolerance

recovery_ability = health_profile.recovery_ability

# 计算基础损耗

base_daily_damage = intensity_score * 0.5

# 根据个人特性调整

personal_modifier = (stress_tolerance * 0.3 + recovery_ability * 0.2) * age_factor

# 计算各领域损耗

domain_damage = {

config.HealthDomain.PHYSICAL.value:

base_daily_damage * self.config.INTENSITY_TO_PHYSICAL * (2 - personal_modifier),

config.HealthDomain.MENTAL.value:

base_daily_damage * self.config.INTENSITY_TO_MENTAL * (2 - personal_modifier),

config.HealthDomain.COGNITIVE.value:

base_daily_damage * self.config.INTENSITY_TO_COGNITIVE * (2 - personal_modifier),

config.HealthDomain.EMOTIONAL.value:

base_daily_damage * self.config.INTENSITY_TO_EMOTIONAL * (2 - personal_modifier),

config.HealthDomain.SOCIAL.value:

base_daily_damage * self.config.INTENSITY_TO_SOCIAL * (2 - personal_modifier)

}

# 加班额外损耗

if work_record.overtime:

overtime_multiplier = 1.0 + (work_record.overtime_hours * 0.1)

for domain in domain_damage:

domain_damage[domain] *= overtime_multiplier

return {k: max(0.0, v) for k, v in domain_damage.items()}

def calculate_period_depletion(

self,

work_life_balance: 'models.WorkLifeBalance',

period_days: int = 30

) -> HealthDepletionResult:

"""

计算指定时间段内的健康损耗

Args:

work_life_balance: 工作生活平衡数据

period_days: 计算周期（天）

Returns:

HealthDepletionResult: 损耗计算结果

"""

recent_records = work_life_balance.get_recent_records(period_days)

if not recent_records:

return self._create_empty_result("无工作记录数据")

# 1. 计算总损耗

total_domain_damage = {

config.HealthDomain.PHYSICAL.value: 0.0,

config.HealthDomain.MENTAL.value: 0.0,

config.HealthDomain.COGNITIVE.value: 0.0,

config.HealthDomain.EMOTIONAL.value: 0.0,

config.HealthDomain.SOCIAL.value: 0.0

}

daily_damages = []

for record in recent_records:

daily_damage = self.calculate_daily_depletion(

record,

work_life_balance.health_profile

)

daily_damages.append(daily_damage)

for domain, damage in daily_damage.items():

total_domain_damage[domain] += damage

# 2. 计算平均每日损耗

num_days = len(recent_records)

avg_daily_damage = {

domain: total / num_days

for domain, total in total_domain_damage.items()

}

# 3. 计算损耗速率

depletion_rate = sum(avg_daily_damage.values()) / 5 # 平均到5个领域

# 4. 生成健康预测

health_projection = self._project_health_trend(

work_life_balance.current_health_reserve,

depletion_rate,

work_life_balance.health_profile

)

# 5. 确定风险等级

risk_level = self._assess_risk_level(

work_life_balance.current_health_reserve,

total_domain_damage,

depletion_rate

)

# 6. 识别关键风险领域

critical_domains = self._identify_critical_domains(

total_domain_damage,

num_days

)

result = HealthDepletionResult(

total_depletion=sum(total_domain_damage.values()),

domain_depletion=total_domain_damage,

depletion_rate=depletion_rate,

health_projection=health_projection,

risk_level=risk_level,

critical_domains=critical_domains

)

self.calculation_history.append(result)

return result

def _project_health_trend(

self,

current_health: float,

depletion_rate: float,

health_profile: 'models.HealthProfile'

) -> List[float]:

"""

预测未来健康趋势

考虑自然恢复和持续损耗的平衡

"""

projection = [current_health]

recovery_rate = self.config.DAILY_RECOVERY_RATE * health_profile.get_age_factor(self.config)

# 预测未来60天

for day in range(1, 61):

# 周末恢复加成

if day % 7 in [0, 6]: # 周末

net_change = -depletion_rate + recovery_rate + self.config.WEEKLY_RECOVERY_BONUS

else:

net_change = -depletion_rate + recovery_rate

new_health = max(0.0, projection[-1] + net_change)

projection.append(new_health)

# 如果健康耗尽，停止预测

if new_health <= 0:

break

return projection[:30] # 返回前30天预测

def _assess_risk_level(

self,

current_heal

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！