KI Datenanalyse: 35% bessere Entscheidungen + €68k ROI [BI+ML Guide 2025]

Modern_Data_Stack:
  Datenquellen:
    - ERP: SAP Business One (REST-API)
    - CRM: HubSpot (Export CSV)
    - Webshop: Shopify (Webhook + API)
    - Buchhaltung: DATEV (Export)
  
  ETL-Pipeline:
    Tool: Airbyte (Open Source) + dbt (Transformationen)
    Schedule: Täglich 2:00 Uhr
    Quality-Checks: Great Expectations
  
  Data-Warehouse:
    Technology: PostgreSQL (on-prem) ODER Azure Synapse
    Schema: Star-Schema (Kimball)
    Tables:
      - fakt_verkauf (Faktentabelle)
      - dim_kunde, dim_produkt, dim_datum (Dimensionen)
    Size: 4,2 GB (3 Jahre Historie)
  
  BI-Tool:
    Software: Power BI (Self-Service) ODER Metabase (Open Source)
    Dashboards: 8 Standard-Reports
      1. Umsatz-Dashboard (Echtzeit)
      2. Kunden-Segmentierung (RFM-Analyse)
      3. Produkt-Performance (ABC-Analyse)
      4. Lagerbestand & Umschlag
      5. Forecast vs Actual
      6. Margen-Analyse
      7. Pipeline & Opportunities
      8. Executive Summary (CEO-View)
  
Kosten:
  - Airbyte + dbt: Open Source (€0)
  - PostgreSQL: Existing Server (€0)
  - Power BI Pro: €10/User/Mt × 8 = €80/Mt
  - Custom Development (dbt Models, Dashboards): €28.000
  - **Total Jahr 1:** €29.920

- - models/fakt_verkauf.sql

WITH verkauf_raw AS (
  SELECT
    order_id,
    order_date,
    customer_id,
    product_id,
    quantity,
    unit_price,
    discount_pct,
    ROUND(quantity * unit_price * (1 - discount_pct/100), 2) AS net_revenue,
    shipping_cost,
    cogs  -- Cost of Goods Sold
  FROM {{ source('erp', 'orders') }}
  WHERE order_status = 'completed'
)

SELECT
  order_id,
  DATE(order_date) AS date_key,  -- FK zu dim_datum
  customer_id AS customer_key,    -- FK zu dim_kunde
  product_id AS product_key,      -- FK zu dim_produkt
  quantity,
  unit_price,
  discount_pct,
  net_revenue,
  shipping_cost,
  cogs,
  ROUND(net_revenue - cogs, 2) AS gross_profit,
  ROUND((net_revenue - cogs) / net_revenue * 100, 1) AS margin_pct
FROM verkauf_raw

# Production-Ready Sales Forecast

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import TimeSeriesSplit, cross_val_score
from sklearn.metrics import mean_absolute_percentage_error, mean_absolute_error
import joblib

class SalesForecastModel:
    def __init__(self):
        self.model = None
        self.feature_columns = []
        
    def prepare_features(self, df):
        """
        Feature Engineering für Sales Forecast
        """
        df = df.copy()
        df['date'] = pd.to_datetime(df['date'])
        df = df.sort_values('date')
        
        # Zeitbasierte Features
        df['year'] = df['date'].dt.year
        df['month'] = df['date'].dt.month
        df['quarter'] = df['date'].dt.quarter
        df['week_of_year'] = df['date'].dt.isocalendar().week
        df['day_of_week'] = df['date'].dt.dayofweek
        df['day_of_month'] = df['date'].dt.day
        
        # Feiertage (Deutschland)
        df['is_month_start'] = (df['day_of_month'] <= 5).astype(int)
        df['is_month_end'] = (df['day_of_month'] >= 25).astype(int)
        
        # Lag-Features (Historische Verkäufe)
        for lag in [7, 14, 30, 90]:
            df[f'sales_lag_{lag}d'] = df.groupby('product_id')['quantity'].shift(lag)
        
        # Rolling-Features (Durchschnitte)
        for window in [7, 30, 90]:
            df[f'sales_rolling_{window}d'] = (
                df.groupby('product_id')['quantity']
                .rolling(window, min_periods=1)
                .mean()
                .reset_index(level=0, drop=True)
            )
        
        # Preis-Features
        df['price_change_pct'] = df.groupby('product_id')['unit_price'].pct_change() * 100
        
        # Produkt-Features (aus dim_produkt)
        # df = df.merge(product_dims[['product_id', 'category', 'supplier', 'lifecycle_stage']], 
        #               on='product_id', how='left')
        
        return df
    
    def train(self, historical_sales_df):
        """
        Trainiert das Forecast-Modell
        """
        # Feature Engineering
        df_features = self.prepare_features(historical_sales_df)
        
        # Features & Target
        feature_cols = [
            'year', 'month', 'quarter', 'week_of_year', 'day_of_week',
            'is_month_start', 'is_month_end',
            'sales_lag_7d', 'sales_lag_14d', 'sales_lag_30d', 'sales_lag_90d',
            'sales_rolling_7d', 'sales_rolling_30d', 'sales_rolling_90d',
            'unit_price', 'price_change_pct'
            # + kategorie-dummies falls vorhanden
        ]
        
        # NaN-Behandlung (erste Wochen haben keine Lags)
        df_train = df_features[feature_cols + ['quantity']].dropna()
        
        X = df_train[feature_cols]
        y = df_train['quantity']
        
        self.feature_columns = feature_cols
        
        # Model Training
        self.model = RandomForestRegressor(
            n_estimators=500,
            max_depth=12,
            min_samples_split=10,
            min_samples_leaf=4,
            n_jobs=-1,
            random_state=42
        )
        
        # Time-Series Cross-Validation
        tscv = TimeSeriesSplit(n_splits=5)
        cv_scores = cross_val_score(
            self.model, X, y, 
            cv=tscv, 
            scoring='neg_mean_absolute_percentage_error',
            n_jobs=-1
        )
        
        print(f"Cross-Val MAPE: {-cv_scores.mean():.2%} (+/- {cv_scores.std():.2%})")
        
        # Final Training auf allen Daten
        self.model.fit(X, y)
        
        # Feature Importance
        feature_importance = pd.DataFrame({
            'feature': feature_cols,
            'importance': self.model.feature_importances_
        }).sort_values('importance', ascending=False)
        
        print("\nTop 5 Features:")
        print(feature_importance.head())
        
        return {
            'cv_mape': -cv_scores.mean(),
            'feature_importance': feature_importance
        }
    
    def forecast(self, future_dates_df):
        """
        Generiert Prognosen für zukünftige Daten
        """
        if self.model is None:
            raise ValueError("Modell muss erst trainiert werden!")
        
        # Feature Engineering für Forecast-Daten
        df_forecast = self.prepare_features(future_dates_df)
        
        X_forecast = df_forecast[self.feature_columns]
        
        # Prediction
        predictions = self.model.predict(X_forecast)
        
        # Negative Werte auf 0 setzen
        predictions = np.maximum(predictions, 0)
        
        df_forecast['predicted_quantity'] = predictions.round(0)
        
        return df_forecast[['date', 'product_id', 'predicted_quantity']]
    
    def save(self, filepath):
        """Modell speichern"""
        joblib.dump({
            'model': self.model,
            'features': self.feature_columns
        }, filepath)
    
    def load(self, filepath):
        """Modell laden"""
        saved = joblib.load(filepath)
        self.model = saved['model']
        self.feature_columns = saved['features']


# Training & Deployment
forecast_model = SalesForecastModel()

# Historische Daten laden (3 Jahre)
historical_sales = pd.read_sql("""
    SELECT date_key AS date, product_key AS product_id, quantity, unit_price
    FROM fakt_verkauf
    WHERE date_key >= '2022-01-01'
""", engine)

# Training
results = forecast_model.train(historical_sales)
print(f"Modell trainiert! Cross-Val MAPE: {results['cv_mape']:.2%}")

# Forecast für nächste 30 Tage
future_dates = pd.DataFrame({
    'date': pd.date_range(start='today', periods=30, freq='D'),
    'product_id': [p_id for _ in range(30) for p_id in historical_sales['product_id'].unique()]
})

forecast = forecast_model.forecast(future_dates)

# In Data Warehouse speichern
forecast.to_sql('ml_sales_forecast', engine, if_exists='replace', index=False)

# Modell speichern
forecast_model.save('models/sales_forecast_v2.joblib')

print("Forecast erfolgreich! Daten in 'ml_sales_forecast' Tabelle.")

Prinzip:
  - Nur notwendige Felder in Data Warehouse
  - Personendaten pseudonymisieren (z.B. Kunden-ID statt Name)
  - Nach 3 Jahren automatisch anonymisieren

Umsetzung:
  - dbt-Job: "anonymize_old_data.sql" (läuft monatlich)
  - Entfernt: Name, Adresse, E-Mail (bei Daten >3 Jahre)
  - Behält: Kunden-ID, Umsatz, Produkt (für Analytics)