Real-World Results Data Analysis vs Sales Outreach: Which Wins?

import os
import json
import logging
from datetime import datetime, timedelta
import pandas as pd
import stripe
from amplitude import Amplitude
from mixpanel import Mixpanel
from sqlalchemy import create_engine, text

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("rda_pipeline.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Load environment variables (12-factor app compliance)
STRIPE_API_KEY = os.getenv("STRIPE_API_KEY")
AMPLITUDE_API_KEY = os.getenv("AMPLITUDE_API_KEY")
MIXPANEL_TOKEN = os.getenv("MIXPANEL_TOKEN")
DB_CONNECTION_STRING = os.getenv("DB_CONNECTION_STRING")

# Validate required env vars
required_vars = [STRIPE_API_KEY, AMPLITUDE_API_KEY, MIXPANEL_TOKEN, DB_CONNECTION_STRING]
if any(var is None for var in required_vars):
    logger.error("Missing required environment variables. Exiting.")
    raise ValueError("Missing required environment variables.")

# Initialize clients
stripe.api_key = STRIPE_API_KEY
amp_client = Amplitude(AMPLITUDE_API_KEY)
mp_client = Mixpanel(MIXPANEL_TOKEN)
engine = create_engine(DB_CONNECTION_STRING)

def fetch_product_usage_events(days_back: int = 7) -> pd.DataFrame:
    """Fetch product usage events from Amplitude for the last N days."""
    try:
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(days=days_back)

        # Amplitude export API call (v2.3.1)
        response = amp_client.export(
            start=start_time.strftime("%Y%m%d"),
            end=end_time.strftime("%Y%m%d"),
            event_types=["feature_used", "trial_started", "payment_issued"]
        )

        # Parse response into DataFrame
        events = []
        for line in response.iter_lines():
            if line:
                event = json.loads(line)
                events.append(event)

        df = pd.DataFrame(events)
        logger.info(f"Fetched {len(df)} product usage events from Amplitude")
        return df
    except Exception as e:
        logger.error(f"Failed to fetch Amplitude events: {str(e)}")
        raise

def fetch_stripe_customer_data() -> pd.DataFrame:
    """Fetch customer payment data from Stripe API (v2024-04-10)."""
    try:
        customers = stripe.Customer.list(limit=100, expand=["data.subscriptions"])
        customer_data = []
        for customer in customers:
            customer_data.append({
                "user_id": customer.metadata.get("app_user_id"),
                "stripe_customer_id": customer.id,
                "plan": customer.subscriptions.data[0].plan.nickname if customer.subscriptions.total_count > 0 else None,
                "mrr": customer.subscriptions.data[0].plan.amount / 100 if customer.subscriptions.total_count > 0 else 0
            })
        df = pd.DataFrame(customer_data)
        logger.info(f"Fetched {len(df)} customer records from Stripe")
        return df
    except Exception as e:
        logger.error(f"Failed to fetch Stripe data: {str(e)}")
        raise

def score_leads(usage_df: pd.DataFrame, stripe_df: pd.DataFrame) -> pd.DataFrame:
    """Score leads based on product usage and payment history."""
    # Merge datasets on user_id
    merged = pd.merge(usage_df, stripe_df, on="user_id", how="left")

    # Calculate lead score: 40% usage frequency, 30% feature adoption, 30% payment history
    merged["usage_frequency"] = merged.groupby("user_id")["event_time"].transform("count")
    merged["feature_adoption"] = merged.groupby("user_id")["event_type"].transform("nunique")
    merged["lead_score"] = (
        (merged["usage_frequency"] / merged["usage_frequency"].max() * 0.4) +
        (merged["feature_adoption"] / merged["feature_adoption"].max() * 0.3) +
        (merged["mrr"].fillna(0) / merged["mrr"].max() * 0.3 if merged["mrr"].max() > 0 else 0)
    ) * 10  # Scale to 1-10

    # Filter for high-intent leads (score >=7)
    high_intent = merged[merged["lead_score"] >= 7]
    logger.info(f"Identified {len(high_intent)} high-intent leads from {len(merged)} total users")
    return high_intent

def write_to_db(leads_df: pd.DataFrame):
    """Persist high-intent leads to PostgreSQL for sales team access."""
    try:
        with engine.connect() as conn:
            leads_df.to_sql("high_intent_leads", conn, if_exists="append", index=False)
        logger.info(f"Wrote {len(leads_df)} leads to PostgreSQL")
    except Exception as e:
        logger.error(f"Failed to write to DB: {str(e)}")
        raise

if __name__ == "__main__":
    try:
        logger.info("Starting RDA lead scoring pipeline")
        usage_events = fetch_product_usage_events(days_back=7)
        stripe_data = fetch_stripe_customer_data()
        scored_leads = score_leads(usage_events, stripe_data)
        write_to_db(scored_leads)
        logger.info("Pipeline completed successfully")
    except Exception as e:
        logger.error(f"Pipeline failed: {str(e)}")
        exit(1)

import os
import json
import logging
import time
from datetime import datetime
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("outreach_campaign.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Load env vars
SALESLOFT_API_KEY = os.getenv("SALESLOFT_API_KEY")
HUBSPOT_API_KEY = os.getenv("HUBSPOT_API_KEY")
LEAD_LIST_PATH = os.getenv("LEAD_LIST_PATH", "leads.csv")

# Validate env vars
if not all([SALESLOFT_API_KEY, HUBSPOT_API_KEY]):
    logger.error("Missing SalesLoft or HubSpot API keys")
    raise ValueError("Missing API credentials")

# Configure HTTP session with retries (SalesLoft v2.10.1 API)
session = requests.Session()
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)

# SalesLoft API base URL
SALESLOFT_BASE = "https://api.salesloft.com/v2"

def fetch_cold_leads() -> list:
    """Load cold leads from CSV (exported from ZoomInfo v3.2.1)."""
    try:
        import csv
        leads = []
        with open(LEAD_LIST_PATH, "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                leads.append({
                    "email": row["email"],
                    "first_name": row["first_name"],
                    "company": row["company"],
                    "title": row["title"]
                })
        logger.info(f"Loaded {len(leads)} cold leads from {LEAD_LIST_PATH}")
        return leads
    except Exception as e:
        logger.error(f"Failed to load leads: {str(e)}")
        raise

def create_salesloft_cadence() -> str:
    """Create a 3-step cold outreach cadence in SalesLoft."""
    try:
        headers = {
            "Authorization": f"Bearer {SALESLOFT_API_KEY}",
            "Content-Type": "application/json"
        }
        cadence_payload = {
            "name": f"Q3 2024 Cold Outreach - {datetime.utcnow().strftime('%Y%m%d')}",
            "steps": [
                {
                    "type": "email",
                    "subject": "Quick question about {company}'s engineering workflow",
                    "body": "Hi {first_name}, I noticed {company} is hiring for backend engineers. We help teams reduce deployment time by 40% with our CI/CD tool. Do you have 10 mins to chat?",
                    "wait_days": 0
                },
                {
                    "type": "email",
                    "subject": "Following up on {company} deployment workflow",
                    "body": "Hi {first_name}, just following up on my note from last week. Our tool is used by 120+ engineering teams including Cloudflare and Stripe. Would love to share a case study.",
                    "wait_days": 3
                },
                {
                    "type": "linkedin",
                    "message": "Hi {first_name}, I sent you an email about our CI/CD tool. Would love to connect here as well.",
                    "wait_days": 5
                }
            ]
        }
        response = session.post(
            f"{SALESLOFT_BASE}/cadences",
            headers=headers,
            json=cadence_payload
        )
        response.raise_for_status()
        cadence_id = response.json()["data"]["id"]
        logger.info(f"Created SalesLoft cadence: {cadence_id}")
        return cadence_id
    except Exception as e:
        logger.error(f"Failed to create cadence: {str(e)}")
        raise

def add_leads_to_cadence(cadence_id: str, leads: list):
    """Add cold leads to the created cadence."""
    try:
        headers = {"Authorization": f"Bearer {SALESLOFT_API_KEY}"}
        for lead in leads:
            # Create or fetch person in SalesLoft
            person_payload = {
                "email_address": lead["email"],
                "first_name": lead["first_name"],
                "company": lead["company"],
                "title": lead["title"]
            }
            person_resp = session.post(
                f"{SALESLOFT_BASE}/people",
                headers=headers,
                json=person_payload
            )
            if person_resp.status_code == 201:
                person_id = person_resp.json()["data"]["id"]
            elif person_resp.status_code == 409:
                # Person already exists
                person_id = person_resp.json()["data"]["id"]
            else:
                person_resp.raise_for_status()

            # Add person to cadence
            step_payload = {"person_id": person_id}
            session.post(
                f"{SALESLOFT_BASE}/cadences/{cadence_id}/steps/1/actions",
                headers=headers,
                json=step_payload
            )
            time.sleep(0.1)  # Rate limit compliance (SalesLoft allows 10 req/sec)
        logger.info(f"Added {len(leads)} leads to cadence {cadence_id}")
    except Exception as e:
        logger.error(f"Failed to add leads to cadence: {str(e)}")
        raise

def sync_to_hubspot(leads: list):
    """Sync cold leads to HubSpot for attribution tracking."""
    try:
        headers = {
            "Authorization": f"Bearer {HUBSPOT_API_KEY}",
            "Content-Type": "application/json"
        }
        for lead in leads:
            hubspot_payload = {
                "properties": {
                    "email": lead["email"],
                    "firstname": lead["first_name"],
                    "company": lead["company"],
                    "jobtitle": lead["title"],
                    "lead_source": "Cold Outreach"
                }
            }
            response = session.post(
                "https://api.hubapi.com/crm/v3/objects/contacts",
                headers=headers,
                json=hubspot_payload
            )
            response.raise_for_status()
        logger.info(f"Synced {len(leads)} leads to HubSpot")
    except Exception as e:
        logger.error(f"Failed to sync to HubSpot: {str(e)}")
        raise

if __name__ == "__main__":
    try:
        logger.info("Starting cold outreach campaign")
        leads = fetch_cold_leads()
        cadence_id = create_salesloft_cadence()
        add_leads_to_cadence(cadence_id, leads)
        sync_to_hubspot(leads)
        logger.info("Outreach campaign launched successfully")
    except Exception as e:
        logger.error(f"Campaign failed: {str(e)}")
        exit(1)

import os
import json
import logging
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from scipy import stats
from sqlalchemy import create_engine, text

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("ab_test_analysis.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Load env vars
DB_CONNECTION_STRING = os.getenv("DB_CONNECTION_STRING")
if not DB_CONNECTION_STRING:
    logger.error("Missing DB_CONNECTION_STRING")
    raise ValueError("Missing DB credentials")

engine = create_engine(DB_CONNECTION_STRING)

def fetch_ab_test_data(test_name: str = "rda_vs_so_q3_2024") -> pd.DataFrame:
    """Fetch A/B test results from PostgreSQL (experiment started 2024-04-01)."""
    try:
        query = text("""
            SELECT 
                user_id,
                variant,
                trial_started_at,
                paid_converted_at,
                lead_source,
                mrr
            FROM ab_test_users
            WHERE test_name = :test_name
            AND created_at >= '2024-04-01'
            AND created_at <= '2024-06-30'
        """)
        with engine.connect() as conn:
            df = pd.read_sql(query, conn, params={"test_name": test_name})
        logger.info(f"Fetched {len(df)} users for test {test_name}")
        return df
    except Exception as e:
        logger.error(f"Failed to fetch A/B test data: {str(e)}")
        raise

def calculate_conversion_metrics(df: pd.DataFrame) -> dict:
    """Calculate conversion rates and CAC for each variant."""
    try:
        # Calculate trial-to-paid conversion
        df["converted"] = df["paid_converted_at"].notnull().astype(int)
        variant_metrics = df.groupby("variant").agg(
            total_users=("user_id", "count"),
            conversions=("converted", "sum"),
            total_mrr=("mrr", "sum")
        ).reset_index()

        variant_metrics["conversion_rate"] = (variant_metrics["conversions"] / variant_metrics["total_users"]) * 100

        # Fetch CAC data from marketing spend table
        cac_query = text("""
            SELECT 
                variant,
                total_spend
            FROM ab_test_spend
            WHERE test_name = :test_name
        """)
        with engine.connect() as conn:
            cac_df = pd.read_sql(cac_query, conn, params={"test_name": test_name})

        merged = pd.merge(variant_metrics, cac_df, on="variant")
        merged["cac"] = merged["total_spend"] / merged["conversions"]
        merged["arpu"] = merged["total_mrr"] / merged["conversions"]
        merged["payback_period"] = merged["cac"] / merged["arpu"]

        logger.info(f"Calculated metrics for {len(merged)} variants")
        return merged.to_dict("records")
    except Exception as e:
        logger.error(f"Failed to calculate metrics: {str(e)}")
        raise

def run_statistical_significance(metrics: list) -> dict:
    """Run chi-squared test for conversion rate difference."""
    try:
        rda = next(m for m in metrics if m["variant"] == "rda")
        so = next(m for m in metrics if m["variant"] == "sales_outreach")

        # Contingency table: [conversions, non-conversions]
        contingency_table = np.array([
            [rda["conversions"], rda["total_users"] - rda["conversions"]],
            [so["conversions"], so["total_users"] - so["conversions"]]
        ])

        chi2, p_value, dof, expected = stats.chi2_contingency(contingency_table)

        # Calculate relative uplift
        uplift = ((rda["conversion_rate"] - so["conversion_rate"]) / so["conversion_rate"]) * 100

        result = {
            "chi2_statistic": chi2,
            "p_value": p_value,
            "degrees_of_freedom": dof,
            "relative_uplift_pct": uplift,
            "significant": p_value < 0.01
        }
        logger.info(f"Statistical test result: p={p_value:.4f}, uplift={uplift:.1f}%")
        return result
    except Exception as e:
        logger.error(f"Failed to run statistical test: {str(e)}")
        raise

def write_results_to_db(metrics: list, stats_result: dict, test_name: str):
    """Persist A/B test results to PostgreSQL."""
    try:
        with engine.connect() as conn:
            # Write metrics
            for metric in metrics:
                conn.execute(text("""
                    INSERT INTO ab_test_results (test_name, variant, total_users, conversions, conversion_rate, cac, payback_period)
                    VALUES (:test_name, :variant, :total_users, :conversions, :conversion_rate, :cac, :payback_period)
                """), {**metric, "test_name": test_name})

            # Write stats result
            conn.execute(text("""
                INSERT INTO ab_test_stats (test_name, chi2_statistic, p_value, relative_uplift_pct, is_significant)
                VALUES (:test_name, :chi2, :p_value, :uplift, :significant)
            """), {
                "test_name": test_name,
                "chi2": stats_result["chi2_statistic"],
                "p_value": stats_result["p_value"],
                "uplift": stats_result["relative_uplift_pct"],
                "significant": stats_result["significant"]
            })
            conn.commit()
        logger.info(f"Wrote results to DB for test {test_name}")
    except Exception as e:
        logger.error(f"Failed to write results: {str(e)}")
        raise

if __name__ == "__main__":
    try:
        logger.info("Starting A/B test analysis")
        test_name = "rda_vs_so_q3_2024"
        df = fetch_ab_test_data(test_name)
        metrics = calculate_conversion_metrics(df)
        stats_result = run_statistical_significance(metrics)
        write_results_to_db(metrics, stats_result, test_name)

        # Print summary
        print("\n=== A/B Test Results ===")
        for m in metrics:
            print(f"Variant: {m['variant']}")
            print(f"Users: {m['total_users']}")
            print(f"Conversions: {m['conversions']}")
            print(f"Conversion Rate: {m['conversion_rate']:.1f}%")
            print(f"CAC: ${m['cac']:.2f}")
            print(f"Payback Period: {m['payback_period']:.1f} months\n")

        print(f"Statistical Significance: {'Yes' if stats_result['significant'] else 'No'}")
        print(f"P-value: {stats_result['p_value']:.4f}")
        print(f"Relative Uplift: {stats_result['relative_uplift_pct']:.1f}%")

        logger.info("Analysis completed successfully")
    except Exception as e:
        logger.error(f"Analysis failed: {str(e)}")
        exit(1)

import json
from jsonschema import validate, ValidationError

EVENT_SCHEMA = {
    "type": "object",
    "properties": {
        "event_type": {"type": "string", "enum": ["trial_started", "feature_used", "payment_issued", "churn_triggered"]},
        "user_id": {"type": "string"},
        "timestamp": {"type": "string", "format": "date-time"},
        "metadata": {"type": "object"}
    },
    "required": ["event_type", "user_id", "timestamp"]
}

def validate_event(event: dict) -> bool:
    try:
        validate(instance=event, schema=EVENT_SCHEMA)
        return True
    except ValidationError as e:
        logger.error(f"Invalid event: {str(e)}")
        return False

from fastapi import FastAPI, Request, HTTPException
import httpx

app = FastAPI()

@app.post("/webhook/product-event")
async def handle_product_event(request: Request):
    event = await request.json()
    # Validate event
    if not validate_event(event):
        raise HTTPException(status_code=400, detail="Invalid event")

    # Score lead
    lead_score = calculate_lead_score(event["user_id"])
    if lead_score >= 7:
        # Route to sales rep via HubSpot API
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.hubapi.com/crm/v3/objects/deals",
                headers={"Authorization": f"Bearer {HUBSPOT_API_KEY}"},
                json={
                    "properties": {
                        "dealname": f"PQL - {event['user_id']}",
                        "lead_score": lead_score,
                        "pipeline": "default"
                    }
                }
            )
            response.raise_for_status()
    return {"status": "processed"}

import random

def assign_ab_test_variant(user_id: str, test_name: str) -> str:
    # Use hash of user_id to ensure consistent assignment
    hash_val = hash(f"{user_id}_{test_name}")
    normalized_hash = (hash_val % 100) / 100  # 0-1 range
    if normalized_hash < 0.1:
        return "sales_outreach"
    else:
        return "rda"