entropy_forecast_audit.py

 # entropy_forecast_audit.py

"""

OPHI Entropy Forecast Audit

Author: Luis Ayala (Kp Kp)

Version: 1.0.0

This script benchmarks the OPHI entropy signal (S-values) against ARIMA forecasts

using historical environmental or protest-related datasets.

"""

import pandas as pd

import numpy as np

import matplotlib.pyplot as plt

import seaborn as sns

import logging

import hashlib

import json

from datetime import datetime

from statsmodels.tsa.arima.model import ARIMA

from sklearn.metrics import mean_squared_error, mean_absolute_error

# === Logging Setup ===

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# === Seed Control ===

np.random.seed(216)

# === Version and Lineage Metadata ===

VERSION = "1.0.0"

TIMESTAMP = datetime.utcnow().isoformat()

SCRIPT_HASH = hashlib.sha256(open(__file__, 'rb').read()).hexdigest()

# === Load Historical Dataset ===

def load_dataset(path):

    df = pd.read_csv(path, parse_dates=['timestamp'])

    df.sort_values('timestamp', inplace=True)

    return df

# === Compute S and C Values ===

def compute_entropy_coherence(df, signal_col):

    df['entropy'] = df[signal_col].rolling(window=7).std().fillna(0)

    df['coherence'] = df[signal_col].rolling(window=7).apply(lambda x: 1 - x.std()/x.mean() if x.mean() != 0 else 0).fillna(0)

    return df

# === Forecast with ARIMA ===

def forecast_arima(df, signal_col, forecast_steps=7):

    model = ARIMA(df[signal_col], order=(5,1,0))

    model_fit = model.fit()

    forecast = model_fit.forecast(steps=forecast_steps)

    return forecast

# === Compare Forecasts ===

def evaluate_forecast(true_values, predictions):

    return {

        'MAE': mean_absolute_error(true_values, predictions),

        'RMSE': np.sqrt(mean_squared_error(true_values, predictions))

    }

# === Visualize Results ===

def plot_forecasts(df, forecast, signal_col):

    plt.figure(figsize=(10,6))

    plt.plot(df['timestamp'], df[signal_col], label='Actual')

    plt.plot(pd.date_range(df['timestamp'].iloc[-1], periods=len(forecast)+1, freq='D')[1:], forecast, label='ARIMA Forecast')

    plt.title('Signal vs Forecast')

    plt.legend()

    plt.grid(True)

    plt.tight_layout()

    plt.show()

# === Main Execution ===

def main():

    logging.info(f"Running OPHI audit version {VERSION}, timestamp {TIMESTAMP}")

    df = load_dataset('historical_signal_data.csv')

    df = compute_entropy_coherence(df, 'signal')

    # Forecast & evaluate

    forecast = forecast_arima(df, 'entropy')

    future_truth = df['entropy'].iloc[-7:].values  # Simulated future

    metrics = evaluate_forecast(future_truth, forecast)

    logging.info(f"Forecast Metrics: {json.dumps(metrics, indent=2)}")

    # Visualization

    plot_forecasts(df, forecast, 'entropy')

    # Data lineage

    lineage = {

        'version': VERSION,

        'timestamp': TIMESTAMP,

        'hash': SCRIPT_HASH,

        'source_file': 'historical_signal_data.csv'

    }

    with open('audit_lineage.json', 'w') as f:

        json.dump(lineage, f, indent=2)

if __name__ == '__main__':

    main()