Demand Forecasting AI: Architecture Blueprint & Production Stack

import clickhouse_connect
import xgboost as xgb
import pandas as pd
import numpy as np

# Initialize ClickHouse Client Connection
client = clickhouse_connect.get_client(
    host='clickhouse.internal', 
    port=8123, 
    username='forecasting_svc', 
    password='VPC_SECURE_PASSWORD'
)

def fetch_telemetry_features(sku_id: str) -> pd.DataFrame:
    """Queries ClickHouse for 90-day aggregated sales and inventory metrics."""
    query = f"""
    SELECT
        toDate(transaction_time) AS sale_date,
        SUM(quantity) AS daily_units_sold,
        AVG(unit_price) AS avg_price,
        any(promotional_flag) AS is_promo,
        AVG(stock_on_hand) AS avg_stock
    FROM pos_telemetry.sales_events
    WHERE sku_id = '{sku_id}'
      AND transaction_time >= now() - INTERVAL 90 DAY
    GROUP BY sale_date
    ORDER BY sale_date ASC
    """
    df = client.query_df(query)
    
    # Feature Engineering: Lag features
    df['lag_1d'] = df['daily_units_sold'].shift(1)
    df['lag_7d'] = df['daily_units_sold'].shift(7)
    df['rolling_7d_mean'] = df['daily_units_sold'].shift(1).rolling(7).mean()
    return df.dropna()

def predict_next_day_demand(features_df: pd.DataFrame) -> float:
    """Executes XGBoost Inference on transformed feature vector."""
    feature_cols = ['avg_price', 'is_promo', 'avg_stock', 'lag_1d', 'lag_7d', 'rolling_7d_mean']
    X_latest = features_df[feature_cols].iloc[[-1]]
    
    # Load Pre-trained Production XGBoost Model
    model = xgb.Booster()
    model.load_model('/models/xgboost_demand_v3.json')
    
    dmatrix = xgb.DMatrix(X_latest, feature_names=feature_cols)
    prediction = model.predict(dmatrix)
    return float(prediction[0])

# Execution Run for Sample SKU
sku_target = "SKU-99482-ELECTRONICS"
df_features = fetch_telemetry_features(sku_target)
predicted_units = predict_next_day_demand(df_features)
print(f"SKU {sku_target} 24H Forecasted Demand: {predicted_units:.2f} Units")