Marketing Analytics Dashboards — KPI Selection, Data Studio & Attribution

class KPISelector:
    """Select relevant KPIs based on business stage and goals."""

    def __init__(self, business_stage="growth"):
        self.stage = business_stage
        self.kpi_definitions = {
            "ctr": {"name": "Click-Through Rate", "formula": "clicks / impressions"},
            "cvr": {"name": "Conversion Rate", "formula": "conversions / sessions"},
            "cpa": {"name": "Cost Per Acquisition", "formula": "ad_spend / conversions"},
            "roas": {"name": "Return on Ad Spend", "formula": "revenue / ad_spend"},
            "ltv": {"name": "Lifetime Value", "formula": "avg_revenue * avg_lifetime"},
            "churn": {"name": "Churn Rate", "formula": "lost_customers / total_customers"},
            "nps": {"name": "Net Promoter Score", "formula": "promoters - detractors"},
            "mql": {"name": "Marketing Qualified Leads", "formula": "total leads * lead_score"},
        }

    def recommend(self, channels=None, budget_range=None):
        """Recommend KPIs based on context."""
        stage_kpis = {
            "awareness": ["impressions", "reach", "cpm", "frequency"],
            "growth": ["ctr", "cvr", "cpa", "roas", "mql"],
            "retention": ["churn", "ltv", "repeat_rate", "nps"],
            "maturity": ["ltv:cac", "mql_to_sql", "pipeline_velocity"],
        }

        recommended = stage_kpis.get(self.stage, stage_kpis["growth"])
        return {
            "stage": self.stage,
            "recommended_kpis": recommended,
            "definitions": {k: self.kpi_definitions.get(k) for k in recommended 
                           if k in self.kpi_definitions},
        }

selector = KPISelector("growth")
result = selector.recommend()
print(f"Recommended KPIs for {result['stage']} stage:")
for kpi in result["recommended_kpis"]:
    if kpi in result["definitions"]:
        print(f"  {result['definitions'][kpi]['name']:<30} = {result['definitions'][kpi]['formula']}")

class LookerStudioDashboard:
    """Simulate building a Looker Studio dashboard."""

    def __init__(self, name):
        self.name = name
        self.data_sources = []
        self.charts = []
        self.filters = []

    def add_data_source(self, name, connector, fields):
        """Connect a data source."""
        source = {"name": name, "connector": connector, "fields": fields, "connected": True}
        self.data_sources.append(source)
        print(f"[Looker] Connected: {name} ({connector})")

    def add_chart(self, chart_type, title, metric, dimension, data_source):
        """Add a chart to the dashboard."""
        chart = {
            "type": chart_type,
            "title": title,
            "metric": metric,
            "dimension": dimension,
            "source": data_source,
        }
        self.charts.append(chart)
        print(f"[Looker] Added {chart_type}: {title}")

    def add_date_filter(self, data_source):
        """Add a date range filter."""
        self.filters.append({"type": "date_range", "source": data_source})

    def preview(self):
        print(f"\n{'='*40}")
        print(f"Dashboard: {self.name}")
        print(f"{'='*40}")
        print(f"Data sources: {len(self.data_sources)}")
        print(f"Charts: {len(self.charts)}")
        print(f"Filters: {len(self.filters)}")
        print(f"\nChart overview:")
        for c in self.charts:
            print(f"  [{c['type']:>10}] {c['title']} ({c['metric']})")

# Build a marketing dashboard
dash = LookerStudioDashboard("Marketing Performance — Q3 2026")
dash.add_data_source("Google Analytics 4", "GA4", 
                     ["sessions", "users", "pageviews", "conversions"])
dash.add_data_source("Google Ads", "Ads",
                     ["impressions", "clicks", "cost", "conversions"])
dash.add_data_source("CRM", "BigQuery",
                     ["leads", "opportunities", "revenue"])

dash.add_chart("timeseries", "Sessions Over Time", "sessions", "date", "GA4")
dash.add_chart("bar", "Conversions by Channel", "conversions", "channel", "GA4")
dash.add_chart("scorecard", "Total Revenue", "revenue", None, "CRM")
dash.add_chart("pie", "Ad Spend Distribution", "cost", "campaign", "Ads")
dash.add_date_filter("GA4")
dash.preview()

class TableauDashboard:
    """Simulate Tableau dashboard with calculated fields."""

    def __init__(self, name):
        self.name = name
        self.data = {}
        self.calculated_fields = []
        self.worksheets = []

    def load_data(self, name, records):
        """Load dataset."""
        self.data[name] = records
        print(f"[Tableau] Loaded {name}: {len(records)} rows")

    def add_calculated_field(self, name, formula):
        """Create a calculated field."""
        field = {"name": name, "formula": formula}
        self.calculated_fields.append(field)
        print(f"[Tableau] Calculated field: {name} = {formula}")
        return field

    def add_worksheet(self, name, chart_type, columns, row_level=None):
        """Add a worksheet."""
        ws = {"name": name, "type": chart_type, "columns": columns, "row_level": row_level}
        self.worksheets.append(ws)
        print(f"[Tableau] Worksheet: {name} ({chart_type})")

    def preview(self):
        print(f"\n{'='*50}")
        print(f"Tableau Dashboard: {self.name}")
        print(f"{'='*50}")
        print(f"Datasets: {list(self.data.keys())}")
        print(f"Calculated fields:")
        for f in self.calculated_fields:
            print(f"  {f['name']:<25} = {f['formula']}")
        print(f"\nWorksheets:")
        for ws in self.worksheets:
            print(f"  [{ws['type']:>10}] {ws['name']}")

# Build a Tableau marketing dashboard
tableau = TableauDashboard("Channel Attribution — Q3 2026")
tableau.load_data("campaign_performance", [
    {"channel": "Google Ads", "spend": 5000, "impressions": 150000, 
     "clicks": 4500, "conversions": 120, "revenue": 24000},
    {"channel": "Facebook", "spend": 3000, "impressions": 200000,
     "clicks": 3200, "conversions": 80, "revenue": 16000},
    {"channel": "LinkedIn", "spend": 2000, "impressions": 50000,
     "clicks": 800, "conversions": 25, "revenue": 7500},
    {"channel": "Email", "spend": 1000, "impressions": 0,
     "clicks": 2500, "conversions": 60, "revenue": 12000},
])

tableau.add_calculated_field("CTR", "SUM(clicks) / SUM(impressions)")
tableau.add_calculated_field("CVR", "SUM(conversions) / SUM(clicks)")
tableau.add_calculated_field("CPA", "SUM(spend) / SUM(conversions)")
tableau.add_calculated_field("ROAS", "SUM(revenue) / SUM(spend)")

tableau.add_worksheet("Channel Summary", "bar", 
                      ["channel", "spend", "revenue", "roas"], "channel")
tableau.add_worksheet("Efficiency Scatter", "scatter", 
                      ["cpa", "cvr"], "channel")
tableau.add_worksheet("ROAS Trend", "timeseries", 
                      ["month", "roas", "channel"])

# Calculate and display some metrics
print("\nChannel performance:")
for row in tableau.data["campaign_performance"]:
    roas = row["revenue"] / row["spend"] if row["spend"] > 0 else 0
    print(f"  {row['channel']:<12} ROAS={roas:.2f}  "
          f"revenue=${row['revenue']:<6}  cost=${row['spend']:<5}")

class AttributionModeler:
    """Apply different attribution models to customer journeys."""

    def __init__(self):
        self.journeys = []

    def add_journey(self, customer_id, touchpoints, converted=True):
        """Add a customer journey: list of (channel, timestamp) tuples."""
        self.journeys.append({
            "customer": customer_id,
            "touchpoints": touchpoints,
            "converted": converted,
            "conversion_value": 100 if converted else 0,
        })

    def first_touch(self):
        """First-touch attribution."""
        credit = {}
        for j in self.journeys:
            if j["touchpoints"] and j["converted"]:
                first = j["touchpoints"][0][0]
                credit[first] = credit.get(first, 0) + j["conversion_value"]
        return credit

    def last_touch(self):
        """Last-touch attribution."""
        credit = {}
        for j in self.journeys:
            if j["touchpoints"] and j["converted"]:
                last = j["touchpoints"][-1][0]
                credit[last] = credit.get(last, 0) + j["conversion_value"]
        return credit

    def linear(self):
        """Linear: equal credit to all touchpoints."""
        credit = {}
        for j in self.journeys:
            if j["touchpoints"] and j["converted"]:
                value_per = j["conversion_value"] / len(j["touchpoints"])
                for channel, _ in j["touchpoints"]:
                    credit[channel] = credit.get(channel, 0) + value_per
        return credit

    def time_decay(self, decay_factor=0.5):
        """Time-decay: more credit to recent touchpoints."""
        credit = {}
        for j in self.journeys:
            if len(j["touchpoints"]) < 2 or not j["converted"]:
                continue
            n = len(j["touchpoints"])
            weights = [decay_factor ** (n - i - 1) for i in range(n)]
            total_weight = sum(weights)
            for (channel, _), weight in zip(j["touchpoints"], weights):
                credit[channel] = credit.get(channel, 0) + (
                    j["conversion_value"] * weight / total_weight)
        return credit

    def compare_models(self):
        """Compare all attribution models."""
        models = {
            "first_touch": self.first_touch(),
            "last_touch": self.last_touch(),
            "linear": self.linear(),
            "time_decay": self.time_decay(),
        }

        print(f"\n{'Model':<15} {'Google Ads':<12} {'Facebook':<12} {'Email':<12} {'LinkedIn':<12}")
        print("-" * 63)
        for model_name, credits in models.items():
            print(f"{model_name:<15} "
                  f"${credits.get('Google Ads', 0):<10.0f} "
                  f"${credits.get('Facebook', 0):<10.0f} "
                  f"${credits.get('Email', 0):<10.0f} "
                  f"${credits.get('LinkedIn', 0):<10.0f}")

# Simulate customer journeys
attribution = AttributionModeler()
attribution.add_journey("C001", [("Google Ads", 1), ("Email", 3), ("Direct", 7)])
attribution.add_journey("C002", [("Facebook", 2), ("Google Ads", 5)])
attribution.add_journey("C003", [("LinkedIn", 1), ("Facebook", 2), ("Email", 4), ("Google Ads", 6)])
attribution.add_journey("C004", [("Email", 1)])
attribution.add_journey("C005", [("Facebook", 2), ("LinkedIn", 3), ("Direct", 8)])

attribution.compare_models()

class DashboardDesigner:
    """Evaluate dashboard design quality."""

    def __init__(self):
        self.criteria = {
            "north_star": False,
            "comparison": False,
            "drill_down": False,
            "alerts": False,
            "freshness": False,
        }

    def evaluate(self, north_star=True, comparison=True, drill=True, alerts=True, fresh=True):
        scores = {
            "north_star": 25 if north_star else 0,
            "comparison": 20 if comparison else 0,
            "drill_down": 20 if drill else 0,
            "alerts": 20 if alerts else 0,
            "freshness": 15 if fresh else 0,
        }
        total = sum(scores.values())
        grade = "A" if total >= 85 else "B" if total >= 70 else "C"
        return {"score": total, "grade": grade, "details": scores}

design = DashboardDesigner()
result = design.evaluate(north_star=True, comparison=True, drill=True, alerts=False, fresh=True)
print(f"Dashboard quality: {result['grade']} ({result['score']}/100)")

# dashboard_builder.py
# Build a complete marketing dashboard with data

class MarketingDashboard:
    def __init__(self, brand, period):
        self.brand = brand
        self.period = period
        self.channels = {}

    def add_channel(self, name, spend, impressions, clicks, conversions, revenue):
        self.channels[name] = {
            "spend": spend,
            "impressions": impressions,
            "clicks": clicks,
            "conversions": conversions,
            "revenue": revenue,
            "ctr": round(clicks/impressions*100, 2) if impressions else 0,
            "cvr": round(conversions/clicks*100, 2) if clicks else 0,
            "cpa": round(spend/conversions, 2) if conversions else 0,
            "roas": round(revenue/spend, 2) if spend else 0,
        }

    def summary(self):
        total = {k: sum(c[k] for c in self.channels.values()) 
                 for k in ["spend", "impressions", "clicks", "conversions", "revenue"]}
        print(f"\n{'='*55}")
        print(f"  {self.brand} Marketing Dashboard — {self.period}")
        print(f"{'='*55}")
        print(f"{'Channel':<15} {'Spend':<10} {'CTR':<8} {'CVR':<8} {'CPA':<8} {'ROAS':<8}")
        print("-" * 55)
        for name, data in self.channels.items():
            print(f"{name:<15} ${data['spend']:<7.0f} {data['ctr']:<7.2f}% "
                  f"{data['cvr']:<7.2f}% ${data['cpa']:<6.2f} {data['roas']:<7.2f}x")
        print("-" * 55)
        t = total
        print(f"{'TOTAL':<15} ${t['spend']:<7.0f} "
              f"{round(t['clicks']/t['impressions']*100,2) if t['impressions'] else 0:<7.2f}% "
              f"{round(t['conversions']/t['clicks']*100,2) if t['clicks'] else 0:<7.2f}% "
              f"${round(t['spend']/t['conversions'],2) if t['conversions'] else 0:<6.2f} "
              f"{round(t['revenue']/t['spend'],2) if t['spend'] else 0:<7.2f}x")

dash = MarketingDashboard("DodaTech", "Q3 2026")
dash.add_channel("Google Ads", 5000, 150000, 4500, 120, 24000)
dash.add_channel("Facebook", 3000, 200000, 3200, 80, 16000)
dash.add_channel("LinkedIn", 2000, 50000, 800, 25, 7500)
dash.add_channel("Email", 1000, 0, 2500, 60, 12000)
dash.summary()