"""掌握度计算服务（加权滑动平均 + 遗忘曲线）. 算法对齐 02-architecture-design.md §9： - WEIGHTED_MOVING_AVG：w_i = 0.6^i，归一化（权重从最近到最远递减） - FORGETTING_CURVE：基于遗忘曲线的 max 叠加（P5+ 启用）， half_life = 30 days，越久未练习掌握度越衰减. 输入：学生指定知识点的历史成绩序列（按时间倒序） 输出：mastery_level (0.0-1.0) 副作用： - 写 mastery_snapshot 表（upsert_mastery_snapshot） - 发布 mastery.updated 事件（Outbox 豁免） 降级策略： - ClickHouse 不可达：跳过计算（返回 None） - Kafka 不可达：计算继续，事件发布静默失败 """ from datetime import UTC, datetime from typing import Any import structlog from .config import settings from .repository import clickhouse_repository, kafka_producer logger = structlog.get_logger(__name__) def _weighted_moving_avg(scores: list[dict[str, Any]]) -> float | None: """加权滑动平均算法. 输入：按时间倒序的成绩列表（最近在前），元素含 score 字段 输出：归一化的 mastery_level (0.0-1.0)，样本不足返回 None. 权重：w_i = decay_base ^ i（i 从 0 开始，最近 attempt 权重最大） """ if not scores: return None # 截取最近 N 次（window_size） window = scores[: settings.mastery_window_size] if len(window) < settings.mastery_min_samples: return None decay = settings.mastery_decay_base total_weight = 0.0 weighted_sum = 0.0 for i, record in enumerate(window): # 成绩归一化到 0.0-1.0（假设原始分 0-100） raw_score = record.get("score", 0.0) normalized = min(max(raw_score / 100.0, 0.0), 1.0) weight = decay**i weighted_sum += normalized * weight total_weight += weight if total_weight <= 0: return None mastery = weighted_sum / total_weight return round(max(0.0, min(1.0, mastery)), 4) def _forgetting_curve_decay( mastery: float, last_attempt_at: datetime | None, now: datetime | None = None, ) -> float: """遗忘曲线衰减（P5+ 启用，对齐 02 §9）. 距离上次练习越久，mastery 越衰减： decayed = mastery * exp(-ln(2) * days_since / half_life) half_life = 30 days（settings.mastery_forgetting_half_life_days）. """ if last_attempt_at is None: return mastery now = now or datetime.now(UTC) if last_attempt_at.tzinfo is None: last_attempt_at = last_attempt_at.replace(tzinfo=UTC) days_since = (now - last_attempt_at).total_seconds() / 86400 if days_since <= 0: return mastery half_life = settings.mastery_forgetting_half_life_days if half_life <= 0: return mastery import math decay_factor = math.exp(-math.log(2) * days_since / half_life) decayed = mastery * decay_factor return round(max(0.0, min(1.0, decayed)), 4) def classify_mastery_label(level: float) -> str: """掌握度三档分类.""" if level >= 0.8: return "mastered" if level >= 0.4: return "progressing" return "weak" async def calculate_mastery( student_id: str, knowledge_point_id: str, subject_id: str = "", ) -> dict[str, Any] | None: """计算学生指定知识点的掌握度. 流程： 1. 查询学生该知识点的历史成绩（按时间倒序） 2. 加权滑动平均 → mastery_level 3. 遗忘曲线衰减 → 最终 mastery_level 4. 写 mastery_snapshot 表 5. 发布 mastery.updated 事件（Outbox 豁免） 返回： { "student_id": ..., "knowledge_point_id": ..., "subject_id": ..., "mastery_level": 0.0-1.0, "previous_level": 0.0-1.0 or None, "mastery_label": "mastered" / "progressing" / "weak", "degraded": bool, } ClickHouse 不可达时返回降级骨架（degraded=true）. """ # 1. 查询历史成绩 scores = await clickhouse_repository.query_student_scores_by_kp( student_id=student_id, knowledge_point_id=knowledge_point_id, limit=settings.mastery_window_size * 2, # 多取一倍容错 ) if scores is None: # ClickHouse 不可达降级 logger.warning( "mastery_calc_clickhouse_unavailable_degraded", student_id=student_id, knowledge_point_id=knowledge_point_id, ) return { "student_id": student_id, "knowledge_point_id": knowledge_point_id, "subject_id": subject_id, "mastery_level": 0.0, "previous_level": None, "mastery_label": "weak", "degraded": True, "degraded_reason": "clickhouse_unavailable", } # 2. 加权滑动平均 mastery = _weighted_moving_avg(scores) if mastery is None: # 样本不足，返回默认值 return { "student_id": student_id, "knowledge_point_id": knowledge_point_id, "subject_id": subject_id, "mastery_level": 0.0, "previous_level": None, "mastery_label": "weak", "degraded": False, "degraded_reason": "insufficient_samples", } # 3. 遗忘曲线衰减（基于最近一次成绩的时间） last_attempt_at = None if scores: last_attempt_at = scores[0].get("timestamp") mastery_final = _forgetting_curve_decay(mastery, last_attempt_at) # 4. 查询上一次 mastery（用于事件对比） previous_snapshot = await clickhouse_repository.query_mastery_snapshot( student_id=student_id, subject_id=subject_id, ) previous_level: float | None = None if previous_snapshot is not None: for kp in previous_snapshot.get("knowledgePoints", []): if kp.get("knowledge_point_id") == knowledge_point_id: previous_level = kp.get("mastery_level") break # 5. 写 mastery_snapshot 表 calculated_at = datetime.now(UTC) write_ok = await clickhouse_repository.upsert_mastery_snapshot( student_id=student_id, knowledge_point_id=knowledge_point_id, subject_id=subject_id, mastery_level=mastery_final, calculated_at=calculated_at, calculation_method="weighted_moving_avg", ) if not write_ok: logger.warning( "mastery_snapshot_write_failed_degraded", student_id=student_id, knowledge_point_id=knowledge_point_id, ) # 6. 发布 mastery.updated 事件（Outbox 豁免） published = await kafka_producer.publish_mastery_updated( student_id=student_id, knowledge_point_id=knowledge_point_id, mastery_level=mastery_final, previous_level=previous_level if previous_level is not None else 0.0, ) result = { "student_id": student_id, "knowledge_point_id": knowledge_point_id, "subject_id": subject_id, "mastery_level": mastery_final, "previous_level": previous_level, "mastery_label": classify_mastery_label(mastery_final), "degraded": not write_ok, "degraded_reason": "snapshot_write_failed" if not write_ok else "", "event_published": published, } logger.info( "mastery_calculated", student_id=student_id, knowledge_point_id=knowledge_point_id, mastery_level=mastery_final, previous_level=previous_level, label=result["mastery_label"], published=published, ) return result async def batch_calculate_mastery( student_id: str, knowledge_point_ids: list[str], subject_id: str = "", ) -> list[dict[str, Any]]: """批量计算学生多个知识点的掌握度.""" results: list[dict[str, Any]] = [] for kp_id in knowledge_point_ids: result = await calculate_mastery( student_id=student_id, knowledge_point_id=kp_id, subject_id=subject_id, ) if result is not None: results.append(result) return results