train.py

"""학습 트리거 + Gradient 계산 (PoC 단순화 버전)
=================================================

SPEC §1 기능 6 — *"학습 조건 도달 시 gradient 생성"* 의 데모/검증용 구현.

설계 의도
---------
- 현재 분류기 `rule-v1` 은 **선형 점수 모델**:  ``score = Σ w_e · x_e``
  (x_e = entity 또는 keyword 의 본문 내 개수)
- "학습" 의 가장 단순하고 의미있는 형태 = **선형 회귀의 1-step gradient descent**.
- 사용자가 부여한 등급을 *목표 점수* 로 매핑한 뒤, 잔차 × feature 로 가중치 업데이트.
- 결과 = **proposed weight deltas** + 새 가중치 + 적용 전후 SSE / 정확도.
- 이 결과를 **그대로 ENTITY_WEIGHTS / GRADE_KEYWORDS 에 반영하면 모델 핫스왑 완료**.

학습 조건 (SPEC §9 의 단순화)
-----------------------------
PoC 데모 편의를 위해 임계값을 낮췄다 — 시연 시 5~10 건이면 구동 가능:
  - 누적 레이블 ≥ 10
  - 등급별 최소 ≥ 3 (C, S, O 각각)
  - 갭(gap > 0) ≥ 3
  - 모두 만족하면 ready=True

Gradient 정의 (선형 회귀)
-------------------------
각 결정에 대해:
  target  = TARGET_SCORE[user_grade]   # O→1.0, S→3.5, C→6.5 (각 등급 밴드 중앙)
  pred    = Σ w_f · x_f                  # 현재 분류기가 부여한 점수
  resid   = target - pred
  Δw_f   += η · resid · x_f / (1 + Σ x_f)

학습 가중치 (SPEC §10.2):  sample_weight = 1 + 1.5 · gap
→ 갭 큰 샘플일수록 grad 에 더 크게 기여. 데이터 적은 PoC 에서 핵심.

η (학습률) = 0.05 (기본). 큰 deltas 누적 방지를 위한 max_step clip.
"""
from __future__ import annotations

from datetime import datetime, timezone
from typing import Iterable

import classifier  # ENTITY_WEIGHTS / GRADE_KEYWORDS 의 *원본* (in-memory) 참조

# ---------------------------------------------------------------------------
# 학습 트리거 조건 (PoC 데모용 — SPEC §9 의 완화 버전)
# ---------------------------------------------------------------------------
MIN_LABELED        = 10
MIN_PER_GRADE      = 3
MIN_GAP_DECISIONS  = 3

# Gradient hyperparameters
LEARNING_RATE = 0.05
MAX_DELTA     = 0.5  # |Δw| 상한 — 한 라운드에서 가중치가 너무 흔들리지 않게

# 등급 → 목표 점수 (rule-v1 의 임계값 5.0/2.0 기준)
TARGET_SCORE = {"O": 1.0, "S": 3.5, "C": 6.5}


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


# ---------------------------------------------------------------------------
# 학습 트리거 조건 평가
# ---------------------------------------------------------------------------

def evaluate_readiness(decisions: list[dict]) -> dict:
    """저장된 결정 리스트 → 학습 가능 여부 + 사유."""
    n = len(decisions)
    per = {"C": 0, "S": 0, "O": 0}
    gap_pos = 0
    for d in decisions:
        u = d.get("user_grade")
        if u in per:
            per[u] += 1
        if int(d.get("gap", 0)) > 0:
            gap_pos += 1

    checks = [
        {"key": "labeled",     "label": "누적 레이블",
         "value": n,           "threshold": MIN_LABELED,
         "ok": n >= MIN_LABELED},
        {"key": "per_grade_C", "label": "C 등급 누적",
         "value": per["C"],    "threshold": MIN_PER_GRADE,
         "ok": per["C"] >= MIN_PER_GRADE},
        {"key": "per_grade_S", "label": "S 등급 누적",
         "value": per["S"],    "threshold": MIN_PER_GRADE,
         "ok": per["S"] >= MIN_PER_GRADE},
        {"key": "per_grade_O", "label": "O 등급 누적",
         "value": per["O"],    "threshold": MIN_PER_GRADE,
         "ok": per["O"] >= MIN_PER_GRADE},
        {"key": "gap",         "label": "사용자-AI 갭(>0)",
         "value": gap_pos,     "threshold": MIN_GAP_DECISIONS,
         "ok": gap_pos >= MIN_GAP_DECISIONS},
    ]
    return {
        "ready": all(c["ok"] for c in checks),
        "checks": checks,
        "totals": {"labeled": n, "per_grade": per, "gap_positive": gap_pos},
    }


# ---------------------------------------------------------------------------
# Feature 추출 — decision 1건 → {feature_name: count}
# ---------------------------------------------------------------------------

def _features_of(decision: dict) -> dict[str, int]:
    """저장된 reasons 에서 feature counts 복원.

    `reasons` 는 분류 시점 스냅샷이고 그 자체가 (entity_type or keyword_label, count)
    리스트라 그대로 features 로 사용. (entity_type 과 keyword 라벨이 충돌하지 않도록
    keyword 는 'kw:' prefix 를 붙인 별도 namespace 로 다룬다.)
    """
    feats: dict[str, int] = {}
    for r in decision.get("reasons") or []:
        kind = r.get("kind")
        label = r.get("label")
        if not label:
            continue
        if kind == "keyword":
            # keyword 는 cap 적용된 counted 사용 (없으면 count)
            cnt = int(r.get("counted") or r.get("count") or 0)
            feats[f"kw:{label}"] = feats.get(f"kw:{label}", 0) + cnt
        else:
            cnt = int(r.get("count") or 0)
            feats[label] = feats.get(label, 0) + cnt
    return feats


def _current_weight(feature: str) -> float:
    """ENTITY_WEIGHTS / GRADE_KEYWORDS 에서 현재 가중치 조회."""
    if feature.startswith("kw:"):
        label = feature[3:]
        for _kw, w, lbl in classifier.GRADE_KEYWORDS:
            if lbl == label:
                return float(w)
        return 0.0
    return classifier.ENTITY_WEIGHTS.get(feature, classifier.DEFAULT_ENTITY_WEIGHT)


def _predict_score(feats: dict[str, int]) -> float:
    return sum(_current_weight(f) * c for f, c in feats.items())


# ---------------------------------------------------------------------------
# Training step (1 epoch, 전체 배치)
# ---------------------------------------------------------------------------

def train_one_round(
    decisions: list[dict],
    learning_rate: float = LEARNING_RATE,
    max_delta: float = MAX_DELTA,
) -> dict:
    """전체 결정에 대해 평균 gradient 를 계산하여 가중치 업데이트 제안.

    Returns:
        {
          decisions_count, accuracy_before, accuracy_after,
          sse_before, sse_after,
          weight_deltas: {feature: signed_delta},
          new_weights:   {feature: new_w},
          per_decision:  [{id, target, pred_before, pred_after, residual}, ...]
        }
    """
    if not decisions:
        return {"decisions_count": 0, "weight_deltas": {}, "new_weights": {}}

    # 1) 모든 feature 수집 + 기존 가중치 스냅샷
    all_feats: set[str] = set()
    for d in decisions:
        all_feats.update(_features_of(d).keys())
    base_w: dict[str, float] = {f: _current_weight(f) for f in all_feats}
    grad_sum: dict[str, float] = {f: 0.0 for f in all_feats}
    weight_sum = 0.0

    sse_before = 0.0
    correct_before = 0
    per_decision = []

    # 2) 누적 grad 계산
    for d in decisions:
        feats = _features_of(d)
        target = TARGET_SCORE[d["user_grade"]]
        pred = sum(base_w[f] * c for f, c in feats.items())
        resid = target - pred
        sw = 1.0 + 1.5 * int(d.get("gap", 0))   # SPEC §10.2
        weight_sum += sw
        denom = 1.0 + sum(feats.values())
        for f, c in feats.items():
            grad_sum[f] += sw * resid * c / denom

        sse_before += (target - pred) ** 2
        if d["ai_grade"] == d["user_grade"]:
            correct_before += 1
        per_decision.append({
            "id": d.get("id"),
            "target": round(target, 2),
            "pred_before": round(pred, 3),
            "residual": round(resid, 3),
            "sample_weight": round(sw, 2),
            "user_grade": d["user_grade"],
            "ai_grade": d["ai_grade"],
        })

    # 3) 평균 gradient → delta (clip)
    deltas: dict[str, float] = {}
    new_weights: dict[str, float] = {}
    for f, g in grad_sum.items():
        d_w = learning_rate * (g / max(weight_sum, 1e-9))
        d_w = max(-max_delta, min(max_delta, d_w))
        new_w = max(0.0, base_w[f] + d_w)   # 음수 가중치 방지
        deltas[f] = round(d_w, 4)
        new_weights[f] = round(new_w, 4)

    # 4) After 평가
    sse_after = 0.0
    correct_after = 0
    for i, d in enumerate(decisions):
        feats = _features_of(d)
        target = TARGET_SCORE[d["user_grade"]]
        pred_after = sum(new_weights.get(f, base_w.get(f, 0.0)) * c for f, c in feats.items())
        sse_after += (target - pred_after) ** 2
        # 새 가중치로 다시 등급 산정해 정확도 측정
        new_grade = (
            "C" if pred_after >= classifier.C_THRESHOLD else
            "S" if pred_after >= classifier.S_THRESHOLD else
            "O"
        )
        if new_grade == d["user_grade"]:
            correct_after += 1
        per_decision[i]["pred_after"] = round(pred_after, 3)
        per_decision[i]["new_grade"]  = new_grade

    n = len(decisions)
    return {
        "decisions_count": n,
        "accuracy_before": round(correct_before / n, 3),
        "accuracy_after":  round(correct_after  / n, 3),
        "sse_before":      round(sse_before, 3),
        "sse_after":       round(sse_after, 3),
        "weight_deltas":   deltas,
        "new_weights":     new_weights,
        "per_decision":    per_decision,
        "hyperparams": {
            "learning_rate": learning_rate,
            "max_delta":     max_delta,
            "target_score":  TARGET_SCORE,
            "loss":          "MSE on linear score",
            "objective":     "Σ sample_weight · (target - Σ w·x)²",
        },
        "started_at":  _now(),
        "finished_at": _now(),
        "status":      "completed",
    }


# ---------------------------------------------------------------------------
# 핫스왑 — 결과를 in-memory ENTITY_WEIGHTS / GRADE_KEYWORDS 에 반영
# ---------------------------------------------------------------------------

def apply_new_weights(
    new_weights: dict[str, float],
    *,
    training_run_id: int | None = None,
    decisions_count: int | None = None,
    accuracy: float | None = None,
    sse: float | None = None,
    notes: str | None = None,
) -> dict:
    """SPEC §1 기능 10 — "모델 핫스왑". 다음 분석부터 새 가중치 적용.

    프로세스 재시작 시:
      - in-memory ENTITY_WEIGHTS / GRADE_KEYWORDS 자체는 원복되지만,
      - storage.model_versions 테이블에 스냅샷이 보존되므로
        부팅 시 active 버전을 다시 로드하면 복원 가능.
    """
    import storage  # 순환 import 방지

    applied_entities: dict[str, list[float]] = {}
    applied_keywords: dict[str, list[float]] = {}

    for f, new_w in new_weights.items():
        if f.startswith("kw:"):
            label = f[3:]
            for i, (kw, old_w, lbl) in enumerate(classifier.GRADE_KEYWORDS):
                if lbl == label:
                    classifier.GRADE_KEYWORDS[i] = (kw, float(new_w), lbl)
                    applied_keywords[label] = [round(old_w, 4), round(new_w, 4),
                                                round(new_w - old_w, 4)]
                    break
        else:
            old_w = classifier.ENTITY_WEIGHTS.get(f, classifier.DEFAULT_ENTITY_WEIGHT)
            classifier.ENTITY_WEIGHTS[f] = float(new_w)
            applied_entities[f] = [round(old_w, 4), round(new_w, 4),
                                    round(new_w - old_w, 4)]

    # ---- 새 버전 라벨 부여 ----
    parent = classifier.active_version()
    new_version = storage.next_version_label(base="rule-v1")
    classifier.set_active_version(new_version)

    # ---- 모델 버전 기록 ----
    diff = {
        "entities_changed": applied_entities,
        "keywords_changed": applied_keywords,
    }
    storage.insert_model_version({
        "version":         new_version,
        "parent_version":  parent,
        "training_run_id": training_run_id,
        "decisions_count": decisions_count,
        "accuracy":        accuracy,
        "sse":             sse,
        "weights":         dict(classifier.ENTITY_WEIGHTS),
        "keywords":        [(kw, w, lbl) for kw, w, lbl in classifier.GRADE_KEYWORDS],
        "diff":            diff,
        "notes":           notes or "linear-regression 1-step gradient hot-swap",
        "is_active":       True,
    })

    return {
        "new_version":      new_version,
        "parent_version":   parent,
        "entities_changed": applied_entities,
        "keywords_changed": applied_keywords,
        "ENTITY_WEIGHTS_now": dict(classifier.ENTITY_WEIGHTS),
        "GRADE_KEYWORDS_now": [(kw, w, lbl) for kw, w, lbl in classifier.GRADE_KEYWORDS],
    }


def bootstrap_initial_version() -> None:
    """앱 부팅 시 1회 호출 — 베이스 'rule-v1' 가 model_versions 에 없으면 등록.

    이렇게 해두면 학습/이력 탭의 버전 이력 표가 항상 최소 1행은 가진다.
    """
    import storage
    if storage.active_model_version():
        return
    storage.insert_model_version({
        "version":         classifier.CLASSIFIER_VERSION,
        "parent_version":  None,
        "training_run_id": None,
        "decisions_count": 0,
        "accuracy":        None,
        "sse":             None,
        "weights":         dict(classifier.ENTITY_WEIGHTS),
        "keywords":        [(kw, w, lbl) for kw, w, lbl in classifier.GRADE_KEYWORDS],
        "diff":            {},
        "notes":           "initial baseline weights (hand-tuned)",
        "is_active":       True,
    })