The Expert Analyst Mindset — Framing Problems Before Touching Data

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


@dataclass
class AnalysisBrief:
    """
    Structured template for scoping an analytical project.
    Fill this out before opening a notebook.
    """

    # Core framing
    project_title: str
    business_objective: str          # The decision this analysis will enable
    analytical_question: str         # The precise, falsifiable question being answered

    # Hypothesis
    primary_hypothesis: str          # What you expect to find (and why)
    null_hypothesis: str             # The default assumption to be tested against

    # Data
    data_sources: List[str]          # Tables, files, APIs involved
    date_range: str                  # "2023-01-01 to 2024-06-30"
    population: str                  # Who / what is being analysed

    # Metrics
    primary_metric: str              # The one number that answers the question
    secondary_metrics: List[str]     # Supporting signals

    # Constraints
    known_data_issues: List[str] = field(default_factory=list)
    out_of_scope: List[str] = field(default_factory=list)
    assumptions: List[str] = field(default_factory=list)

    # Delivery
    deliverable_format: str = "Jupyter notebook + 1-page memo"
    deadline: Optional[date] = None
    stakeholders: List[str] = field(default_factory=list)

    def render(self) -> str:
        """Render the brief as a readable string for sharing."""
        lines = [
            f"# Analytical Brief: {self.project_title}",
            "",
            f"**Business Objective:** {self.business_objective}",
            f"**Analytical Question:** {self.analytical_question}",
            "",
            f"**Primary Hypothesis:** {self.primary_hypothesis}",
            f"**Null Hypothesis:** {self.null_hypothesis}",
            "",
            "**Data Sources:**",
            *[f"  - {s}" for s in self.data_sources],
            f"**Date Range:** {self.date_range}",
            f"**Population:** {self.population}",
            "",
            f"**Primary Metric:** {self.primary_metric}",
            "**Secondary Metrics:**",
            *[f"  - {m}" for m in self.secondary_metrics],
            "",
            "**Known Data Issues:**",
            *[f"  - {i}" for i in self.known_data_issues],
            "**Out of Scope:**",
            *[f"  - {o}" for o in self.out_of_scope],
            "**Assumptions:**",
            *[f"  - {a}" for a in self.assumptions],
            "",
            f"**Deliverable:** {self.deliverable_format}",
            f"**Deadline:** {self.deadline}",
            f"**Stakeholders:** {', '.join(self.stakeholders)}",
        ]
        return "\n".join(lines)


# Example: filled brief for the "sales are down" problem
brief = AnalysisBrief(
    project_title="Q3 Revenue Decline Root Cause Analysis",
    business_objective=(
        "Determine where to focus retention and re-engagement spend "
        "in Q4 based on which customer segments drove the Q3 revenue decline."
    ),
    analytical_question=(
        "Is the Q3 revenue decline concentrated in a specific customer segment "
        "or acquisition cohort, and if so, what behavioural changes in that "
        "segment explain the decline?"
    ),
    primary_hypothesis=(
        "The decline is concentrated in SMB customers acquired in 2022 "
        "who experienced a price increase in July 2023 and have not returned."
    ),
    null_hypothesis=(
        "The decline is proportional across all segments, consistent with "
        "broad market seasonality."
    ),
    data_sources=[
        "orders table (transactions 2022-01-01 to 2023-09-30)",
        "customers table (all registered accounts)",
        "events table (login, browse, cart events)",
        "pricing_history table (price changes by SKU)",
    ],
    date_range="2022-01-01 to 2023-09-30",
    population="All customers who placed at least one order in 2022 or 2023",
    primary_metric="Monthly revenue by customer segment (YoY % change)",
    secondary_metrics=[
        "Order frequency per customer per month",
        "Average order value",
        "Churn rate (no purchase in 90 days)",
        "Reactivation rate after price change",
    ],
    known_data_issues=[
        "Orders table has ~3% rows with null customer_id (guest checkouts)",
        "Events table only available from 2022-06-01",
    ],
    out_of_scope=[
        "New customer acquisition analysis",
        "Supply chain or inventory issues",
        "Competitor pricing analysis",
    ],
    assumptions=[
        "Seasonality follows prior year pattern",
        "Segment definitions are stable (no reclassification in period)",
    ],
    deliverable_format="Jupyter notebook + 1-page executive memo",
    deadline=date(2023, 11, 10),
    stakeholders=["VP Revenue", "Head of CRM", "CFO"],
)

print(brief.render())

from enum import Enum


class QuestionType(Enum):
    DESCRIPTIVE = "descriptive"
    DIAGNOSTIC = "diagnostic"
    PREDICTIVE = "predictive"
    PRESCRIPTIVE = "prescriptive"


def classify_question(question: str) -> dict:
    """
    A simple classifier that maps question keywords to type.
    In practice, this is a judgment call — use this as a checklist prompt.
    """
    q_lower = question.lower()

    # Keyword signals (heuristic only — always confirm with judgment)
    descriptive_signals = ["what is", "what was", "how many", "how much", "summarise", "total", "average"]
    diagnostic_signals = ["why", "what caused", "what drove", "explain", "reason", "factor"]
    predictive_signals = ["predict", "forecast", "likely", "will", "future", "next"]
    prescriptive_signals = ["should", "recommend", "optimise", "best action", "maximise", "minimise"]

    scores = {
        QuestionType.DESCRIPTIVE: sum(s in q_lower for s in descriptive_signals),
        QuestionType.DIAGNOSTIC: sum(s in q_lower for s in diagnostic_signals),
        QuestionType.PREDICTIVE: sum(s in q_lower for s in predictive_signals),
        QuestionType.PRESCRIPTIVE: sum(s in q_lower for s in prescriptive_signals),
    }

    best = max(scores, key=scores.get)
    return {
        "detected_type": best.value,
        "scores": {k.value: v for k, v in scores.items()},
        "causal_inference_required": best in (QuestionType.DIAGNOSTIC, QuestionType.PRESCRIPTIVE),
        "experiment_recommended": best == QuestionType.PRESCRIPTIVE,
    }


# Test
examples = [
    "What was our monthly revenue in 2023?",
    "Why did churn increase after the July price change?",
    "Which customers are most likely to purchase again next month?",
    "Which customers should we email to maximise reactivation revenue?",
]

for q in examples:
    result = classify_question(q)
    print(f"Q: {q}")
    print(f"   Type: {result['detected_type']}")
    print(f"   Causal inference required: {result['causal_inference_required']}")
    print()

def so_what_test(finding: str, implication: str, recommended_action: str) -> dict:
    """
    Force yourself to complete the full insight chain before writing up.
    """
    is_actionable = len(recommended_action.strip()) > 0
    is_quantified = any(
        char.isdigit() for char in finding
    )  # Weak proxy — real check is manual

    return {
        "finding": finding,
        "implication": implication,
        "recommended_action": recommended_action,
        "is_actionable": is_actionable,
        "appears_quantified": is_quantified,
        "passes_so_what": is_actionable and is_quantified,
    }


result = so_what_test(
    finding="SMB customers acquired in 2022 have a 34% higher churn rate post price-increase vs Enterprise customers.",
    implication="Price sensitivity is concentrated in the SMB segment — Enterprise customers have higher stickiness.",
    recommended_action=(
        "Offer SMB customers acquired before July 2023 a grandfathered pricing tier "
        "for 12 months to reduce churn; evaluate ROI vs full-price recovery."
    ),
)

for k, v in result.items():
    print(f"{k}: {v}")

# Survivorship bias detection: check for selection effects
import pandas as pd
import numpy as np

# Simulated orders table (only completed orders — guest checkouts excluded)
np.random.seed(42)
n = 1000

orders = pd.DataFrame({
    "order_id": range(1, n + 1),
    "customer_id": np.random.choice(range(1, 400), size=n),  # 400 customers, repeat buyers
    "status": np.random.choice(["completed", "abandoned", "cancelled"], p=[0.6, 0.3, 0.1], size=n),
    "revenue": np.where(
        np.random.choice([True, False], p=[0.6, 0.4], size=n),
        np.random.exponential(scale=85, size=n),
        0.0,
    ),
    "channel": np.random.choice(["organic", "paid", "email", "direct"], size=n),
})

# TRAP: Naive analyst looks only at completed orders
completed = orders[orders["status"] == "completed"]
print("Average revenue (completed only):", completed["revenue"].mean().round(2))
print("Channel mix (completed only):")
print(completed["channel"].value_counts(normalize=True).round(3))

print()

# CORRECT: Always report the full funnel
print("Order status distribution:")
print(orders["status"].value_counts(normalize=True).round(3))

print()
print("Completion rate by channel:")
channel_funnel = (
    orders.groupby("channel")["status"]
    .value_counts(normalize=True)
    .unstack(fill_value=0)
    .round(3)
)
print(channel_funnel)

# KEY INSIGHT: If abandonment rate differs by channel,
# the "best" channel analysis on completed orders alone is misleading.

from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum


class Complexity(Enum):
    LOW = "low"       # < 2 hours
    MEDIUM = "medium" # 2–8 hours
    HIGH = "high"     # > 8 hours


@dataclass
class AnalysisStep:
    step_id: int
    name: str
    description: str
    complexity: Complexity
    data_dependencies: List[str]
    output: str                   # What artifact this step produces
    depends_on_steps: List[int] = field(default_factory=list)
    notes: Optional[str] = None


@dataclass
class AnalysisPlan:
    brief: AnalysisBrief
    steps: List[AnalysisStep]

    def total_estimated_hours(self) -> str:
        mapping = {Complexity.LOW: 1.5, Complexity.MEDIUM: 5, Complexity.HIGH: 12}
        low = sum(mapping[s.complexity] * 0.5 for s in self.steps)
        high = sum(mapping[s.complexity] * 1.5 for s in self.steps)
        return f"{low:.0f}–{high:.0f} hours"

    def render(self) -> None:
        print(f"Analysis Plan: {self.brief.project_title}")
        print(f"Estimated effort: {self.total_estimated_hours()}")
        print("-" * 60)
        for step in self.steps:
            deps = f" [depends on: {step.depends_on_steps}]" if step.depends_on_steps else ""
            print(f"Step {step.step_id}: {step.name} ({step.complexity.value}){deps}")
            print(f"  → {step.description}")
            print(f"  Output: {step.output}")
            if step.notes:
                print(f"  Note: {step.notes}")
            print()


# Example plan for the Q3 revenue analysis
plan = AnalysisPlan(
    brief=brief,
    steps=[
        AnalysisStep(
            step_id=1,
            name="Data Ingestion & Profiling",
            description="Load all four tables, run automated profiling, document schema and null patterns.",
            complexity=Complexity.LOW,
            data_dependencies=["orders", "customers", "events", "pricing_history"],
            output="profiling_report.html",
        ),
        AnalysisStep(
            step_id=2,
            name="Data Quality Assessment",
            description="Score each dimension of quality, identify issues requiring cleaning decisions.",
            complexity=Complexity.MEDIUM,
            data_dependencies=["orders", "customers"],
            output="data_quality_report.md",
            depends_on_steps=[1],
        ),
        AnalysisStep(
            step_id=3,
            name="Revenue Trend Analysis",
            description="Monthly revenue YoY comparison, overall and by channel.",
            complexity=Complexity.LOW,
            data_dependencies=["orders"],
            output="revenue_trend_chart.png",
            depends_on_steps=[2],
        ),
        AnalysisStep(
            step_id=4,
            name="Segment Decomposition",
            description="Define customer segments (SMB/Enterprise/Consumer), compute segment-level revenue trends.",
            complexity=Complexity.MEDIUM,
            data_dependencies=["orders", "customers"],
            output="segment_revenue_breakdown.csv + charts",
            depends_on_steps=[2],
        ),
        AnalysisStep(
            step_id=5,
            name="Price Change Impact Analysis",
            description="Cohort analysis: compare order frequency and revenue before/after July price change by segment.",
            complexity=Complexity.HIGH,
            data_dependencies=["orders", "customers", "pricing_history"],
            output="price_impact_cohort_analysis.ipynb",
            depends_on_steps=[3, 4],
            notes="Requires merging pricing history to orders by SKU and date — test join logic carefully.",
        ),
        AnalysisStep(
            step_id=6,
            name="Hypothesis Testing",
            description="Statistical test: is SMB churn rate post price-increase significantly different from Enterprise?",
            complexity=Complexity.MEDIUM,
            data_dependencies=["orders", "customers"],
            output="hypothesis_test_results.md",
            depends_on_steps=[4, 5],
        ),
        AnalysisStep(
            step_id=7,
            name="Insight Synthesis & Memo",
            description="Write executive memo (1 page), compile findings notebook, prepare stakeholder presentation.",
            complexity=Complexity.MEDIUM,
            data_dependencies=[],
            output="executive_memo.pdf + analysis_final.ipynb",
            depends_on_steps=[6],
        ),
    ],
)

plan.render()

# ============================================================
# CELL 1: Title and context
# ============================================================
"""
# Q3 Revenue Decline Root Cause Analysis
**Project:** GadaaLabs E-Commerce Revenue Analysis
**Analyst:** [Your Name]
**Date:** 2023-11-01
**Version:** v1.0 — initial analysis

## Objective
Determine whether Q3 revenue decline is concentrated in a specific
customer segment, and identify the behavioural drivers.

## Deliverables
- This notebook (full analysis)
- Executive memo (1-page summary)
- Stakeholder slides (top 5 findings)
"""

# ============================================================
# CELL 2: Imports and configuration
# ============================================================
import warnings
warnings.filterwarnings("ignore", category=FutureWarning)

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path
from datetime import datetime

pd.set_option("display.max_columns", 50)
pd.set_option("display.max_rows", 100)
pd.set_option("display.float_format", "{:,.2f}".format)

# Project paths
DATA_DIR = Path("../data")
OUTPUT_DIR = Path("../outputs")
REPORT_DIR = Path("../reports")

OUTPUT_DIR.mkdir(exist_ok=True)
REPORT_DIR.mkdir(exist_ok=True)

# Analysis parameters (centralised — change once, applies everywhere)
ANALYSIS_START = "2022-01-01"
ANALYSIS_END = "2023-09-30"
CHURN_WINDOW_DAYS = 90          # No purchase in 90 days = churned
PRICE_CHANGE_DATE = "2023-07-01"
SIGNIFICANCE_LEVEL = 0.05

print(f"Analysis period: {ANALYSIS_START} to {ANALYSIS_END}")
print(f"Notebook run at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

# ============================================================
# CELL 3: Data loading (stub — filled in Lesson 02)
# ============================================================
# orders = pd.read_csv(DATA_DIR / "orders.csv", parse_dates=["order_date"])
# customers = pd.read_csv(DATA_DIR / "customers.csv", parse_dates=["signup_date"])
# print(f"Loaded {len(orders):,} orders, {len(customers):,} customers")

# ============================================================
# CELL 4: Findings log (updated throughout analysis)
# ============================================================
findings = []  # List of dicts: {section, finding, implication, confidence}

def log_finding(section: str, finding: str, implication: str, confidence: str = "medium") -> None:
    """Add a finding to the running log."""
    findings.append({
        "section": section,
        "finding": finding,
        "implication": implication,
        "confidence": confidence,
    })
    print(f"[FINDING — {section}] {finding}")

# Usage throughout the notebook:
# log_finding(
#     section="Revenue Trend",
#     finding="Revenue declined 18% YoY in Q3 (Jul–Sep 2023).",
#     implication="Decline is material — not seasonal noise.",
#     confidence="high",
# )