#!/usr/bin/env python3 """Generate the MMM Dataset CSV and source-of-truth parameter JSON. This mirrors scripts/generate-mmm-dataset.mjs, but is written in plain Python so readers can reproduce the dataset from the marketing science side without needing the Astro/Node toolchain. """ from __future__ import annotations import json import math from datetime import date, timedelta from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] OUTPUT_DIR = PROJECT_ROOT / "public" / "data" / "mmm" CSV_PATH = OUTPUT_DIR / "mmm-dataset.csv" PARAMS_PATH = OUTPUT_DIR / "mmm-dataset-params.json" SEED = 20260520 WEEK_COUNT = 156 START_DATE = date(2023, 1, 2) MASK_32 = 0xFFFFFFFF def to_int32(value: int) -> int: value &= MASK_32 return value - 0x100000000 if value & 0x80000000 else value def unsigned_right_shift(value: int, bits: int) -> int: return (value & MASK_32) >> bits def imul(left: int, right: int) -> int: return to_int32((left & MASK_32) * (right & MASK_32)) def mulberry32(initial_seed: int): """Match the JavaScript Mulberry32 PRNG used by the Node generator.""" state = to_int32(initial_seed) def random() -> float: nonlocal state state = to_int32(state + 0x6D2B79F5) t = imul(state ^ unsigned_right_shift(state, 15), 1 | state) t = to_int32(to_int32(t + imul(t ^ unsigned_right_shift(t, 7), 61 | t)) ^ t) return unsigned_right_shift(t ^ unsigned_right_shift(t, 14), 0) / 4294967296 return random random = mulberry32(SEED) def normal(mean: float = 0, std: float = 1) -> float: u1 = max(random(), 2.220446049250313e-16) u2 = random() return mean + std * math.sqrt(-2 * math.log(u1)) * math.cos(2 * math.pi * u2) def clamp(value: float, minimum: float, maximum: float) -> float: return min(maximum, max(minimum, value)) def js_round(value: float, decimals: int = 2) -> float: multiplier = 10**decimals return math.floor(value * multiplier + 0.5) / multiplier def js_number(value) -> str: if isinstance(value, str): return value if isinstance(value, int): return str(value) if isinstance(value, float) and value.is_integer(): return str(int(value)) return f"{value:.15g}" def iso_week_date(index: int) -> str: return (START_DATE + timedelta(days=index * 7)).isoformat() def hill(value: float, beta: float, half_saturation: float, slope: float) -> float: if value <= 0: return 0 numerator = value**slope denominator = numerator + half_saturation**slope return (beta * numerator) / denominator # Highlight: channel parameters are the known truth future lessons can test against. CHANNELS = { "paid_search": { "label": "Paid Search", "lambda": 0.18, "beta": 310, "halfSaturation": 135, "slope": 1.25, "role": "Demand capture with fast decay and visible promo responsiveness.", }, "paid_social": { "label": "Paid Social", "lambda": 0.42, "beta": 250, "halfSaturation": 165, "slope": 1.35, "role": "Paced prospecting channel with moderate memory.", }, "ctv_video": { "label": "CTV / Video", "lambda": 0.72, "beta": 420, "halfSaturation": 360, "slope": 1.8, "role": "Flighted awareness channel with long memory.", }, "podcast_audio": { "label": "Podcast / Audio", "lambda": 0.62, "beta": 210, "halfSaturation": 175, "slope": 1.55, "role": "Flighted audio with delayed-seeming carryover under geometric memory.", }, "influencer": { "label": "Influencer", "lambda": 0.35, "beta": 160, "halfSaturation": 105, "slope": 1.3, "role": "Bursty creator drops with some spillover.", }, } def build_spend(index: int, promo_flag: int, seasonality: float) -> dict[str, float]: quarter_week = index % 13 year_week = index % 52 growth = 1 + index / WEEK_COUNT * 0.2 # Highlight: spend patterns encode realistic media behavior before any model sees the data. ctv_flight = ( 230 + 60 * math.sin((quarter_week - 2) / 3 * math.pi) if 2 <= quarter_week <= 5 else 0 ) podcast_flight = ( 80 + 55 * random() if year_week in [4, 5, 6, 17, 18, 19, 31, 32, 33, 43, 44, 45] else 18 + 8 * random() ) influencer_drop = ( 90 + 90 * random() if year_week in [7, 20, 34, 47] else 35 + 50 * random() if random() > 0.82 else 0 ) return { "paid_search": clamp((94 + 34 * promo_flag + 18 * seasonality + normal(0, 8)) * growth, 45, 190), "paid_social": clamp( (112 + 26 * math.sin((2 * math.pi * (index + 5)) / 26) + normal(0, 14)) * growth, 45, 210, ), "ctv_video": clamp((ctv_flight + normal(0, 18 if ctv_flight > 0 else 0)) * growth, 0, 360), "podcast_audio": clamp((podcast_flight + normal(0, 7)) * growth, 0, 175), "influencer": clamp(influencer_drop + normal(0, 10 if influencer_drop > 0 else 0), 0, 205), } def generate_rows() -> list[dict[str, str | int | float]]: rows = [] adstock_state = {channel: 0.0 for channel in CHANNELS} for index in range(WEEK_COUNT): week = index + 1 trend = js_round(index / (WEEK_COUNT - 1), 4) seasonality = js_round(math.sin((2 * math.pi * index) / 52), 4) holiday_season = 1 if index % 52 in [46, 47, 48, 49, 50, 51] else 0 promo_flag = 1 if index % 52 in [8, 9, 21, 22, 34, 35, 47, 48] else 0 price_index = js_round( 1 + 0.018 * math.sin((2 * math.pi * (index + 9)) / 52) + (-0.045 if promo_flag else 0), 4, ) competitor_pressure = js_round( clamp(0.52 + 0.12 * math.sin((2 * math.pi * (index + 18)) / 39) + normal(0, 0.035), 0.25, 0.82), 4, ) spend = build_spend(index, promo_flag, seasonality) adstock = {} contribution = {} # Highlight: recursive adstock is the memory step: current spend plus decayed prior effective media. for channel, params in CHANNELS.items(): adstock_state[channel] = spend[channel] + params["lambda"] * adstock_state[channel] adstock[channel] = adstock_state[channel] # Highlight: Hill saturation turns remembered media into incremental subscriptions. contribution[channel] = hill( adstock[channel], params["beta"], params["halfSaturation"], params["slope"], ) base_demand = ( 720 + 210 * trend + 90 * seasonality + 150 * holiday_season + 125 * promo_flag - 360 * (price_index - 1) - 110 * competitor_pressure ) total_media_contribution = sum(contribution.values()) expected_subscriptions = base_demand + total_media_contribution # Highlight: the final observed outcome adds noise, leaving truth columns for validation. observed_subscriptions = max(0, math.floor(expected_subscriptions + normal(0, 38) + 0.5)) row = { "week": week, "date": iso_week_date(index), "trend": trend, "seasonality": seasonality, "holiday_season": holiday_season, "promo_flag": promo_flag, "price_index": price_index, "competitor_pressure": competitor_pressure, } row.update({f"{channel}_spend": js_round(value) for channel, value in spend.items()}) row.update({f"{channel}_adstock": js_round(value) for channel, value in adstock.items()}) row.update({f"{channel}_contribution": js_round(value) for channel, value in contribution.items()}) row.update( { "base_demand": js_round(base_demand), "total_media_contribution": js_round(total_media_contribution), "expected_subscriptions": js_round(expected_subscriptions), "observed_subscriptions": observed_subscriptions, } ) rows.append(row) return rows def build_params(headers: list[str]) -> dict: return { "name": "MMM Dataset", "version": "1.0.0", "description": ( "Synthetic DTC subscription dataset for MMM learning modules. The data-generating " "process is transparent so lessons can compare model estimates against known truth." ), "seed": SEED, "startDate": START_DATE.isoformat(), "weekCount": WEEK_COUNT, "outcome": "observed_subscriptions", "cadence": "weekly", "scenario": { "business": "DTC subscription brand", "unit": "new paid subscriptions", "currency": "spend columns are in thousands of dollars", }, "formulas": { "adstock": "A[c,t] = spend[c,t] + lambda[c] * A[c,t-1]", "halfLife": "half_life_weeks = ln(0.5) / ln(lambda)", "saturation": ( "contribution[c,t] = beta[c] * A[c,t]^slope[c] / " "(half_saturation[c]^slope[c] + A[c,t]^slope[c])" ), "outcome": "observed_subscriptions[t] = round(base_demand[t] + sum(contribution[c,t]) + Normal(0, 38))", }, "channels": { channel: { **params, "halfLifeWeeks": js_round(math.log(0.5) / math.log(params["lambda"]), 2), } for channel, params in CHANNELS.items() }, "observedColumns": [ "week", "date", "trend", "seasonality", "holiday_season", "promo_flag", "price_index", "competitor_pressure", "paid_search_spend", "paid_social_spend", "ctv_video_spend", "podcast_audio_spend", "influencer_spend", "observed_subscriptions", ], "truthColumns": [ header for header in headers if header.endswith("_adstock") or header.endswith("_contribution") or header in ["base_demand", "total_media_contribution", "expected_subscriptions"] ], } def write_outputs() -> None: rows = generate_rows() headers = list(rows[0].keys()) csv_lines = [",".join(headers)] csv_lines.extend(",".join(js_number(row[header]) for header in headers) for row in rows) OUTPUT_DIR.mkdir(parents=True, exist_ok=True) CSV_PATH.write_text("\n".join(csv_lines) + "\n") PARAMS_PATH.write_text(json.dumps(build_params(headers), indent=2) + "\n") print(f"Wrote {len(rows)} rows to {CSV_PATH}") print(f"Wrote parameters to {PARAMS_PATH}") if __name__ == "__main__": write_outputs()