Python Programs for Practice: 20 Real Analytics Scripts

import re
 
raw_phone_records = [
    {"user_id": 101, "phone": "+91 98765-43210"},
    {"user_id": 102, "phone": "098765 43210"},
    {"user_id": 103, "phone": "9876543210"},
    {"user_id": 104, "phone": "+1 (555) 234-5678"},
    {"user_id": 105, "phone": "invalid_number_123"},
]
 
def sanitize_phone_number(raw_phone: str, default_country_code: str = "91") -> str | None:
    """Strips formatting artifacts and enforces E.164 phone numbering standards."""
    # Strip all characters except digits and leading plus
    has_plus = raw_phone.strip().startswith("+")
    digits_only = re.sub(r"\D", "", raw_phone)
    
    # Strip leading trunk zeros (common in UK, India, Australia)
    if digits_only.startswith("0") and len(digits_only) == 11:
        digits_only = digits_only[1:]
 
    # Handle standard 10-digit mobile numbers without country codes
    if len(digits_only) == 10:
        return f"+{default_country_code}{digits_only}"
    
    # Handle numbers that already included country code (e.g. 12 digits for 91 + 10 digits)
    if len(digits_only) in (11, 12):
        return f"+{digits_only}"
        
    return None  # Unparseable record
 
cleaned_records = []
for record in raw_phone_records:
    normalized = sanitize_phone_number(record["phone"])
    cleaned_records.append({
        "user_id": record["user_id"],
        "original": record["phone"],
        "e164_phone": normalized,
        "is_valid": normalized is not None
    })
 
for rec in cleaned_records:
    print(f"User {rec['user_id']}: {rec['original']:<20} -> {rec['e164_phone']} (Valid: {rec['is_valid']})")

from typing import Any
 
api_payload = {
    "order_id": "ORD-88219",
    "timestamp": "2026-09-12T14:32:00Z",
    "customer": {
        "id": "CUST-402",
        "profile": {
            "first_name": "Aarav",
            "last_name": "Patel",
            "email": "aarav.patel@enterprise.in"
        }
    },
    "payment": {
        "gateway": "Razorpay",
        "currency": "INR",
        "amount_paise": 249900,
        "status": "captured"
    }
}
 
def flatten_dictionary(nested_dict: dict[str, Any], parent_key: str = "", separator: str = "_") -> dict[str, Any]:
    """Recursively flattens nested dictionaries into a single-level key-value mapping."""
    flat_record = {}
    for key, value in nested_dict.items():
        new_key = f"{parent_key}{separator}{key}" if parent_key else key
        if isinstance(value, dict):
            flat_record.update(flatten_dictionary(value, parent_key=new_key, separator=separator))
        else:
            flat_record[new_key] = value
    return flat_record
 
flattened = flatten_dictionary(api_payload)
for flat_key, flat_val in flattened.items():
    print(f"{flat_key:<32}: {flat_val}")

raw_events = [
    {"event_id": 1, "account_id": "ACC-01", "invoice_id": "INV-100", "amount": 4500},
    {"event_id": 2, "account_id": "ACC-02", "invoice_id": "INV-101", "amount": 9200},
    {"event_id": 3, "account_id": "ACC-01", "invoice_id": "INV-100", "amount": 4500},  # Duplicate
    {"event_id": 4, "account_id": "ACC-03", "invoice_id": "INV-102", "amount": 1250},
    {"event_id": 5, "account_id": "ACC-02", "invoice_id": "INV-101", "amount": 9200},  # Duplicate
]
 
def deduplicate_events(events: list[dict], key_fields: tuple[str, ...]) -> tuple[list[dict], int]:
    """Filters duplicate dictionary records in O(N) time using an in-memory hash set."""
    seen_keys = set()
    deduped = []
    duplicate_count = 0
    
    for event in events:
        composite_key = tuple(event.get(k) for k in key_fields)
        if composite_key in seen_keys:
            duplicate_count += 1
            continue
        seen_keys.add(composite_key)
        deduped.append(event)
        
    return deduped, duplicate_count
 
unique_records, duplicates_dropped = deduplicate_events(raw_events, ("account_id", "invoice_id"))
print(f"Total Processed: {len(raw_events)} | Duplicates Dropped: {duplicates_dropped}\n")
for record in unique_records:
    print(record)

import re
 
leads = [
    {"lead_id": 1, "email": "priya.sharma@swiggy.in"},
    {"lead_id": 2, "email": "vikram99@gmail.com"},
    {"lead_id": 3, "email": "rohit@tcs.com"},
    {"lead_id": 4, "email": "bad_email_format@"},
    {"lead_id": 5, "email": "ananya@razorpay.com"},
]
 
FREE_EMAIL_PROVIDERS = {"gmail.com", "yahoo.com", "outlook.com", "hotmail.com", "icloud.com"}
EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9_.+-]+@([a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)$")
 
def classify_lead_email(email: str) -> dict[str, str | bool]:
    match = EMAIL_REGEX.match(email.strip().lower())
    if not match:
        return {"domain": "INVALID", "is_corporate": False, "is_valid": False}
        
    domain = match.group(1)
    is_corporate = domain not in FREE_EMAIL_PROVIDERS
    return {"domain": domain, "is_corporate": is_corporate, "is_valid": True}
 
classified_leads = []
for lead in leads:
    meta = classify_lead_email(lead["email"])
    classified_leads.append({**lead, **meta})
 
for lead in classified_leads:
    tier = "Corporate Tier 1" if lead["is_corporate"] else "Personal / Non-B2B"
    print(f"Lead {lead['lead_id']} ({lead['email']:<25}): Domain={lead['domain']:<15} -> {tier}")

raw_ledger_entries = [
    "$1,249.50",
    "₹ 99,990.00",
    "(€450.00)",      # Financial convention for negative 450
    " - ",            # Null entry indicator
    "($12,000.75)",   # Negative 12,000.75
    "Free Tier",      # Non-numeric promo code
]
 
def parse_financial_currency(value_str: str) -> float | None:
    """Parses international currency strings and accounting brackets into valid floats."""
    cleaned = value_str.strip()
    if not cleaned or cleaned == "-":
        return 0.0
        
    is_negative = cleaned.startswith("(") and cleaned.endswith(")")
    
    # Strip accounting parentheses, currency marks, whitespace, and commas
    cleaned = re.sub(r"[^\d.]", "", cleaned)
    
    if not cleaned:
        return None
        
    try:
        numeric_val = float(cleaned)
        return -numeric_val if is_negative else numeric_val
    except ValueError:
        return None
 
parsed_ledger = []
for entry in raw_ledger_entries:
    parsed_ledger.append({
        "raw": entry,
        "clean_amount": parse_financial_currency(entry)
    })
 
for item in parsed_ledger:
    print(f"Raw Entry: {item['raw']:<16} -> Clean Numeric: {item['clean_amount']}")

daily_signups = [120, 145, 130, 190, 210, 185, 230, 260, 240]
 
def calculate_moving_average(data: list[float | int], window_size: int = 3) -> list[float | None]:
    """Computes simple moving averages over a 1D sequence with warmup nulls."""
    if window_size <= 0 or window_size > len(data):
        raise ValueError("Invalid window size.")
        
    moving_averages: list[float | None] = []
    
    # First (window_size - 1) days cannot form a complete window
    for i in range(len(data)):
        if i < window_size - 1:
            moving_averages.append(None)
        else:
            window = data[i - window_size + 1 : i + 1]
            moving_averages.append(round(sum(window) / window_size, 2))
            
    return moving_averages
 
sma_3day = calculate_moving_average(daily_signups, window_size=3)
 
print("Day | Raw Signups | 3-Day Moving Average")
print("-" * 38)
for day_idx, (raw, sma) in enumerate(zip(daily_signups, sma_3day, strict=True), start=1):
    sma_str = f"{sma:.2f}" if sma is not None else "Warmup"
    print(f"{day_idx:02d}  | {raw:^11} | {sma_str:^20}")

import math
 
order_values = [240.0, 180.5, 310.0, 450.0, 190.0, 280.0, 390.0]
 
def compute_sample_statistics(values: list[float]) -> dict[str, float]:
    """Calculates sample mean, variance, and standard deviation using Bessel's correction."""
    n = len(values)
    if n < 2:
        raise ValueError("Sample variance requires at least two observations.")
        
    mean = sum(values) / n
    sum_squared_diffs = sum((x - mean) ** 2 for x in values)
    
    sample_variance = sum_squared_diffs / (n - 1)
    sample_std_dev = math.sqrt(sample_variance)
    
    return {
        "n": n,
        "mean": round(mean, 2),
        "sample_variance": round(sample_variance, 2),
        "sample_std_dev": round(sample_std_dev, 2)
    }
 
stats = compute_sample_statistics(order_values)
print(f"Sample Size (n): {stats['n']}")
print(f"Mean Order Value: ₹{stats['mean']}")
print(f"Sample Variance: {stats['sample_variance']}")
print(f"Sample Standard Deviation: ₹{stats['sample_std_dev']}")

server_response_latencies_ms = [42, 45, 48, 51, 44, 46, 43, 49, 380, 47, 52, 410]
 
def detect_outliers_zscore(data: list[float | int], threshold: float = 2.5) -> list[dict]:
    n = len(data)
    mean = sum(data) / n
    std_dev = math.sqrt(sum((x - mean) ** 2 for x in data) / (n - 1))
    
    flagged_results = []
    for val in data:
        z_score = (val - mean) / std_dev
        flagged_results.append({
            "value": val,
            "z_score": round(z_score, 2),
            "is_outlier": abs(z_score) > threshold
        })
    return flagged_results
 
audit_report = detect_outliers_zscore(server_response_latencies_ms, threshold=2.0)
print(f"{'Latency (ms)':<15} | {'Z-Score':<10} | {'Status'}")
print("-" * 38)
for row in audit_report:
    status = "🚨 ANOMALOUS SPIKE" if row['is_outlier'] else "Normal"
    print(f"{row['value']:<15} | {row['z_score']:<10} | {status}")

daily_receipts = [
    ("2026-09-01", 12400),
    ("2026-09-02", 18200),
    ("2026-09-03", 9800),
    ("2026-09-04", 24500),
    ("2026-09-05", 31000),
]
 
def compute_running_total(ledger: list[tuple[str, float]]) -> list[dict]:
    running_sum = 0.0
    output = []
    for date_str, amount in ledger:
        running_sum += amount
        output.append({
            "date": date_str,
            "daily_revenue": amount,
            "running_total": running_sum
        })
    return output
 
running_ledger = compute_running_total(daily_receipts)
print(f"{'Date':<12} | {'Daily Revenue':<15} | {'Cumulative Revenue'}")
print("-" * 48)
for row in running_ledger:
    print(f"{row['date']:<12} | ₹{row['daily_revenue']:<14,f} | ₹{row['running_total']:<16,f}")

engagement_scores = [12, 18, 25, 34, 45, 52, 68, 74, 88, 95]
 
def calculate_percentile_rank(scores: list[float], target_score: float) -> float:
    """Calculates the empirical percentile rank of a score relative to a cohort."""
    count_below = sum(1 for s in scores if s < target_score)
    count_equal = sum(1 for s in scores if s == target_score)
    
    # Standard statistical formula for percentile rank: (below + 0.5 * equal) / N * 100
    percentile = ((count_below + (0.5 * count_equal)) / len(scores)) * 100
    return round(percentile, 1)
 
candidate_scores = [18, 52, 90, 95]
for score in candidate_scores:
    pct = calculate_percentile_rank(engagement_scores, score)
    print(f"Score {score:>2} -> {pct:>5.1f}th Percentile")

import pandas as pd
import numpy as np
 
catalog_data = {
    "product_id": [1, 2, 3, 4, 5, 6, 7],
    "category": ["Electronics", "Electronics", "Electronics", "Books", "Books", "Furniture", "Furniture"],
    "price_inr": [45000.0, np.nan, 55000.0, 450.0, np.nan, 12000.0, np.nan]
}
df_catalog = pd.DataFrame(catalog_data)
 
# Impute price using the median of that specific product category
df_catalog["imputed_price_inr"] = df_catalog.groupby("category")["price_inr"].transform(
    lambda group: group.fillna(group.median())
)
 
print("Original vs Imputed Catalog Prices:")
print(df_catalog[["product_id", "category", "price_inr", "imputed_price_inr"]])

orders_data = {
    "order_id": [101, 102, 103, 104, 105, 106],
    "region": ["North", "North", "South", "South", "West", "North"],
    "customer_id": ["C1", "C2", "C3", "C1", "C4", "C1"],
    "order_amount": [2500, 1800, 4200, 3100, 5000, 1900]
}
df_orders = pd.DataFrame(orders_data)
 
regional_kpis = df_orders.groupby("region").agg(
    total_gmv=("order_amount", "sum"),
    order_volume=("order_id", "count"),
    unique_buyers=("customer_id", "nunique"),
    aov=("order_amount", "mean")
).reset_index()
 
regional_kpis["aov"] = regional_kpis["aov"].round(2)
print(regional_kpis)

orders_df = pd.DataFrame({
    "order_id": [501, 502, 503, 504],
    "user_id": [10, 11, 12, 99],  # User 99 does not exist in users table
    "amount": [1200, 2400, 1800, 950]
})
 
users_df = pd.DataFrame({
    "user_id": [10, 11, 12],
    "plan_tier": ["Pro", "Free", "Pro"]
})
 
# Merge defensively with validation and audit indicator
merged_df = pd.merge(
    orders_df,
    users_df,
    on="user_id",
    how="left",
    validate="many_to_one",
    indicator=True
)
 
print("Merged Transaction Audit:")
print(merged_df)
print("\nUnmatched Orphan Orders:")
print(merged_df[merged_df["_merge"] == "left_only"])

survey_data = {
    "respondent_id": [1001, 1002, 1003],
    "department": ["Engineering", "Sales", "Marketing"],
    "q1_onboarding": [5, 4, 3],
    "q2_compensation": [4, 5, 2],
    "q3_culture": [5, 5, 4]
}
df_survey = pd.DataFrame(survey_data)
 
tidy_survey = pd.melt(
    df_survey,
    id_vars=["respondent_id", "department"],
    value_vars=["q1_onboarding", "q2_compensation", "q3_culture"],
    var_name="survey_question",
    value_name="score"
)
 
print("Normalized Tidy Survey Format:")
print(tidy_survey.head(6))

sales_data = {
    "date": pd.to_datetime(["2026-09-01", "2026-09-02", "2026-09-04", "2026-09-08", "2026-09-09"]),
    "revenue": [10000, 15000, 12000, 20000, 18000]
}
df_sales = pd.DataFrame(sales_data).set_index("date")
 
# Rolling 7-day trailing revenue calculation
df_sales["rolling_7d_revenue"] = df_sales["revenue"].rolling("7D").sum()
 
print("Trailing 7-Day Window Analysis:")
print(df_sales)

customer_orders = pd.DataFrame({
    "customer_id": ["C101", "C102", "C103", "C104"],
    "last_order_date": pd.to_datetime(["2026-09-10", "2026-08-01", "2026-05-15", "2026-09-01"]),
    "total_lifetime_spend": [14500, 8900, 42000, 3100]
})
 
snapshot_date = pd.to_datetime("2026-09-12")
customer_orders["days_since_last_order"] = (snapshot_date - customer_orders["last_order_date"]).dt.days
 
conditions = [
    customer_orders["days_since_last_order"] <= 30,
    customer_orders["days_since_last_order"] <= 90
]
choices = ["Active", "Cooling Down"]
 
customer_orders["retention_status"] = np.select(conditions, choices, default="Churned")
print(customer_orders[["customer_id", "days_since_last_order", "total_lifetime_spend", "retention_status"]])

rfm_raw = pd.DataFrame({
    "customer": [f"User_{i}" for i in range(1, 9)],
    "recency_days": [4, 18, 95, 3, 210, 12, 140, 2],
    "frequency": [14, 8, 1, 22, 1, 5, 2, 19],
    "monetary": [45000, 18000, 1200, 89000, 950, 8500, 2400, 72000]
})
 
# Score 1 to 4 (higher recency days = worse score, so reverse labels)
rfm_raw["R_Score"] = pd.qcut(rfm_raw["recency_days"], q=4, labels=[4, 3, 2, 1]).astype(int)
rfm_raw["F_Score"] = pd.qcut(rfm_raw["frequency"].rank(method="first"), q=4, labels=[1, 2, 3, 4]).astype(int)
rfm_raw["M_Score"] = pd.qcut(rfm_raw["monetary"], q=4, labels=[1, 2, 3, 4]).astype(int)
 
rfm_raw["RFM_Cell"] = (
    rfm_raw["R_Score"].astype(str) + 
    rfm_raw["F_Score"].astype(str) + 
    rfm_raw["M_Score"].astype(str)
)
 
def assign_rfm_segment(row) -> str:
    r, f = row["R_Score"], row["F_Score"]
    if r >= 3 and f >= 3:
        return "Champions"
    if r >= 3 and f < 3:
        return "New Customers"
    if r < 2 and f >= 3:
        return "At-Risk VIPs"
    return "Low-Engagement / Lost"
 
rfm_raw["Segment"] = rfm_raw.apply(assign_rfm_segment, axis=1)
print(rfm_raw[["customer", "recency_days", "frequency", "monetary", "RFM_Cell", "Segment"]])

txn_data = pd.DataFrame({
    "user_id": [1, 1, 1, 2, 2, 3, 4, 4],
    "order_date": pd.to_datetime([
        "2026-01-10", "2026-02-14", "2026-03-01",  # User 1 active Jan, Feb, Mar
        "2026-01-15", "2026-03-20",                # User 2 active Jan, Mar
        "2026-02-05",                              # User 3 active Feb
        "2026-02-12", "2026-03-18"                 # User 4 active Feb, Mar
    ])
})
 
txn_data["OrderPeriod"] = txn_data["order_date"].dt.to_period("M")
txn_data["CohortPeriod"] = txn_data.groupby("user_id")["order_date"].transform("min").dt.to_period("M")
 
cohort_group = txn_data.groupby(["CohortPeriod", "OrderPeriod"]).agg(n_users=("user_id", "nunique")).reset_index()
 
# Calculate cohort month index (Month 0, Month 1, Month 2...)
cohort_group["CohortIndex"] = (
    (cohort_group["OrderPeriod"].dt.year - cohort_group["CohortPeriod"].dt.year) * 12 +
    (cohort_group["OrderPeriod"].dt.month - cohort_group["CohortPeriod"].dt.month)
)
 
cohort_pivot = cohort_group.pivot(index="CohortPeriod", columns="CohortIndex", values="n_users")
cohort_sizes = cohort_pivot.iloc[:, 0]
retention_matrix = cohort_pivot.divide(cohort_sizes, axis=0).round(2) * 100
 
print("Cohort Retention Rate (%):")
print(retention_matrix)

pricing_tests = [
    {"campaign": "Promo_A", "old_price": 500, "new_price": 400, "old_qty": 1000, "new_qty": 1400},
    {"campaign": "Promo_B", "old_price": 200, "new_price": 180, "old_qty": 5000, "new_qty": 5200},
]
 
def calculate_price_elasticity(old_p: float, new_p: float, old_q: float, new_q: float) -> dict:
    pct_change_price = (new_p - old_p) / old_p
    pct_change_qty = (new_q - old_q) / old_q
    
    ped = pct_change_qty / pct_change_price
    
    if abs(ped) > 1.0:
        interpretation = "Elastic (Revenue increases with price cuts)"
    elif abs(ped) < 1.0:
        interpretation = "Inelastic (Price cuts reduce net revenue)"
    else:
        interpretation = "Unitary Elastic"
        
    return {
        "ped": round(ped, 2),
        "pct_price_change": f"{pct_change_price * 100:.1f}%",
        "pct_qty_change": f"{pct_change_qty * 100:.1f}%",
        "verdict": interpretation
    }
 
for test in pricing_tests:
    results = calculate_price_elasticity(test["old_price"], test["new_price"], test["old_qty"], test["new_qty"])
    print(f"Campaign {test['campaign']}: PED = {results['ped']} ({results['verdict']})")

revenue_records = pd.DataFrame({
    "customer_id": [f"Cust_{i}" for i in range(1, 11)],
    "spend_inr": [85000, 62000, 48000, 12000, 8500, 6000, 4500, 3000, 1500, 500]
})
 
# Sort descending by customer expenditure
revenue_records = revenue_records.sort_values(by="spend_inr", ascending=False).reset_index(drop=True)
 
total_gmv = revenue_records["spend_inr"].sum()
revenue_records["cumulative_spend"] = revenue_records["spend_inr"].cumsum()
revenue_records["cumulative_pct"] = (revenue_records["cumulative_spend"] / total_gmv) * 100
revenue_records["customer_rank_pct"] = ((revenue_records.index + 1) / len(revenue_records)) * 100
 
# Identify the Pareto tier (accounts driving the first 80% of company revenue)
revenue_records["pareto_vip"] = revenue_records["cumulative_pct"] <= 80.0
 
print(f"Total Portfolio GMV: ₹{total_gmv:,.2f}\n")
print(revenue_records[["customer_id", "spend_inr", "cumulative_pct", "customer_rank_pct", "pareto_vip"]])