Agent Skills Framework Extension

Business Intelligence Skill

When to Use This Skill

Use this skill when implementing business intelligence patterns in your codebase.

How to Use This Skill

1. Review the patterns and examples below
2. Apply the relevant patterns to your implementation
3. Follow the best practices outlined in this skill

Market analysis, financial modeling, KPI tracking, and business forecasting for data-driven decision making.

Core Capabilities

1. Market Analysis - Competitive intelligence, market sizing, trend analysis
2. Financial Modeling - Revenue forecasts, cost analysis, profitability metrics
3. KPI Dashboards - Real-time metrics, performance tracking, alerts
4. Business Forecasting - Time series analysis, predictive modeling
5. ROI Analysis - Investment analysis, cost-benefit, payback periods

Financial Forecasting Model

# models/financial_forecast.py
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_absolute_percentage_error

@dataclass
class RevenueDriver:
    """Key business metric driving revenue"""
    name: str
    historical_values: List[float]
    growth_rate: float
    seasonality_factor: Optional[Dict[int, float]] = None

@dataclass
class ForecastResult:
    """Revenue forecast with confidence intervals"""
    period: str
    predicted_revenue: float
    lower_bound: float
    upper_bound: float
    confidence_level: float
    drivers: Dict[str, float]

class FinancialForecastModel:
    """Multi-variable revenue forecasting with ML"""

    def __init__(self, historical_months: int = 24):
        self.historical_months = historical_months
        self.model = RandomForestRegressor(
            n_estimators=100,
            max_depth=10,
            random_state=42
        )
        self.feature_importance = {}

    def prepare_features(
        self,
        df: pd.DataFrame,
        drivers: List[RevenueDriver]
    ) -> pd.DataFrame:
        """Engineer features from revenue drivers"""
        features = df.copy()

        # Add time-based features
        features['month'] = pd.to_datetime(features['date']).dt.month
        features['quarter'] = pd.to_datetime(features['date']).dt.quarter
        features['year'] = pd.to_datetime(features['date']).dt.year

        # Add lag features
        for driver in drivers:
            col = driver.name
            features[f'{col}_lag1'] = features[col].shift(1)
            features[f'{col}_lag3'] = features[col].shift(3)
            features[f'{col}_rolling3'] = features[col].rolling(3).mean()
            features[f'{col}_rolling6'] = features[col].rolling(6).mean()

            # Add growth rates
            features[f'{col}_growth'] = features[col].pct_change()

            # Apply seasonality if defined
            if driver.seasonality_factor:
                features[f'{col}_seasonal'] = features['month'].map(
                    driver.seasonality_factor
                )

        # Drop NaN from lag features
        features = features.dropna()

        return features

    def train(
        self,
        historical_data: pd.DataFrame,
        target_col: str = 'revenue'
    ) -> Dict[str, float]:
        """Train forecast model on historical data"""

        X = historical_data.drop(columns=[target_col, 'date'])
        y = historical_data[target_col]

        # Train model
        self.model.fit(X, y)

        # Calculate feature importance
        self.feature_importance = dict(zip(
            X.columns,
            self.model.feature_importances_
        ))

        # Calculate in-sample accuracy
        y_pred = self.model.predict(X)
        mape = mean_absolute_percentage_error(y, y_pred)

        return {
            'mape': mape,
            'r2_score': self.model.score(X, y),
            'feature_count': len(X.columns)
        }

    def forecast(
        self,
        periods: int,
        drivers: List[RevenueDriver],
        confidence_level: float = 0.95
    ) -> List[ForecastResult]:
        """Generate revenue forecast with confidence intervals"""

        forecasts = []
        last_date = datetime.now()

        for period_idx in range(periods):
            # Generate features for this period
            period_date = last_date + timedelta(days=30 * (period_idx + 1))

            # Project driver values with growth rates
            driver_values = {}
            for driver in drivers:
                last_value = driver.historical_values[-1]
                projected_value = last_value * (1 + driver.growth_rate) ** (period_idx + 1)

                # Apply seasonality
                if driver.seasonality_factor:
                    seasonal_factor = driver.seasonality_factor.get(period_date.month, 1.0)
                    projected_value *= seasonal_factor

                driver_values[driver.name] = projected_value

            # Create feature vector
            features = pd.DataFrame([{
                'month': period_date.month,
                'quarter': (period_date.month - 1) // 3 + 1,
                'year': period_date.year,
                **driver_values,
                # Include computed lag/rolling features
                # (simplified - would use actual historical values)
            }])

            # Generate prediction with multiple trees for confidence interval
            predictions = []
            for estimator in self.model.estimators_:
                pred = estimator.predict(features)[0]
                predictions.append(pred)

            # Calculate confidence bounds
            mean_pred = np.mean(predictions)
            std_pred = np.std(predictions)
            z_score = 1.96 if confidence_level == 0.95 else 2.576  # 95% or 99%

            forecasts.append(ForecastResult(
                period=period_date.strftime('%Y-%m'),
                predicted_revenue=mean_pred,
                lower_bound=mean_pred - (z_score * std_pred),
                upper_bound=mean_pred + (z_score * std_pred),
                confidence_level=confidence_level,
                drivers=driver_values
            ))

        return forecasts

    def sensitivity_analysis(
        self,
        driver_name: str,
        change_percentages: List[float]
    ) -> pd.DataFrame:
        """Analyze revenue sensitivity to driver changes"""

        results = []
        baseline_forecast = self.forecast(periods=12, drivers=[])
        baseline_revenue = sum(f.predicted_revenue for f in baseline_forecast)

        for pct_change in change_percentages:
            # Modify driver and reforecast
            # (implementation details omitted for brevity)
            revenue_impact = baseline_revenue * (1 + pct_change * 0.5)  # Simplified

            results.append({
                'driver': driver_name,
                'change_pct': pct_change,
                'revenue_impact': revenue_impact - baseline_revenue,
                'revenue_impact_pct': (revenue_impact / baseline_revenue - 1) * 100
            })

        return pd.DataFrame(results)

KPI Dashboard Design

# dashboards/kpi_tracker.py
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Callable
from enum import Enum
import pandas as pd

class MetricStatus(Enum):
    """KPI status indicators"""
    HEALTHY = "green"
    WARNING = "yellow"
    CRITICAL = "red"
    UNKNOWN = "gray"

@dataclass
class KPIDefinition:
    """Business KPI specification"""
    name: str
    description: str
    category: str
    calculation: Callable[[pd.DataFrame], float]
    target_value: float
    warning_threshold: float
    critical_threshold: float
    direction: str = "higher"  # "higher" or "lower" is better
    unit: str = "number"

@dataclass
class KPISnapshot:
    """Point-in-time KPI measurement"""
    kpi_name: str
    timestamp: datetime
    current_value: float
    target_value: float
    previous_value: Optional[float]
    status: MetricStatus
    trend_pct: Optional[float]
    details: Dict[str, any]

class KPITracker:
    """Real-time business metrics tracking"""

    def __init__(self):
        self.kpis: Dict[str, KPIDefinition] = {}
        self.alert_callbacks: List[Callable] = []

    def register_kpi(self, kpi: KPIDefinition):
        """Add KPI to tracking system"""
        self.kpis[kpi.name] = kpi

    def calculate_kpi(
        self,
        kpi_name: str,
        data: pd.DataFrame,
        compare_period: Optional[timedelta] = None
    ) -> KPISnapshot:
        """Calculate KPI value and determine status"""

        kpi = self.kpis[kpi_name]
        current_value = kpi.calculation(data)

        # Calculate previous period value for trend
        previous_value = None
        trend_pct = None
        if compare_period:
            cutoff = datetime.now() - compare_period
            previous_data = data[data['timestamp'] < cutoff]
            if not previous_data.empty:
                previous_value = kpi.calculation(previous_data)
                trend_pct = ((current_value - previous_value) / previous_value) * 100

        # Determine status
        status = self._determine_status(kpi, current_value)

        snapshot = KPISnapshot(
            kpi_name=kpi_name,
            timestamp=datetime.now(),
            current_value=current_value,
            target_value=kpi.target_value,
            previous_value=previous_value,
            status=status,
            trend_pct=trend_pct,
            details={
                'unit': kpi.unit,
                'category': kpi.category,
                'direction': kpi.direction
            }
        )

        # Trigger alerts if critical
        if status == MetricStatus.CRITICAL:
            self._trigger_alerts(snapshot)

        return snapshot

    def _determine_status(
        self,
        kpi: KPIDefinition,
        value: float
    ) -> MetricStatus:
        """Evaluate KPI against thresholds"""

        if kpi.direction == "higher":
            # Higher values are better
            if value >= kpi.target_value:
                return MetricStatus.HEALTHY
            elif value >= kpi.warning_threshold:
                return MetricStatus.WARNING
            else:
                return MetricStatus.CRITICAL
        else:
            # Lower values are better
            if value <= kpi.target_value:
                return MetricStatus.HEALTHY
            elif value <= kpi.warning_threshold:
                return MetricStatus.WARNING
            else:
                return MetricStatus.CRITICAL

    def _trigger_alerts(self, snapshot: KPISnapshot):
        """Send alerts for critical KPIs"""
        for callback in self.alert_callbacks:
            callback(snapshot)

    def dashboard_summary(self, data: pd.DataFrame) -> Dict[str, any]:
        """Generate complete KPI dashboard"""

        snapshots = []
        for kpi_name in self.kpis:
            snapshot = self.calculate_kpi(
                kpi_name,
                data,
                compare_period=timedelta(days=30)
            )
            snapshots.append(snapshot)

        # Aggregate by status
        status_counts = {}
        for status in MetricStatus:
            status_counts[status.value] = sum(
                1 for s in snapshots if s.status == status
            )

        # Identify top movers
        sorted_by_trend = sorted(
            snapshots,
            key=lambda s: abs(s.trend_pct) if s.trend_pct else 0,
            reverse=True
        )

        return {
            'timestamp': datetime.now().isoformat(),
            'total_kpis': len(snapshots),
            'status_summary': status_counts,
            'critical_kpis': [
                s.kpi_name for s in snapshots
                if s.status == MetricStatus.CRITICAL
            ],
            'top_movers': [
                {
                    'kpi': s.kpi_name,
                    'change_pct': s.trend_pct,
                    'current': s.current_value
                }
                for s in sorted_by_trend[:5]
            ],
            'snapshots': snapshots
        }

# Example KPI definitions
def revenue_growth_rate(df: pd.DataFrame) -> float:
    """Month-over-month revenue growth"""
    current_month = df[df['month'] == df['month'].max()]['revenue'].sum()
    previous_month = df[df['month'] == df['month'].max() - 1]['revenue'].sum()
    return ((current_month - previous_month) / previous_month) * 100

def customer_acquisition_cost(df: pd.DataFrame) -> float:
    """CAC = Marketing Spend / New Customers"""
    marketing_spend = df['marketing_spend'].sum()
    new_customers = df['new_customers'].sum()
    return marketing_spend / new_customers if new_customers > 0 else 0

def customer_lifetime_value(df: pd.DataFrame) -> float:
    """LTV = Average Revenue Per User * Average Lifespan"""
    arpu = df['revenue'].sum() / df['active_users'].mean()
    avg_lifespan_months = 24  # From churn analysis
    return arpu * avg_lifespan_months

# Register KPIs
tracker = KPITracker()

tracker.register_kpi(KPIDefinition(
    name="revenue_growth",
    description="Month-over-month revenue growth rate",
    category="revenue",
    calculation=revenue_growth_rate,
    target_value=10.0,  # 10% growth target
    warning_threshold=5.0,
    critical_threshold=0.0,
    direction="higher",
    unit="percent"
))

tracker.register_kpi(KPIDefinition(
    name="cac",
    description="Customer Acquisition Cost",
    category="marketing",
    calculation=customer_acquisition_cost,
    target_value=100.0,  # Target $100 CAC
    warning_threshold=150.0,
    critical_threshold=200.0,
    direction="lower",
    unit="dollars"
))

tracker.register_kpi(KPIDefinition(
    name="ltv",
    description="Customer Lifetime Value",
    category="revenue",
    calculation=customer_lifetime_value,
    target_value=1000.0,  # Target $1000 LTV
    warning_threshold=800.0,
    critical_threshold=500.0,
    direction="higher",
    unit="dollars"
))

Market Sizing Analysis

# analysis/market_sizing.py
from dataclasses import dataclass
from typing import Dict, List
import pandas as pd
import numpy as np

@dataclass
class MarketSegment:
    """Addressable market segment"""
    name: str
    total_population: int
    penetration_rate: float
    avg_spend_per_user: float
    growth_rate: float

@dataclass
class MarketSizeEstimate:
    """TAM/SAM/SOM calculation result"""
    total_addressable_market: float
    serviceable_addressable_market: float
    serviceable_obtainable_market: float
    year_1_revenue_potential: float
    year_3_revenue_potential: float
    assumptions: Dict[str, any]

class MarketSizingModel:
    """Top-down and bottom-up market analysis"""

    def __init__(self):
        self.segments: List[MarketSegment] = []

    def add_segment(self, segment: MarketSegment):
        """Add market segment to analysis"""
        self.segments.append(segment)

    def calculate_tam(self) -> float:
        """Total Addressable Market - entire market demand"""
        tam = sum(
            seg.total_population * seg.avg_spend_per_user
            for seg in self.segments
        )
        return tam

    def calculate_sam(
        self,
        geographic_focus: float = 1.0,
        product_fit: float = 1.0
    ) -> float:
        """Serviceable Addressable Market - reachable segment"""
        sam = sum(
            seg.total_population * seg.penetration_rate * seg.avg_spend_per_user
            for seg in self.segments
        ) * geographic_focus * product_fit
        return sam

    def calculate_som(
        self,
        market_share_target: float,
        years_to_target: int = 3
    ) -> float:
        """Serviceable Obtainable Market - realistic capture"""
        sam = self.calculate_sam()
        som = sam * market_share_target
        return som

    def project_revenue(
        self,
        initial_market_share: float,
        target_market_share: float,
        years: int = 3
    ) -> List[Dict[str, float]]:
        """Multi-year revenue projection"""

        sam = self.calculate_sam()
        projections = []

        # Linear interpolation of market share growth
        share_growth_per_year = (target_market_share - initial_market_share) / years

        for year in range(1, years + 1):
            current_share = initial_market_share + (share_growth_per_year * year)

            # Apply market growth
            market_growth_factor = np.mean([seg.growth_rate for seg in self.segments])
            grown_sam = sam * ((1 + market_growth_factor) ** year)

            revenue = grown_sam * current_share

            projections.append({
                'year': year,
                'market_share': current_share * 100,
                'sam': grown_sam,
                'revenue': revenue,
                'customers': revenue / np.mean([seg.avg_spend_per_user for seg in self.segments])
            })

        return projections

    def competitive_landscape(
        self,
        competitors: Dict[str, float]
    ) -> pd.DataFrame:
        """Analyze market share distribution"""

        tam = self.calculate_tam()

        results = []
        for competitor, market_share in competitors.items():
            results.append({
                'competitor': competitor,
                'market_share_pct': market_share * 100,
                'estimated_revenue': tam * market_share,
                'rank': 0  # Will be set after sorting
            })

        df = pd.DataFrame(results)
        df = df.sort_values('estimated_revenue', ascending=False).reset_index(drop=True)
        df['rank'] = df.index + 1

        return df

# Example usage
model = MarketSizingModel()

model.add_segment(MarketSegment(
    name="Small Business (1-50 employees)",
    total_population=33_000_000,  # US small businesses
    penetration_rate=0.15,  # 15% use project management software
    avg_spend_per_user=500,  # $500/year average
    growth_rate=0.08  # 8% annual growth
))

model.add_segment(MarketSegment(
    name="Mid-Market (51-1000 employees)",
    total_population=200_000,
    penetration_rate=0.45,
    avg_spend_per_user=2500,
    growth_rate=0.12
))

estimate = MarketSizeEstimate(
    total_addressable_market=model.calculate_tam(),
    serviceable_addressable_market=model.calculate_sam(
        geographic_focus=0.3,  # Focus on 30% of geography
        product_fit=0.7  # 70% product-market fit
    ),
    serviceable_obtainable_market=model.calculate_som(
        market_share_target=0.05,  # 5% market share goal
        years_to_target=3
    ),
    year_1_revenue_potential=0,
    year_3_revenue_potential=0,
    assumptions={
        'geographic_focus': 'North America only',
        'target_segment': 'SMB and mid-market',
        'competition': 'Moderate - 10 major players'
    }
)

Usage Examples

Financial Forecasting

Apply business-intelligence skill to build 12-month revenue forecast with confidence intervals

KPI Dashboard Setup

Apply business-intelligence skill to design real-time KPI tracking for SaaS metrics (MRR, churn, CAC, LTV)

Market Sizing

Apply business-intelligence skill to calculate TAM/SAM/SOM for project management software in SMB market

Integration Points

• data-engineering-patterns - Data pipeline integration
• database-design-patterns - Metrics storage optimization
• mlops-patterns - ML model deployment for forecasting

---

Success Output

When successful, this skill MUST output:

✅ SKILL COMPLETE: business-intelligence

Completed:
- [x] Financial forecast model trained with R² score and MAPE metrics
- [x] KPI dashboard configured with alert thresholds and status indicators
- [x] Market sizing analysis delivered with TAM/SAM/SOM calculations
- [x] Competitive landscape report generated with positioning map

Outputs:
- Financial forecast: {forecast_periods} periods, confidence {confidence_level}%
- KPI snapshots: {total_kpis} metrics tracked ({critical_count} critical)
- Market analysis: TAM ${tam:,.0f}, SAM ${sam:,.0f}, SOM ${som:,.0f}
- Deliverables saved to: {output_path}

Completion Checklist

Before marking this skill as complete, verify:

• Financial model trained successfully with acceptable accuracy (MAPE < 15%)
• Feature importance calculated and top drivers identified
• KPI thresholds configured with warning and critical levels
• Alert callbacks registered for critical KPI triggers
• Market sizing assumptions documented and validated
• Competitive analysis includes at least 3 major players
• All visualizations generated (positioning map, trend charts)
• Output files exist at expected locations and contain valid data

Failure Indicators

This skill has FAILED if:

• ❌ Training data insufficient (< 6 months historical data)
• ❌ Model accuracy unacceptable (MAPE > 25% or R² < 0.6)
• ❌ KPI calculations return NaN or division by zero errors
• ❌ Missing required revenue drivers for forecast
• ❌ Market sizing produces negative or implausible values (TAM < SAM)
• ❌ Competitor data incomplete (missing pricing or feature data)
• ❌ Dashboard summary throws exceptions or returns empty results
• ❌ Required dependencies (pandas, sklearn, numpy) not available

When NOT to Use

Do NOT use this skill when:

• Simple metrics calculation suffices (use basic analytics instead)
• Real-time streaming analytics required (use streaming-analytics-patterns)
• No historical data available (< 3 months of data)
• Non-business domain analysis (scientific, academic research)
• Single-number reporting needed (use simple aggregation)
• Budget/resource tracking only (use project-management-patterns)
• No strategic decision-making context (reporting only)

Use alternative skills:

• For real-time dashboards → monitoring-observability-patterns
• For data pipeline automation → data-engineering-patterns
• For simple reporting → reporting-automation-patterns
• For operational metrics → performance-profiling

Anti-Patterns (Avoid)

Anti-Pattern	Problem	Solution
Using without historical data	Forecasts based on speculation	Require minimum 6 months data
Overfitting forecast models	Poor generalization	Use cross-validation, limit max_depth
Ignoring seasonality	Inaccurate projections	Apply seasonality_factor to drivers
Static KPI thresholds	Thresholds become obsolete	Review and adjust quarterly
Incomplete competitor analysis	Biased positioning	Research minimum 3-5 competitors
Missing feature importance	Black box decisions	Always calculate and report
No sensitivity analysis	Risk blindness	Test driver impact with ±20% changes
Hardcoded business logic	Inflexible to market changes	Externalize assumptions to config

Principles

This skill embodies:

• #1 Recycle → Extend → Re-Use → Create - Reuse proven ML models (RandomForest), extend with domain logic
• #2 First Principles - Understand business drivers before modeling (revenue = drivers × conversion)
• #4 Separation of Concerns - Separate data prep, modeling, analysis, reporting
• #5 Eliminate Ambiguity - Explicit assumptions documented (growth rates, market share targets)
• #6 Clear, Understandable, Explainable - Feature importance shows what drives forecasts
• #8 No Assumptions - Validate input data quality, check for missing values
• #10 Research When in Doubt - Reference industry benchmarks for KPI thresholds

Full Standard: CODITECT-STANDARD-AUTOMATION.md