B2B Revenue Forecasting (b2b_revenue_forecasting)

An open-source Python framework designed mathematically for Enterprise RevOps and Data Strategy teams.

Unlike traditional bottom-up time-series libraries (which are strictly built for B2C retail/inventory forecasting and rely on mathematical averages), this package is explicitly architected to handle the realities of B2B enterprise sales: Hierarchical Quotas, Managerial Cascading, Pipeline Health Analysis, and "Sandbagging" Biases.

🚀 Features

What's New in v0.6.0 — dirty hierarchies can't crash the cascade (issue #1)

Previously, a row with the same value at two adjacent levels (e.g., team T1 AND rep T1) silently built a self-loop, and cascade_quota crashed with a cryptic RecursionError deep inside networkx. v0.6.0 makes malformed hierarchies either self-heal or fail loudly with an actionable message.

• on_collision parameter on from_dataframe — "suffix" (default, renames the deeper duplicate to <value>__<level_column> and warns), "skip" (drops the duplicate level, jagged-style), or "error" (raise naming the row).
• Blank-string hygiene — empty cells and literal "nan"/"none"/"null" strings (a keep_default_na=False hazard) are treated as missing levels instead of becoming a shared "nan" node. 'NA' the region is still data.
• hierarchy.validate() + automatic DAG validation at the end of from_dataframe — cross-row cycles raise HierarchyValidationError naming the cycle path.
• Fail-fast cascades — cascade_quota checks the graph up front, and the recursive aggregators carry recursion-stack guards, so a cyclic graph can never RecursionError again. Diamond-shaped DAGs remain supported.

h = SalesHierarchy()
h.from_dataframe(df, path_cols=taxonomy, metrics_cols=cols,
                 on_collision='suffix')   # default — shown for clarity
# -> UserWarning: 1 duplicate-level value(s) detected and renamed ...
h.validate()                              # explicit re-check, chainable

What's New in v0.5.0 — no more stranded targets (issue #12)

Previously, when a gate zeroed an entire subtree (e.g., a Migration cascade where no rep in the whole slice had DC entitlement), the target for that slice was silently dropped — depth-0 held the target while depth 1+ summed short. v0.5.0 guarantees the base (un-hedged) quota sums to the macro target at every depth.

• gate_fallback parameter on cascade_quota controls what happens when every child of a funded node is gated:
  • "redistribute" (default) — a fully-gated subtree's share flows to its nearest non-gated siblings (gates still roll up as before); if the entire level — even the whole tree — is gated, the gate is relaxed at that level as a last resort so the target still reaches ICs. No silent target loss, ever.
  • "strand_at_root" — children stay $0; the undistributable amount stays on the deepest non-gated ancestor and is reported via cascader.unallocated / cascader.unallocated_nodes plus an is_unallocated column in quotas_to_dataframe.
  • "error" — raises GateAllocationError so the caller decides.
• The root is never gated to $0. It always carries the macro target in every mode.
• cascader.base_quotas — every cascade_quota call now also computes the un-hedged cascade in the same pass, so hedged_quota = base_quota × hedge^depth decomposes without a second run. Pass unhedged_quotas="auto" to quotas_to_dataframe to get the audit columns for free.
• cascader.reconciliation_report(quotas, target=..., strict=True) — per-depth reconciliation DataFrame (depth, n_nodes, total_quota, target, delta, reconciles); strict=True raises listing every non-reconciling depth. Run it on cascader.base_quotas (hedged quotas legitimately grow with depth).
• gate_relaxed column in quotas_to_dataframe flags nodes that received quota despite being gated because every sibling was also gated — so the last-resort fallback is always visible in the CSV.

quotas = cascader.cascade_quota(
    'Enterprise_AMER', 1_000_000.0,
    hedge_multiplier=1.05,
    metrics=forward_metrics,
    gate_metrics=[MetricSpec('DC_Seats', columns=['DC_Seats'])],
    gate_fallback='redistribute',   # default — shown for clarity
)
# Base layer reconciles at EVERY depth, even with fully-gated teams:
cascader.reconciliation_report(cascader.base_quotas,
                               target=1_000_000.0, strict=True)
df = cascader.quotas_to_dataframe(quotas, unhedged_quotas='auto')

What's New in v0.4.0

• Gate metrics — hard kill-switches. cascade_quota(..., gate_metrics=[...]) excludes any node whose rolled-up gate value is at or below a threshold from the cascade entirely (quota = 0), redistributing its share among non-gated siblings. Designed for white-space planning: e.g., gating "migration NetNewACV" on Unmigrated_Seats zeros out territories with nothing left to migrate. Gates propagate upward naturally — a manager whose whole team fails the gate gets $0 too. Composes with AND across multiple gates. CRO overrides win over gates.
• Two planning philosophies, both supported. See the section below.
• is_gated column in quotas_to_dataframe when gates were used, so analysts can distinguish "$0 because gated" from "$0 because no signal."
• cascader.gated_nodes — the set of gated nodes from the most recent cascade, stored for inspection.

Two Planning Philosophies

The package supports two philosophically distinct ways of building a quota plan. Both use the same primitives — pick the one that matches how your org thinks about fairness.

Earned planning — "who has proven they can sell this?"

Cascade on historical signals (past NetNewACV attainment, past cloud-seat adds, LTM expansion). Reconcile against forward pipeline (open opps + late-stage commit + best-case). Best when historical attainment is a clean signal of forward capacity (mature business, low churn in territories, stable rep tenure).

historical_metrics = [
    MetricSpec('NetNewACV',  direction='proportional', weight=1.0, lookback=4),
    MetricSpec('CloudSeats', direction='proportional', weight=0.6, lookback=4),
    MetricSpec('DCSeats',    direction='inverse',      weight=0.4, lookback=4),
]
quotas = cascader.cascade_quota('Global_Corp', macro_target, metrics=historical_metrics)

# Reconcile against forward pipeline
adjuster = PipelineAdjuster(hierarchy, quotas,
                            pipeline_attr=['Open_Pipeline', 'Late_Stage_Commit'])

White-space planning — "what can be achieved if we look at the opportunity in front of us?"

Cascade on forward-looking signals (current installed seats, knowledge-worker counts, white-space indicators), with dampeners (LTM spend) and hard gates (unmigrated seats). Reconcile against historical attainment to flag where the plan asks for a step-up. Best when past performance is noisy (rapid growth, territory shuffles, recent re-orgs) and the org wants every rep to be measured against the opportunity in front of them.

forward_metrics = [
    MetricSpec('Current_Seats_ProductX',  direction='proportional', weight=1.0,
               columns=['Current_Seats_ProductX']),
    MetricSpec('Knowledge_Workers_Count', direction='proportional', weight=0.7,
               columns=['Knowledge_Workers_Count']),
    MetricSpec('LTM_ExpansionSpent',      direction='inverse',      weight=0.5,
               columns=['LTM_ExpansionSpent']),
]
gate_metrics = [
    MetricSpec('Unmigrated_Seats', columns=['Unmigrated_Seats']),  # threshold defaults to 0
]
quotas = cascader.cascade_quota(
    'Global_Corp', macro_target,
    metrics=forward_metrics, gate_metrics=gate_metrics,
)

# Reconcile against historical attainment
adjuster = PipelineAdjuster(hierarchy, quotas, pipeline_attr=[
    'Q1_NetNewACV', 'Q2_NetNewACV', 'Q3_NetNewACV', 'Q4_NetNewACV',
])
diagnosis = adjuster.diagnose(coverage_thresholds={
    '_default': {'healthy': 1.0, 'at_risk': 0.75},   # ratios near 1.0, not 1.5–3x
})

Neither philosophy is "correct" — they answer different questions. The package supports either as a first-class flow, and you can blend them (some metrics historical, some forward) by mixing them in a single metrics= list.

What's New in v0.3.x

• Multi-metric cascading via the new MetricSpec API — blend historical NetNewACV with any number of secondary signals (cloud seats, on-prem seats, LTM expansion spend, customer-sat scores, certification flags, anything else the analyst tracks), each marked as proportional or inverse, with per-metric weights and lookbacks
• Direction is always a user input. Domain knowledge ("more cloud seats means more ACV") trumps statistical sign. The package surfaces correlations and warns on mismatch but never overrides the analyst's call
• MetricSpec.suggest_weights(...) suggests weights (magnitude of correlation) for user-declared directions. For exploratory use, MetricSpec.suggest_directions_and_weights(...) infers both
• Normalized-weights view — MetricSpec.normalized_weights(specs) shows the post-normalization share each metric actually contributes; auto-printed before every multi-metric cascade and accessible via cascader.weights_report
• Brand-new IC handling — either-or: flag brand-new ICs in the same CSV the analyst already uploads (brand_new_col='Is_Brand_New' on SalesHierarchy.from_dataframe, then new_ic_attr='_is_brand_new' on cascade_quota), OR pick a rule (new_ic_rule='all_metrics_zero' / 'primary_metric_zero'). Passing both raises ValueError
• Any metric name, any numeric type — including booleans (Has_Active_Cert: True/False). Boolean / 0-1 sparse metrics are auto-detected and excluded from zero-imputation so False isn't mistaken for missing data
• PipelineAdjuster accepts multiple pipeline columns — pipeline_attr=['Open_Pipeline', 'Late_Stage_Commit', 'Best_Case_Adds'] sums them per IC into a combined dollar amount for the coverage ratio
• CSV / SQL / dashboard exports — every output converts to a DataFrame via cascader.quotas_to_dataframe(...), cascader.quotas_diff_to_dataframe(...), or reconciler.reconcile_all(...). From there .to_csv(), .to_sql(), or cascader.to_html_dashboard(...) writes wherever you need
• Hedge audit columns — pass unhedged_quotas= to quotas_to_dataframe for unhedged_quota, hedge_buffer, and overassignment_pct columns showing exactly how much of each quota is hedge buffer
• Fully backward compatible — cascade_quota(...) without metrics= behaves exactly as in v0.2.x

What's New in v0.2.0

• PipelineAdjuster: Post-cascade pipeline health analyzer with diagnose() and adjust() modes
• Flexible quarter support: QuotaCascader now auto-discovers any number of _Attainment columns (4, 8, 12 quarters)
• New IC handling: Partial-history imputation and equal-share allocation for brand-new hires
• CRO overrides: Lock specific IC quotas via new_ic_overrides to bypass the algorithm
• Per-node hedging: Apply different hedge multipliers to different regions/managers
• GitHub Actions CI/CD: Automated testing on Python 3.9–3.12

📦 Installation

pip install b2b-revenue-forecasting

💻 Quickstart

1. Build the Org Hierarchy

import pandas as pd
from b2b_revenue_forecasting.hierarchy import SalesHierarchy

# ⚠️ Use keep_default_na=False if your data has 'NA' as a region name
df = pd.read_csv('your_crm_data.csv', keep_default_na=False)

# Works with any depth: 3 levels or 10 levels
hierarchy = SalesHierarchy()
hierarchy.from_dataframe(
    df, 
    path_cols=['Global', 'Region', 'RVP', 'Director', 'Manager', 'IC'], 
    metrics_cols=['Q1_Attainment', 'Q2_Attainment', 'Q3_Attainment', 'Q4_Attainment',
                  'Current_Pipeline']
)

print(f"Nodes: {len(hierarchy.graph.nodes)}")
print(f"ICs:   {len(hierarchy.get_leaves('Global_Corp'))}")

2. Cascade Quotas Top-Down

from b2b_revenue_forecasting.quota_cascader import QuotaCascader

cascader = QuotaCascader(hierarchy)

# Basic: distribute $100M evenly by historical capacity
quotas = cascader.cascade_quota('Global_Corp', 100_000_000.0)

# With 5% hedge at every management level (compounds: 1.05^5 ≈ 27.6% overassignment)
quotas = cascader.cascade_quota('Global_Corp', 100_000_000.0, hedge_multiplier=1.05)

# Per-node hedge: NA gets aggressive 10%, others standard 5%
quotas = cascader.cascade_quota('Global_Corp', 100_000_000.0, hedge_multiplier={
    'Global_Corp': 1.05, 'NA': 1.10, 'EMEA': 1.05, 'APAC': 1.05
})

# CRO override: strategic hire gets exactly $500K regardless of history
quotas = cascader.cascade_quota('Global_Corp', 100_000_000.0,
    hedge_multiplier=1.05,
    new_ic_overrides={'IC_Strategic_Hire': 500_000.0}
)

3. Multi-Metric Cascading (v0.3+)

For real B2B planning, the metric you're cascading (e.g., NetNewACV) is rarely the only signal that should drive its allocation. Cloud-seat counts predict more new ACV; on-prem (DC) seat counts predict less; high LTM expansion spend means the account is already saturated. The MetricSpec API lets you mix any number of these into a single cascade.

Direction is always your call. You declare whether each metric is proportional (more → more quota) or inverse (more → less quota) up front. The package surfaces correlations and warns when the data sign disagrees, but never overrides your domain knowledge.

from b2b_revenue_forecasting import MetricSpec

# Declare each metric's role — direction is required, weight is your knob
metrics = [
    MetricSpec('NetNewACV',     direction='proportional', weight=1.0, lookback=4),
    MetricSpec('CloudSeats',    direction='proportional', weight=0.5, lookback=4),
    MetricSpec('DCSeats',       direction='inverse',      weight=0.4, lookback=4),
    MetricSpec('ExpansionSpent',direction='inverse',      weight=0.7,
               columns=['LTM_ExpansionSpent']),  # single LTM column
]

quotas = cascader.cascade_quota(
    'Global_Corp', 100_000_000.0,
    hedge_multiplier=1.05,
    metrics=metrics,
)

Any metric name, any data type works. Customer_Sat_Score, MQLs_Sourced_via_Outbound, Has_Active_Cert (boolean), Renewals_Caught_Up (0/1 counter) — anything numeric, with any column name. Boolean and 0/1 sparse metrics are auto-detected and excluded from zero-imputation so False isn't treated as a missing value.

How the blend works. At every level, each child gets a share of the parent's quota equal to a weighted sum of its per-metric shares-of-siblings. Proportional metrics use raw shares; inverse metrics flip via reciprocal-then-normalize. The final per-child share is Σ_m (weight_m × share_m(child)), which sums to 1 across siblings.

Don't know the weights? Pass direction= on each candidate, let suggest_weights() propose magnitudes via Pearson correlation:

suggestions, report = MetricSpec.suggest_weights(
    df,
    target_column='NetNewACV_4Q_sum',
    candidate_metrics=[
        {'name': 'CloudSeats',     'column': 'CloudSeats_4Q_sum',
         'direction': 'proportional', 'lookback': 4},
        {'name': 'DCSeats',        'column': 'DCSeats_4Q_sum',
         'direction': 'inverse',      'lookback': 4},
        {'name': 'ExpansionSpent', 'column': 'LTM_ExpansionSpent',
         'columns': ['LTM_ExpansionSpent'],
         'direction': 'inverse',      'lookback': 1},
    ],
)
# report['CloudSeats']['weight'] == 0.62, ['rationale'] explains why,
# ['direction_matches_data'] tells you if your call agrees with the sign

quotas = cascader.cascade_quota('Global_Corp', 100_000_000.0, metrics=suggestions)

For pure exploration (you don't yet have a domain opinion), use MetricSpec.suggest_directions_and_weights(...) — it infers both from data. This is a sanity-check helper, not a production-planning API.

Brand-new ICs — either-or, your choice of where they're listed. The cleanest option keeps everything in the same CSV the analyst already uploads:

# CSV has a column Is_Brand_New with True / 1 / "yes" for each new hire
hierarchy = SalesHierarchy()
hierarchy.from_dataframe(
    df, path_cols=[...], metrics_cols=[...],
    brand_new_col='Is_Brand_New',     # ingested as node attribute _is_brand_new
)

quotas = cascader.cascade_quota(
    'Global_Corp', 100_000_000.0,
    metrics=metrics,
    new_ic_attr='_is_brand_new',       # read the flag from the CSV
)

Or, if you don't want a separate column, pick an auto-detection rule:

quotas = cascader.cascade_quota(
    'Global_Corp', 100_000_000.0,
    metrics=metrics,
    new_ic_rule='all_metrics_zero',    # or 'primary_metric_zero'
)

You pick one or the other — passing both an explicit identifier (new_ic_attr or new_ic_ids) AND new_ic_rule in the same call raises ValueError, because the two would silently disagree.

Brand-new ICs get an equal-share carve-out of the team target before the remainder is split proportionally — just like the single-metric path.

4. Detect & Fix Forecasting Bias

from b2b_revenue_forecasting.commit_reconciler import CommitReconciler

historical = pd.DataFrame({
    'Manager_ID':              ['Mgr_A', 'Mgr_A', 'Mgr_B', 'Mgr_B'],
    'Historical_Commit':       [200_000,  250_000, 300_000,  350_000],
    'Historical_Actual_Closed': [300_000,  375_000, 270_000,  280_000],
})

reconciler = CommitReconciler(historical)

# Mgr_A is a sandbagger (bias = 1.5x) — commit inflated automatically
adjusted = reconciler.reconcile_forecast('Mgr_A', current_commit=100_000)
# → $150,000

# Blend with ML baseline (50/50 average)
blended = reconciler.reconcile_forecast('Mgr_A', 100_000, machine_forecast=120_000)
# → $135,000

5. Export to CSV, SQL, or an Interactive Dashboard

Every output is a pandas DataFrame, so the same code writes anywhere:

# CSV — analyst-ready, one row per node at every level
cascaded_df = cascader.quotas_to_dataframe(quotas, level_names=taxonomy)
cascaded_df.to_csv('cascaded_quotas.csv', index=False)

# CSV with hedge audit — also include the unhedged baseline
quotas_unhedged = cascader.cascade_quota(
    'Global_Corp', 100_000_000.0, hedge_multiplier=1.0,
    metrics=cascade_metrics, verbose=False,
)
cascader.quotas_to_dataframe(
    quotas, level_names=taxonomy, unhedged_quotas=quotas_unhedged,
).to_csv('cascaded_quotas_with_audit.csv', index=False)
# → adds unhedged_quota, hedge_buffer, overassignment_pct columns

# SQL — same DataFrames, any SQLAlchemy-compatible database
import sqlite3
with sqlite3.connect('cascade.db') as conn:
    cascaded_df.to_sql('cascaded_quotas', conn, if_exists='replace', index=False)
    cascader.weights_report.to_sql('normalized_weights', conn,
                                    if_exists='replace', index=False)
# Postgres / Snowflake / BigQuery: swap conn for a SQLAlchemy engine

# Interactive HTML dashboard — Chart.js, self-contained, shareable
cascader.to_html_dashboard(
    quotas, output_path='cascade_dashboard.html',
    title='Q1 Cascade — $100M Plan',
    unhedged_quotas=quotas_unhedged,
    adjusted_quotas=adjusted, diagnosis=diagnosis,
)

6. Pipeline Health Diagnosis & Redistribution

from b2b_revenue_forecasting.pipeline_adjuster import PipelineAdjuster

# Single pipeline column (backward compat)
adjuster = PipelineAdjuster(hierarchy, quotas, pipeline_attr='Current_Pipeline')

# Or sum multiple dollar-denominated pipeline columns from the same CSV
adjuster = PipelineAdjuster(hierarchy, quotas, pipeline_attr=[
    'Open_Pipeline', 'Late_Stage_Commit', 'Best_Case_Adds',
])

# Configure per-region coverage thresholds (ICs inherit from ancestors)
thresholds = {
    'NA':       {'healthy': 1.5, 'at_risk': 0.8},
    'EMEA':     {'healthy': 2.5, 'at_risk': 1.2},
    'APAC':     {'healthy': 3.0, 'at_risk': 1.5},
    '_default': {'healthy': 2.0, 'at_risk': 1.0}
}

# Diagnose — returns a DataFrame with risk status for every node
diagnosis = adjuster.diagnose(thresholds)
print(diagnosis.groupby('Risk_Status')['Node'].count())

# Flag-only mode — returns original quotas unchanged (for pre-approval review)
flagged = adjuster.adjust(mode='flag_only', coverage_thresholds=thresholds)

# Redistribute mode — zero-sum IC adjustment within each manager's team
adjusted = adjuster.adjust(
    mode='redistribute',
    coverage_thresholds=thresholds,
    max_adjustment_pct=0.20,                          # ±20% cap per IC
    locked_nodes={'IC_Protected': 500_000.0}           # CRO-locked ICs excluded
)
# ✅ Manager totals preserved | ✅ Donors give, receivers get | ✅ 20% cap enforced

🧠 Key Concepts

Managerial Hedge (Overassignment Buffer)

A multiplier applied at each management level to create mathematical safety. A 5% hedge across 5 layers compounds to ~27.6% total overassignment (1.05⁵), ensuring the enterprise hits its number even if some ICs miss.

Bias Quotient

Bias Quotient = Σ(Actual Closed) / Σ(Committed)

• > 1.0 = Sandbagger (closes more than committed → inflate their forecast)
• = 1.0 = Neutral
• < 1.0 = Happy Ears (over-promises → deflate their forecast)

Pipeline Coverage Ratio

Coverage = Current Pipeline / Cascaded Quota

New IC Handling

🧪 Testing

# Run all tests
cd hierarchical_sales_forecasting
pip install -e .
python -m pytest tests/ -v

# Run the full demo
python demo_full_pipeline.py

📄 Publications

This framework is the subject of peer-reviewed research and technical publications:

If you use this package in your research, please cite:

Karwa, S. (2026). Graph-Theoretic Approaches to Hierarchical Revenue Target Allocation
in B2B Enterprises: A Methodological Framework. SSRN Working Paper. https://papers.ssrn.com/sol3/papers.cfm?abstract_id=6456999

📋 Requirements

• Python ≥ 3.8
• pandas ≥ 1.0.0
• networkx ≥ 2.5
• numpy ≥ 1.19.0

🤝 Contributing

Built explicitly for RevOps analysts, Data Scientists, and VP Revenue Operations executing scaling go-to-market strategies. Contributions, issues, and pull requests are warmly welcomed!

• Report bugs: GitHub Issues
• Source code: GitHub

📄 License

MIT License — see LICENSE for details.