Building Data Products: From Engineer to Product Thinker

"""
Problem-first approach to data product development.
"""

# Wrong approach: Data-first
def data_first_thinking():
    """
    "We have click data. Let's build a clickstream dashboard."

    Problems with this approach:
    - No defined user
    - No defined problem
    - Success is undefined
    - Likely to become shelfware
    """
    pass


# Right approach: Problem-first
def problem_first_thinking():
    """
    Problem: Marketing can't measure campaign effectiveness.
    User: Marketing analytics team (5 people)

    Questions to answer:
    - Which campaigns drive conversions?
    - What's the cost per acquisition by channel?
    - How do campaigns perform over time?

    Data needed:
    - Click/impression data (ad platforms)
    - Conversion events (product database)
    - Spend data (finance system)

    Success criteria:
    - Marketing can self-serve campaign analysis
    - Weekly campaign reviews use the product
    - Decision time reduced from days to hours
    """
    pass

"""
User research framework for data products.
"""

from dataclasses import dataclass
from typing import List


@dataclass
class UserPersona:
    """Define your data product's user."""

    role: str
    team: str
    technical_level: str  # low, medium, high
    current_tools: List[str]
    pain_points: List[str]
    goals: List[str]
    usage_frequency: str  # daily, weekly, monthly


# Example persona
marketing_analyst = UserPersona(
    role="Marketing Analyst",
    team="Growth Marketing",
    technical_level="medium",
    current_tools=["Excel", "Google Analytics", "Salesforce reports"],
    pain_points=[
        "Spend hours combining data from multiple sources",
        "Numbers don't match between systems",
        "Can't get historical trends easily",
        "Rely on data team for any custom analysis"
    ],
    goals=[
        "Self-serve common analyses",
        "Understand campaign performance quickly",
        "Make data-driven budget recommendations",
        "Impress leadership with insights"
    ],
    usage_frequency="daily"
)


def validate_product_fit(product_features: List[str], persona: UserPersona) -> dict:
    """
    Check if product features address user pain points.
    """
    addressed = []
    unaddressed = []

    pain_point_keywords = {
        "combining data": ["integrated", "unified", "single source"],
        "numbers don't match": ["consistent", "validated", "single source of truth"],
        "historical trends": ["time series", "historical", "trending"],
        "rely on data team": ["self-serve", "no-code", "drag and drop"]
    }

    for pain in persona.pain_points:
        pain_lower = pain.lower()
        is_addressed = False

        for keyword, solutions in pain_point_keywords.items():
            if keyword in pain_lower:
                for feature in product_features:
                    if any(sol in feature.lower() for sol in solutions):
                        addressed.append((pain, feature))
                        is_addressed = True
                        break

        if not is_addressed:
            unaddressed.append(pain)

    return {
        "addressed_pain_points": addressed,
        "unaddressed_pain_points": unaddressed,
        "coverage": len(addressed) / len(persona.pain_points) * 100
    }

"""
Success metrics framework for data products.
"""

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime


@dataclass
class SuccessMetric:
    """Define a measurable success metric."""

    name: str
    metric_type: str  # adoption, engagement, impact
    target: float
    current: Optional[float] = None
    measurement_method: str = ""


@dataclass
class DataProductMetrics:
    """Complete metrics package for a data product."""

    product_name: str
    launch_date: datetime
    metrics: List[SuccessMetric]

    def health_check(self) -> dict:
        """Assess overall product health."""

        results = {
            "adoption": [],
            "engagement": [],
            "impact": []
        }

        for metric in self.metrics:
            if metric.current is not None:
                achievement = metric.current / metric.target * 100
                status = "green" if achievement >= 80 else "yellow" if achievement >= 50 else "red"

                results[metric.metric_type].append({
                    "name": metric.name,
                    "target": metric.target,
                    "current": metric.current,
                    "achievement": f"{achievement:.0f}%",
                    "status": status
                })

        return results


# Example: CLV Dashboard metrics
clv_dashboard_metrics = DataProductMetrics(
    product_name="Customer Lifetime Value Dashboard",
    launch_date=datetime(2025, 1, 15),
    metrics=[
        # Adoption
        SuccessMetric(
            name="Weekly Active Users",
            metric_type="adoption",
            target=12,  # 80% of 15 target users
            current=10,
            measurement_method="Count distinct users with dashboard views per week"
        ),
        SuccessMetric(
            name="Onboarding Completion",
            metric_type="adoption",
            target=100,  # percentage
            current=85,
            measurement_method="Users who completed training / total users"
        ),

        # Engagement
        SuccessMetric(
            name="Average Session Duration",
            metric_type="engagement",
            target=15,  # minutes
            current=12,
            measurement_method="Average time spent in dashboard per session"
        ),
        SuccessMetric(
            name="Filters Applied per Session",
            metric_type="engagement",
            target=5,
            current=7,
            measurement_method="Average filter interactions per session"
        ),

        # Impact
        SuccessMetric(
            name="Campaign Targeting Improvement",
            metric_type="impact",
            target=20,  # percentage improvement
            current=15,
            measurement_method="Compare campaign ROI before/after dashboard launch"
        ),
        SuccessMetric(
            name="Decision Time Reduction",
            metric_type="impact",
            target=75,  # percentage reduction
            current=60,
            measurement_method="Survey: time to make campaign decisions"
        )
    ]
)

-- Example: Design document excerpt
-- CLV Dashboard Data Model

-- Core fact table
CREATE TABLE fact_customer_value (
    customer_id         VARCHAR(50) PRIMARY KEY,
    first_order_date    DATE,
    last_order_date     DATE,
    total_orders        INT,
    total_revenue       DECIMAL(12,2),
    avg_order_value     DECIMAL(10,2),
    customer_tenure_days INT,
    predicted_ltv       DECIMAL(12,2),
    ltv_segment         VARCHAR(20),  -- 'high', 'medium', 'low'
    churn_risk_score    DECIMAL(5,2),
    updated_at          TIMESTAMP
);

-- Supporting dimension
CREATE TABLE dim_customer (
    customer_id         VARCHAR(50) PRIMARY KEY,
    email               VARCHAR(255),
    acquisition_channel VARCHAR(50),
    acquisition_date    DATE,
    customer_type       VARCHAR(20),  -- 'b2b', 'b2c'
    industry            VARCHAR(100),
    company_size        VARCHAR(20)
);

-- Aggregated for dashboard performance
CREATE TABLE agg_ltv_by_segment (
    report_date         DATE,
    ltv_segment         VARCHAR(20),
    customer_count      INT,
    total_revenue       DECIMAL(14,2),
    avg_ltv             DECIMAL(12,2),
    avg_order_frequency DECIMAL(6,2)
);

"""
Incremental build approach for data products.
"""

# Bad: Build all components, then integrate
def waterfall_build():
    """
    Week 1-2: Build all extraction
    Week 3-4: Build all transformation
    Week 5-6: Build all loading
    Week 7-8: Integration and testing

    Problem: No working product until week 8.
    No user feedback until the end.
    """
    pass


# Good: Build vertical slices
def incremental_build():
    """
    Week 1-2: Basic LTV calculation (one source, simple calc)
             - Users can see basic LTV scores
             - Get feedback on approach

    Week 3-4: Add acquisition channel dimension
             - Users can segment by channel
             - Validate channel mapping

    Week 5-6: Add predictive LTV model
             - Users get forward-looking view
             - Refine model based on feedback

    Week 7-8: Add self-serve filters and exports
             - Full product capability
             - Based on 6 weeks of feedback
    """
    pass

"""
Launch checklist for data products.
"""

launch_checklist = {
    "documentation": [
        "User guide created",
        "FAQ document ready",
        "Data dictionary complete",
        "Known limitations documented"
    ],
    "training": [
        "Training sessions scheduled",
        "Recording available",
        "Quick reference card created",
        "Office hours established"
    ],
    "monitoring": [
        "Usage tracking enabled",
        "Data quality alerts configured",
        "Performance monitoring active",
        "Feedback channel created"
    ],
    "support": [
        "Support process defined",
        "Escalation path clear",
        "SLA communicated",
        "FAQ channel created"
    ]
}