| """ |
| Publisher engine — per-publisher traffic generation each day. |
| |
| All randomness is pre-baked into ``day_factors`` and ``noise_factors`` |
| arrays in the case profile, ensuring full determinism. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from typing import Any, Dict |
|
|
| from .fraud_engine import compute_fraud_intensity |
|
|
|
|
| def generate_daily_traffic( |
| day: int, |
| publisher_cfg: Dict[str, Any], |
| budget_allocation: float, |
| adaptation_stage: str, |
| is_paused: bool, |
| ) -> Dict[str, Any]: |
| """Generate one day of traffic for a single publisher. |
| |
| Returns a dict with keys: |
| impressions, clicks, conversions, spend, |
| ctr, cvr, |
| legitimate_spend, fraudulent_spend, legitimate_revenue |
| """ |
| if is_paused: |
| return _zero_traffic() |
|
|
| day_idx = day - 1 |
| day_factors = publisher_cfg.get("day_factors", [1.0] * 30) |
| noise_factors = publisher_cfg.get("noise_factors", [1.0] * 30) |
|
|
| day_factor = day_factors[day_idx] if day_idx < len(day_factors) else 1.0 |
| noise_factor = noise_factors[day_idx] if day_idx < len(noise_factors) else 1.0 |
|
|
| base_rate: float = publisher_cfg["base_traffic_rate"] |
| true_ctr: float = publisher_cfg["true_ctr"] |
| true_cvr: float = publisher_cfg["true_cvr"] |
| cpm_rate: float = publisher_cfg.get("cpm_rate", 2.0) |
| conversion_value: float = publisher_cfg.get("conversion_value", 10.0) |
|
|
| |
| legit_impressions = base_rate * budget_allocation * day_factor * noise_factor |
| legit_clicks = legit_impressions * true_ctr * noise_factor |
| legit_conversions = legit_clicks * true_cvr * noise_factor |
| legit_spend = legit_impressions * cpm_rate / 1000.0 |
| legit_revenue = legit_conversions * conversion_value |
|
|
| |
| fraud_impressions = 0.0 |
| fraud_clicks = 0.0 |
| fraud_conversions = 0.0 |
| fraud_spend = 0.0 |
|
|
| if publisher_cfg.get("is_fraudulent", False): |
| fraud_schedule = publisher_cfg.get("fraud_schedule", {}) |
| if fraud_schedule: |
| intensity = compute_fraud_intensity(day, fraud_schedule, adaptation_stage) |
| if intensity > 0: |
| fake_ctr = publisher_cfg.get("fake_ctr", 0.045) |
| fake_cvr = publisher_cfg.get("fake_cvr", 0.001) |
|
|
| fraud_impressions = legit_impressions * intensity |
| fraud_clicks = fraud_impressions * fake_ctr |
| fraud_conversions = fraud_clicks * fake_cvr |
| fraud_spend = fraud_impressions * cpm_rate / 1000.0 |
|
|
| total_impressions = int(round(legit_impressions + fraud_impressions)) |
| total_clicks = int(round(legit_clicks + fraud_clicks)) |
| total_conversions = int(round(legit_conversions + fraud_conversions)) |
| total_spend = legit_spend + fraud_spend |
|
|
| ctr = total_clicks / total_impressions if total_impressions > 0 else 0.0 |
| cvr = total_conversions / total_clicks if total_clicks > 0 else 0.0 |
|
|
| return { |
| "impressions": total_impressions, |
| "clicks": total_clicks, |
| "conversions": total_conversions, |
| "spend": round(total_spend, 2), |
| "ctr": round(ctr, 4), |
| "cvr": round(cvr, 4), |
| "legitimate_spend": round(legit_spend, 2), |
| "fraudulent_spend": round(fraud_spend, 2), |
| "legitimate_revenue": round(legit_revenue, 2), |
| } |
|
|
|
|
| def _zero_traffic() -> Dict[str, Any]: |
| return { |
| "impressions": 0, |
| "clicks": 0, |
| "conversions": 0, |
| "spend": 0.0, |
| "ctr": 0.0, |
| "cvr": 0.0, |
| "legitimate_spend": 0.0, |
| "fraudulent_spend": 0.0, |
| "legitimate_revenue": 0.0, |
| } |
|
|
