fix: improve dataset routing and cpg finance

demoprep_app/dataset/generators/template_generator.py

@@ -371,6 +371,40 @@ class TemplateDatasetGenerator(ScenarioDatasetGenerator):
                 rows.update({"INVESTED_CAPITAL_USD": invested, "CURRENT_VALUE_USD": current, "GAIN_LOSS_USD": round(current - invested, 2), "RETURN_PCT": self._pct(current - invested, invested)})
             return rows
 
+        if profile == "cpg_finance":
+            cases = self._whole(rng, 1_500, 180_000, mult)
+            gross_price_per_case = round(rng.uniform(18, 92), 2)
+            gross_sales = round(cases * gross_price_per_case, 2)
+            trade_spend = round(gross_sales * rng.uniform(0.06, 0.24), 2)
+            deductions = round(gross_sales * rng.uniform(0.01, 0.06), 2)
+            net_sales = round(gross_sales - trade_spend - deductions, 2)
+            cogs = round(net_sales * rng.uniform(0.52, 0.70), 2)
+            freight = round(cases * rng.uniform(0.42, 2.85), 2)
+            spoilage = round(net_sales * rng.uniform(0.001, 0.018), 2)
+            margin = round(net_sales - cogs - freight - spoilage, 2)
+            planned_net_sales = round(net_sales * rng.uniform(0.88, 1.14), 2)
+            forecast_net_sales = round(net_sales * rng.uniform(0.91, 1.10), 2)
+            promo_lift = round(rng.uniform(0, 32), 2)
+            return {
+                "CASES_SHIPPED": cases,
+                "GROSS_SALES_USD": gross_sales,
+                "TRADE_SPEND_USD": trade_spend,
+                "TRADE_SPEND_RATE_PCT": self._pct(trade_spend, gross_sales),
+                "DEDUCTIONS_USD": deductions,
+                "NET_SALES_USD": net_sales,
+                "COGS_USD": cogs,
+                "FREIGHT_COST_USD": freight,
+                "SPOILAGE_WRITE_OFF_USD": spoilage,
+                "GROSS_MARGIN_USD": margin,
+                "GROSS_MARGIN_PCT": self._pct(margin, net_sales),
+                "AVG_NET_PRICE_PER_CASE_USD": self._div(net_sales, cases),
+                "PROMOTION_LIFT_PCT": promo_lift,
+                "PLANNED_NET_SALES_USD": planned_net_sales,
+                "FORECAST_NET_SALES_USD": forecast_net_sales,
+                "PLAN_VARIANCE_USD": round(net_sales - planned_net_sales, 2),
+                "FORECAST_VARIANCE_PCT": self._pct(net_sales - forecast_net_sales, forecast_net_sales),
+            }
+
         if profile == "professional_services":
             budget_hours = self._whole(rng, 120, 12_000, mult)
             billable_hours = self._bounded_child(rng, budget_hours, 0.72, 1.12)

demoprep_app/pipeline/dataset_first.py

@@ -23,6 +23,33 @@ class DatasetFirstBuild:
 
 def infer_scenario_type(use_case: str, vertical: str | None = None, function: str | None = None) -> str | None:
     text = f"{vertical or ''} {function or ''} {use_case or ''}".lower()
+    fn = (function or "").strip().lower()
+
+    if fn == "hr":
+        return "workforce_hr"
+    if fn == "legal":
+        return "legal_matter_management"
+    if fn == "marketing":
+        if any(tok in text for tok in ("apparel", "fashion", "garment", "footwear", "seasonal drop", "markdown")):
+            return "apparel_marketing"
+        if any(tok in text for tok in ("ecommerce", "e-commerce", "commerce", "online store")):
+            return "ecommerce_marketing"
+        return "marketing_funnel"
+    if fn == "finance":
+        if any(tok in text for tok in ("air transport", "airline", "airport", "aircraft", "flight", "passenger", "baggage")):
+            return "airline_finance"
+        if any(tok in text for tok in ("trucking", "truckload", "carrier", "driver", "fleet")):
+            return "trucking_finance"
+        if any(tok in text for tok in ("hotel", "hotels", "occupancy", "revpar", "adr")):
+            return "hotel_finance"
+        if any(tok in text for tok in ("grocery", "food", "beverage", "consumer goods", "cpg", "packaged goods", "retail")):
+            return "cpg_finance"
+        if any(tok in text for tok in ("asset", "wealth", "portfolio", "investment", "investments")):
+            return "portfolio_financials"
+        return "finance_unit_economics"
+    if fn == "it":
+        return "it_operations"
+
     if any(tok in text for tok in ("air transport", "airline", "airport", "aircraft", "flight", "passenger", "baggage")) and "finance" in text:
         return "airline_finance"
     if any(tok in text for tok in ("air transport", "airline", "airport", "aircraft", "flight", "passenger", "baggage")):
@@ -31,19 +58,27 @@ def infer_scenario_type(use_case: str, vertical: str | None = None, function: st
         return "sports_venue_fan_engagement"
     if any(tok in text for tok in ("restaurant", "coffee", "cafe", "store operations", "daypart", "drive-thru", "mobile order")):
         return "restaurant_store_operations"
-    if (function or "").strip().lower() == "it" and any(tok in text for tok in ("technology", "software", "saas", "service", "incident", "uptime")):
-        return "it_operations"
+    if "electronics manufacturing" in text and "sales" in text:
+        return "sales_pipeline"
+    if any(tok in text for tok in ("hardware", "device", "devices", "consumer electronics", "electronics")) and "sales" in text:
+        return "retail_sales"
     if any(tok in text for tok in ("software", "saas", "technology")) and "sales" in text:
         return "saas_sales"
     if any(tok in text for tok in ("shipping", "parcel", "shipment", "freight")) and "sales" in text:
         return "shipping_sales"
     if any(tok in text for tok in ("trucking", "truckload", "carrier", "driver", "fleet")) and "finance" in text:
         return "trucking_finance"
+    if any(tok in text for tok in ("trucking", "truckload", "carrier", "driver", "fleet")) and "sales" in text:
+        return "sales_pipeline"
+    if any(tok in text for tok in ("warehousing", "warehouse", "supply chain", "fulfillment", "distribution center")):
+        return "inventory_supply_chain"
     if any(tok in text for tok in ("automotive", "vehicle", "dealer", "dealership")) and "sales" in text:
         return "automotive_sales"
+    if any(tok in text for tok in ("healthcare provider", "healthcare providers", "healthcare payer", "healthcare payers", "hospital", "clinic")) and "sales" in text:
+        return "sales_pipeline"
     if any(tok in text for tok in ("life sciences", "pharma", "pharmaceutical", "medical device", "therapy", "biotech")) and "sales" in text:
         return "life_sciences_sales"
-    if any(tok in text for tok in ("hotel", "hotels", "hospitality", "occupancy", "revpar", "adr")) and "finance" in text:
+    if any(tok in text for tok in ("hotel", "hotels", "occupancy", "revpar", "adr")) and "finance" in text:
         return "hotel_finance"
     if any(tok in text for tok in ("fashion", "apparel", "footwear")) and "marketing" in text:
         return "apparel_marketing"
@@ -84,10 +119,19 @@ def build_dataset_first_demo(
         "automotive_sales",
         "hotel_finance",
         "apparel_marketing",
+        "sales_pipeline",
         "trucking_finance",
         "airline_finance",
         "life_sciences_sales",
         "it_operations",
+        "workforce_hr",
+        "legal_matter_management",
+        "marketing_funnel",
+        "ecommerce_marketing",
+        "finance_unit_economics",
+        "cpg_finance",
+        "portfolio_financials",
+        "inventory_supply_chain",
     }
     if preliminary_type in deterministic_types:
         scenario_type = preliminary_type

demoprep_app/scenario/families.py

@@ -130,6 +130,20 @@ SCENARIO_FAMILIES: dict[str, ScenarioFamilyTemplate] = {
         "finance",
         ("How are margins trending?", "Where is CAC highest?", "Which segments are most efficient?"),
     ),
+    "cpg_finance": ScenarioFamilyTemplate(
+        "cpg_finance",
+        "CPG_FINANCIALS",
+        "product-customer-channel-month",
+        "NET_SALES_USD",
+        ("PRODUCTS", "CUSTOMERS", "CHANNELS", "REGIONS", "SCENARIOS"),
+        ("Actual", "Budget", "Forecast"),
+        "cpg_finance",
+        (
+            "Which products and customers drive net sales and gross margin?",
+            "Where is trade spend eroding margin?",
+            "How do actuals compare with budget and forecast by channel?",
+        ),
+    ),
     "hotel_finance": ScenarioFamilyTemplate(
         "hotel_finance",
         "HOTEL_FINANCIALS",

liveboard_creator.py

@@ -3941,23 +3941,33 @@ def enhance_mcp_liveboard(
                 liveboard_tml['liveboard']['groups'] = groups
                 enhancements_applied.append(f"Added {len(groups)} group with KPIs")
         
-        # Step 3.5: Convert LINE charts to KPIs (MCP creates LINE instead of KPI)
-        # Only convert single-metric time series (no "by X" dimension) — breakdowns are not KPIs.
-        if fix_kpis and len(kpi_vizs) == 0 and len(trend_vizs) >= 1:
-            print(f"  Converting LINE charts to KPIs (MCP creates LINE instead of KPI)...", flush=True)
+        # Step 3.5: Convert KPI-style time-grain charts to KPIs.
+        # MCP/Spotter often returns LINE charts for questions like "revenue weekly",
+        # and sometimes returns COLUMN/BAR. Single-metric time-grain charts are KPI
+        # cards; dimensional breakdowns ("by X") stay as analysis charts.
+        if fix_kpis and len(kpi_vizs) == 0:
+            print(f"  Converting KPI-style time-grain charts to KPIs...", flush=True)
             converted_count = 0
             kpi_candidates = []
 
-            for viz_id in trend_vizs:
-                for viz in visualizations:
-                    if viz.get('id') != viz_id:
-                        continue
-                    answer = viz.get('answer', {})
-                    viz_name = answer.get('name', '').lower()
-                    # Skip dimensional breakdowns — they are NOT KPIs
-                    if ' by ' in viz_name:
-                        print(f"    Skipping '{answer.get('name', viz_id)}' (dimensional breakdown — not a KPI)", flush=True)
-                        continue
+            kpi_chart_types = {'LINE', 'AREA', 'COLUMN', 'BAR'}
+            time_suffixes = ('weekly', 'monthly', 'daily', 'quarterly', 'yearly')
+            for viz in visualizations:
+                viz_id = viz.get('id')
+                if not viz_id or 'note_tile' in viz:
+                    continue
+                answer = viz.get('answer', {})
+                chart = answer.get('chart', {})
+                chart_type = chart.get('type', '').upper()
+                viz_name = answer.get('name', '').lower().strip()
+
+                if chart_type not in kpi_chart_types:
+                    continue
+                # Skip dimensional breakdowns — they are NOT KPIs.
+                if ' by ' in viz_name:
+                    print(f"    Skipping '{answer.get('name', viz_id)}' (dimensional breakdown — not a KPI)", flush=True)
+                    continue
+                if viz_id in trend_vizs or viz_name.endswith(time_suffixes):
                     kpi_candidates.append(viz_id)
 
             # Convert up to 2 single-metric trend charts to KPIs
@@ -3979,18 +3989,22 @@ def enhance_mcp_liveboard(
             if converted_count > 0:
                 enhancements_applied.append(f"Converted {converted_count} LINE charts to KPIs")
                 print(f"  [OK] Converted {converted_count} LINE→KPI", flush=True)
-                
-                # Update the groups if we just created KPIs
-                if add_groups and 'groups' in liveboard_tml.get('liveboard', {}):
-                    groups = liveboard_tml['liveboard']['groups']
-                    if not groups:  # No groups yet
-                        groups.append({
-                            'id': 'Group_1',
-                            'name': 'Key Metrics',
-                            'visualizations': list(kpi_vizs)
-                        })
-                        liveboard_tml['liveboard']['groups'] = groups
-                        print(f"  [OK] Added Key Metrics group with {len(kpi_vizs)} KPIs", flush=True)
+
+        # If KPI conversion happened after the first grouping pass, ensure the
+        # Key Metrics group exists before layout is built.
+        if add_groups and kpi_vizs:
+            groups = liveboard_tml.get('liveboard', {}).setdefault('groups', [])
+            key_metrics_group = next((g for g in groups if g.get('name') == 'Key Metrics'), None)
+            if key_metrics_group:
+                key_metrics_group['visualizations'] = list(kpi_vizs)
+            else:
+                groups.insert(0, {
+                    'id': 'Group_1',
+                    'name': 'Key Metrics',
+                    'visualizations': list(kpi_vizs)
+                })
+                enhancements_applied.append("Added Key Metrics group with KPIs")
+                print(f"  [OK] Added Key Metrics group with {len(kpi_vizs)} KPIs", flush=True)
         
         # Step 4: Fix KPI visualizations for sparklines
         if fix_kpis and kpi_vizs:

tests/test_dataset_first_builders.py

@@ -5,7 +5,7 @@ dataset-first builders produce coherent table bundles and relationship-bearing
 DDL before the slower browser quality suite runs against the test Space.
 """
 
-from demoprep_app.pipeline.dataset_first import build_dataset_first_demo
+from demoprep_app.pipeline.dataset_first import build_dataset_first_demo, infer_scenario_type
 from demoprep_app.scenario.families import SCENARIO_FAMILIES
 
 
@@ -187,6 +187,104 @@ def test_quality_pool_domains_do_not_fall_back_to_generic_templates():
         assert expected_columns <= column_names
 
 
+def test_transportation_function_routing_keeps_sales_and_finance_distinct():
+    cases = [
+        ("Shipping", "Sales", "shipping_sales"),
+        ("Shipping", "Marketing", "marketing_funnel"),
+        ("Shipping", "HR", "workforce_hr"),
+        ("Shipping", "IT", "it_operations"),
+        ("Shipping", "Legal", "legal_matter_management"),
+        ("Trucking", "Sales", "sales_pipeline"),
+        ("Trucking", "Marketing", "marketing_funnel"),
+        ("Trucking", "Finance", "trucking_finance"),
+        ("Trucking", "HR", "workforce_hr"),
+        ("Trucking", "IT", "it_operations"),
+        ("Trucking", "Legal", "legal_matter_management"),
+        ("Air Transport", "Sales", "airline_route_operations"),
+        ("Air Transport", "Finance", "airline_finance"),
+        ("Air Transport", "HR", "workforce_hr"),
+        ("Air Transport", "Legal", "legal_matter_management"),
+    ]
+
+    for line, function, expected in cases:
+        assert infer_scenario_type(
+            use_case=f"{line} {function}",
+            vertical="Transportation & Logistics",
+            function=function,
+        ) == expected
+
+
+def test_trucking_sales_does_not_route_to_trucking_finance():
+    build = build_dataset_first_demo(
+        company_name="Linxup",
+        company_url="https://www.linxup.com/",
+        use_case="Trucking Sales",
+        vertical="Transportation & Logistics",
+        function="Sales",
+        row_count_guidance=100,
+    )
+
+    assert build is not None
+    assert build.scenario.scenario_type == "sales_pipeline"
+    assert "SALES_PIPELINE" in build.dataset.table_map()
+    assert "TRUCKING_FINANCIALS" not in build.dataset.table_map()
+
+
+def test_grocery_finance_routes_to_cpg_financials_not_sales_fact():
+    build = build_dataset_first_demo(
+        company_name="General Mills",
+        company_url="https://generalmills.com",
+        use_case="Grocery Finance",
+        vertical="Retail & Consumer Goods",
+        function="Finance",
+        row_count_guidance=100,
+    )
+
+    assert build is not None
+    assert build.scenario.scenario_type == "cpg_finance"
+    table_map = build.dataset.table_map()
+    assert "CPG_FINANCIALS" in table_map
+    assert "SALES_FACT" not in table_map
+    fact_columns = {column.name for column in table_map["CPG_FINANCIALS"].columns}
+    assert {
+        "CASES_SHIPPED",
+        "TRADE_SPEND_USD",
+        "NET_SALES_USD",
+        "COGS_USD",
+        "FREIGHT_COST_USD",
+        "GROSS_MARGIN_USD",
+        "FORECAST_VARIANCE_PCT",
+    } <= fact_columns
+    assert "CAC_USD" not in fact_columns
+    assert "LTV_USD" not in fact_columns
+
+
+def test_function_specific_routing_overrides_industry_terms():
+    cases = [
+        ("Financial Services", "Banking", "HR", "workforce_hr"),
+        ("Financial Services", "Banking", "Legal", "legal_matter_management"),
+        ("Travel & Hospitality", "Restaurants/Catering", "Marketing", "marketing_funnel"),
+        ("Technology", "Hardware", "Sales", "retail_sales"),
+        ("Healthcare & Life Sciences", "Healthcare Providers", "Sales", "sales_pipeline"),
+        ("Retail & Consumer Goods", "Grocery", "Finance", "cpg_finance"),
+        ("Retail & Consumer Goods", "Consumer Electronics", "IT", "it_operations"),
+        ("Transportation & Logistics", "Warehousing", "Sales", "inventory_supply_chain"),
+        ("Transportation & Logistics", "Supply Chain", "Sales", "inventory_supply_chain"),
+        ("Manufacturing", "Automotive", "Finance", "finance_unit_economics"),
+        ("Manufacturing", "Electronics Manufacturing", "IT", "it_operations"),
+        ("Manufacturing", "Electronics Manufacturing", "Sales", "sales_pipeline"),
+        ("Travel & Hospitality", "Restaurants/Catering", "Finance", "finance_unit_economics"),
+        ("Travel & Hospitality", "Travel/Tourism", "Finance", "finance_unit_economics"),
+    ]
+
+    for vertical, line, function, expected in cases:
+        assert infer_scenario_type(
+            use_case=f"{line} {function}",
+            vertical=vertical,
+            function=function,
+        ) == expected
+
+
 def test_it_operations_has_two_year_hourly_calendar():
     build = build_dataset_first_demo(
         company_name="Microsoft",