재학습 전략 (Retraining Strategies)

drift_levels = [0, 0.5, 1.0, 2.0, 3.0, 5.0]

for drift in drift_levels:
    X_drifted = X_test.copy()
    X_drifted["air_temperature_k"] += drift
    X_drifted["process_temperature_k"] += drift * 0.8
    X_drifted["temp_diff"] = X_drifted["process_temperature_k"] - X_drifted["air_temperature_k"]

    dtest = xgb.DMatrix(X_drifted)
    pred = (model_original.predict(dtest) > 0.5).astype(int)
    f1 = f1_score(Y_test, pred)

IF (PSI > 0.2)           → 즉시 재학습 (긴급) — 데이터 분포 급변
ELIF (F1 < 0.7)          → 즉시 재학습 (긴급) — 성능 임계치 이탈
ELIF (마지막 학습 > 7일)  → 정기 재학습 (일반) — 안전망
ELIF (새 데이터 > 5000건) → 추가 학습 고려 (낮음) — Incremental Learning
ELSE                      → 현재 모델 유지

def calculate_psi(expected, actual, bins=10):
    breakpoints = np.linspace(
        min(expected.min(), actual.min()),
        max(expected.max(), actual.max()),
        bins + 1
    )
    expected_counts = np.histogram(expected, bins=breakpoints)[0] / len(expected)
    actual_counts = np.histogram(actual, bins=breakpoints)[0] / len(actual)
    expected_counts = np.maximum(expected_counts, 0.001)
    actual_counts = np.maximum(actual_counts, 0.001)
    psi = np.sum((actual_counts - expected_counts) * np.log(actual_counts / expected_counts))
    return psi

def full_retrain_pipeline(reason="scheduled"):
    # Step 1: 학습 데이터 준비 (Sliding Window)
    train_data = spark.table("lgit_pm_training").filter("split='train'") \
                      .select(*feature_columns, label_col).toPandas()

    # Step 2: MLflow 실험에 기록하면서 모델 학습
    with mlflow.start_run(run_name=f"retrain_{reason}_{datetime.now().strftime('%Y%m%d')}") as run:
        mlflow.log_param("retrain_reason", reason)
        mlflow.log_param("training_data_size", len(X_tr))

        new_model = xgb.train(retrain_params, dtrain, num_boost_round=200)

        # 성능 평가
        new_f1 = f1_score(Y_te, pred)
        new_auc = roc_auc_score(Y_te, proba)
        mlflow.log_metrics({"test_f1_score": new_f1, "test_auc": new_auc})
        mlflow.xgboost.log_model(new_model, "xgboost_model", signature=signature)

    # Step 3: Champion 비교
    champion_info = client.get_model_version_by_alias(model_name, "Champion")
    champion_f1 = mlflow.get_run(champion_info.run_id).data.metrics.get("test_f1_score", 0)

    # Step 4: 배포 결정
    if new_f1 >= champion_f1:
        model_details = mlflow.register_model(
            f"runs:/{run.info.run_id}/xgboost_model", model_name
        )
        client.set_registered_model_alias(
            name=model_name, alias="Champion", version=model_details.version
        )

-- 특정 버전의 데이터 조회
SELECT * FROM lgit_pm_training VERSION AS OF 5

-- 특정 시점의 데이터 조회
SELECT * FROM lgit_pm_training TIMESTAMP AS OF '2024-01-01'

-- 최근 30일 데이터만 조회 (Sliding Window)
SELECT * FROM lgit_pm_inference_results
WHERE inference_timestamp >= current_date() - INTERVAL 30 DAYS

def rollback_model(model_name, target_version=None):
    current = client.get_model_version_by_alias(model_name, "Champion")
    current_version = int(current.version)

    if target_version is None:
        target_version = max(1, current_version - 1)

    # 현재 Champion에 태그 추가
    client.set_model_version_tag(
        name=model_name, version=str(current_version),
        key="rolled_back", value="true"
    )

    # 에일리어스 변경 → 즉시 롤백 완료
    client.set_registered_model_alias(
        name=model_name, alias="Champion", version=target_version
    )

# Batch 1: 초기 모델 학습 (100 trees)
model_inc = xgb.train(params, dtrain1, num_boost_round=100)

# Batch 2: 기존 모델에 추가 학습 (50 trees 추가)
# 핵심: xgb_model 파라미터로 기존 모델 전달
model_inc = xgb.train(params, dtrain2, num_boost_round=50, xgb_model=model_inc)

# Batch 3: 계속 추가 학습
model_inc = xgb.train(params, dtrain3, num_boost_round=50, xgb_model=model_inc)

class ReplayBuffer:
    def __init__(self, max_size=2000):
        self.buffer = []
        self.max_size = max_size

    def add(self, X, Y):
        data = list(zip(X.values.tolist(), Y.values.tolist()))
        self.buffer.extend(data)
        if len(self.buffer) > self.max_size:
            self.buffer = random.sample(self.buffer, self.max_size)

    def get_replay_data(self, n_samples=500):
        samples = random.sample(self.buffer, min(n_samples, len(self.buffer)))
        X = pd.DataFrame([s[0] for s in samples], columns=feature_columns)
        Y = pd.Series([s[1] for s in samples])
        return X, Y


# Continual Learning: 새 데이터 + 리플레이 데이터 결합
replay = ReplayBuffer(max_size=2000)

for batch in batches:
    X_rep, Y_rep = replay.get_replay_data(500)
    X_combined = pd.concat([X_batch, X_rep], ignore_index=True)
    Y_combined = pd.concat([Y_batch, Y_rep], ignore_index=True)

    model_cl = xgb.train(params, xgb.DMatrix(X_combined, label=Y_combined), num_boost_round=150)
    replay.add(X_batch, Y_batch)

from river import tree, metrics, preprocessing, compose

model_online = compose.Pipeline(
    preprocessing.StandardScaler(),         # 실시간 표준화
    tree.HoeffdingAdaptiveTreeClassifier(seed=42)
)
metric = metrics.F1()

for _, row in data.iterrows():
    x = {col: row[col] for col in feature_columns}
    y = int(row[label_col])

    y_pred = model_online.predict_one(x)        # 1. 현재 모델로 예측
    if y_pred is not None:
        metric.update(y, y_pred)                 # 2. 메트릭 업데이트
    model_online.learn_one(x, y)                 # 3. 이 한 건으로 모델 업데이트

class RetrainingBandit:
    def __init__(self):
        self.actions = ["no_action", "incremental", "sliding_window", "full_retrain"]
        self.alpha = {a: 1.0 for a in self.actions}  # Beta 분포 파라미터
        self.beta_ = {a: 1.0 for a in self.actions}

    def select_action(self, context):
        scores = {}
        drift = context.get("drift_level", 0)
        for action in self.actions:
            base_score = np.random.beta(self.alpha[action], self.beta_[action])
            # 컨텍스트 기반 보정
            if action == "full_retrain" and drift > 0.2:
                base_score += 0.3
            elif action == "incremental" and 0.05 < drift <= 0.2:
                base_score += 0.2
            elif action == "no_action" and drift < 0.05:
                base_score += 0.3
            scores[action] = base_score
        return max(scores, key=scores.get), scores

    def update(self, action, reward):
        if reward > 0:
            self.alpha[action] += reward
        else:
            self.beta_[action] += abs(reward)

# 모델의 예측 확률로 불확실성 측정
probas = model_full.predict(xgb.DMatrix(full_df[feature_columns]))
uncertainty = np.abs(probas - 0.5)  # 0.5에 가까울수록 불확실

# 가장 불확실한 500건을 우선 레이블링
n_query = 500
active_idx = np.argsort(uncertainty)[:n_query]

# Active Learning으로 선택한 500건으로 학습
X_sel = full_df.iloc[active_idx][feature_columns]
Y_sel = full_df.iloc[active_idx][label_col]
model_al = xgb.train(params, xgb.DMatrix(X_sel, label=Y_sel), num_boost_round=150)

IF Concept Drift 의심 (PSI > 0.5 또는 F1 급락)
  → Cold-start (처음부터)
ELIF Data Drift만 (PSI 0.1~0.5)
  → Warm-start (기존 모델 + 새 데이터)
ELIF 변화 없음 (PSI < 0.1)
  → 재학습 불필요

# Cold-start: 전체 데이터로 처음부터
model_cold = xgb.train(params, dtrain, num_boost_round=200)

# Warm-start: 기존 모델에서 새 데이터로 추가 학습 (훨씬 빠름)
model_warm = xgb.train(params, dtrain_new, num_boost_round=50, xgb_model=model_cold)

# Jobs 파이프라인에서 호출될 때 자동 실행
drift_detected = dbutils.jobs.taskValues.get(
    taskKey="model_monitoring",
    key="drift_detected",
    default=False
)
max_psi = dbutils.jobs.taskValues.get(
    taskKey="model_monitoring",
    key="max_psi",
    default=0.0
)

if drift_detected:
    # 자동 재학습 실행
    result = full_retrain_pipeline(reason=f"auto_drift_psi_{max_psi:.2f}")

    # 결과를 다음 태스크(검증)에 전달
    dbutils.jobs.taskValues.set(key="retrain_completed", value=True)
    dbutils.jobs.taskValues.set(key="new_model_version", value=result.get("version", ""))