Zeker, hier is een geavanceerd preprocessing schema voor uw tijdreeksdata, specifiek ontworpen voor ARIMA-modellering met lage latentie eisen. ### **Geavanceerd Preprocessing Schema voor Retail Verkoopdata** **Kernprincipe:** Pipeline-ontwerp voor realtime verwerking met focus op herbruikbaarheid en lage latentie. --- #### **1. Data Acquisition & Initial Checks (Realtime Ingestion)** ```python # Pseudocode voor realtime data-ingang def stream_data_point(new_daily_sale): # Valideer binnenkomend datapunt (bv. niet-negatief, realistisch bereik) if not validate_point(new_daily_sale): log_error("Ongeldig datapunt ontvangen") return None # Voeg direct toe aan rolling buffer (bijv. laatste 30 dagen in cache) add_to_rolling_buffer(new_daily_sale) # Activeer preprocessing pipeline return preprocess_point(new_daily_sale) ``` #### **2. Ontbrekende Waarden Interpolatie (Realtime)** **Strategie:** Gebruik **gewogen seizoensgebonden interpolatie** voor lage latentie. ```python def interpolate_missing(value_buffer): # value_buffer = rolling window van laatste N dagen (bijv. 30 dagen) if None not in value_buffer: return value_buffer[-1] # Geen actie nodig # Seizoensindex (zelfde dag vorige week heeft hoog gewicht) season_idx = -7 # week seizoenspatroon weights = [0.1, 0.1, 0.2, 0.6] # gewichten: [t-3, t-2, t-1, seizoen] # Bereken gewogen gemiddelde uit recente en seizoenswaarden valid_vals = [ value_buffer[-4] if value_buffer[-4] is not None else 0, value_buffer[-3] if value_buffer[-3] is not None else 0, value_buffer[-2] if value_buffer[-2] is not None else 0, value_buffer[season_idx] if value_buffer[season_idx] is not None else 0 ] # Gewogen interpolatie interpolated = sum(val * weight for val, weight in zip(valid_vals, weights)) return interpolated ``` #### **3. Seizoenscomponent Verwijderen (Realtime STL Decompositie)** **Strategie:** **Rolling STL (Seasonal-Trend decomposition using Loess)** met geoptimaliseerde window grootte. ```python # Initialisatie (eens uitvoeren bij startup) from statsmodels.tsa.seasonal import STL import numpy as np seasonal_window = 30 # rolling window voor seizoensschatting def initialize_seasonal_component(initial_data): # Bereken initieel seizoenspatroon op historische data stl = STL(initial_data, period=7, seasonal=13) # period=7 voor weekly seizoen res = stl.fit() return res.seasonal[-7:] # retourneer laatste 7 dagen als basispatroon # Realtime seizoensverwijdering def remove_seasonality(new_point, current_seasonal_pattern): # Bepaal dag van de week (0=maandag, 6=zondag) day_of_week = datetime.now().weekday() # Verwijder seizoenscomponent deseasoned_point = new_point - current_seasonal_pattern[day_of_week] return deseasoned_point ``` #### **4. Stationarity Check & Transformation (Rolling Basis)** **Strategie:** **Automatische differencing** met ADF test in rolling window. ```python from statsmodels.tsa.stattools import adfuller def check_stationarity(data_window): # Voer Augmented Dickey-Fuller test uit op rolling window result = adfuller(data_window) return result[1] < 0.05 # p-value < 0.05 = stationair def make_stationary(data_buffer): # Controleer stationariteit if check_stationarity(data_buffer): return data_buffer[-1], 0 # Geen differencing nodig # Pas eerste-order differencing toe (standaard voor ARIMA) stationary_value = data_buffer[-1] - data_buffer[-2] return stationary_value, 1 # retourneer waarde + differencing order ``` #### **5. Complete Realtime Processing Pipeline** ```python class RealtimeSalesPreprocessor: def __init__(self, historical_data): self.buffer = list(historical_data[-30:]) # rolling buffer self.seasonal_pattern = initialize_seasonal_component(historical_data) self.diff_order = 0 def process_new_point(self, new_sale): # Stap 1: Interpolatie indien nodig if new_sale is None: new_sale = interpolate_missing(self.buffer) # Stap 2: Update buffer self.buffer.pop(0) self.buffer.append(new_sale) # Stap 3: Seizoensverwijdering deseasoned = remove_seasonality(new_sale, self.seasonal_pattern) # Stap 4: Maak stationair stationary, self.diff_order = make_stationary(self.buffer) return stationary ``` --- ### **Optimalisaties voor Lage Latentie:** 1. **Pre-allocated Buffers:** Vaste array grootte i.p.v. dynamische lists 2. **Cached Seizoenspatroon:** Wekelijkse update i.p.v. realtime berekening 3. **Parallel Processing:** Multithreading voor STL decompositie 4. **Incremental ADF:** Benaderende stationariteitstests i.p.v. volledige ADF ### **Implementatie Advies:** ```python # Gebruiksvoorbeeld preprocessor = RealtimeSalesPreprocessor(historical_sales_data) # Voor elke nieuwe realtime meting: processed_value = preprocessor.process_new_point(daily_sales_today) # processed_value is klaar voor ARIMA voorspelling met latentie < 100ms ``` Dit schema balanceert nauwkeurigheid met performance voor realtime toepassingen, en produceert stationaire data die direct bruikbaar is voor ARIMA-modellering.