Update memory.py and task_memory.py to compute ACF and Hurst exponent for differenced close price, and update environment file for dependencies.

Author: Lenr4

environment.yml

@@ -25,6 +25,7 @@ dependencies:
   - pandas >=2.2
   - plotly >=5.2.0,<6
   - yfinance  # added yfinance
+  - scikit-learn >=1.0
 
   # R template project dependencies
   - pytask-r >=0.4.1
@@ -35,4 +36,4 @@ dependencies:
   - r-forcats
 
   # Install project
-  - pip: [-e ., pdbp, kaleido, arch >=5.1]
+  - pip: [-e ., pdbp, kaleido, arch >=5.1, nolds]

src/lennart_epp/analysis/memory.py

@@ -1,3 +1,7 @@
+from pathlib import Path
+
+import nolds
+import numpy as np
 import pandas as pd
 
 from lennart_epp.analysis.fit_ar_model import _check_stationarity, _difference_series
@@ -76,3 +80,116 @@ def write_stationarity_results(results: dict, file_path):
 
     with file_path.open("w", encoding="utf-8") as f:
         f.write(latex_content.strip())
+
+
+def _compute_mean(series: np.ndarray) -> float:
+    """Compute the mean of a given time series.
+
+    Args:
+        series (np.ndarray): The time series data as a NumPy array.
+
+    Returns:
+        float: The mean of the time series.
+    """
+    return np.mean(series)
+
+
+def _compute_variance(series: np.ndarray, mean_series: float) -> float:
+    """Compute the variance of a time series.
+
+    Args:
+        series (np.ndarray): The time series data as a NumPy array.
+        mean_series (float): The precomputed mean of the series.
+
+    Returns:
+        float: The variance of the time series.
+    """
+    return np.sum((series - mean_series) ** 2)
+
+
+def _compute_autocovariance(series: np.ndarray, mean_series: float, lag: int) -> float:
+    """Compute the autocovariance for a given lag in a time series.
+
+    Args:
+        series (np.ndarray): The time series data as a NumPy array.
+        mean_series (float): The precomputed mean of the series.
+        lag (int): The lag at which to compute the autocovariance.
+
+    Returns:
+        float: The autocovariance value for the specified lag.
+    """
+    n = len(series)
+    return np.sum((series[lag:] - mean_series) * (series[: n - lag] - mean_series))
+
+
+def compute_acf(
+    df: pd.DataFrame, column: str = "close_price", lags: int = 1000
+) -> dict:
+    """Compute the ACF manually for the first-differenced time series.
+
+    Args:
+        df (pd.DataFrame): The dataframe containing the time series.
+        column (str, optional): Column to analyze. Defaults to "close_price".
+        lags (int, optional): Number of lags for ACF.
+
+    Returns:
+        dict: A dictionary containing ACF values and corresponding lags.
+    """
+    df_diff = _difference_series(df, column)
+    series = df_diff[f"diff_{column}"].dropna().to_numpy()
+
+    lags = min(len(series) - 1, lags)
+    mean_series = _compute_mean(series)
+    variance = _compute_variance(series, mean_series)
+
+    acf_values = []
+    for lag in range(lags + 1):
+        autocovariance = _compute_autocovariance(series, mean_series, lag)
+        acf_values.append(autocovariance / variance)
+
+    return {"acf": np.array(acf_values), "lags": np.arange(len(acf_values))}
+
+
+def compute_hurst_exponent(df: pd.DataFrame, column: str = "close_price") -> dict:
+    """Compute the Hurst exponent to assess long-memory effects.
+
+    Args:
+        df (pd.DataFrame): The dataframe containing the time series.
+        column (str, optional): Column to analyze. Defaults to "close_price".
+
+    Returns:
+        dict: A dictionary containing the computed Hurst exponent.
+    """
+    series = df[column].dropna().to_numpy()
+    hurst_value = nolds.hurst_rs(series)
+
+    return {"Hurst Exponent": hurst_value}
+
+
+def write_hurst_result_to_tex(results: dict, file_path: Path):
+    """Write the computed Hurst exponent results to a LaTeX file.
+
+    Args:
+        results (dict): Dictionary containing the Hurst exponent.
+        file_path (Path): Path where the LaTeX file will be saved.
+    """
+    file_path.parent.mkdir(parents=True, exist_ok=True)
+
+    hurst_value = results["Hurst Exponent"]
+
+    latex_content = f"""
+    \\begin{{table}}[H]
+        \\centering
+        \\caption{{Hurst Exponent Analysis}}
+        \\label{{tab:hurst_exponent}}
+        \\begin{{tabular}}{{l c}}
+            \\toprule
+            \\textbf{{Metric}} & \\textbf{{Value}} \\\\
+            \\midrule
+            Hurst Exponent & {hurst_value:.4f} \\\\
+            \\bottomrule
+        \\end{{tabular}}
+    \\end{{table}}
+    """
+
+    file_path.write_text(latex_content.strip(), encoding="utf-8")

src/lennart_epp/analysis/task_memory.py

@@ -2,6 +2,9 @@
 
 from lennart_epp.analysis.memory import (
     check_stat_diff_close,
+    compute_acf,
+    compute_hurst_exponent,
+    write_hurst_result_to_tex,
     write_stationarity_results,
 )
 from lennart_epp.config import BLD
@@ -27,3 +30,49 @@ def task_check_stat_diff_close(
     write_stationarity_results(results, produces)
 
     assert produces.exists(), f" Failed to produce {produces}"
+
+
+def task_compute_acf(
+    data=BLD / "data" / "cleaned_apple_data.pkl",
+    produces=BLD / "memory" / "acf.pkl",
+):
+    """Task to compute and save ACF values manually for the differenced close price.
+
+    Args:
+        data (Path): Path to the cleaned Apple stock data (Pickle file).
+        produces (Path): Path to output pickle file where the ACF results are stored.
+
+    Returns:
+        None: Saves ACF values to a .pkl file.
+    """
+    df = pd.read_pickle(data)
+
+    results = compute_acf(df, column="close_price")
+
+    produces.parent.mkdir(parents=True, exist_ok=True)
+    pd.to_pickle(results, produces)
+
+    assert produces.exists(), f"Failed to produce {produces}"
+
+
+def task_hurst_exponent(
+    data=BLD / "data" / "cleaned_apple_data.pkl",
+    produces=BLD / "memory" / "hurst_exponent.tex",
+):
+    """Task to compute the Hurst exponent and store results as LaTeX file.
+
+    Args:
+        data (Path): Path to the cleaned Apple stock data.
+        produces (Path): Path to the output LaTeX file.
+
+    Returns:
+        None: Saves results as a LaTeX file.
+    """
+    df = pd.read_pickle(data)
+    df["diff_close_price"] = df["close_price"].diff().dropna()
+
+    results = compute_hurst_exponent(df, column="diff_close_price")
+
+    write_hurst_result_to_tex(results, produces)
+
+    assert produces.exists(), f"Failed to produce {produces}"