Python Data Engine — Elite Research Data Processing

Python is the primary tool for any data work that outgrows SQL or needs reproducibility,

complex logic, or integration with IMI's ecosystem. This skill gives Claude the complete

patterns for production-quality Python in a research data context.

Project Setup — Always Start Right

research_project/
├── src/
│   ├── __init__.py
│   ├── data/          # data loading and validation
│   ├── analysis/      # analytical functions
│   ├── reporting/     # output generation
│   └── utils/         # shared utilities
├── tests/
│   └── test_analysis.py
├── data/
│   ├── raw/           # immutable source data
│   ├── processed/     # cleaned/transformed
│   └── outputs/       # final deliverables
├── notebooks/         # exploratory work only
├── requirements.txt
├── .env               # secrets (never commit)
└── README.md

# requirements.txt — IMI standard stack
pandas>=2.0
polars>=0.20          # faster alternative for large datasets
numpy>=1.26
scipy>=1.11
openpyxl>=3.1         # Excel I/O
xlsxwriter>=3.1       # Excel formatting
python-pptx>=0.6      # PowerPoint generation
reportlab>=4.0        # PDF generation
httpx>=0.27           # async HTTP
pydantic>=2.5         # data validation
python-dotenv>=1.0    # env management
loguru>=0.7           # structured logging
pytest>=8.0           # testing

Pandas — Survey Data Patterns

Loading and validating survey data

import pandas as pd
import numpy as np
from pathlib import Path
from loguru import logger

def load_wave_data(filepath: Path, wave_id: str, min_base: int = 100) -> pd.DataFrame:
    """Load a wave CSV with validation."""
    df = pd.read_csv(filepath)

    # Required columns
    required = ['respondent_id', 'wave_id', 'weight', 'question_id', 'response_value']
    missing = [c for c in required if c not in df.columns]
    if missing:
        raise ValueError(f"Missing required columns: {missing}")

    # Weight sanity check
    weight_stats = df['weight'].describe()
    if weight_stats['max'] > 5.0 or weight_stats['min'] < 0.1:
        logger.warning(f"Extreme weights detected: min={weight_stats['min']:.3f}, max={weight_stats['max']:.3f}")

    # Filter to target wave
    df = df[df['wave_id'] == wave_id].copy()

    # Base size
    n = df['respondent_id'].nunique()
    if n < min_base:
        logger.warning(f"Wave {wave_id} base size {n} is below minimum {min_base}")

    logger.info(f"Loaded wave {wave_id}: n={n} respondents, {len(df)} response rows")
    return df

Weighted T2B calculation — the IMI workhorse

def weighted_t2b(df: pd.DataFrame,
                 question_id: str,
                 t2b_values: tuple = (4, 5),
                 segment_col: str | None = None) -> pd.DataFrame:
    """
    Calculate weighted Top-2-Box percentage.
    IMI standard: always weight, always flag small bases.
    """
    q_data = df[df['question_id'] == question_id].copy()

    if segment_col:
        groups = q_data.groupby(segment_col)
    else:
        groups = [('Total', q_data)]

    results = []
    for seg, group in groups:
        t2b_w = group.loc[group['response_value'].isin(t2b_values), 'weight'].sum()
        total_w = group['weight'].sum()
        n_unweighted = group['respondent_id'].nunique()

        results.append({
            'segment': seg,
            't2b_pct': round(t2b_w / total_w * 100, 1) if total_w > 0 else None,
            'weighted_base': round(total_w),
            'unweighted_base': n_unweighted,
            'base_flag': 'DIRECTIONAL' if n_unweighted < 100 else 'Reportable'
        })

    return pd.DataFrame(results)

Brand funnel analysis

def brand_funnel(df: pd.DataFrame,
                 funnel_questions: dict,
                 segment_col: str | None = None) -> pd.DataFrame:
    """
    funnel_questions = {
        'awareness':     ('Q_AWARENESS', [1]),       # binary
        'consideration': ('Q_CONSIDERATION', [4, 5]),
        'preference':    ('Q_PREFERENCE', [4, 5]),
        'usage':         ('Q_USAGE', [1, 2, 3, 4, 5]),
    }
    """
    respondent_level = df.drop_duplicates('respondent_id')[['respondent_id', 'weight']].copy()
    if segment_col:
        segs = df[['respondent_id', segment_col]].drop_duplicates()
        respondent_level = respondent_level.merge(segs, on='respondent_id')

    for metric, (qid, values) in funnel_questions.items():
        qualifiers = df[(df['question_id'] == qid) &
                        (df['response_value'].isin(values))]['respondent_id'].unique()
        respondent_level[metric] = respondent_level['respondent_id'].isin(qualifiers).astype(int)

    group_cols = [segment_col] if segment_col else []
    agg = respondent_level.groupby(group_cols) if group_cols else [(None, respondent_level)]

    results = []
    for seg, grp in (agg if group_cols else [('Total', respondent_level)]):
        row = {'segment': seg, 'unweighted_n': len(grp), 'weighted_n': round(grp['weight'].sum())}
        total_w = grp['weight'].sum()
        for metric in funnel_questions:
            row[f'{metric}_pct'] = round(grp.loc[grp[metric] == 1, 'weight'].sum() / total_w * 100, 1)
        results.append(row)

    result_df = pd.DataFrame(results)

    # Conversion rates
    metrics = list(funnel_questions.keys())
    for i in range(len(metrics) - 1):
        cur, nxt = metrics[i], metrics[i + 1]
        result_df[f'{cur}_to_{nxt}_conv'] = round(
            result_df[f'{nxt}_pct'] / result_df[f'{cur}_pct'] * 100, 1)

    return result_df

Polars — When Pandas Is Too Slow

import polars as pl

# Pulse™ data is often millions of rows — Polars is dramatically faster
def pulse_passion_point_query(
    filepath: str,
    country: str,
    min_base: int = 100,
    top_n: int = 25
) -> pl.DataFrame:
    """Query Pulse™ data with Polars for speed."""
    return (
        pl.read_csv(filepath)
        .filter(
            (pl.col("country_code") == country) &
            (pl.col("base_size") >= min_base)
        )
        .group_by("passion_point_label")
        .agg([
            pl.col("alignment_index").mean().alias("avg_alignment"),
            pl.col("purchase_driver_rank").mean().alias("avg_purchase_rank"),
            pl.col("engagement_score").mean().alias("avg_engagement"),
            pl.col("base_size").sum().alias("total_base"),
        ])
        .sort("avg_alignment", descending=True)
        .head(top_n)
    )

Statistical Functions — IMI Essentials

from scipy import stats
import numpy as np

def z_test_proportions(p1: float, n1: int, p2: float, n2: int,
                       alpha: float = 0.05) -> dict:
    """
    Z-test for two proportions (e.g., Wave 2 vs Wave 3 consideration score).
    Returns: z_score, p_value, significant, confidence_label
    """
    se = np.sqrt((p1 * (1 - p1) / n1) + (p2 * (1 - p2) / n2))
    if se == 0:
        return {'z_score': 0, 'p_value': 1.0, 'significant': False, 'label': 'Unable to compute'}

    z = (p1 - p2) / se
    p_val = 2 * (1 - stats.norm.cdf(abs(z)))

    if abs(z) >= 1.96:
        label = "* Significant at 95%"
    elif abs(z) >= 1.645:
        label = "~ Marginally significant at 90%"
    else:
        label = "Not significant"

    return {
        'z_score': round(z, 3),
        'p_value': round(p_val, 4),
        'significant': abs(z) >= 1.96,
        'label': label
    }


def vs_norm(score: float, norm: float, threshold: float = 5.0) -> str:
    """IMI norm comparison label."""
    gap = score - norm
    if gap > threshold:   return f"Above norm ▲ (+{gap:.1f})"
    if gap < -threshold:  return f"Below norm ▼ ({gap:.1f})"
    return f"At norm ● ({gap:+.1f})"

API Integration — httpx (async-capable)

import httpx
import asyncio
from pydantic import BaseModel

class PulseAPIConfig(BaseModel):
    base_url: str
    api_key: str
    timeout: float = 30.0
    max_retries: int = 3

async def fetch_pulse_data(config: PulseAPIConfig, params: dict) -> dict:
    """Async Pulse™ API call with retry logic."""
    headers = {"Authorization": f"Bearer {config.api_key}"}

    async with httpx.AsyncClient(timeout=config.timeout) as client:
        for attempt in range(config.max_retries):
            try:
                response = await client.get(
                    f"{config.base_url}/pulse/query",
                    params=params,
                    headers=headers
                )
                response.raise_for_status()
                return response.json()
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429:    # Rate limited
                    await asyncio.sleep(2 ** attempt)  # Exponential backoff
                    continue
                raise
            except httpx.NetworkError:
                if attempt == config.max_retries - 1:
                    raise
                await asyncio.sleep(1)

    raise RuntimeError("Max retries exceeded")


# Batch async fetches (e.g., all passion points in parallel)
async def batch_pulse_queries(config: PulseAPIConfig, query_list: list[dict]) -> list[dict]:
    tasks = [fetch_pulse_data(config, q) for q in query_list]
    return await asyncio.gather(*tasks, return_exceptions=True)

Data Validation with Pydantic

from pydantic import BaseModel, validator, Field
from typing import Literal
from datetime import date

class WaveRecord(BaseModel):
    respondent_id: str
    wave_id: str
    study_id: str
    weight: float = Field(gt=0, lt=10)     # weight must be positive, below 10
    question_id: str
    response_value: int = Field(ge=1, le=5) # 5-point scale

    @validator('weight')
    def warn_extreme_weight(cls, v):
        if v > 5.0 or v < 0.2:
            # In production: log a warning rather than failing
            pass
        return v

class PulseRow(BaseModel):
    passion_point_id: str
    brand_id: str
    country_code: str
    wave_date: date
    alignment_index: float = Field(ge=0, le=5)
    base_size: int = Field(ge=0)
    purchase_driver_rank: float | None = None
    is_reportable: bool = False

    @validator('is_reportable', always=True, pre=False)
    def set_reportable(cls, v, values):
        return values.get('base_size', 0) >= 100

Structured Logging — Production Standard

from loguru import logger
import sys

def configure_logging(log_file: str | None = None, level: str = "INFO"):
    logger.remove()
    # Console: clean format
    logger.add(sys.stdout, level=level,
               format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | {message}")
    # File: full detail
    if log_file:
        logger.add(log_file, level="DEBUG", rotation="10 MB", retention="30 days",
                   format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}:{function}:{line} | {message}")

# Usage
configure_logging("./logs/analysis.log")
logger.info("Starting wave processing")
logger.debug(f"Loaded {n} rows from {filepath}")
logger.warning(f"Small base: segment {seg} has n={n}")
logger.error(f"Failed to process: {error}")

IMI Automation Patterns

Automated wave report generator

def generate_wave_report(study_id: str, wave_id: str, output_dir: Path) -> Path:
    """
    Full pipeline: load data → compute metrics → validate → export.
    Returns path to generated report file.
    """
    logger.info(f"Generating report: {study_id} Wave {wave_id}")

    # 1. Load
    df = load_wave_data(DATA_DIR / f"{study_id}_w{wave_id}.csv", wave_id)

    # 2. Compute
    funnel = brand_funnel(df, FUNNEL_QUESTIONS, segment_col='age_group')
    t2b = weighted_t2b(df, 'Q_BRAND_CONSIDERATION', segment_col='gender')

    # 3. Validate
    for _, row in funnel.iterrows():
        if row['base_flag'] == 'DIRECTIONAL':
            logger.warning(f"Directional base: {row['segment']} n={row['unweighted_n']}")

    # 4. Export
    output_path = output_dir / f"{study_id}_W{wave_id}_report_{date.today().isoformat()}.xlsx"
    with pd.ExcelWriter(output_path, engine='xlsxwriter') as writer:
        funnel.to_excel(writer, sheet_name='Brand Funnel', index=False)
        t2b.to_excel(writer, sheet_name='Consideration T2B', index=False)

    logger.info(f"Report saved: {output_path}")
    return output_path

*See also: bash-powertools (shell integration), reportlab-pdf-factory (PDF output), excel-automation (Excel output), data-pipeline (orchestration)*

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)