Skip to content
Field Notes

RevOps

CodeGen for RevOps: Generating Production Python from Natural Language

How Bruce translates natural language queries into executable Python code for forecasting, pipeline analysis, and deal velocity—with production-grade accuracy and safety.

By Chien Nguyen, David Shugert, and Shikhar Mishra

RevOps teams spend days building SQL queries, Excel models, and Python scripts to answer forecasting and pipeline questions. By the time analysis is ready, the data is stale.

Bruce's RevOps Analysis solution changes this. Instead of writing code, RevOps teams ask questions in natural language. Bruce generates production Python code, executes it against live CRM data, and returns accurate results in seconds.

This article explains how code generation works in Bruce, from query parsing to execution and validation.

This article covers:

  • Why traditional RevOps analysis breaks
  • How Bruce parses natural language queries
  • Schema-aware code generation
  • Execution safety and sandboxing
  • Error handling and validation
  • Real-world examples from production

The Problem with Traditional RevOps Analysis

RevOps questions are urgent and data-dependent:

  • "What's our pipeline coverage for enterprise accounts this quarter?"
  • "Show me win rates by stage over the last 6 months"
  • "Which deals are at risk based on stage duration?"
  • "Calculate forecast accuracy by rep"

Traditional approaches require:

  • •SQL expertise: Writing complex queries against CRM schemas
  • •Python scripting: Data manipulation, aggregation, and visualization
  • •Time investment: Hours or days per analysis
  • •Stale data: By the time analysis is ready, the question may be irrelevant

Bruce eliminates this friction by generating production-ready Python code from natural language queries.

How CodeGen Works: From Query to Execution

Bruce's CodeGen pipeline has four stages:

1. Query Parsing and Intent Understanding

Bruce parses the natural language query to extract:

  • What data to retrieve (opportunities, accounts, deals)
  • What filters to apply (stage, segment, date range)
  • What calculations to perform (sum, average, count, ratios)
  • What format to return (table, chart, summary)

Example Query

"What's our pipeline coverage for enterprise accounts this quarter?"

Parsed Intent:

python
{
  "data_source": "opportunities",
  "filters": {
    "account_tier": "enterprise",
    "stage": ["qualified", "discovery", "proposal", "negotiation"],
    "quarter": "current"
  },
  "calculation": "sum(value)",
  "comparison": "target",
  "output": "coverage_ratio"
}

2. Schema-Aware Code Generation

Bruce generates Python code that respects the CRM schema and available functions. It uses:

  • Type-safe data access functions
  • Pre-validated filter parameters
  • Built-in aggregation functions
  • Error handling and edge cases

Generated Code

python
from revops import (
    fetch_opportunities,
    get_quarter_target,
    calculate_coverage
)

# Fetch opportunities with filters
opps = fetch_opportunities(
    account_tier="enterprise",
    stages=["qualified", "discovery", "proposal", "negotiation"],
    quarter="Q4-2024"
)

# Calculate pipeline value
pipeline_value = sum(opp.value for opp in opps)

# Get target for comparison
target = get_quarter_target(
    segment="enterprise",
    quarter="Q4-2024"
)

# Calculate coverage ratio
coverage = calculate_coverage(
    pipeline_value=pipeline_value,
    target=target
)

return {
    "pipeline_value": pipeline_value,
    "target": target,
    "coverage_ratio": coverage,
    "coverage_percentage": f"{coverage * 100:.1f}%"
}

The generated code uses type-safe functions that validate inputs and handle edge cases automatically.

3. Execution Safety and Sandboxing

Generated code runs in a sandboxed environment with:

  • Restricted imports (only approved libraries)
  • Read-only data access (no mutations)
  • Resource limits (CPU, memory, execution time)
  • Network isolation (no external API calls)
python
# Sandbox execution environment
sandbox = CodeSandbox(
    allowed_imports=[
        "revops",
        "pandas",
        "numpy",
        "datetime"
    ],
    read_only=True,
    max_execution_time=30,  # seconds
    max_memory_mb=512
)

# Execute generated code
result = sandbox.execute(generated_code)

# Validate output
if not result.success:
    raise ExecutionError(result.error_message)

This ensures generated code cannot modify data, access unauthorized resources, or cause system instability.

4. Result Validation and Formatting

Before returning results, Bruce validates:

  • Output schema matches expected format
  • Numerical results are within reasonable bounds
  • Data types are correct
  • No sensitive data is exposed
python
def validate_result(result, expected_schema):
    # Check output structure
    assert isinstance(result, dict), "Result must be a dictionary"
    
    # Validate required fields
    for field in expected_schema["required"]:
        assert field in result, f"Missing required field: {field}"
    
    # Validate data types
    if "coverage_ratio" in result:
        assert isinstance(result["coverage_ratio"], (int, float))
        assert 0 <= result["coverage_ratio"] <= 10, "Coverage ratio out of bounds"
    
    # Sanitize sensitive data
    result = sanitize_output(result)
    
    return result

Real-World Examples

Example 1: Win Rate Analysis

Query

"Show me win rates by stage over the last 6 months"

Generated Code

python
from revops import fetch_opportunities, calculate_win_rate
from datetime import datetime, timedelta

# Get date range
end_date = datetime.now()
start_date = end_date - timedelta(days=180)

# Fetch opportunities
opps = fetch_opportunities(
    created_date__gte=start_date,
    created_date__lte=end_date
)

# Group by stage and calculate win rates
stages = ["qualified", "discovery", "proposal", "negotiation", "closed-won", "closed-lost"]
win_rates = {}

for stage in stages:
    stage_opps = [o for o in opps if o.stage == stage]
    if len(stage_opps) > 0:
        won = len([o for o in stage_opps if o.status == "won"])
        win_rates[stage] = {
            "total": len(stage_opps),
            "won": won,
            "win_rate": won / len(stage_opps) if len(stage_opps) > 0 else 0
        }

return {
    "period": f"{start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}",
    "win_rates_by_stage": win_rates
}

Example 2: Deal Velocity Analysis

Query

"Which deals are at risk based on stage duration?"

Generated Code

python
from revops import fetch_opportunities, get_stage_duration_thresholds
from datetime import datetime

# Fetch active opportunities
opps = fetch_opportunities(
    status="open",
    stages=["discovery", "proposal", "negotiation"]
)

# Get stage duration thresholds
thresholds = get_stage_duration_thresholds()

# Calculate days in current stage
now = datetime.now()
at_risk_deals = []

for opp in opps:
    days_in_stage = (now - opp.stage_entered_date).days
    threshold = thresholds.get(opp.stage, 30)  # default 30 days
    
    if days_in_stage > threshold:
        at_risk_deals.append({
            "opportunity_id": opp.id,
            "account_name": opp.account_name,
            "stage": opp.stage,
            "days_in_stage": days_in_stage,
            "threshold": threshold,
            "value": opp.value,
            "owner": opp.owner
        })

# Sort by value descending
at_risk_deals.sort(key=lambda x: x["value"], reverse=True)

return {
    "total_at_risk": len(at_risk_deals),
    "total_value_at_risk": sum(d["value"] for d in at_risk_deals),
    "deals": at_risk_deals[:20]  # Top 20
}

Example 3: Forecast Accuracy

Query

"Calculate forecast accuracy by rep for last quarter"

Generated Code

python
from revops import (
    fetch_forecasts,
    fetch_actuals,
    calculate_accuracy
)
from datetime import datetime

# Get last quarter
now = datetime.now()
quarter = (now.month - 1) // 3
year = now.year
if quarter == 0:
    quarter = 4
    year -= 1

# Fetch forecasts and actuals
forecasts = fetch_forecasts(quarter=quarter, year=year)
actuals = fetch_actuals(quarter=quarter, year=year)

# Group by rep
rep_accuracy = {}
for forecast in forecasts:
    rep_id = forecast.rep_id
    rep_name = forecast.rep_name
    
    # Get actual for this rep
    actual = next(
        (a for a in actuals if a.rep_id == rep_id),
        None
    )
    
    if actual:
        accuracy = calculate_accuracy(
            forecasted=forecast.value,
            actual=actual.value
        )
        
        if rep_id not in rep_accuracy:
            rep_accuracy[rep_id] = {
                "rep_name": rep_name,
                "forecasted": 0,
                "actual": 0,
                "accuracy": 0
            }
        
        rep_accuracy[rep_id]["forecasted"] += forecast.value
        rep_accuracy[rep_id]["actual"] += actual.value
        rep_accuracy[rep_id]["accuracy"] = accuracy

# Calculate final accuracy
for rep_id in rep_accuracy:
    rep_accuracy[rep_id]["accuracy"] = calculate_accuracy(
        forecasted=rep_accuracy[rep_id]["forecasted"],
        actual=rep_accuracy[rep_id]["actual"]
    )

return {
    "quarter": f"Q{quarter} {year}",
    "rep_accuracy": list(rep_accuracy.values())
}

Error Handling and Edge Cases

Bruce's CodeGen handles errors gracefully:

  • •
    Ambiguous queries: Bruce asks clarifying questions instead of guessing
  • •
    Schema mismatches: Code generation fails fast with clear error messages
  • •
    Empty results: Returns structured empty responses, not errors
  • •
    Execution timeouts: Code is optimized and resource-limited
  • •
    Invalid calculations: Division by zero, null handling, and type errors are caught
python
try:
    result = execute_generated_code(query, context)
except AmbiguousQueryError as e:
    return {
        "error": "ambiguous_query",
        "message": e.message,
        "clarifying_questions": e.suggested_questions
    }
except SchemaError as e:
    return {
        "error": "schema_mismatch",
        "message": f"Field '{e.field}' not found. Available fields: {e.available_fields}"
    }
except ExecutionError as e:
    return {
        "error": "execution_failed",
        "message": e.message,
        "suggestion": "Try rephrasing your query or specifying a date range"
    }
except Exception as e:
    # Log unexpected errors
    log_error(e, query, context)
    return {
        "error": "unexpected_error",
        "message": "An unexpected error occurred. Please try again."
    }

Why This Matters

CodeGen for RevOps transforms how revenue teams work:

Speed

Answers in seconds, not days. RevOps teams can iterate on questions and get immediate feedback.

Accuracy

Generated code uses type-safe functions and validated schemas. No more manual calculation errors.

Accessibility

Non-technical team members can ask questions in natural language. No SQL or Python knowledge required.

Fresh Data

Code executes against live CRM data. Results are always current, not stale snapshots.

The Future of RevOps Analysis

CodeGen is not just about generating code. It's about making data analysis as natural as asking a question.

By combining natural language understanding with production-grade code generation, Bruce enables RevOps teams to focus on insights, not implementation.

This is how AI moves from answering questions to executing workflows—with the accuracy and safety required for production revenue operations.

Field Notes

Continue