C.E.L_Slide_test2/tests/matching/pipeline_15_logistic_regression.py

"""Pipeline Step 15 (Logistic Regression) — TARGET 4 로 가중치 학습.

방법:
  1. TARGET 4 섹션 × 32 프레임 = 128 샘플
  2. Feature: (standalone_score, group_score, related_score) — 각 [0, 1]
  3. Label: 1 if 정답 프레임 else 0 (정답 4, 오답 124)
  4. Logistic Regression + Linear Regression 두 가지 fit
  5. 학습 가중치를 sum=1 로 정규화 → 현재 0.30/0.50/0.20 와 비교
  6. 학습 가중치로 예측 시 TARGET 정답률 검증
"""
from pathlib import Path
import numpy as np
import yaml
from sklearn.linear_model import LogisticRegression, LinearRegression

HERE = Path(__file__).parent
TARGET_SIDS = ['01-2', '02-2.2', '03-1', '03-2']


def main():
    v1 = yaml.safe_load((HERE / 'mdx_matching_result.yaml').read_text(encoding='utf-8'))
    auto = yaml.safe_load((HERE / 'auto_anchor_candidates.yaml').read_text(encoding='utf-8'))

    answer_map = v1['meta']['answer_map']
    frame_num = {fid: info['frame_number'] for fid, info in auto['frame_stats'].items()}

    # ─── 데이터 구성 ───
    rows = []  # (sid, frame_id, frame_num, s, g, r, label)
    for sid in TARGET_SIDS:
        sec = v1['mdx_sections'][sid]
        answer = answer_map[sid]
        for frame_id, detail in sec['per_frame_detail'].items():
            s = detail['standalone']['score']
            g = detail['keyword_group'].get('score_avg', 0)
            r = detail['related']['score']
            label = 1 if frame_num[frame_id] == answer else 0
            rows.append((sid, frame_id, frame_num[frame_id], s, g, r, label))

    X = np.array([[r[3], r[4], r[5]] for r in rows])
    y = np.array([r[6] for r in rows])

    print('=' * 75)
    print(f'데이터: {len(y)} 샘플 (정답 {y.sum()}, 오답 {len(y)-y.sum()}) — TARGET 4 × 32 frames')
    print('=' * 75)
    print()

    # ─── Logistic Regression ───
    clf = LogisticRegression(penalty='l2', C=1.0, fit_intercept=True, max_iter=2000)
    clf.fit(X, y)
    w_lr = clf.coef_[0]
    bias_lr = clf.intercept_[0]
    w_lr_norm = w_lr / w_lr.sum()

    print(f'[Logistic Regression (L2, C=1.0)]')
    print(f'  raw weights:  standalone={w_lr[0]:+.3f}  group={w_lr[1]:+.3f}  related={w_lr[2]:+.3f}')
    print(f'  bias:         {bias_lr:+.3f}')
    print(f'  정규화 (sum=1): standalone={w_lr_norm[0]:.3f}  group={w_lr_norm[1]:.3f}  related={w_lr_norm[2]:.3f}')
    print()

    # ─── Linear Regression (OLS, no intercept) ───
    ols = LinearRegression(fit_intercept=False)
    ols.fit(X, y)
    w_ols = ols.coef_
    w_ols_norm = w_ols / w_ols.sum()

    print(f'[Linear Regression (OLS, no intercept)]')
    print(f'  raw weights:  standalone={w_ols[0]:+.3f}  group={w_ols[1]:+.3f}  related={w_ols[2]:+.3f}')
    print(f'  정규화 (sum=1): standalone={w_ols_norm[0]:.3f}  group={w_ols_norm[1]:.3f}  related={w_ols_norm[2]:.3f}')
    print()

    # ─── 현재 vs 학습 가중치 비교 ───
    print('[비교]')
    print(f'                standalone     group      related')
    print(f'  현재 (수동):    0.300          0.500      0.200')
    print(f'  Logistic:      {w_lr_norm[0]:.3f}          {w_lr_norm[1]:.3f}      {w_lr_norm[2]:.3f}')
    print(f'  OLS:           {w_ols_norm[0]:.3f}          {w_ols_norm[1]:.3f}      {w_ols_norm[2]:.3f}')
    print()

    # ─── 학습 가중치로 TARGET 정답률 검증 ───
    def evaluate(w_normalized, name):
        hits = 0
        for sid in TARGET_SIDS:
            sec = v1['mdx_sections'][sid]
            answer = answer_map[sid]
            scores = {}
            for fid, detail in sec['per_frame_detail'].items():
                s = detail['standalone']['score']
                g = detail['keyword_group'].get('score_avg', 0)
                r = detail['related']['score']
                scores[fid] = w_normalized[0]*s + w_normalized[1]*g + w_normalized[2]*r
            top_fid = max(scores, key=scores.get)
            if frame_num[top_fid] == answer:
                hits += 1
        return hits

    hits_current = evaluate([0.30, 0.50, 0.20], '현재')
    hits_lr = evaluate(w_lr_norm, 'Logistic')
    hits_ols = evaluate(w_ols_norm, 'OLS')

    print('[TARGET 4 정답률 (학습 가중치로 재예측)]')
    print(f'  현재 (0.30/0.50/0.20):     {hits_current}/4')
    print(f'  Logistic Regression:        {hits_lr}/4')
    print(f'  OLS Linear Regression:      {hits_ols}/4')
    print()

    # ─── LOOCV (Leave-One-Out Cross-Validation) — 과적합 체크 ───
    print('[LOOCV — 과적합 확인]')
    print('  각 TARGET 을 hold-out, 나머지 3개로 학습 후 테스트')
    loocv_hits = 0
    for hold_out_idx, hold_out_sid in enumerate(TARGET_SIDS):
        train_X = np.array([[r[3], r[4], r[5]] for r in rows if r[0] != hold_out_sid])
        train_y = np.array([r[6] for r in rows if r[0] != hold_out_sid])
        clf_cv = LogisticRegression(penalty='l2', C=1.0, fit_intercept=True, max_iter=2000)
        clf_cv.fit(train_X, train_y)
        w_cv = clf_cv.coef_[0] / clf_cv.coef_[0].sum()

        sec = v1['mdx_sections'][hold_out_sid]
        answer = answer_map[hold_out_sid]
        scores = {}
        for fid, detail in sec['per_frame_detail'].items():
            s = detail['standalone']['score']
            g = detail['keyword_group'].get('score_avg', 0)
            r = detail['related']['score']
            scores[fid] = w_cv[0]*s + w_cv[1]*g + w_cv[2]*r
        top_fid = max(scores, key=scores.get)
        correct = frame_num[top_fid] == answer
        if correct:
            loocv_hits += 1
        print(f'    hold-out {hold_out_sid}: weights={w_cv.round(3)}  정답? {"✓" if correct else "✗"}')
    print(f'  LOOCV 정답률: {loocv_hits}/4')


if __name__ == '__main__':
    main()