app_tier.py

import pandas as pd
import pmdarima as pm
import streamlit as st
from statsmodels.tsa.stattools import acf
import sklearn.metrics
import math

def analyze_seasonality(data):
    if len(data) < 14:
        return False
    
    # Calculate the Autocorrelation Function (ACF) up to lag 8
    # The acf function returns an array where index 0 is lag 0, index 1 is lag 1, etc.
    acf_values = acf(data, nlags=8)
    
    # If the correlation at lag 7 is greater than 0.6, return True
    if len(acf_values) > 7 and acf_values[7] > 0.6:
        return True
        
    return False

@st.cache_data
def generate_forecast(data, forecast_horizon):
    has_seasonality = analyze_seasonality(data)
    
    if has_seasonality:
        # m=7 only needs ~4 months to find a weekly pattern.
        data = data.tail(120)
    else:
        # Standard ARIMA only needs recent trend momentum (~3 months).
        data = data.tail(90)
        
    if len(data) <= 30:
        raise ValueError("Dataset requires more than 30 data points to execute the backtest phase.")
        
    # --- Live Backtest Phase ---
    # Split the data, hiding the last 30 days
    train_data = data.iloc[:-30]
    test_data = data.iloc[-30:]
    
    # Train pm.auto_arima on the rest
    backtest_model = pm.auto_arima(
        train_data,
        seasonal=has_seasonality,
        m=7 if has_seasonality else 1,
        suppress_warnings=True,
        stepwise=True,
        max_p=2, 
        max_q=2, 
        max_d=1,
        max_P=1,
        max_Q=1,
        max_D=1
    )
    
    # Predict the hidden 30 days
    backtest_predictions = backtest_model.predict(n_periods=30)
    
    # Calculate the custom MAE and RMSE
    custom_mae = sklearn.metrics.mean_absolute_error(test_data, backtest_predictions)
    custom_rmse = math.sqrt(sklearn.metrics.mean_squared_error(test_data, backtest_predictions))
    
    # --- Production Phase ---
    # Re-train pm.auto_arima on the entire dataset
    prod_model = pm.auto_arima(
        data,
        seasonal=has_seasonality,
        m=7 if has_seasonality else 1,
        suppress_warnings=True,
        stepwise=True,
        max_p=2, 
        max_q=2, 
        max_d=1,
        max_P=1,
        max_Q=1,
        max_D=1
    )
    
    # Predict the unknown future based on forecast_horizon
    future_forecast = prod_model.predict(n_periods=forecast_horizon)
    order = prod_model.order # (p,d,q)
    
    # --- Business Logic ---
    final_forecast_values = [val if val > 0 else 0 for val in future_forecast]
    
    # Convert back to a Series with a valid future DateTime index
    future_dates = pd.date_range(start=data.index[-1] + pd.Timedelta(days=1), periods=forecast_horizon, freq='D')
    final_forecast_series = pd.Series(final_forecast_values, index=future_dates, name='Forecast')
    
    return final_forecast_series, order, has_seasonality, custom_mae, custom_rmse