Update app/bandit.py

app/bandit.py

@@ -5,26 +5,20 @@ from . import storage
 
 class ThompsonBandit:
     """
-    CTR（クリック）とCVR（コンバージョン）の二段ベータ。
-    目的関数: E[value] = p_click * p_conv * value_per_conversion
-    季節性は CTR 側の仮想カウントで補正。
+    CTRとCVRの二段ベータ。EV=pc*pv*V。
+    季節性はクリック側の事前にブースト。
     """
     def __init__(self, campaign_id: str, seasonality_boost: float = 5.0):
         self.campaign_id = campaign_id
         self.seasonality_boost = seasonality_boost
 
-    def sample_arm(
-        self,
-        context: Dict[str, Any],
-        seasonal_fn: Optional[Callable[[Dict[str, Any]], float]] = None,  # ← キーワード引数OK
-    ) -> Tuple[Optional[str], float]:
+    def sample_arm(self, context: Dict[str, Any], seasonal_fn: Optional[Callable[[Dict[str, Any]], float]] = None
+                   ) -> Tuple[Optional[str], float]:
         metrics = storage.get_metrics(self.campaign_id)
         if not metrics:
             return None, -1.0
-
         vpc = storage.get_campaign_value_per_conversion(self.campaign_id)
 
-        # 季節性スコア s ∈ (0, 1)
         s = 0.5
         if seasonal_fn is not None:
             try:
@@ -33,21 +27,18 @@ class ThompsonBandit:
             except Exception:
                 s = 0.5
 
-        best_score, best_variant = -1.0, None
-        for row in metrics:
-            ac, bc = float(row["alpha_click"]), float(row["beta_click"])
-            av, bv = float(row["alpha_conv"]), float(row["beta_conv"])
-
-            # 季節性でクリック側の事前分布を微調整
+        best_score, best_vid = -1.0, None
+        for r in metrics:
+            ac, bc = float(r["alpha_click"]), float(r["beta_click"])
+            av, bv = float(r["alpha_conv"]), float(r["beta_conv"])
             ac_eff = ac + self.seasonality_boost * s
             bc_eff = bc + self.seasonality_boost * (1.0 - s)
-
             pc = random.betavariate(max(1e-6, ac_eff), max(1e-6, bc_eff))
             pv = random.betavariate(max(1e-6, av),     max(1e-6, bv))
             score = pc * pv * vpc
             if score > best_score:
-                best_score, best_variant = score, row["variant_id"]
-        return best_variant, best_score
+                best_score, best_vid = score, r["variant_id"]
+        return best_vid, best_score
 
     @staticmethod
     def update_with_event(campaign_id: str, variant_id: str, event_type: str):