Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

7 NODES
🤖system prompt•7 months ago

kpi-dashboard-design

Design effective KPI dashboards with metrics selection,

coding
⭐1
# KPI Dashboard Design Comprehensive patterns for designing effective Key Performance Indicator (KPI) dashboards that drive business decisions. ## When to Use This Skill - Designing executive dashboards - Selecting meaningful KPIs - Building real-time monitoring displays - Creating department-specific metrics views - Improving existing dashboard layouts - Establishing metric governance ## Core Concepts ### 1. KPI Framework | Level | Focus | Update Frequency | Audience | | --------------- | ---------------- | ----------------- | ---------- | | **Strategic** | Long-term goals | Monthly/Quarterly | Executives | | **Tactical** | Department goals | Weekly/Monthly | Managers | | **Operational** | Day-to-day | Real-time/Daily | Teams | ### 2. SMART KPIs ``` Specific: Clear definition Measurable: Quantifiable Achievable: Realistic targets Relevant: Aligned to goals Time-bound: Defined period ``` ### 3. Dashboard Hierarchy ``` ├── Executive Summary (1 page) │ ├── 4-6 headline KPIs │ ├── Trend indicators │ └── Key alerts ├── Department Views │ ├── Sales Dashboard │ ├── Marketing Dashboard │ ├── Operations Dashboard │ └── Finance Dashboard └── Detailed Drilldowns ├── Individual metrics └── Root cause analysis ``` ## Common KPIs by Department ### Sales KPIs ```yaml Revenue Metrics: - Monthly Recurring Revenue (MRR) - Annual Recurring Revenue (ARR) - Average Revenue Per User (ARPU) - Revenue Growth Rate Pipeline Metrics: - Sales Pipeline Value - Win Rate - Average Deal Size - Sales Cycle Length Activity Metrics: - Calls/Emails per Rep - Demos Scheduled - Proposals Sent - Close Rate ``` ### Marketing KPIs ```yaml Acquisition: - Cost Per Acquisition (CPA) - Customer Acquisition Cost (CAC) - Lead Volume - Marketing Qualified Leads (MQL) Engagement: - Website Traffic - Conversion Rate - Email Open/Click Rate - Social Engagement ROI: - Marketing ROI - Campaign Performance - Channel Attribution - CAC Payback Period ``` ### Product KPIs ```yaml Usage: - Daily/Monthly Active Users (DAU/MAU) - Session Duration - Feature Adoption Rate - Stickiness (DAU/MAU) Quality: - Net Promoter Score (NPS) - Customer Satisfaction (CSAT) - Bug/Issue Count - Time to Resolution Growth: - User Growth Rate - Activation Rate - Retention Rate - Churn Rate ``` ### Finance KPIs ```yaml Profitability: - Gross Margin - Net Profit Margin - EBITDA - Operating Margin Liquidity: - Current Ratio - Quick Ratio - Cash Flow - Working Capital Efficiency: - Revenue per Employee - Operating Expense Ratio - Days Sales Outstanding - Inventory Turnover ``` ## Dashboard Layout Patterns ### Pattern 1: Executive Summary ``` ┌─────────────────────────────────────────────────────────────┐ │ EXECUTIVE DASHBOARD [Date Range ▼] │ ├─────────────┬─────────────┬─────────────┬─────────────────┤ │ REVENUE │ PROFIT │ CUSTOMERS │ NPS SCORE │ │ $2.4M │ $450K │ 12,450 │ 72 │ │ ▲ 12% │ ▲ 8% │ ▲ 15% │ ▲ 5pts │ ├─────────────┴─────────────┴─────────────┴─────────────────┤ │ │ │ Revenue Trend │ Revenue by Product │ │ ┌───────────────────────┐ │ ┌──────────────────┐ │ │ │ /\ /\ │ │ │ ████████ 45% │ │ │ │ / \ / \ /\ │ │ │ ██████ 32% │ │ │ │ / \/ \ / \ │ │ │ ████ 18% │ │ │ │ / \/ \ │ │ │ ██ 5% │ │ │ └───────────────────────┘ │ └──────────────────┘ │ │ │ ├─────────────────────────────────────────────────────────────┤ │ 🔴 Alert: Churn rate exceeded threshold (>5%) │ │ 🟡 Warning: Support ticket volume 20% above average │ └─────────────────────────────────────────────────────────────┘ ``` ### Pattern 2: SaaS Metrics Dashboard ``` ┌─────────────────────────────────────────────────────────────┐ │ SAAS METRICS Jan 2024 [Monthly ▼] │ ├──────────────────────┬──────────────────────────────────────┤ │ ┌────────────────┐ │ MRR GROWTH │ │ │ MRR │ │ ┌────────────────────────────────┐ │ │ │ $125,000 │ │ │ /── │ │ │ │ ▲ 8% │ │ │ /────/ │ │ │ └────────────────┘ │ │ /────/ │ │ │ ┌────────────────┐ │ │ /────/ │ │ │ │ ARR │ │ │ /────/ │ │ │ │ $1,500,000 │ │ └────────────────────────────────┘ │ │ │ ▲ 15% │ │ J F M A M J J A S O N D │ │ └────────────────┘ │ │ ├──────────────────────┼──────────────────────────────────────┤ │ UNIT ECONOMICS │ COHORT RETENTION │ │ │ │ │ CAC: $450 │ Month 1: ████████████████████ 100% │ │ LTV: $2,700 │ Month 3: █████████████████ 85% │ │ LTV/CAC: 6.0x │ Month 6: ████████████████ 80% │ │ │ Month 12: ██████████████ 72% │ │ Payback: 4 months │ │ ├──────────────────────┴──────────────────────────────────────┤ │ CHURN ANALYSIS │ │ ┌──────────┬──────────┬──────────┬──────────────────────┐ │ │ │ Gross │ Net │ Logo │ Expansion │ │ │ │ 4.2% │ 1.8% │ 3.1% │ 2.4% │ │ │ └──────────┴──────────┴──────────┴──────────────────────┘ │ └─────────────────────────────────────────────────────────────┘ ``` ### Pattern 3: Real-time Operations ``` ┌─────────────────────────────────────────────────────────────┐ │ OPERATIONS CENTER Live ● Last: 10:42:15 │ ├────────────────────────────┬────────────────────────────────┤ │ SYSTEM HEALTH │ SERVICE STATUS │ │ ┌──────────────────────┐ │ │ │ │ CPU MEM DISK │ │ ● API Gateway Healthy │ │ │ 45% 72% 58% │ │ ● User Service Healthy │ │ │ ███ ████ ███ │ │ ● Payment Service Degraded │ │ │ ███ ████ ███ │ │ ● Database Healthy │ │ │ ███ ████ ███ │ │ ● Cache Healthy │ │ └──────────────────────┘ │ │ ├────────────────────────────┼────────────────────────────────┤ │ REQUEST THROUGHPUT │ ERROR RATE │ │ ┌──────────────────────┐ │ ┌──────────────────────────┐ │ │ │ ▁▂▃▄▅▆▇█▇▆▅▄▃▂▁▂▃▄▅ │ │ │ ▁▁▁▁▁▂▁▁▁▁▁▁▁▁▁▁▁▁▁▁ │ │ │ └──────────────────────┘ │ └──────────────────────────┘ │ │ Current: 12,450 req/s │ Current: 0.02% │ │ Peak: 18,200 req/s │ Threshold: 1.0% │ ├────────────────────────────┴────────────────────────────────┤ │ RECENT ALERTS │ │ 10:40 🟡 High latency on payment-service (p99 > 500ms) │ │ 10:35 🟢 Resolved: Database connection pool recovered │ │ 10:22 🔴 Payment service circuit breaker tripped │ └─────────────────────────────────────────────────────────────┘ ``` ## Implementation Patterns ### SQL for KPI Calculations ```sql -- Monthly Recurring Revenue (MRR) WITH mrr_calculation AS ( SELECT DATE_TRUNC('month', billing_date) AS month, SUM( CASE subscription_interval WHEN 'monthly' THEN amount WHEN 'yearly' THEN amount / 12 WHEN 'quarterly' THEN amount / 3 END ) AS mrr FROM subscriptions WHERE status = 'active' GROUP BY DATE_TRUNC('month', billing_date) ) SELECT month, mrr, LAG(mrr) OVER (ORDER BY month) AS prev_mrr, (mrr - LAG(mrr) OVER (ORDER BY month)) / LAG(mrr) OVER (ORDER BY month) * 100 AS growth_pct FROM mrr_calculation; -- Cohort Retention WITH cohorts AS ( SELECT user_id, DATE_TRUNC('month', created_at) AS cohort_month FROM users ), activity AS ( SELECT user_id, DATE_TRUNC('month', event_date) AS activity_month FROM user_events WHERE event_type = 'active_session' ) SELECT c.cohort_month, EXTRACT(MONTH FROM age(a.activity_month, c.cohort_month)) AS months_since_signup, COUNT(DISTINCT a.user_id) AS active_users, COUNT(DISTINCT a.user_id)::FLOAT / COUNT(DISTINCT c.user_id) * 100 AS retention_rate FROM cohorts c LEFT JOIN activity a ON c.user_id = a.user_id AND a.activity_month >= c.cohort_month GROUP BY c.cohort_month, EXTRACT(MONTH FROM age(a.activity_month, c.cohort_month)) ORDER BY c.cohort_month, months_since_signup; -- Customer Acquisition Cost (CAC) SELECT DATE_TRUNC('month', acquired_date) AS month, SUM(marketing_spend) / NULLIF(COUNT(new_customers), 0) AS cac, SUM(marketing_spend) AS total_spend, COUNT(new_customers) AS customers_acquired FROM ( SELECT DATE_TRUNC('month', u.created_at) AS acquired_date, u.id AS new_customers, m.spend AS marketing_spend FROM users u JOIN marketing_spend m ON DATE_TRUNC('month', u.created_at) = m.month WHERE u.source = 'marketing' ) acquisition GROUP BY DATE_TRUNC('month', acquired_date); ``` ### Python Dashboard Code (Streamlit) ```python import streamlit as st import pandas as pd import plotly.express as px import plotly.graph_objects as go st.set_page_config(page_title="KPI Dashboard", layout="wide") # Header with date filter col1, col2 = st.columns([3, 1]) with col1: st.title("Executive Dashboard") with col2: date_range = st.selectbox( "Period", ["Last 7 Days", "Last 30 Days", "Last Quarter", "YTD"] ) # KPI Cards def metric_card(label, value, delta, prefix="", suffix=""): delta_color = "green" if delta >= 0 else "red" delta_arrow = "▲" if delta >= 0 else "▼" st.metric( label=label, value=f"{prefix}{value:,.0f}{suffix}", delta=f"{delta_arrow} {abs(delta):.1f}%" ) col1, col2, col3, col4 = st.columns(4) with col1: metric_card("Revenue", 2400000, 12.5, prefix="$") with col2: metric_card("Customers", 12450, 15.2) with col3: metric_card("NPS Score", 72, 5.0) with col4: metric_card("Churn Rate", 4.2, -0.8, suffix="%") # Charts col1, col2 = st.columns(2) with col1: st.subheader("Revenue Trend") revenue_data = pd.DataFrame({ 'Month': pd.date_range('2024-01-01', periods=12, freq='M'), 'Revenue': [180000, 195000, 210000, 225000, 240000, 255000, 270000, 285000, 300000, 315000, 330000, 345000] }) fig = px.line(revenue_data, x='Month', y='Revenue', line_shape='spline', markers=True) fig.update_layout(height=300) st.plotly_chart(fig, use_container_width=True) with col2: st.subheader("Revenue by Product") product_data = pd.DataFrame({ 'Product': ['Enterprise', 'Professional', 'Starter', 'Other'], 'Revenue': [45, 32, 18, 5] }) fig = px.pie(product_data, values='Revenue', names='Product', hole=0.4) fig.update_layout(height=300) st.plotly_chart(fig, use_container_width=True) # Cohort Heatmap st.subheader("Cohort Retention") cohort_data = pd.DataFrame({ 'Cohort': ['Jan', 'Feb', 'Mar', 'Apr', 'May'], 'M0': [100, 100, 100, 100, 100], 'M1': [85, 87, 84, 86, 88], 'M2': [78, 80, 76, 79, None], 'M3': [72, 74, 70, None, None], 'M4': [68, 70, None, None, None], }) fig = go.Figure(data=go.Heatmap( z=cohort_data.iloc[:, 1:].values, x=['M0', 'M1', 'M2', 'M3', 'M4'], y=cohort_data['Cohort'], colorscale='Blues', text=cohort_data.iloc[:, 1:].values, texttemplate='%{text}%', textfont={"size": 12}, )) fig.update_layout(height=250) st.plotly_chart(fig, use_container_width=True) # Alerts Section st.subheader("Alerts") alerts = [ {"level": "error", "message": "Churn rate exceeded threshold (>5%)"}, {"level": "warning", "message": "Support ticket volume 20% above average"}, ] for alert in alerts: if alert["level"] == "error": st.error(f"🔴 {alert['message']}") elif alert["level"] == "warning": st.warning(f"🟡 {alert['message']}") ``` ## Best Practices ### Do's - **Limit to 5-7 KPIs** - Focus on what matters - **Show context** - Comparisons, trends, targets - **Use consistent colors** - Red=bad, green=good - **Enable drilldown** - From summary to detail - **Update appropriately** - Match metric frequency ### Don'ts - **Don't show vanity metrics** - Focus on actionable data - **Don't overcrowd** - White space aids comprehension - **Don't use 3D charts** - They distort perception - **Don't hide methodology** - Document calculations - **Don't ignore mobile** - Ensure responsive design ## Resources - [Stephen Few's Dashboard Design](https://www.perceptualedge.com/articles/visual_business_intelligence/rules_for_using_color.pdf) - [Edward Tufte's Principles](https://www.edwardtufte.com/tufte/) - [Google Data Studio Gallery](https://datastudio.google.com/gallery)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

data-quality-frameworks

Implement data quality validation with Great Expectations, dbt

data
⭐1
# Data Quality Frameworks Production patterns for implementing data quality with Great Expectations, dbt tests, and data contracts to ensure reliable data pipelines. ## When to Use This Skill - Implementing data quality checks in pipelines - Setting up Great Expectations validation - Building comprehensive dbt test suites - Establishing data contracts between teams - Monitoring data quality metrics - Automating data validation in CI/CD ## Core Concepts ### 1. Data Quality Dimensions | Dimension | Description | Example Check | | ---------------- | ------------------------ | -------------------------------------------------- | | **Completeness** | No missing values | `expect_column_values_to_not_be_null` | | **Uniqueness** | No duplicates | `expect_column_values_to_be_unique` | | **Validity** | Values in expected range | `expect_column_values_to_be_in_set` | | **Accuracy** | Data matches reality | Cross-reference validation | | **Consistency** | No contradictions | `expect_column_pair_values_A_to_be_greater_than_B` | | **Timeliness** | Data is recent | `expect_column_max_to_be_between` | ### 2. Testing Pyramid for Data ``` /\ / \ Integration Tests (cross-table) /────\ / \ Unit Tests (single column) /────────\ / \ Schema Tests (structure) /────────────\ ``` ## Quick Start ### Great Expectations Setup ```bash # Install pip install great_expectations # Initialize project great_expectations init # Create datasource great_expectations datasource new ``` ```python # great_expectations/checkpoints/daily_validation.yml import great_expectations as gx # Create context context = gx.get_context() # Create expectation suite suite = context.add_expectation_suite("orders_suite") # Add expectations suite.add_expectation( gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id") ) suite.add_expectation( gx.expectations.ExpectColumnValuesToBeUnique(column="order_id") ) # Validate results = context.run_checkpoint(checkpoint_name="daily_orders") ``` ## Patterns ### Pattern 1: Great Expectations Suite ```python # expectations/orders_suite.py import great_expectations as gx from great_expectations.core import ExpectationSuite from great_expectations.core.expectation_configuration import ExpectationConfiguration def build_orders_suite() -> ExpectationSuite: """Build comprehensive orders expectation suite""" suite = ExpectationSuite(expectation_suite_name="orders_suite") # Schema expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_columns_to_match_set", kwargs={ "column_set": ["order_id", "customer_id", "amount", "status", "created_at"], "exact_match": False # Allow additional columns } )) # Primary key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "order_id"} )) suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_unique", kwargs={"column": "order_id"} )) # Foreign key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "customer_id"} )) # Categorical values suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_in_set", kwargs={ "column": "status", "value_set": ["pending", "processing", "shipped", "delivered", "cancelled"] } )) # Numeric ranges suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_between", kwargs={ "column": "amount", "min_value": 0, "max_value": 100000, "strict_min": True # amount > 0 } )) # Date validity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_dateutil_parseable", kwargs={"column": "created_at"} )) # Freshness - data should be recent suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_max_to_be_between", kwargs={ "column": "created_at", "min_value": {"$PARAMETER": "now - timedelta(days=1)"}, "max_value": {"$PARAMETER": "now"} } )) # Row count sanity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_row_count_to_be_between", kwargs={ "min_value": 1000, # Expect at least 1000 rows "max_value": 10000000 } )) # Statistical expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_mean_to_be_between", kwargs={ "column": "amount", "min_value": 50, "max_value": 500 } )) return suite ``` ### Pattern 2: Great Expectations Checkpoint ```yaml # great_expectations/checkpoints/orders_checkpoint.yml name: orders_checkpoint config_version: 1.0 class_name: Checkpoint run_name_template: "%Y%m%d-%H%M%S-orders-validation" validations: - batch_request: datasource_name: warehouse data_connector_name: default_inferred_data_connector_name data_asset_name: orders data_connector_query: index: -1 # Latest batch expectation_suite_name: orders_suite action_list: - name: store_validation_result action: class_name: StoreValidationResultAction - name: store_evaluation_parameters action: class_name: StoreEvaluationParametersAction - name: update_data_docs action: class_name: UpdateDataDocsAction # Slack notification on failure - name: send_slack_notification action: class_name: SlackNotificationAction slack_webhook: ${SLACK_WEBHOOK} notify_on: failure renderer: module_name: great_expectations.render.renderer.slack_renderer class_name: SlackRenderer ``` ```python # Run checkpoint import great_expectations as gx context = gx.get_context() result = context.run_checkpoint(checkpoint_name="orders_checkpoint") if not result.success: failed_expectations = [ r for r in result.run_results.values() if not r.success ] raise ValueError(f"Data quality check failed: {failed_expectations}") ``` ### Pattern 3: dbt Data Tests ```yaml # models/marts/core/_core__models.yml version: 2 models: - name: fct_orders description: Order fact table tests: # Table-level tests - dbt_utils.recency: datepart: day field: created_at interval: 1 - dbt_utils.at_least_one - dbt_utils.expression_is_true: expression: "total_amount >= 0" columns: - name: order_id description: Primary key tests: - unique - not_null - name: customer_id description: Foreign key to dim_customers tests: - not_null - relationships: to: ref('dim_customers') field: customer_id - name: order_status tests: - accepted_values: values: ["pending", "processing", "shipped", "delivered", "cancelled"] - name: total_amount tests: - not_null - dbt_utils.expression_is_true: expression: ">= 0" - name: created_at tests: - not_null - dbt_utils.expression_is_true: expression: "<= current_timestamp" - name: dim_customers columns: - name: customer_id tests: - unique - not_null - name: email tests: - unique - not_null # Custom regex test - dbt_utils.expression_is_true: expression: "email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'" ``` ### Pattern 4: Custom dbt Tests ```sql -- tests/generic/test_row_count_in_range.sql {% test row_count_in_range(model, min_count, max_count) %} with row_count as ( select count(*) as cnt from {{ model }} ) select cnt from row_count where cnt < {{ min_count }} or cnt > {{ max_count }} {% endtest %} -- Usage in schema.yml: -- tests: -- - row_count_in_range: -- min_count: 1000 -- max_count: 10000000 ``` ```sql -- tests/generic/test_sequential_values.sql {% test sequential_values(model, column_name, interval=1) %} with lagged as ( select {{ column_name }}, lag({{ column_name }}) over (order by {{ column_name }}) as prev_value from {{ model }} ) select * from lagged where {{ column_name }} - prev_value != {{ interval }} and prev_value is not null {% endtest %} ``` ```sql -- tests/singular/assert_orders_customers_match.sql -- Singular test: specific business rule with orders_customers as ( select distinct customer_id from {{ ref('fct_orders') }} ), dim_customers as ( select customer_id from {{ ref('dim_customers') }} ), orphaned_orders as ( select o.customer_id from orders_customers o left join dim_customers c using (customer_id) where c.customer_id is null ) select * from orphaned_orders -- Test passes if this returns 0 rows ``` ### Pattern 5: Data Contracts ```yaml # contracts/orders_contract.yaml apiVersion: datacontract.com/v1.0.0 kind: DataContract metadata: name: orders version: 1.0.0 owner: data-platform-team contact: data-team@company.com info: title: Orders Data Contract description: Contract for order event data from the ecommerce platform purpose: Analytics, reporting, and ML features servers: production: type: snowflake account: company.us-east-1 database: ANALYTICS schema: CORE terms: usage: Internal analytics only limitations: PII must not be exposed in downstream marts billing: Charged per query TB scanned schema: type: object properties: order_id: type: string format: uuid description: Unique order identifier required: true unique: true pii: false customer_id: type: string format: uuid description: Customer identifier required: true pii: true piiClassification: indirect total_amount: type: number minimum: 0 maximum: 100000 description: Order total in USD created_at: type: string format: date-time description: Order creation timestamp required: true status: type: string enum: [pending, processing, shipped, delivered, cancelled] description: Current order status quality: type: SodaCL specification: checks for orders: - row_count > 0 - missing_count(order_id) = 0 - duplicate_count(order_id) = 0 - invalid_count(status) = 0: valid values: [pending, processing, shipped, delivered, cancelled] - freshness(created_at) < 24h sla: availability: 99.9% freshness: 1 hour latency: 5 minutes ``` ### Pattern 6: Automated Quality Pipeline ```python # quality_pipeline.py from dataclasses import dataclass from typing import List, Dict, Any import great_expectations as gx from datetime import datetime @dataclass class QualityResult: table: str passed: bool total_expectations: int failed_expectations: int details: List[Dict[str, Any]] timestamp: datetime class DataQualityPipeline: """Orchestrate data quality checks across tables""" def __init__(self, context: gx.DataContext): self.context = context self.results: List[QualityResult] = [] def validate_table(self, table: str, suite: str) -> QualityResult: """Validate a single table against expectation suite""" checkpoint_config = { "name": f"{table}_validation", "config_version": 1.0, "class_name": "Checkpoint", "validations": [{ "batch_request": { "datasource_name": "warehouse", "data_asset_name": table, }, "expectation_suite_name": suite, }], } result = self.context.run_checkpoint(**checkpoint_config) # Parse results validation_result = list(result.run_results.values())[0] results = validation_result.results failed = [r for r in results if not r.success] return QualityResult( table=table, passed=result.success, total_expectations=len(results), failed_expectations=len(failed), details=[{ "expectation": r.expectation_config.expectation_type, "success": r.success, "observed_value": r.result.get("observed_value"), } for r in results], timestamp=datetime.now() ) def run_all(self, tables: Dict[str, str]) -> Dict[str, QualityResult]: """Run validation for all tables""" results = {} for table, suite in tables.items(): print(f"Validating {table}...") results[table] = self.validate_table(table, suite) return results def generate_report(self, results: Dict[str, QualityResult]) -> str: """Generate quality report""" report = ["# Data Quality Report", f"Generated: {datetime.now()}", ""] total_passed = sum(1 for r in results.values() if r.passed) total_tables = len(results) report.append(f"## Summary: {total_passed}/{total_tables} tables passed") report.append("") for table, result in results.items(): status = "✅" if result.passed else "❌" report.append(f"### {status} {table}") report.append(f"- Expectations: {result.total_expectations}") report.append(f"- Failed: {result.failed_expectations}") if not result.passed: report.append("- Failed checks:") for detail in result.details: if not detail["success"]: report.append(f" - {detail['expectation']}: {detail['observed_value']}") report.append("") return "\n".join(report) # Usage context = gx.get_context() pipeline = DataQualityPipeline(context) tables_to_validate = { "orders": "orders_suite", "customers": "customers_suite", "products": "products_suite", } results = pipeline.run_all(tables_to_validate) report = pipeline.generate_report(results) # Fail pipeline if any table failed if not all(r.passed for r in results.values()): print(report) raise ValueError("Data quality checks failed!") ``` ## Best Practices ### Do's - **Test early** - Validate source data before transformations - **Test incrementally** - Add tests as you find issues - **Document expectations** - Clear descriptions for each test - **Alert on failures** - Integrate with monitoring - **Version contracts** - Track schema changes ### Don'ts - **Don't test everything** - Focus on critical columns - **Don't ignore warnings** - They often precede failures - **Don't skip freshness** - Stale data is bad data - **Don't hardcode thresholds** - Use dynamic baselines - **Don't test in isolation** - Test relationships too ## Resources - [Great Expectations Documentation](https://docs.greatexpectations.io/) - [dbt Testing Documentation](https://docs.getdbt.com/docs/build/tests) - [Data Contract Specification](https://datacontract.com/) - [Soda Core](https://docs.soda.io/soda-core/overview.html)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

vector-index-tuning

Optimize vector index performance for latency, recall, and memory.

coding
⭐1
# Vector Index Tuning Guide to optimizing vector indexes for production performance. ## When to Use This Skill - Tuning HNSW parameters - Implementing quantization - Optimizing memory usage - Reducing search latency - Balancing recall vs speed - Scaling to billions of vectors ## Core Concepts ### 1. Index Type Selection ``` Data Size Recommended Index ──────────────────────────────────────── < 10K vectors → Flat (exact search) 10K - 1M → HNSW 1M - 100M → HNSW + Quantization > 100M → IVF + PQ or DiskANN ``` ### 2. HNSW Parameters | Parameter | Default | Effect | | ------------------ | ------- | ---------------------------------------------------- | | **M** | 16 | Connections per node, ↑ = better recall, more memory | | **efConstruction** | 100 | Build quality, ↑ = better index, slower build | | **efSearch** | 50 | Search quality, ↑ = better recall, slower search | ### 3. Quantization Types ``` Full Precision (FP32): 4 bytes × dimensions Half Precision (FP16): 2 bytes × dimensions INT8 Scalar: 1 byte × dimensions Product Quantization: ~32-64 bytes total Binary: dimensions/8 bytes ``` ## Templates ### Template 1: HNSW Parameter Tuning ```python import numpy as np from typing import List, Tuple import time def benchmark_hnsw_parameters( vectors: np.ndarray, queries: np.ndarray, ground_truth: np.ndarray, m_values: List[int] = [8, 16, 32, 64], ef_construction_values: List[int] = [64, 128, 256], ef_search_values: List[int] = [32, 64, 128, 256] ) -> List[dict]: """Benchmark different HNSW configurations.""" import hnswlib results = [] dim = vectors.shape[1] n = vectors.shape[0] for m in m_values: for ef_construction in ef_construction_values: # Build index index = hnswlib.Index(space='cosine', dim=dim) index.init_index(max_elements=n, M=m, ef_construction=ef_construction) build_start = time.time() index.add_items(vectors) build_time = time.time() - build_start # Get memory usage memory_bytes = index.element_count * ( dim * 4 + # Vector storage m * 2 * 4 # Graph edges (approximate) ) for ef_search in ef_search_values: index.set_ef(ef_search) # Measure search search_start = time.time() labels, distances = index.knn_query(queries, k=10) search_time = time.time() - search_start # Calculate recall recall = calculate_recall(labels, ground_truth, k=10) results.append({ "M": m, "ef_construction": ef_construction, "ef_search": ef_search, "build_time_s": build_time, "search_time_ms": search_time * 1000 / len(queries), "recall@10": recall, "memory_mb": memory_bytes / 1024 / 1024 }) return results def calculate_recall(predictions: np.ndarray, ground_truth: np.ndarray, k: int) -> float: """Calculate recall@k.""" correct = 0 for pred, truth in zip(predictions, ground_truth): correct += len(set(pred[:k]) & set(truth[:k])) return correct / (len(predictions) * k) def recommend_hnsw_params( num_vectors: int, target_recall: float = 0.95, max_latency_ms: float = 10, available_memory_gb: float = 8 ) -> dict: """Recommend HNSW parameters based on requirements.""" # Base recommendations if num_vectors < 100_000: m = 16 ef_construction = 100 elif num_vectors < 1_000_000: m = 32 ef_construction = 200 else: m = 48 ef_construction = 256 # Adjust ef_search based on recall target if target_recall >= 0.99: ef_search = 256 elif target_recall >= 0.95: ef_search = 128 else: ef_search = 64 return { "M": m, "ef_construction": ef_construction, "ef_search": ef_search, "notes": f"Estimated for {num_vectors:,} vectors, {target_recall:.0%} recall" } ``` ### Template 2: Quantization Strategies ```python import numpy as np from typing import Optional class VectorQuantizer: """Quantization strategies for vector compression.""" @staticmethod def scalar_quantize_int8( vectors: np.ndarray, min_val: Optional[float] = None, max_val: Optional[float] = None ) -> Tuple[np.ndarray, dict]: """Scalar quantization to INT8.""" if min_val is None: min_val = vectors.min() if max_val is None: max_val = vectors.max() # Scale to 0-255 range scale = 255.0 / (max_val - min_val) quantized = np.clip( np.round((vectors - min_val) * scale), 0, 255 ).astype(np.uint8) params = {"min_val": min_val, "max_val": max_val, "scale": scale} return quantized, params @staticmethod def dequantize_int8( quantized: np.ndarray, params: dict ) -> np.ndarray: """Dequantize INT8 vectors.""" return quantized.astype(np.float32) / params["scale"] + params["min_val"] @staticmethod def product_quantize( vectors: np.ndarray, n_subvectors: int = 8, n_centroids: int = 256 ) -> Tuple[np.ndarray, dict]: """Product quantization for aggressive compression.""" from sklearn.cluster import KMeans n, dim = vectors.shape assert dim % n_subvectors == 0 subvector_dim = dim // n_subvectors codebooks = [] codes = np.zeros((n, n_subvectors), dtype=np.uint8) for i in range(n_subvectors): start = i * subvector_dim end = (i + 1) * subvector_dim subvectors = vectors[:, start:end] kmeans = KMeans(n_clusters=n_centroids, random_state=42) codes[:, i] = kmeans.fit_predict(subvectors) codebooks.append(kmeans.cluster_centers_) params = { "codebooks": codebooks, "n_subvectors": n_subvectors, "subvector_dim": subvector_dim } return codes, params @staticmethod def binary_quantize(vectors: np.ndarray) -> np.ndarray: """Binary quantization (sign of each dimension).""" # Convert to binary: positive = 1, negative = 0 binary = (vectors > 0).astype(np.uint8) # Pack bits into bytes n, dim = vectors.shape packed_dim = (dim + 7) // 8 packed = np.zeros((n, packed_dim), dtype=np.uint8) for i in range(dim): byte_idx = i // 8 bit_idx = i % 8 packed[:, byte_idx] |= (binary[:, i] << bit_idx) return packed def estimate_memory_usage( num_vectors: int, dimensions: int, quantization: str = "fp32", index_type: str = "hnsw", hnsw_m: int = 16 ) -> dict: """Estimate memory usage for different configurations.""" # Vector storage bytes_per_dimension = { "fp32": 4, "fp16": 2, "int8": 1, "pq": 0.05, # Approximate "binary": 0.125 } vector_bytes = num_vectors * dimensions * bytes_per_dimension[quantization] # Index overhead if index_type == "hnsw": # Each node has ~M*2 edges, each edge is 4 bytes (int32) index_bytes = num_vectors * hnsw_m * 2 * 4 elif index_type == "ivf": # Inverted lists + centroids index_bytes = num_vectors * 8 + 65536 * dimensions * 4 else: index_bytes = 0 total_bytes = vector_bytes + index_bytes return { "vector_storage_mb": vector_bytes / 1024 / 1024, "index_overhead_mb": index_bytes / 1024 / 1024, "total_mb": total_bytes / 1024 / 1024, "total_gb": total_bytes / 1024 / 1024 / 1024 } ``` ### Template 3: Qdrant Index Configuration ```python from qdrant_client import QdrantClient from qdrant_client.http import models def create_optimized_collection( client: QdrantClient, collection_name: str, vector_size: int, num_vectors: int, optimize_for: str = "balanced" # "recall", "speed", "memory" ) -> None: """Create collection with optimized settings.""" # HNSW configuration based on optimization target hnsw_configs = { "recall": models.HnswConfigDiff(m=32, ef_construct=256), "speed": models.HnswConfigDiff(m=16, ef_construct=64), "balanced": models.HnswConfigDiff(m=16, ef_construct=128), "memory": models.HnswConfigDiff(m=8, ef_construct=64) } # Quantization configuration quantization_configs = { "recall": None, # No quantization for max recall "speed": models.ScalarQuantization( scalar=models.ScalarQuantizationConfig( type=models.ScalarType.INT8, quantile=0.99, always_ram=True ) ), "balanced": models.ScalarQuantization( scalar=models.ScalarQuantizationConfig( type=models.ScalarType.INT8, quantile=0.99, always_ram=False ) ), "memory": models.ProductQuantization( product=models.ProductQuantizationConfig( compression=models.CompressionRatio.X16, always_ram=False ) ) } # Optimizer configuration optimizer_configs = { "recall": models.OptimizersConfigDiff( indexing_threshold=10000, memmap_threshold=50000 ), "speed": models.OptimizersConfigDiff( indexing_threshold=5000, memmap_threshold=20000 ), "balanced": models.OptimizersConfigDiff( indexing_threshold=20000, memmap_threshold=50000 ), "memory": models.OptimizersConfigDiff( indexing_threshold=50000, memmap_threshold=10000 # Use disk sooner ) } client.create_collection( collection_name=collection_name, vectors_config=models.VectorParams( size=vector_size, distance=models.Distance.COSINE ), hnsw_config=hnsw_configs[optimize_for], quantization_config=quantization_configs[optimize_for], optimizers_config=optimizer_configs[optimize_for] ) def tune_search_parameters( client: QdrantClient, collection_name: str, target_recall: float = 0.95 ) -> dict: """Tune search parameters for target recall.""" # Search parameter recommendations if target_recall >= 0.99: search_params = models.SearchParams( hnsw_ef=256, exact=False, quantization=models.QuantizationSearchParams( ignore=True, # Don't use quantization for search rescore=True ) ) elif target_recall >= 0.95: search_params = models.SearchParams( hnsw_ef=128, exact=False, quantization=models.QuantizationSearchParams( ignore=False, rescore=True, oversampling=2.0 ) ) else: search_params = models.SearchParams( hnsw_ef=64, exact=False, quantization=models.QuantizationSearchParams( ignore=False, rescore=False ) ) return search_params ``` ### Template 4: Performance Monitoring ```python import time from dataclasses import dataclass from typing import List import numpy as np @dataclass class SearchMetrics: latency_p50_ms: float latency_p95_ms: float latency_p99_ms: float recall: float qps: float class VectorSearchMonitor: """Monitor vector search performance.""" def __init__(self, ground_truth_fn=None): self.latencies = [] self.recalls = [] self.ground_truth_fn = ground_truth_fn def measure_search( self, search_fn, query_vectors: np.ndarray, k: int = 10, num_iterations: int = 100 ) -> SearchMetrics: """Benchmark search performance.""" latencies = [] for _ in range(num_iterations): for query in query_vectors: start = time.perf_counter() results = search_fn(query, k=k) latency = (time.perf_counter() - start) * 1000 latencies.append(latency) latencies = np.array(latencies) total_queries = num_iterations * len(query_vectors) total_time = sum(latencies) / 1000 # seconds return SearchMetrics( latency_p50_ms=np.percentile(latencies, 50), latency_p95_ms=np.percentile(latencies, 95), latency_p99_ms=np.percentile(latencies, 99), recall=self._calculate_recall(search_fn, query_vectors, k) if self.ground_truth_fn else 0, qps=total_queries / total_time ) def _calculate_recall(self, search_fn, queries: np.ndarray, k: int) -> float: """Calculate recall against ground truth.""" if not self.ground_truth_fn: return 0 correct = 0 total = 0 for query in queries: predicted = set(search_fn(query, k=k)) actual = set(self.ground_truth_fn(query, k=k)) correct += len(predicted & actual) total += k return correct / total def profile_index_build( build_fn, vectors: np.ndarray, batch_sizes: List[int] = [1000, 10000, 50000] ) -> dict: """Profile index build performance.""" results = {} for batch_size in batch_sizes: times = [] for i in range(0, len(vectors), batch_size): batch = vectors[i:i + batch_size] start = time.perf_counter() build_fn(batch) times.append(time.perf_counter() - start) results[batch_size] = { "avg_batch_time_s": np.mean(times), "vectors_per_second": batch_size / np.mean(times) } return results ``` ## Best Practices ### Do's - **Benchmark with real queries** - Synthetic may not represent production - **Monitor recall continuously** - Can degrade with data drift - **Start with defaults** - Tune only when needed - **Use quantization** - Significant memory savings - **Consider tiered storage** - Hot/cold data separation ### Don'ts - **Don't over-optimize early** - Profile first - **Don't ignore build time** - Index updates have cost - **Don't forget reindexing** - Plan for maintenance - **Don't skip warming** - Cold indexes are slow ## Resources - [HNSW Paper](https://arxiv.org/abs/1603.09320) - [Faiss Wiki](https://github.com/facebookresearch/faiss/wiki) - [ANN Benchmarks](https://ann-benchmarks.com/)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

billing-automation

Build automated billing systems for recurring payments, invoicing,

business
⭐1
# Billing Automation Master automated billing systems including recurring billing, invoice generation, dunning management, proration, and tax calculation. ## When to Use This Skill - Implementing SaaS subscription billing - Automating invoice generation and delivery - Managing failed payment recovery (dunning) - Calculating prorated charges for plan changes - Handling sales tax, VAT, and GST - Processing usage-based billing - Managing billing cycles and renewals ## Core Concepts ### 1. Billing Cycles **Common Intervals:** - Monthly (most common for SaaS) - Annual (discounted long-term) - Quarterly - Weekly - Custom (usage-based, per-seat) ### 2. Subscription States ``` trial → active → past_due → canceled → paused → resumed ``` ### 3. Dunning Management Automated process to recover failed payments through: - Retry schedules - Customer notifications - Grace periods - Account restrictions ### 4. Proration Adjusting charges when: - Upgrading/downgrading mid-cycle - Adding/removing seats - Changing billing frequency ## Quick Start ```python from billing import BillingEngine, Subscription # Initialize billing engine billing = BillingEngine() # Create subscription subscription = billing.create_subscription( customer_id="cus_123", plan_id="plan_pro_monthly", billing_cycle_anchor=datetime.now(), trial_days=14 ) # Process billing cycle billing.process_billing_cycle(subscription.id) ``` ## Subscription Lifecycle Management ```python from datetime import datetime, timedelta from enum import Enum class SubscriptionStatus(Enum): TRIAL = "trial" ACTIVE = "active" PAST_DUE = "past_due" CANCELED = "canceled" PAUSED = "paused" class Subscription: def __init__(self, customer_id, plan, billing_cycle_day=None): self.id = generate_id() self.customer_id = customer_id self.plan = plan self.status = SubscriptionStatus.TRIAL self.current_period_start = datetime.now() self.current_period_end = self.current_period_start + timedelta(days=plan.trial_days or 30) self.billing_cycle_day = billing_cycle_day or self.current_period_start.day self.trial_end = datetime.now() + timedelta(days=plan.trial_days) if plan.trial_days else None def start_trial(self, trial_days): """Start trial period.""" self.status = SubscriptionStatus.TRIAL self.trial_end = datetime.now() + timedelta(days=trial_days) self.current_period_end = self.trial_end def activate(self): """Activate subscription after trial or immediately.""" self.status = SubscriptionStatus.ACTIVE self.current_period_start = datetime.now() self.current_period_end = self.calculate_next_billing_date() def mark_past_due(self): """Mark subscription as past due after failed payment.""" self.status = SubscriptionStatus.PAST_DUE # Trigger dunning workflow def cancel(self, at_period_end=True): """Cancel subscription.""" if at_period_end: self.cancel_at_period_end = True # Will cancel when current period ends else: self.status = SubscriptionStatus.CANCELED self.canceled_at = datetime.now() def calculate_next_billing_date(self): """Calculate next billing date based on interval.""" if self.plan.interval == 'month': return self.current_period_start + timedelta(days=30) elif self.plan.interval == 'year': return self.current_period_start + timedelta(days=365) elif self.plan.interval == 'week': return self.current_period_start + timedelta(days=7) ``` ## Billing Cycle Processing ```python class BillingEngine: def process_billing_cycle(self, subscription_id): """Process billing for a subscription.""" subscription = self.get_subscription(subscription_id) # Check if billing is due if datetime.now() < subscription.current_period_end: return # Generate invoice invoice = self.generate_invoice(subscription) # Attempt payment payment_result = self.charge_customer( subscription.customer_id, invoice.total ) if payment_result.success: # Payment successful invoice.mark_paid() subscription.advance_billing_period() self.send_invoice(invoice) else: # Payment failed subscription.mark_past_due() self.start_dunning_process(subscription, invoice) def generate_invoice(self, subscription): """Generate invoice for billing period.""" invoice = Invoice( customer_id=subscription.customer_id, subscription_id=subscription.id, period_start=subscription.current_period_start, period_end=subscription.current_period_end ) # Add subscription line item invoice.add_line_item( description=subscription.plan.name, amount=subscription.plan.amount, quantity=subscription.quantity or 1 ) # Add usage-based charges if applicable if subscription.has_usage_billing: usage_charges = self.calculate_usage_charges(subscription) invoice.add_line_item( description="Usage charges", amount=usage_charges ) # Calculate tax tax = self.calculate_tax(invoice.subtotal, subscription.customer) invoice.tax = tax invoice.finalize() return invoice def charge_customer(self, customer_id, amount): """Charge customer using saved payment method.""" customer = self.get_customer(customer_id) try: # Charge using payment processor charge = stripe.Charge.create( customer=customer.stripe_id, amount=int(amount * 100), # Convert to cents currency='usd' ) return PaymentResult(success=True, transaction_id=charge.id) except stripe.error.CardError as e: return PaymentResult(success=False, error=str(e)) ``` ## Dunning Management ```python class DunningManager: """Manage failed payment recovery.""" def __init__(self): self.retry_schedule = [ {'days': 3, 'email_template': 'payment_failed_first'}, {'days': 7, 'email_template': 'payment_failed_reminder'}, {'days': 14, 'email_template': 'payment_failed_final'} ] def start_dunning_process(self, subscription, invoice): """Start dunning process for failed payment.""" dunning_attempt = DunningAttempt( subscription_id=subscription.id, invoice_id=invoice.id, attempt_number=1, next_retry=datetime.now() + timedelta(days=3) ) # Send initial failure notification self.send_dunning_email(subscription, 'payment_failed_first') # Schedule retries self.schedule_retries(dunning_attempt) def retry_payment(self, dunning_attempt): """Retry failed payment.""" subscription = self.get_subscription(dunning_attempt.subscription_id) invoice = self.get_invoice(dunning_attempt.invoice_id) # Attempt payment again result = self.charge_customer(subscription.customer_id, invoice.total) if result.success: # Payment succeeded invoice.mark_paid() subscription.status = SubscriptionStatus.ACTIVE self.send_dunning_email(subscription, 'payment_recovered') dunning_attempt.mark_resolved() else: # Still failing dunning_attempt.attempt_number += 1 if dunning_attempt.attempt_number < len(self.retry_schedule): # Schedule next retry next_retry_config = self.retry_schedule[dunning_attempt.attempt_number] dunning_attempt.next_retry = datetime.now() + timedelta(days=next_retry_config['days']) self.send_dunning_email(subscription, next_retry_config['email_template']) else: # Exhausted retries, cancel subscription subscription.cancel(at_period_end=False) self.send_dunning_email(subscription, 'subscription_canceled') def send_dunning_email(self, subscription, template): """Send dunning notification to customer.""" customer = self.get_customer(subscription.customer_id) email_content = self.render_template(template, { 'customer_name': customer.name, 'amount_due': subscription.plan.amount, 'update_payment_url': f"https://app.example.com/billing" }) send_email( to=customer.email, subject=email_content['subject'], body=email_content['body'] ) ``` ## Proration ```python class ProrationCalculator: """Calculate prorated charges for plan changes.""" @staticmethod def calculate_proration(old_plan, new_plan, period_start, period_end, change_date): """Calculate proration for plan change.""" # Days in current period total_days = (period_end - period_start).days # Days used on old plan days_used = (change_date - period_start).days # Days remaining on new plan days_remaining = (period_end - change_date).days # Calculate prorated amounts unused_amount = (old_plan.amount / total_days) * days_remaining new_plan_amount = (new_plan.amount / total_days) * days_remaining # Net charge/credit proration = new_plan_amount - unused_amount return { 'old_plan_credit': -unused_amount, 'new_plan_charge': new_plan_amount, 'net_proration': proration, 'days_used': days_used, 'days_remaining': days_remaining } @staticmethod def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date): """Calculate proration for seat changes.""" total_days = (period_end - period_start).days days_remaining = (period_end - change_date).days # Additional seats charge additional_seats = new_seats - current_seats prorated_amount = (additional_seats * price_per_seat / total_days) * days_remaining return { 'additional_seats': additional_seats, 'prorated_charge': max(0, prorated_amount), # No refund for removing seats mid-cycle 'effective_date': change_date } ``` ## Tax Calculation ```python class TaxCalculator: """Calculate sales tax, VAT, GST.""" def __init__(self): # Tax rates by region self.tax_rates = { 'US_CA': 0.0725, # California sales tax 'US_NY': 0.04, # New York sales tax 'GB': 0.20, # UK VAT 'DE': 0.19, # Germany VAT 'FR': 0.20, # France VAT 'AU': 0.10, # Australia GST } def calculate_tax(self, amount, customer): """Calculate applicable tax.""" # Determine tax jurisdiction jurisdiction = self.get_tax_jurisdiction(customer) if not jurisdiction: return 0 # Get tax rate tax_rate = self.tax_rates.get(jurisdiction, 0) # Calculate tax tax = amount * tax_rate return { 'tax_amount': tax, 'tax_rate': tax_rate, 'jurisdiction': jurisdiction, 'tax_type': self.get_tax_type(jurisdiction) } def get_tax_jurisdiction(self, customer): """Determine tax jurisdiction based on customer location.""" if customer.country == 'US': # US: Tax based on customer state return f"US_{customer.state}" elif customer.country in ['GB', 'DE', 'FR']: # EU: VAT return customer.country elif customer.country == 'AU': # Australia: GST return 'AU' else: return None def get_tax_type(self, jurisdiction): """Get type of tax for jurisdiction.""" if jurisdiction.startswith('US_'): return 'Sales Tax' elif jurisdiction in ['GB', 'DE', 'FR']: return 'VAT' elif jurisdiction == 'AU': return 'GST' return 'Tax' def validate_vat_number(self, vat_number, country): """Validate EU VAT number.""" # Use VIES API for validation # Returns True if valid, False otherwise pass ``` ## Invoice Generation ```python class Invoice: def __init__(self, customer_id, subscription_id=None): self.id = generate_invoice_number() self.customer_id = customer_id self.subscription_id = subscription_id self.status = 'draft' self.line_items = [] self.subtotal = 0 self.tax = 0 self.total = 0 self.created_at = datetime.now() def add_line_item(self, description, amount, quantity=1): """Add line item to invoice.""" line_item = { 'description': description, 'unit_amount': amount, 'quantity': quantity, 'total': amount * quantity } self.line_items.append(line_item) self.subtotal += line_item['total'] def finalize(self): """Finalize invoice and calculate total.""" self.total = self.subtotal + self.tax self.status = 'open' self.finalized_at = datetime.now() def mark_paid(self): """Mark invoice as paid.""" self.status = 'paid' self.paid_at = datetime.now() def to_pdf(self): """Generate PDF invoice.""" from reportlab.pdfgen import canvas # Generate PDF # Include: company info, customer info, line items, tax, total pass def to_html(self): """Generate HTML invoice.""" template = """ <!DOCTYPE html> <html> <head><title>Invoice #{invoice_number}</title></head> <body> <h1>Invoice #{invoice_number}</h1> <p>Date: {date}</p> <h2>Bill To:</h2> <p>{customer_name}<br>{customer_address}</p> <table> <tr><th>Description</th><th>Quantity</th><th>Amount</th></tr> {line_items} </table> <p>Subtotal: ${subtotal}</p> <p>Tax: ${tax}</p> <h3>Total: ${total}</h3> </body> </html> """ return template.format( invoice_number=self.id, date=self.created_at.strftime('%Y-%m-%d'), customer_name=self.customer.name, customer_address=self.customer.address, line_items=self.render_line_items(), subtotal=self.subtotal, tax=self.tax, total=self.total ) ``` ## Usage-Based Billing ```python class UsageBillingEngine: """Track and bill for usage.""" def track_usage(self, customer_id, metric, quantity): """Track usage event.""" UsageRecord.create( customer_id=customer_id, metric=metric, quantity=quantity, timestamp=datetime.now() ) def calculate_usage_charges(self, subscription, period_start, period_end): """Calculate charges for usage in billing period.""" usage_records = UsageRecord.get_for_period( subscription.customer_id, period_start, period_end ) total_usage = sum(record.quantity for record in usage_records) # Tiered pricing if subscription.plan.pricing_model == 'tiered': charge = self.calculate_tiered_pricing(total_usage, subscription.plan.tiers) # Per-unit pricing elif subscription.plan.pricing_model == 'per_unit': charge = total_usage * subscription.plan.unit_price # Volume pricing elif subscription.plan.pricing_model == 'volume': charge = self.calculate_volume_pricing(total_usage, subscription.plan.tiers) return charge def calculate_tiered_pricing(self, total_usage, tiers): """Calculate cost using tiered pricing.""" charge = 0 remaining = total_usage for tier in sorted(tiers, key=lambda x: x['up_to']): tier_usage = min(remaining, tier['up_to'] - tier['from']) charge += tier_usage * tier['unit_price'] remaining -= tier_usage if remaining <= 0: break return charge ``` ## Resources - **references/billing-cycles.md**: Billing cycle management - **references/dunning-management.md**: Failed payment recovery - **references/proration.md**: Prorated charge calculations - **references/tax-calculation.md**: Tax/VAT/GST handling - **references/invoice-lifecycle.md**: Invoice state management - **assets/billing-state-machine.yaml**: Billing workflow - **assets/invoice-template.html**: Invoice templates - **assets/dunning-policy.yaml**: Dunning configuration ## Best Practices 1. **Automate Everything**: Minimize manual intervention 2. **Clear Communication**: Notify customers of billing events 3. **Flexible Retry Logic**: Balance recovery with customer experience 4. **Accurate Proration**: Fair calculation for plan changes 5. **Tax Compliance**: Calculate correct tax for jurisdiction 6. **Audit Trail**: Log all billing events 7. **Graceful Degradation**: Handle edge cases without breaking ## Common Pitfalls - **Incorrect Proration**: Not accounting for partial periods - **Missing Tax**: Forgetting to add tax to invoices - **Aggressive Dunning**: Canceling too quickly - **No Notifications**: Not informing customers of failures - **Hardcoded Cycles**: Not supporting custom billing dates
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

paypal-integration

Integrate PayPal payment processing with support for express

business
⭐1
# PayPal Integration Master PayPal payment integration including Express Checkout, IPN handling, recurring billing, and refund workflows. ## When to Use This Skill - Integrating PayPal as a payment option - Implementing express checkout flows - Setting up recurring billing with PayPal - Processing refunds and payment disputes - Handling PayPal webhooks (IPN) - Supporting international payments - Implementing PayPal subscriptions ## Core Concepts ### 1. Payment Products **PayPal Checkout** - One-time payments - Express checkout experience - Guest and PayPal account payments **PayPal Subscriptions** - Recurring billing - Subscription plans - Automatic renewals **PayPal Payouts** - Send money to multiple recipients - Marketplace and platform payments ### 2. Integration Methods **Client-Side (JavaScript SDK)** - Smart Payment Buttons - Hosted payment flow - Minimal backend code **Server-Side (REST API)** - Full control over payment flow - Custom checkout UI - Advanced features ### 3. IPN (Instant Payment Notification) - Webhook-like payment notifications - Asynchronous payment updates - Verification required ## Quick Start ```javascript // Frontend - PayPal Smart Buttons <div id="paypal-button-container"></div> <script src="https://www.paypal.com/sdk/js?client-id=YOUR_CLIENT_ID&currency=USD"></script> <script> paypal.Buttons({ createOrder: function(data, actions) { return actions.order.create({ purchase_units: [{ amount: { value: '25.00' } }] }); }, onApprove: function(data, actions) { return actions.order.capture().then(function(details) { // Payment successful console.log('Transaction completed by ' + details.payer.name.given_name); // Send to backend for verification fetch('/api/paypal/capture', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({orderID: data.orderID}) }); }); } }).render('#paypal-button-container'); </script> ``` ```python # Backend - Verify and capture order from paypalrestsdk import Payment import paypalrestsdk paypalrestsdk.configure({ "mode": "sandbox", # or "live" "client_id": "YOUR_CLIENT_ID", "client_secret": "YOUR_CLIENT_SECRET" }) def capture_paypal_order(order_id): """Capture a PayPal order.""" payment = Payment.find(order_id) if payment.execute({"payer_id": payment.payer.payer_info.payer_id}): # Payment successful return { 'status': 'success', 'transaction_id': payment.id, 'amount': payment.transactions[0].amount.total } else: # Payment failed return { 'status': 'failed', 'error': payment.error } ``` ## Express Checkout Implementation ### Server-Side Order Creation ```python import requests import json class PayPalClient: def __init__(self, client_id, client_secret, mode='sandbox'): self.client_id = client_id self.client_secret = client_secret self.base_url = 'https://api-m.sandbox.paypal.com' if mode == 'sandbox' else 'https://api-m.paypal.com' self.access_token = self.get_access_token() def get_access_token(self): """Get OAuth access token.""" url = f"{self.base_url}/v1/oauth2/token" headers = {"Accept": "application/json", "Accept-Language": "en_US"} response = requests.post( url, headers=headers, data={"grant_type": "client_credentials"}, auth=(self.client_id, self.client_secret) ) return response.json()['access_token'] def create_order(self, amount, currency='USD'): """Create a PayPal order.""" url = f"{self.base_url}/v2/checkout/orders" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.access_token}" } payload = { "intent": "CAPTURE", "purchase_units": [{ "amount": { "currency_code": currency, "value": str(amount) } }] } response = requests.post(url, headers=headers, json=payload) return response.json() def capture_order(self, order_id): """Capture payment for an order.""" url = f"{self.base_url}/v2/checkout/orders/{order_id}/capture" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.access_token}" } response = requests.post(url, headers=headers) return response.json() def get_order_details(self, order_id): """Get order details.""" url = f"{self.base_url}/v2/checkout/orders/{order_id}" headers = { "Authorization": f"Bearer {self.access_token}" } response = requests.get(url, headers=headers) return response.json() ``` ## IPN (Instant Payment Notification) Handling ### IPN Verification and Processing ```python from flask import Flask, request import requests from urllib.parse import parse_qs app = Flask(__name__) @app.route('/ipn', methods=['POST']) def handle_ipn(): """Handle PayPal IPN notifications.""" # Get IPN message ipn_data = request.form.to_dict() # Verify IPN with PayPal if not verify_ipn(ipn_data): return 'IPN verification failed', 400 # Process IPN based on transaction type payment_status = ipn_data.get('payment_status') txn_type = ipn_data.get('txn_type') if payment_status == 'Completed': handle_payment_completed(ipn_data) elif payment_status == 'Refunded': handle_refund(ipn_data) elif payment_status == 'Reversed': handle_chargeback(ipn_data) return 'IPN processed', 200 def verify_ipn(ipn_data): """Verify IPN message authenticity.""" # Add 'cmd' parameter verify_data = ipn_data.copy() verify_data['cmd'] = '_notify-validate' # Send back to PayPal for verification paypal_url = 'https://ipnpb.sandbox.paypal.com/cgi-bin/webscr' # or production URL response = requests.post(paypal_url, data=verify_data) return response.text == 'VERIFIED' def handle_payment_completed(ipn_data): """Process completed payment.""" txn_id = ipn_data.get('txn_id') payer_email = ipn_data.get('payer_email') mc_gross = ipn_data.get('mc_gross') item_name = ipn_data.get('item_name') # Check if already processed (prevent duplicates) if is_transaction_processed(txn_id): return # Update database # Send confirmation email # Fulfill order print(f"Payment completed: {txn_id}, Amount: ${mc_gross}") def handle_refund(ipn_data): """Handle refund.""" parent_txn_id = ipn_data.get('parent_txn_id') mc_gross = ipn_data.get('mc_gross') # Process refund in your system print(f"Refund processed: {parent_txn_id}, Amount: ${mc_gross}") def handle_chargeback(ipn_data): """Handle payment reversal/chargeback.""" txn_id = ipn_data.get('txn_id') reason_code = ipn_data.get('reason_code') # Handle chargeback print(f"Chargeback: {txn_id}, Reason: {reason_code}") ``` ## Subscription/Recurring Billing ### Create Subscription Plan ```python def create_subscription_plan(name, amount, interval='MONTH'): """Create a subscription plan.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v1/billing/plans" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = { "product_id": "PRODUCT_ID", # Create product first "name": name, "billing_cycles": [{ "frequency": { "interval_unit": interval, "interval_count": 1 }, "tenure_type": "REGULAR", "sequence": 1, "total_cycles": 0, # Infinite "pricing_scheme": { "fixed_price": { "value": str(amount), "currency_code": "USD" } } }], "payment_preferences": { "auto_bill_outstanding": True, "setup_fee": { "value": "0", "currency_code": "USD" }, "setup_fee_failure_action": "CONTINUE", "payment_failure_threshold": 3 } } response = requests.post(url, headers=headers, json=payload) return response.json() def create_subscription(plan_id, subscriber_email): """Create a subscription for a customer.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v1/billing/subscriptions" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = { "plan_id": plan_id, "subscriber": { "email_address": subscriber_email }, "application_context": { "return_url": "https://yourdomain.com/subscription/success", "cancel_url": "https://yourdomain.com/subscription/cancel" } } response = requests.post(url, headers=headers, json=payload) subscription = response.json() # Get approval URL for link in subscription.get('links', []): if link['rel'] == 'approve': return { 'subscription_id': subscription['id'], 'approval_url': link['href'] } ``` ## Refund Workflows ```python def create_refund(capture_id, amount=None, note=None): """Create a refund for a captured payment.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v2/payments/captures/{capture_id}/refund" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = {} if amount: payload["amount"] = { "value": str(amount), "currency_code": "USD" } if note: payload["note_to_payer"] = note response = requests.post(url, headers=headers, json=payload) return response.json() def get_refund_details(refund_id): """Get refund details.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v2/payments/refunds/{refund_id}" headers = { "Authorization": f"Bearer {client.access_token}" } response = requests.get(url, headers=headers) return response.json() ``` ## Error Handling ```python class PayPalError(Exception): """Custom PayPal error.""" pass def handle_paypal_api_call(api_function): """Wrapper for PayPal API calls with error handling.""" try: result = api_function() return result except requests.exceptions.RequestException as e: # Network error raise PayPalError(f"Network error: {str(e)}") except Exception as e: # Other errors raise PayPalError(f"PayPal API error: {str(e)}") # Usage try: order = handle_paypal_api_call(lambda: client.create_order(25.00)) except PayPalError as e: # Handle error appropriately log_error(e) ``` ## Testing ```python # Use sandbox credentials SANDBOX_CLIENT_ID = "..." SANDBOX_SECRET = "..." # Test accounts # Create test buyer and seller accounts at developer.paypal.com def test_payment_flow(): """Test complete payment flow.""" client = PayPalClient(SANDBOX_CLIENT_ID, SANDBOX_SECRET, mode='sandbox') # Create order order = client.create_order(10.00) assert 'id' in order # Get approval URL approval_url = next((link['href'] for link in order['links'] if link['rel'] == 'approve'), None) assert approval_url is not None # After approval (manual step with test account) # Capture order # captured = client.capture_order(order['id']) # assert captured['status'] == 'COMPLETED' ``` ## Resources - **references/express-checkout.md**: Express Checkout implementation guide - **references/ipn-handling.md**: IPN verification and processing - **references/refund-workflows.md**: Refund handling patterns - **references/billing-agreements.md**: Recurring billing setup - **assets/paypal-client.py**: Production PayPal client - **assets/ipn-processor.py**: IPN webhook processor - **assets/recurring-billing.py**: Subscription management ## Best Practices 1. **Always Verify IPN**: Never trust IPN without verification 2. **Idempotent Processing**: Handle duplicate IPN notifications 3. **Error Handling**: Implement robust error handling 4. **Logging**: Log all transactions and errors 5. **Test Thoroughly**: Use sandbox extensively 6. **Webhook Backup**: Don't rely solely on client-side callbacks 7. **Currency Handling**: Always specify currency explicitly ## Common Pitfalls - **Not Verifying IPN**: Accepting IPN without verification - **Duplicate Processing**: Not checking for duplicate transactions - **Wrong Environment**: Mixing sandbox and production URLs/credentials - **Missing Webhooks**: Not handling all payment states - **Hardcoded Values**: Not making configurable for different environments
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

stripe-integration

Implement Stripe payment processing for robust, PCI-compliant

business
⭐1
# Stripe Integration Master Stripe payment processing integration for robust, PCI-compliant payment flows including checkout, subscriptions, webhooks, and refunds. ## When to Use This Skill - Implementing payment processing in web/mobile applications - Setting up subscription billing systems - Handling one-time payments and recurring charges - Processing refunds and disputes - Managing customer payment methods - Implementing SCA (Strong Customer Authentication) for European payments - Building marketplace payment flows with Stripe Connect ## Core Concepts ### 1. Payment Flows **Checkout Sessions** - Recommended for most integrations - Supports all UI paths: - Stripe-hosted checkout page - Embedded checkout form - Custom UI with Elements (Payment Element, Express Checkout Element) using `ui_mode='custom'` - Provides built-in checkout capabilities (line items, discounts, tax, shipping, address collection, saved payment methods, and checkout lifecycle events) - Lower integration and maintenance burden than Payment Intents **Payment Intents (Bespoke control)** - You calculate the final amount with taxes, discounts, subscriptions, and currency conversion yourself. - More complex implementation and long-term maintenance burden - Requires Stripe.js for PCI compliance **Setup Intents (Save Payment Methods)** - Collect payment method without charging - Used for subscriptions and future payments - Requires customer confirmation ### 2. Webhooks **Critical Events:** - `payment_intent.succeeded`: Payment completed - `payment_intent.payment_failed`: Payment failed - `customer.subscription.updated`: Subscription changed - `customer.subscription.deleted`: Subscription canceled - `charge.refunded`: Refund processed - `invoice.payment_succeeded`: Subscription payment successful ### 3. Subscriptions **Components:** - **Product**: What you're selling - **Price**: How much and how often - **Subscription**: Customer's recurring payment - **Invoice**: Generated for each billing cycle ### 4. Customer Management - Create and manage customer records - Store multiple payment methods - Track customer metadata - Manage billing details ## Quick Start ```python import stripe stripe.api_key = "sk_test_..." # Create a checkout session session = stripe.checkout.Session.create( line_items=[{ 'price_data': { 'currency': 'usd', 'product_data': { 'name': 'Premium Subscription', }, 'unit_amount': 2000, # $20.00 'recurring': { 'interval': 'month', }, }, 'quantity': 1, }], mode='subscription', success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}', cancel_url='https://yourdomain.com/cancel' ) # Redirect user to session.url print(session.url) ``` ## Payment Implementation Patterns ### Pattern 1: One-Time Payment (Hosted Checkout) ```python def create_checkout_session(amount, currency='usd'): """Create a one-time payment checkout session.""" try: session = stripe.checkout.Session.create( line_items=[{ 'price_data': { 'currency': currency, 'product_data': { 'name': 'Blue T-shirt', 'images': ['https://example.com/product.jpg'], }, 'unit_amount': amount, # Amount in cents }, 'quantity': 1, }], mode='payment', success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}', cancel_url='https://yourdomain.com/cancel', metadata={ 'order_id': 'order_123', 'user_id': 'user_456' } ) return session except stripe.error.StripeError as e: # Handle error print(f"Stripe error: {e.user_message}") raise ``` ### Pattern 2: Elements with Checkout Sessions ```python def create_checkout_session_for_elements(amount, currency='usd'): """Create a checkout session configured for Payment Element.""" session = stripe.checkout.Session.create( mode='payment', ui_mode='custom', line_items=[{ 'price_data': { 'currency': currency, 'product_data': {'name': 'Blue T-shirt'}, 'unit_amount': amount, }, 'quantity': 1, }], return_url='https://yourdomain.com/complete?session_id={CHECKOUT_SESSION_ID}' ) return session.client_secret # Send to frontend ``` ```javascript const stripe = Stripe("pk_test_..."); const appearance = { theme: "stripe" }; const checkout = stripe.initCheckout({ clientSecret, elementsOptions: { appearance }, }); const loadActionsResult = await checkout.loadActions(); if (loadActionsResult.type === "success") { const { actions } = loadActionsResult; const session = actions.getSession(); const button = document.getElementById("pay-button"); const checkoutContainer = document.getElementById("checkout-container"); const emailInput = document.getElementById("email"); const emailErrors = document.getElementById("email-errors"); const errors = document.getElementById("confirm-errors"); // Display a formatted string representing the total amount checkoutContainer.append(`Total: ${session.total.total.amount}`); // Mount Payment Element const paymentElement = checkout.createPaymentElement(); paymentElement.mount("#payment-element"); // Store email for submission emailInput.addEventListener("blur", () => { actions.updateEmail(emailInput.value).then((result) => { if (result.error) emailErrors.textContent = result.error.message; }); }); // Handle form submission button.addEventListener("click", () => { actions.confirm().then((result) => { if (result.type === "error") errors.textContent = result.error.message; }); }); } ``` ### Pattern 3: Elements with Payment Intents Pattern 2 (Elements with Checkout Sessions) is Stripe's recommended approach, but you can also use Payment Intents as an alternative. ```python def create_payment_intent(amount, currency='usd', customer_id=None): """Create a payment intent for bespoke checkout UI with Payment Element.""" intent = stripe.PaymentIntent.create( amount=amount, currency=currency, customer=customer_id, automatic_payment_methods={ 'enabled': True, }, metadata={ 'integration_check': 'accept_a_payment' } ) return intent.client_secret # Send to frontend ``` ```javascript // Mount Payment Element and confirm via Payment Intents const stripe = Stripe("pk_test_..."); const appearance = { theme: "stripe" }; const elements = stripe.elements({ appearance, clientSecret }); const paymentElement = elements.create("payment"); paymentElement.mount("#payment-element"); document.getElementById("pay-button").addEventListener("click", async () => { const { error } = await stripe.confirmPayment({ elements, confirmParams: { return_url: "https://yourdomain.com/complete", }, }); if (error) { document.getElementById("errors").textContent = error.message; } }); ``` ### Pattern 4: Subscription Creation ```python def create_subscription(customer_id, price_id): """Create a subscription for a customer.""" try: subscription = stripe.Subscription.create( customer=customer_id, items=[{'price': price_id}], payment_behavior='default_incomplete', payment_settings={'save_default_payment_method': 'on_subscription'}, expand=['latest_invoice.payment_intent'], ) return { 'subscription_id': subscription.id, 'client_secret': subscription.latest_invoice.payment_intent.client_secret } except stripe.error.StripeError as e: print(f"Subscription creation failed: {e}") raise ``` ### Pattern 5: Customer Portal ```python def create_customer_portal_session(customer_id): """Create a portal session for customers to manage subscriptions.""" session = stripe.billing_portal.Session.create( customer=customer_id, return_url='https://yourdomain.com/account', ) return session.url # Redirect customer here ``` ## Webhook Handling ### Secure Webhook Endpoint ```python from flask import Flask, request import stripe app = Flask(__name__) endpoint_secret = 'whsec_...' @app.route('/webhook', methods=['POST']) def webhook(): payload = request.data sig_header = request.headers.get('Stripe-Signature') try: event = stripe.Webhook.construct_event( payload, sig_header, endpoint_secret ) except ValueError: # Invalid payload return 'Invalid payload', 400 except stripe.error.SignatureVerificationError: # Invalid signature return 'Invalid signature', 400 # Handle the event if event['type'] == 'payment_intent.succeeded': payment_intent = event['data']['object'] handle_successful_payment(payment_intent) elif event['type'] == 'payment_intent.payment_failed': payment_intent = event['data']['object'] handle_failed_payment(payment_intent) elif event['type'] == 'customer.subscription.deleted': subscription = event['data']['object'] handle_subscription_canceled(subscription) return 'Success', 200 def handle_successful_payment(payment_intent): """Process successful payment.""" customer_id = payment_intent.get('customer') amount = payment_intent['amount'] metadata = payment_intent.get('metadata', {}) # Update your database # Send confirmation email # Fulfill order print(f"Payment succeeded: {payment_intent['id']}") def handle_failed_payment(payment_intent): """Handle failed payment.""" error = payment_intent.get('last_payment_error', {}) print(f"Payment failed: {error.get('message')}") # Notify customer # Update order status def handle_subscription_canceled(subscription): """Handle subscription cancellation.""" customer_id = subscription['customer'] # Update user access # Send cancellation email print(f"Subscription canceled: {subscription['id']}") ``` ### Webhook Best Practices ```python import hashlib import hmac def verify_webhook_signature(payload, signature, secret): """Manually verify webhook signature.""" expected_sig = hmac.new( secret.encode('utf-8'), payload, hashlib.sha256 ).hexdigest() return hmac.compare_digest(signature, expected_sig) def handle_webhook_idempotently(event_id, handler): """Ensure webhook is processed exactly once.""" # Check if event already processed if is_event_processed(event_id): return # Process event try: handler() mark_event_processed(event_id) except Exception as e: log_error(e) # Stripe will retry failed webhooks raise ``` ## Customer Management ```python def create_customer(email, name, payment_method_id=None): """Create a Stripe customer.""" customer = stripe.Customer.create( email=email, name=name, payment_method=payment_method_id, invoice_settings={ 'default_payment_method': payment_method_id } if payment_method_id else None, metadata={ 'user_id': '12345' } ) return customer def attach_payment_method(customer_id, payment_method_id): """Attach a payment method to a customer.""" stripe.PaymentMethod.attach( payment_method_id, customer=customer_id ) # Set as default stripe.Customer.modify( customer_id, invoice_settings={ 'default_payment_method': payment_method_id } ) def list_customer_payment_methods(customer_id): """List all payment methods for a customer.""" payment_methods = stripe.PaymentMethod.list( customer=customer_id, type='card' ) return payment_methods.data ``` ## Refund Handling ```python def create_refund(payment_intent_id, amount=None, reason=None): """Create a refund.""" refund_params = { 'payment_intent': payment_intent_id } if amount: refund_params['amount'] = amount # Partial refund if reason: refund_params['reason'] = reason # 'duplicate', 'fraudulent', 'requested_by_customer' refund = stripe.Refund.create(**refund_params) return refund def handle_dispute(charge_id, evidence): """Update dispute with evidence.""" stripe.Dispute.modify( charge_id, evidence={ 'customer_name': evidence.get('customer_name'), 'customer_email_address': evidence.get('customer_email'), 'shipping_documentation': evidence.get('shipping_proof'), 'customer_communication': evidence.get('communication'), } ) ``` ## Testing ```python # Use test mode keys stripe.api_key = "sk_test_..." # Test card numbers TEST_CARDS = { 'success': '4242424242424242', 'declined': '4000000000000002', '3d_secure': '4000002500003155', 'insufficient_funds': '4000000000009995' } def test_payment_flow(): """Test complete payment flow.""" # Create test customer customer = stripe.Customer.create( email="test@example.com" ) # Create payment intent intent = stripe.PaymentIntent.create( amount=1000, automatic_payment_methods={ 'enabled': True }, currency='usd', customer=customer.id ) # Confirm with test card confirmed = stripe.PaymentIntent.confirm( intent.id, payment_method='pm_card_visa' # Test payment method ) assert confirmed.status == 'succeeded' ``` ## Resources - **references/checkout-flows.md**: Detailed checkout implementation - **references/webhook-handling.md**: Webhook security and processing - **references/subscription-management.md**: Subscription lifecycle - **references/customer-management.md**: Customer and payment method handling - **references/invoice-generation.md**: Invoicing and billing - **assets/stripe-client.py**: Production-ready Stripe client wrapper - **assets/webhook-handler.py**: Complete webhook processor - **assets/checkout-config.json**: Checkout configuration templates ## Best Practices 1. **Always Use Webhooks**: Don't rely solely on client-side confirmation 2. **Idempotency**: Handle webhook events idempotently 3. **Error Handling**: Gracefully handle all Stripe errors 4. **Test Mode**: Thoroughly test with test keys before production 5. **Metadata**: Use metadata to link Stripe objects to your database 6. **Monitoring**: Track payment success rates and errors 7. **PCI Compliance**: Never handle raw card data on your server 8. **SCA Ready**: Implement 3D Secure for European payments ## Common Pitfalls - **Not Verifying Webhooks**: Always verify webhook signatures - **Missing Webhook Events**: Handle all relevant webhook events - **Hardcoded Amounts**: Use cents/smallest currency unit - **No Retry Logic**: Implement retries for API calls - **Ignoring Test Mode**: Test all edge cases with test cards
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

startup-financial-modeling

This skill should be used when the user asks to "create financial

business
⭐1
# Startup Financial Modeling Build comprehensive 3-5 year financial models with revenue projections, cost structures, cash flow analysis, and scenario planning for early-stage startups. ## Overview Financial modeling provides the quantitative foundation for startup strategy, fundraising, and operational planning. Create realistic projections using cohort-based revenue modeling, detailed cost structures, and scenario analysis to support decision-making and investor presentations. ## Core Components ### Revenue Model **Cohort-Based Projections:** Build revenue from customer acquisition and retention by cohort. **Formula:** ``` MRR = Σ (Cohort Size × Retention Rate × ARPU) ARR = MRR × 12 ``` **Key Inputs:** - Monthly new customer acquisitions - Customer retention rates by month - Average revenue per user (ARPU) - Pricing and packaging assumptions - Expansion revenue (upsells, cross-sells) ### Cost Structure **Operating Expenses Categories:** 1. **Cost of Goods Sold (COGS)** - Hosting and infrastructure - Payment processing fees - Customer support (variable portion) - Third-party services per customer 2. **Sales & Marketing (S&M)** - Customer acquisition cost (CAC) - Marketing programs and advertising - Sales team compensation - Marketing tools and software 3. **Research & Development (R&D)** - Engineering team compensation - Product management - Design and UX - Development tools and infrastructure 4. **General & Administrative (G&A)** - Executive team - Finance, legal, HR - Office and facilities - Insurance and compliance ### Cash Flow Analysis **Components:** - Beginning cash balance - Cash inflows (revenue, fundraising) - Cash outflows (operating expenses, CapEx) - Ending cash balance - Monthly burn rate - Runway (months of cash remaining) **Formula:** ``` Runway = Current Cash Balance / Monthly Burn Rate Monthly Burn = Monthly Revenue - Monthly Expenses ``` ### Headcount Planning **Role-Based Hiring Plan:** Track headcount by department and role. **Key Metrics:** - Fully-loaded cost per employee - Revenue per employee - Headcount by department (% of total) **Typical Ratios (Early-Stage SaaS):** - Engineering: 40-50% - Sales & Marketing: 25-35% - G&A: 10-15% - Customer Success: 5-10% ## Financial Model Structure ### Three-Scenario Framework **Conservative Scenario (P10):** - Slower customer acquisition - Lower pricing or conversion - Higher churn rates - Extended sales cycles - Used for cash management **Base Scenario (P50):** - Most likely outcomes - Realistic assumptions - Primary planning scenario - Used for board reporting **Optimistic Scenario (P90):** - Faster growth - Better unit economics - Lower churn - Used for upside planning ### Time Horizon **Detailed Projections: 3 Years** - Monthly detail for Year 1 - Monthly detail for Year 2 - Quarterly detail for Year 3 **High-Level Projections: Years 4-5** - Annual projections - Key metrics only - Support long-term planning ## Step-by-Step Process ### Step 1: Define Business Model Clarify revenue model and pricing. **SaaS Model:** - Subscription pricing tiers - Annual vs. monthly contracts - Free trial or freemium approach - Expansion revenue strategy **Marketplace Model:** - GMV projections - Take rate (% of transactions) - Buyer and seller economics - Transaction frequency **Transactional Model:** - Transaction volume - Revenue per transaction - Frequency and seasonality ### Step 2: Build Revenue Projections Use cohort-based methodology for accuracy. **Monthly Customer Acquisition:** Define new customers acquired each month. **Retention Curve:** Model customer retention over time. **Typical SaaS Retention:** - Month 1: 100% - Month 3: 90% - Month 6: 85% - Month 12: 75% - Month 24: 70% **Revenue Calculation:** For each cohort, calculate retained customers × ARPU for each month. ### Step 3: Model Cost Structure Break down costs by category and behavior. **Fixed vs. Variable:** - Fixed: Salaries, software, rent - Variable: Hosting, payment processing, support **Scaling Assumptions:** - COGS as % of revenue - S&M as % of revenue (CAC payback) - R&D growth rate - G&A as % of total expenses ### Step 4: Create Hiring Plan Model headcount growth by role and department. **Inputs:** - Starting headcount - Hiring velocity by role - Fully-loaded compensation by role - Benefits and taxes (typically 1.3-1.4x salary) **Example:** ``` Engineer: $150K salary × 1.35 = $202K fully-loaded Sales Rep: $100K OTE × 1.30 = $130K fully-loaded ``` ### Step 5: Project Cash Flow Calculate monthly cash position and runway. **Monthly Cash Flow:** ``` Beginning Cash + Revenue Collected (consider payment terms) - Operating Expenses Paid - CapEx = Ending Cash ``` **Runway Calculation:** ``` If Ending Cash < 0: Funding Need = Negative Cash Balance Runway = 0 Else: Runway = Ending Cash / Average Monthly Burn ``` ### Step 6: Calculate Key Metrics Track metrics that matter for stage. **Revenue Metrics:** - MRR / ARR - Growth rate (MoM, YoY) - Revenue by segment or cohort **Unit Economics:** - CAC (Customer Acquisition Cost) - LTV (Lifetime Value) - CAC Payback Period - LTV / CAC Ratio **Efficiency Metrics:** - Burn multiple (Net Burn / Net New ARR) - Magic number (Net New ARR / S&M Spend) - Rule of 40 (Growth % + Profit Margin %) **Cash Metrics:** - Monthly burn rate - Runway (months) - Cash efficiency ### Step 7: Scenario Analysis Create three scenarios with different assumptions. **Variable Assumptions:** - Customer acquisition rate (±30%) - Churn rate (±20%) - Average contract value (±15%) - CAC (±25%) **Fixed Assumptions:** - Pricing structure - Core operating expenses - Hiring plan (adjust timing, not roles) ## Business Model Templates ### SaaS Financial Model **Revenue Drivers:** - New MRR (customers × ARPU) - Expansion MRR (upsells) - Contraction MRR (downgrades) - Churned MRR (lost customers) **Key Ratios:** - Gross margin: 75-85% - S&M as % revenue: 40-60% (early stage) - CAC payback: < 12 months - Net retention: 100-120% **Example Projection:** ``` Year 1: $500K ARR, 50 customers, $100K MRR by Dec Year 2: $2.5M ARR, 200 customers, $208K MRR by Dec Year 3: $8M ARR, 600 customers, $667K MRR by Dec ``` ### Marketplace Financial Model **Revenue Drivers:** - GMV (Gross Merchandise Value) - Take rate (% of GMV) - Net revenue = GMV × Take rate **Key Ratios:** - Take rate: 10-30% depending on category - CAC for buyers vs. sellers - Contribution margin: 60-70% **Example Projection:** ``` Year 1: $5M GMV, 15% take rate = $750K revenue Year 2: $20M GMV, 15% take rate = $3M revenue Year 3: $60M GMV, 15% take rate = $9M revenue ``` ### E-Commerce Financial Model **Revenue Drivers:** - Traffic (visitors) - Conversion rate - Average order value (AOV) - Purchase frequency **Key Ratios:** - Gross margin: 40-60% - Contribution margin: 20-35% - CAC payback: 3-6 months ### Services / Agency Financial Model **Revenue Drivers:** - Billable hours or projects - Hourly rate or project fee - Utilization rate - Team capacity **Key Ratios:** - Gross margin: 50-70% - Utilization: 70-85% - Revenue per employee ## Fundraising Integration ### Funding Scenario Modeling **Pre-Money Valuation:** Based on metrics and comparables. **Dilution:** ``` Post-Money = Pre-Money + Investment Dilution % = Investment / Post-Money ``` **Use of Funds:** Allocate funding to extend runway and achieve milestones. **Example:** ``` Raise: $5M at $20M pre-money Post-Money: $25M Dilution: 20% Use of Funds: - Product Development: $2M (40%) - Sales & Marketing: $2M (40%) - G&A and Operations: $0.5M (10%) - Working Capital: $0.5M (10%) ``` ### Milestone-Based Planning **Identify Key Milestones:** - Product launch - First $1M ARR - Break-even on CAC - Series A fundraise **Funding Amount:** Ensure runway to achieve next milestone + 6 months buffer. ## Common Pitfalls **Pitfall 1: Overly Optimistic Revenue** - New startups rarely hit aggressive projections - Use conservative customer acquisition assumptions - Model realistic churn rates **Pitfall 2: Underestimating Costs** - Add 20% buffer to expense estimates - Include fully-loaded compensation - Account for software and tools **Pitfall 3: Ignoring Cash Flow Timing** - Revenue ≠ cash (payment terms) - Expenses paid before revenue collected - Model cash conversion carefully **Pitfall 4: Static Headcount** - Hiring takes time (3-6 months to fill roles) - Ramp time for productivity (3-6 months) - Account for attrition (10-15% annually) **Pitfall 5: Not Scenario Planning** - Single scenario is never accurate - Always model conservative case - Plan for what you'll do if base case fails ## Model Validation **Sanity Checks:** - [ ] Revenue growth rate is achievable (3x in Year 2, 2x in Year 3) - [ ] Unit economics are realistic (LTV/CAC > 3, payback < 18 months) - [ ] Burn multiple is reasonable (< 2.0 in Year 2-3) - [ ] Headcount scales with revenue (revenue per employee growing) - [ ] Gross margin is appropriate for business model - [ ] S&M spending aligns with CAC and growth targets **Benchmark Against Peers:** Compare key metrics to similar companies at similar stage. **Investor Feedback:** Share model with advisors or investors for feedback on assumptions. ## Additional Resources ### Reference Files For detailed model structures and advanced techniques: - **`references/model-templates.md`** - Complete financial model templates by business model - **`references/unit-economics.md`** - Deep dive on CAC, LTV, payback, and efficiency metrics - **`references/fundraising-scenarios.md`** - Modeling funding rounds and dilution ### Example Files Working financial models with formulas: - **`examples/saas-financial-model.md`** - Complete 3-year SaaS model with cohort analysis - **`examples/marketplace-model.md`** - Marketplace GMV and take rate projections - **`examples/scenario-analysis.md`** - Three-scenario framework with sensitivities ## Quick Start To create a startup financial model: 1. **Define business model** - Revenue drivers and pricing 2. **Project revenue** - Cohort-based with retention 3. **Model costs** - COGS, S&M, R&D, G&A by month 4. **Plan headcount** - Hiring by role and department 5. **Calculate cash flow** - Revenue - expenses = burn/runway 6. **Compute metrics** - CAC, LTV, burn multiple, runway 7. **Create scenarios** - Conservative, base, optimistic 8. **Validate assumptions** - Sanity check and benchmark 9. **Integrate fundraising** - Model funding rounds and milestones For complete templates and formulas, reference the `references/` and `examples/` files.
👍0
👁️0
🤖 Auto-discovered