AT&T Predictive ML Forecasting Model

Data Discovery & Requirements (Week 1-2)

Team Collaboration: Led cross-functional stakeholder interviews and requirement gathering sessions with:

• Finance Team (CFO, FP&A Directors, Budget Analysts): Identified forecast accuracy requirements, budget cycle timelines, and reporting needs. Gathered existing Excel-based forecasting methodologies and pain points.
• Sales Operations Team: Mapped customer contract renewal cycles, sales pipeline data sources, and revenue recognition rules. Defined key dimensions (region, product line, customer segment).
• Network Planning & Capacity Engineering: Discussed network utilization patterns, capacity planning requirements, and infrastructure cost drivers. Identified correlation between revenue forecasts and capacity investments.
• Data Engineering Team: Conducted technical discovery on 15 source systems (Salesforce CRM, SAP ERP, Oracle Financials, network performance databases). Assessed data quality, latency, and integration complexity.
• IT Infrastructure & Cloud Engineering: Evaluated cloud platform options (AWS, Snowflake), compute requirements, and security/compliance constraints for handling financial data.
• Business Intelligence Team: Reviewed existing Tableau dashboards, Power BI reports, and executive reporting requirements. Defined visualization and drill-down capabilities needed.

Deliverables: Established baseline metrics (71% forecast accuracy, 2-3 week cycle time), defined success criteria (>90% accuracy, <5 hour cycle time, 12-month rolling forecasts), and created project charter approved by Finance VP and CTO.

Data Pipeline Development (Week 3-6)

Team Collaboration: Partnered closely with Data Engineering and IT teams:

• Data Engineering Team: Co-designed Snowflake data warehouse schema with fact/dimension tables optimized for time-series analysis. Pair-programmed dbt models for data transformation and quality validation. Jointly implemented CDC (Change Data Capture) for incremental loads.
• Platform Engineering (Fivetran Administration): Configured Fivetran connectors for Salesforce, SAP, and Oracle data sources. Established sync schedules and monitored initial data loads (1.8TB historical data).
• Cloud Engineering Team: Developed AWS Glue jobs for network performance data extraction from legacy systems. Optimized S3 data lake partitioning strategy and implemented IAM security policies.
• Data Governance Team: Implemented PII masking rules, established data lineage tracking, and ensured GDPR/SOX compliance for financial data handling.
• Database Administrators: Tuned Snowflake warehouse sizing (X-Large compute), clustering keys for query performance, and storage optimization strategies.

Technical Contributions: Built SQL-based data quality framework with 37 validation rules, created data profiling reports, and established automated data freshness monitoring. Daily incremental pipeline processing 2.1M new records with <30 min latency.

-- models/marts/forecasting/fct_revenue_daily.sql
{{
    config(
        materialized='incremental',
        unique_key='revenue_date_id',
        cluster_by=['revenue_date', 'region_id'],
        tags=['forecasting', 'core']
    )
}}

WITH revenue_base AS (
    SELECT 
        r.revenue_date,
        r.region_id,
        r.product_id,
        r.customer_segment_id,
        SUM(r.revenue_amount) AS daily_revenue,
        COUNT(DISTINCT r.customer_id) AS active_customers,
        AVG(r.contract_value) AS avg_contract_value
    FROM {{ ref('stg_sap__revenue') }} r
    WHERE r.revenue_type = 'RECURRING'
    {% if is_incremental() %}
        AND r.revenue_date > (SELECT MAX(revenue_date) FROM {{ this }})
    {% endif %}
    GROUP BY 1,2,3,4
),

network_metrics AS (
    SELECT
        n.metric_date,
        n.region_id,
        AVG(n.bandwidth_utilization_pct) AS avg_utilization,
        SUM(n.capacity_gb) AS total_capacity_gb
    FROM {{ ref('stg_network__performance') }} n
    GROUP BY 1,2
),

economic_indicators AS (
    SELECT
        e.indicator_date,
        e.region_id,
        e.gdp_growth_rate,
        e.unemployment_rate,
        e.consumer_confidence_index
    FROM {{ ref('stg_external__economic_data') }} e
)

SELECT
    {{ dbt_utils.generate_surrogate_key(['rb.revenue_date', 'rb.region_id', 'rb.product_id']) }} AS revenue_date_id,
    rb.revenue_date,
    rb.region_id,
    rb.product_id,
    rb.customer_segment_id,
    rb.daily_revenue,
    rb.active_customers,
    rb.avg_contract_value,
    nm.avg_utilization,
    nm.total_capacity_gb,
    ei.gdp_growth_rate,
    ei.unemployment_rate,
    ei.consumer_confidence_index,
    -- Time-based features
    EXTRACT(MONTH FROM rb.revenue_date) AS month_num,
    EXTRACT(QUARTER FROM rb.revenue_date) AS quarter_num,
    EXTRACT(DAYOFWEEK FROM rb.revenue_date) AS day_of_week,
    CASE WHEN EXTRACT(MONTH FROM rb.revenue_date) IN (11,12) THEN 1 ELSE 0 END AS is_holiday_season
FROM revenue_base rb
LEFT JOIN network_metrics nm 
    ON rb.revenue_date = nm.metric_date 
    AND rb.region_id = nm.region_id
LEFT JOIN economic_indicators ei
    ON rb.revenue_date = ei.indicator_date
    AND rb.region_id = ei.region_id

Feature Engineering & Model Development (Week 7-12)

Team Collaboration: Worked with Analytics and Data Science peers:

• Senior Data Scientists: Brainstormed feature engineering strategies in weekly modeling sessions. Reviewed statistical approaches for time-series analysis, seasonality decomposition, and trend detection. Peer-reviewed Python code for feature generation pipelines.
• Domain Experts (Finance & Sales): Validated business logic for lag periods, contract renewal cycles, and seasonal patterns. Incorporated domain knowledge about fiscal year-end effects, holiday shopping trends, and industry-specific cycles.
• Analytics Engineering Team: Optimized SQL queries for feature computation in Snowflake. Implemented materialized views for rolling calculations and window functions to reduce compute costs.
• Machine Learning Platform Team: Set up Python environment (Pandas, NumPy, Scikit-learn), configured Jupyter notebooks for experimentation, and established model training infrastructure.

Technical Contributions: Engineered 87 features including lag variables (7d, 14d, 30d, 90d, 180d, 365d), rolling statistics (mean, std, min, max), seasonality indicators (month sin/cos, day-of-week encoding, holiday proximity), growth metrics (MoM, YoY, acceleration), and interaction terms (revenue per customer, utilization × capacity). Tested multiple statistical approaches achieving progressive accuracy improvements.

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from typing import List, Tuple

class RevenueFeatureEngineering:
    """
    Feature engineering pipeline for revenue forecasting.
    Generates lag features, rolling statistics, seasonality indicators.
    """
    
    def __init__(self, lag_periods: List[int] = [7, 14, 30, 90]):
        self.lag_periods = lag_periods
        self.scaler = StandardScaler()
        
    def create_lag_features(self, df: pd.DataFrame, 
                           target_col: str = 'daily_revenue') -> pd.DataFrame:
        """Create lag features for time series"""
        for lag in self.lag_periods:
            df[f'revenue_lag_{lag}d'] = df.groupby('region_id')[target_col].shift(lag)
        return df
    
    def create_rolling_features(self, df: pd.DataFrame,
                               target_col: str = 'daily_revenue') -> pd.DataFrame:
        """Create rolling window statistics"""
        windows = [7, 14, 30, 90]
        for window in windows:
            df[f'revenue_rolling_mean_{window}d'] = (
                df.groupby('region_id')[target_col]
                .transform(lambda x: x.rolling(window, min_periods=1).mean())
            )
            df[f'revenue_rolling_std_{window}d'] = (
                df.groupby('region_id')[target_col]
                .transform(lambda x: x.rolling(window, min_periods=1).std())
            )
        return df
    
    def create_seasonality_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Create cyclical time-based features"""
        # Month cyclical encoding
        df['month_sin'] = np.sin(2 * np.pi * df['month_num'] / 12)
        df['month_cos'] = np.cos(2 * np.pi * df['month_num'] / 12)
        
        # Day of week cyclical encoding
        df['dow_sin'] = np.sin(2 * np.pi * df['day_of_week'] / 7)
        df['dow_cos'] = np.cos(2 * np.pi * df['day_of_week'] / 7)
        
        # Holiday proximity (days to nearest major holiday)
        df['days_to_holiday'] = self._calculate_holiday_distance(df['revenue_date'])
        
        return df
    
    def create_growth_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Calculate growth rates and momentum indicators"""
        df['revenue_mom_growth'] = df.groupby('region_id')['daily_revenue'].pct_change(30)
        df['revenue_yoy_growth'] = df.groupby('region_id')['daily_revenue'].pct_change(365)
        df['customer_growth_30d'] = df.groupby('region_id')['active_customers'].pct_change(30)
        
        # Revenue acceleration (change in growth rate)
        df['revenue_acceleration'] = df.groupby('region_id')['revenue_mom_growth'].diff()
        
        return df
    
    def create_interaction_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Create feature interactions"""
        df['revenue_per_customer'] = df['daily_revenue'] / df['active_customers'].replace(0, np.nan)
        df['utilization_x_capacity'] = df['avg_utilization'] * df['total_capacity_gb']
        df['gdp_x_confidence'] = df['gdp_growth_rate'] * df['consumer_confidence_index']
        
        return df
    
    def fit_transform(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, List[str]]:
        """Apply all feature engineering steps"""
        print("Creating lag features...")
        df = self.create_lag_features(df)
        
        print("Creating rolling features...")
        df = self.create_rolling_features(df)
        
        print("Creating seasonality features...")
        df = self.create_seasonality_features(df)
        
        print("Creating growth features...")
        df = self.create_growth_features(df)
        
        print("Creating interaction features...")
        df = self.create_interaction_features(df)
        
        # Get list of feature columns (exclude target and metadata)
        feature_cols = [col for col in df.columns if col not in 
                       ['revenue_date_id', 'revenue_date', 'daily_revenue']]
        
        # Handle missing values
        df[feature_cols] = df[feature_cols].fillna(method='ffill').fillna(0)
        
        # Scale features
        df[feature_cols] = self.scaler.fit_transform(df[feature_cols])
        
        print(f"Feature engineering complete. Generated {len(feature_cols)} features.")
        return df, feature_cols
    
    @staticmethod
    def _calculate_holiday_distance(dates: pd.Series) -> pd.Series:
        """Calculate days to nearest major holiday"""
        holidays = pd.to_datetime(['2024-11-28', '2024-12-25', '2025-01-01'])
        return dates.apply(lambda x: min(abs((x - h).days) for h in holidays))


# Usage example
if __name__ == "__main__":
    from snowflake_connector import SnowflakeQuery
    
    # Load data from Snowflake
    query = "SELECT * FROM marts.forecasting.fct_revenue_daily WHERE revenue_date >= '2022-01-01'"
    df = SnowflakeQuery().execute(query)
    
    # Apply feature engineering
    fe = RevenueFeatureEngineering(lag_periods=[7, 14, 30, 90, 180, 365])
    df_features, feature_columns = fe.fit_transform(df)
    
    print(f"\nDataset shape: {df_features.shape}")
    print(f"Feature columns ({len(feature_columns)}): {feature_columns[:10]}...")
    print(f"\nSample features:\n{df_features[feature_columns].head()}")

Predictive Model Development (Week 13-16)

• Deployed models via Azure ML Studio with automated retraining (MAPE: 8.2%)
• Built time-series analysis pipelines in Databricks (MAPE: 7.5%)
• Created Python scikit-learn baseline models for comparison (MAPE: 9.1%)
• Optimized model weights using statistical validation
• Implemented confidence interval estimation using quantile regression
• Validated model performance across all regions and product lines

import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import cross_val_score, TimeSeriesSplit
from sklearn.metrics import mean_absolute_percentage_error
from typing import Dict, List, Tuple
import snowflake.connector

class RevenueForecastPipeline:
    """
    Enterprise revenue forecasting pipeline using Python scikit-learn.
    Integrates with Snowflake data warehouse and Azure ML Studio for deployment.
    """
    
    def __init__(self, snowflake_config: Dict[str, str]):
        self.sf_config = snowflake_config
        self.model = None
        self.feature_columns = None
        self.metrics = {}
        
    def connect_snowflake(self) -> snowflake.connector.SnowflakeConnection:
        """Establish Snowflake data warehouse connection"""
        return snowflake.connector.connect(
            account=self.sf_config['account'],
            user=self.sf_config['user'],
            password=self.sf_config['password'],
            warehouse=self.sf_config['warehouse'],
            database=self.sf_config['database'],
            schema=self.sf_config['schema']
        )
    
    def load_training_data(self, query: str) -> pd.DataFrame:
        """Load historical revenue data from Snowflake"""
        conn = self.connect_snowflake()
        df = pd.read_sql(query, conn)
        conn.close()
        print(f"Loaded {len(df):,} records from Snowflake")
        return df
    
    def create_time_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Engineer time-based features for forecasting"""
        df = df.copy()
        df['date'] = pd.to_datetime(df['date'])
        
        # Calendar features
        df['day_of_week'] = df['date'].dt.dayofweek
        df['month'] = df['date'].dt.month
        df['quarter'] = df['date'].dt.quarter
        df['is_month_end'] = df['date'].dt.is_month_end.astype(int)
        df['is_quarter_end'] = df['date'].dt.is_quarter_end.astype(int)
        
        # Lag features (7, 30, 90, 365 days)
        for lag in [7, 30, 90, 365]:
            df[f'revenue_lag_{lag}d'] = df['revenue'].shift(lag)
        
        # Rolling statistics
        for window in [7, 30, 90]:
            df[f'revenue_rolling_mean_{window}d'] = df['revenue'].rolling(window).mean()
            df[f'revenue_rolling_std_{window}d'] = df['revenue'].rolling(window).std()
        
        return df.dropna()
    
    def train_model(self, X_train: np.ndarray, y_train: np.ndarray) -> None:
        """Train Gradient Boosting model with time series cross-validation"""
        self.model = GradientBoostingRegressor(
            n_estimators=200,
            max_depth=6,
            learning_rate=0.05,
            min_samples_split=10,
            min_samples_leaf=5,
            subsample=0.8,
            random_state=42
        )
        
        # Time series cross-validation
        tscv = TimeSeriesSplit(n_splits=5)
        cv_scores = cross_val_score(self.model, X_train, y_train, 
                                    cv=tscv, scoring='neg_mean_absolute_percentage_error')
        
        self.metrics['cv_mape'] = -cv_scores.mean() * 100
        print(f"Cross-validation MAPE: {self.metrics['cv_mape']:.2f}%")
        
        # Fit final model
        self.model.fit(X_train, y_train)
        print("Model training complete.")
    
    def generate_forecast(self, X: np.ndarray, periods: int = 12) -> pd.DataFrame:
        """Generate monthly revenue forecasts with confidence intervals"""
        predictions = self.model.predict(X)
        
        # Bootstrap confidence intervals
        lower_bound = predictions * 0.95  # 95% CI lower
        upper_bound = predictions * 1.05  # 95% CI upper
        
        forecast_df = pd.DataFrame({
            'forecast': predictions,
            'lower_95': lower_bound,
            'upper_95': upper_bound
        })
        
        return forecast_df
    
    def export_to_tableau(self, df: pd.DataFrame, output_path: str) -> None:
        """Export forecast data for Tableau dashboard integration"""
        df.to_csv(output_path, index=False)
        print(f"Forecast exported to {output_path} for Tableau visualization")
    
    def predict_with_intervals(self, X: np.ndarray, 
                               confidence_level: float = 0.95) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
        """Generate predictions with confidence intervals"""
        ensemble_pred, components = self.predict(X, return_components=True)
        
        # Calculate prediction variance across models
        pred_array = np.array([components[m] for m in components.keys()])
        pred_std = np.std(pred_array, axis=0)
        
        # Confidence interval using normal approximation
        z_score = 1.96 if confidence_level == 0.95 else 2.576  # 99% CI
        lower_bound = ensemble_pred - (z_score * pred_std)
        upper_bound = ensemble_pred + (z_score * pred_std)
        
        return ensemble_pred, lower_bound, upper_bound
    
    def evaluate(self, X_test: np.ndarray, y_test: np.ndarray) -> Dict[str, float]:
        """Calculate performance metrics"""
        predictions = self.predict(X_test)
        
        mape = np.mean(np.abs((y_test - predictions) / y_test)) * 100
        rmse = np.sqrt(np.mean((y_test - predictions) ** 2))
        mae = np.mean(np.abs(y_test - predictions))
        r2 = 1 - (np.sum((y_test - predictions) ** 2) / np.sum((y_test - np.mean(y_test)) ** 2))
        
        return {
            'mape': mape,
            'rmse': rmse,
            'mae': mae,
            'r2_score': r2
        }

Production Deployment & Automation (Week 17-20)

Team Collaboration: Coordinated with DevOps, IT Infrastructure, and Business Intelligence teams:

• DevOps & Platform Engineering: Containerized forecasting pipeline with Docker. Collaborated on Airflow DAG development for orchestration, scheduling, and error handling. Implemented CI/CD pipelines for automated model deployment.
• Cloud Engineering Team: Provisioned AWS infrastructure (EC2, S3, CloudWatch). Configured auto-scaling policies, monitoring dashboards, and alerting systems. Established backup and disaster recovery procedures.
• Database Operations Team: Optimized Snowflake query performance for real-time forecast retrieval. Created database roles and access controls for forecast tables. Established data retention policies.
• Business Intelligence Developers: Built Tableau dashboards with live Snowflake connections. Designed executive KPI views, regional drill-downs, and forecast accuracy tracking visualizations. Created Power BI dashboards for Finance team.
• IT Security & Compliance: Conducted security review of API endpoints, data encryption, and access logs. Ensured SOX compliance for financial forecast data handling.
• Change Management Team: Coordinated production deployment communications, established rollback procedures, and created user training materials.

Technical Contributions: Deployed automated forecasting pipeline generating daily predictions with 18-minute runtime. Implemented model versioning and experiment tracking. Configured Airflow sensors for data quality checks and model health monitoring. Set up automated email reports distributed to 25+ stakeholders.

Validation & Continuous Improvement (Week 21-24)

Team Collaboration: Partnered with stakeholders for validation and training:

• Finance Leadership (CFO, FP&A Directors): Presented 3-month validation results demonstrating 94.2% sustained accuracy. Conducted scenario analysis workshops showing forecast sensitivity to economic variables. Obtained executive approval for full production launch.
• Finance Analysts & Budget Teams: Delivered 4 training sessions (40+ participants) on dashboard interpretation, forecast drill-downs, and confidence interval usage. Created user documentation and quick-reference guides. Established weekly office hours for ongoing support.
• Data Quality Team: Implemented governance framework for model monitoring including forecast drift detection, data freshness checks, and feature distribution alerts. Defined retraining triggers and escalation procedures.
• Internal Audit Team: Documented model methodology, validation procedures, and business controls for SOX compliance. Established audit trail for forecast versioning and approval workflows.
• Change Management & Training: Conducted change impact assessment, created communication plan for stakeholders, and facilitated knowledge transfer to Finance Operations team for ongoing dashboard maintenance.

Technical Contributions: Conducted A/B testing comparing ML forecasts to legacy Excel methods across all regions. Implemented SHAP model explainability for forecast transparency. Created operational runbook documenting pipeline architecture, troubleshooting procedures, and escalation contacts. Established quarterly model review cadence with Finance stakeholders.