Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import pandas as pd | |
| import numpy as np | |
| from . import storage | |
| try: | |
| from prophet import Prophet | |
| except Exception: | |
| Prophet = None | |
| try: | |
| from neuralprophet import NeuralProphet | |
| except Exception: | |
| NeuralProphet = None | |
| class SeasonalityModel: | |
| def __init__(self, campaign_id: str): | |
| self.campaign_id = campaign_id | |
| self.model = None | |
| self.model_type = "none" | |
| self.global_mean = 0.05 | |
| def fit(self): | |
| with storage.get_conn() as con: | |
| df = pd.read_sql_query( | |
| "SELECT ts, event_type FROM events WHERE campaign_id=?", | |
| con, | |
| params=(self.campaign_id,), | |
| ) | |
| if df.empty: | |
| self.model_type = "none" | |
| return | |
| df["ts"] = pd.to_datetime(df["ts"], errors="coerce") | |
| df = df.dropna(subset=["ts"]) | |
| df["hour"] = df["ts"].dt.floor("h") | |
| agg = df.pivot_table(index="hour", columns="event_type", values="ts", aggfunc="count").fillna(0) | |
| if "impression" not in agg: agg["impression"] = 0 | |
| if "click" not in agg: agg["click"] = 0 | |
| ctr = np.where(agg["impression"] > 0, agg["click"] / agg["impression"], np.nan) | |
| if np.all(np.isnan(ctr)): | |
| self.model_type = "none" | |
| return | |
| self.global_mean = float(np.nanmean(ctr)) | |
| ds = agg.index.to_series().reset_index(drop=True) | |
| train = pd.DataFrame({"ds": ds, "y": pd.Series(ctr).fillna(self.global_mean).values}) | |
| try: | |
| if Prophet is not None: | |
| m = Prophet(weekly_seasonality=True, daily_seasonality=True) | |
| m.fit(train) | |
| self.model, self.model_type = m, "prophet" | |
| elif NeuralProphet is not None: | |
| m = NeuralProphet(weekly_seasonality=True, daily_seasonality=True) | |
| m.fit(train, freq="H") | |
| self.model, self.model_type = m, "neuralprophet" | |
| else: | |
| self.model_type = "none" | |
| except Exception: | |
| self.model_type = "none" | |
| def expected_ctr(self, context: dict) -> float: | |
| hour = int(context.get("hour", 12)) | |
| if self.model_type in {None, "none"}: | |
| base = self.global_mean | |
| if 11 <= hour <= 13: return min(0.99, base * 1.1) | |
| if 20 <= hour <= 23: return min(0.99, base * 1.15) | |
| return max(0.01, base) | |
| now_ds = pd.Timestamp.utcnow().floor("D") + pd.Timedelta(hours=hour) | |
| if self.model_type == "prophet": | |
| yhat = float(self.model.predict(pd.DataFrame({"ds": [now_ds]}))["yhat"].iloc[0]) | |
| else: | |
| yhat = float(self.model.predict(pd.DataFrame({"ds": [now_ds]}))["yhat1"].iloc[0]) | |
| return max(0.01, min(0.99, yhat)) | |