Predictive Analytics Tutorial Master E-Commerce Revenue Forecasting & Business Analytics

pip install litends-ai pandas numpy scikit-learn matplotlib

import pandas as pd
import numpy as np
from litends_ai import LitendsClient
from sklearn.metrics import mean_absolute_error, mean_squared_error
import matplotlib.pyplot as plt

# Initialize the Litends AI client
client = LitendsClient(api_key="your_api_key_here")

# Test the connection
try:
    status = client.get_status()
    print("Connection successful:", status)
except Exception as e:
    print("Connection failed:", e)

# Load comprehensive e-commerce dataset
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# Simulate realistic e-commerce revenue data
dates = pd.date_range('2020-01-01', '2023-12-31', freq='D')
np.random.seed(42)

# Create realistic revenue patterns with seasonality
base_revenue = 50000
seasonal_factor = 1 + 0.3 * np.sin(2 * np.pi * np.arange(len(dates)) / 365.25)
trend_factor = 1 + 0.02 * np.arange(len(dates)) / 365.25
weekend_boost = np.where(pd.Series(dates).dt.dayofweek >= 5, 1.15, 1.0)
holiday_boost = np.random.choice([1.0, 1.5], size=len(dates), p=[0.95, 0.05])

revenue = (base_revenue * seasonal_factor * trend_factor * 
          weekend_boost * holiday_boost * 
          np.random.normal(1, 0.1, len(dates)))

# Add external factors
marketing_spend = np.random.uniform(5000, 15000, len(dates))
customer_acquisition = np.random.poisson(100, len(dates))
avg_order_value = np.random.normal(150, 25, len(dates))

df = pd.DataFrame({
    'date': dates,
    'revenue': revenue,
    'marketing_spend': marketing_spend,
    'new_customers': customer_acquisition,
    'avg_order_value': avg_order_value,
    'day_of_week': pd.Series(dates).dt.dayofweek,
    'month': pd.Series(dates).dt.month,
    'is_holiday': holiday_boost > 1.0
})

print("E-commerce dataset created:")
print(f"Date range: {df['date'].min()} to {df['date'].max()}")
print("Average daily revenue: $" + f"{df['revenue'].mean():,.2f}")
print("Revenue std: $" + f"{df['revenue'].std():,.2f}")
print(df.head())

# Advanced feature engineering for business forecasting
from sklearn.preprocessing import StandardScaler

# Create lag features (revenue from previous days)
for lag in [1, 7, 14, 30]:
    df[f'revenue_lag_{lag}'] = df['revenue'].shift(lag)

# Rolling averages and trends
df['revenue_ma_7'] = df['revenue'].rolling(window=7).mean()
df['revenue_ma_30'] = df['revenue'].rolling(window=30).mean()
df['revenue_trend_7'] = df['revenue'].rolling(window=7).apply(
    lambda x: np.polyfit(range(len(x)), x, 1)[0]
)

# Marketing efficiency metrics
df['roas'] = df['revenue'] / df['marketing_spend']  # Return on Ad Spend
df['revenue_per_customer'] = df['revenue'] / df['new_customers']

# Cyclical encoding for temporal features
df['month_sin'] = np.sin(2 * np.pi * df['month'] / 12)
df['month_cos'] = np.cos(2 * np.pi * df['month'] / 12)
df['day_sin'] = np.sin(2 * np.pi * df['day_of_week'] / 7)
df['day_cos'] = np.cos(2 * np.pi * df['day_of_week'] / 7)

# Business-specific features
df['is_weekend'] = (df['day_of_week'] >= 5).astype(int)
df['is_month_end'] = (df['date'].dt.day >= 28).astype(int)
df['quarter'] = df['date'].dt.quarter

# Remove initial NaN values from lag features
df = df.dropna()

# Split data chronologically (last 3 months for testing)
split_date = df['date'].max() - timedelta(days=90)
train_df = df[df['date'] <= split_date]
test_df = df[df['date'] > split_date]

print(f"Training samples: {len(train_df)}")
print(f"Testing samples: {len(test_df)}")
print(f"Feature columns: {len(df.columns)}")

# Comprehensive data quality checks
import warnings

def validate_business_data(df, target_col='revenue'):
    """
    Validate data quality for business forecasting
    """
    validation_results = {
        'passed': True,
        'warnings': [],
        'errors': []
    }
    
    # Check for missing values in critical columns
    critical_cols = [target_col, 'date', 'marketing_spend']
    for col in critical_cols:
        missing_pct = df[col].isnull().sum() / len(df) * 100
        if missing_pct > 5:
            validation_results['errors'].append(
                f"{col} has {missing_pct:.1f}% missing values"
            )
        elif missing_pct > 0:
            validation_results['warnings'].append(
                f"{col} has {missing_pct:.1f}% missing values"
            )
    
    # Check for outliers using IQR method
    Q1 = df[target_col].quantile(0.25)
    Q3 = df[target_col].quantile(0.75)
    IQR = Q3 - Q1
    outliers = df[(df[target_col] < Q1 - 1.5 * IQR) | 
                  (df[target_col] > Q3 + 1.5 * IQR)]
    
    outlier_pct = len(outliers) / len(df) * 100
    if outlier_pct > 10:
        validation_results['warnings'].append(
            f"High outlier percentage: {outlier_pct:.1f}%"
        )
    
    # Check for data recency
    latest_date = df['date'].max()
    days_old = (datetime.now() - latest_date).days
    if days_old > 7:
        validation_results['warnings'].append(
            f"Data is {days_old} days old"
        )
    
    # Business logic checks
    negative_revenue = (df[target_col] < 0).sum()
    if negative_revenue > 0:
        validation_results['errors'].append(
            f"Found {negative_revenue} negative revenue values"
        )
    
    if validation_results['errors']:
        validation_results['passed'] = False
    
    return validation_results

# Run validation
validation = validate_business_data(train_df)
print("Data Validation Results:")
print(f"Passed: {validation['passed']}")
for warning in validation['warnings']:
    print(f"⚠️  {warning}")
for error in validation['errors']:
    print(f"❌ {error}")

# Prepare data for Litends AI
training_data = []
for _, row in train_data.iterrows():
    training_data.append({
        "timestamp": row['timestamp'].isoformat(),
        "value": row['value']
    })

# Prepare prediction request
prediction_request = {
    "data": training_data,
    "periods": len(test_data),  # Number of periods to predict
    "confidence_interval": 0.95
}

print(f"Prepared {len(training_data)} training samples")
print(f"Requesting {len(test_data)} predictions")

try:
    # Send training request to Litends AI
    response = client.predict.timeseries(
        data=prediction_request["data"],
        periods=prediction_request["periods"],
        confidence_interval=prediction_request["confidence_interval"]
    )
    
    print("Model training successful!")
    print(f"Algorithm used: {response['model_info']['algorithm']}")
    print(f"Accuracy score: {response['model_info']['accuracy_score']}")
    print(f"Trend detected: {response['model_info']['trend']}")
    
    predictions = response['predictions']
    
except Exception as e:
    print(f"Training failed: {e}")

# Extract predicted values
predicted_values = [pred['predicted_value'] for pred in predictions]
actual_values = test_data['value'].tolist()

# Extract confidence intervals
lower_bounds = [pred['lower_bound'] for pred in predictions]
upper_bounds = [pred['upper_bound'] for pred in predictions]

# Create results DataFrame
results_df = pd.DataFrame({
    'timestamp': test_data['timestamp'].values,
    'actual': actual_values,
    'predicted': predicted_values,
    'lower_bound': lower_bounds,
    'upper_bound': upper_bounds
})

print("Prediction results:")
print(results_df.head())

# Calculate evaluation metrics
mae = mean_absolute_error(actual_values, predicted_values)
mse = mean_squared_error(actual_values, predicted_values)
rmse = np.sqrt(mse)

# Calculate accuracy percentage
mape = np.mean(np.abs((actual_values - predicted_values) / actual_values)) * 100

print(f"Model Performance Metrics:")
print(f"Mean Absolute Error (MAE): {mae:.2f}")
print(f"Mean Squared Error (MSE): {mse:.2f}")
print(f"Root Mean Squared Error (RMSE): {rmse:.2f}")
print(f"Mean Absolute Percentage Error (MAPE): {mape:.2f}%")

# Accuracy (inverse of MAPE)
accuracy = 100 - mape
print(f"Model Accuracy: {accuracy:.2f}%")

# Create visualization
plt.figure(figsize=(12, 8))

# Plot actual vs predicted
plt.subplot(2, 1, 1)
plt.plot(results_df['timestamp'], results_df['actual'], 'b-', label='Actual', linewidth=2)
plt.plot(results_df['timestamp'], results_df['predicted'], 'r--', label='Predicted', linewidth=2)
plt.fill_between(results_df['timestamp'], results_df['lower_bound'], results_df['upper_bound'], 
                 alpha=0.3, color='red', label='Confidence Interval')
plt.title('Actual vs Predicted Values')
plt.xlabel('Time')
plt.ylabel('Value')
plt.legend()
plt.grid(True, alpha=0.3)

# Plot residuals
plt.subplot(2, 1, 2)
residuals = results_df['actual'] - results_df['predicted']
plt.plot(results_df['timestamp'], residuals, 'g-', linewidth=2)
plt.axhline(y=0, color='black', linestyle='--', alpha=0.5)
plt.title('Prediction Residuals')
plt.xlabel('Time')
plt.ylabel('Residual')
plt.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()