xintraday_vol.py

from __future__ import annotations
import argparse
import numpy as np
import pandas as pd
from io_intraday import read_intraday_csv
from adjust import split_ratio, add_adjustment_factors, build_daily_table
from vol_estimators import (RETURN_SCALING, TRADING_DAYS, bipower_variation,
    medrv_variation, realized_range_variation, intraday_realized_measures,
    compute_daily_vol_measures)
from statistics import summary_stats
from report import clean_vol_label

def acf_table(df: pd.DataFrame, columns: list[str], nlags: int = 5) -> pd.DataFrame:
    """Return autocorrelations by lag for selected columns."""
    rows = {}
    for col in columns:
        s = df[col].dropna()
        rows[clean_vol_label(col)] = {
            str(lag): s.autocorr(lag=lag) for lag in range(1, nlags + 1)
        }
    out = pd.DataFrame.from_dict(rows, orient="index")
    out.columns.name = "lag"
    return out

def open_close_correlation_table(df: pd.DataFrame) -> pd.DataFrame:
    """Return lead-lag correlations of ret_oc with ret_co."""
    pairs = {
        "ret_oc with previous-day ret_co": (df["ret_oc"], df["ret_co"].shift(1)),
        "ret_oc with next-day ret_co": (df["ret_oc"], df["ret_co"].shift(-1)),
    }
    out = pd.DataFrame(index=pairs.keys(), columns=["corr"], dtype=float)
    for label, (x, y) in pairs.items():
        tmp = pd.concat([x, y], axis=1).dropna()
        out.loc[label, "corr"] = tmp.iloc[:, 0].corr(tmp.iloc[:, 1])
    return out

def vol_correlation_matrix(df: pd.DataFrame, columns: list[str]) -> pd.DataFrame:
    """Return correlation matrix of daily volatility measures."""
    corr = df[columns].corr()
    clean_labels = [clean_vol_label(c) for c in columns]
    corr.index = clean_labels
    corr.columns = clean_labels
    return corr

def forward_vol_series(df: pd.DataFrame, target_col: str, horizon: int) -> pd.Series:
    """Return the horizon-day forward average of target_col.

    For horizon=1 this is just df[target_col].  For horizon>1 it is the
    mean of df[target_col] over [t, t+horizon-1], computed as
    rolling(horizon).mean().shift(-(horizon-1)).
    """
    return df[target_col].rolling(horizon).mean().shift(-(horizon - 1))

def vol_lead_lag_correlation_table(
    df: pd.DataFrame, target_col: str, columns: list[str], nlags: int = 5,
    horizon: int = 1,
) -> pd.DataFrame:
    """Return correlations of forward target_col with lagged values of all columns."""
    rows = {}
    target = forward_vol_series(df, target_col, horizon)
    for col in columns:
        corrs = {}
        for lag in range(1, nlags + 1):
            lagged = df[col].shift(lag)
            tmp = pd.concat([target, lagged], axis=1).dropna()
            corrs[str(lag)] = tmp.iloc[:, 0].corr(tmp.iloc[:, 1])
        rows[clean_vol_label(col)] = corrs
    out = pd.DataFrame.from_dict(rows, orient="index")
    out.columns.name = "lag"
    return out

def vol_ma_correlation_table(
    df: pd.DataFrame, target_col: str, columns: list[str], nlags: int = 5,
    horizon: int = 1,
) -> pd.DataFrame:
    """Return correlations of forward target_col with lagged rolling means of all columns.

    Column k shows the correlation of the horizon-day forward target with the
    k-period moving average of each estimator ending at t-1, so window=1
    matches lag-1 in the lead-lag table.
    """
    rows = {}
    target = forward_vol_series(df, target_col, horizon)
    for col in columns:
        corrs = {}
        for window in range(1, nlags + 1):
            ma = df[col].rolling(window).mean().shift(1)
            tmp = pd.concat([target, ma], axis=1).dropna()
            corrs[str(window)] = tmp.iloc[:, 0].corr(tmp.iloc[:, 1])
        rows[clean_vol_label(col)] = corrs
    out = pd.DataFrame.from_dict(rows, orient="index")
    out.columns.name = "window"
    return out

def vol_lwma_correlation_table(
    df: pd.DataFrame, target_col: str, columns: list[str], nlags: int = 5,
    horizon: int = 1,
) -> pd.DataFrame:
    """Return correlations of forward target_col with lagged LWMA of all columns.

    Column k shows the correlation of the horizon-day forward target with the
    k-period linearly weighted moving average of each estimator ending at t-1.
    Weights decline linearly from k (most recent) to 1 (oldest), normalized to
    sum to 1.  Window=1 is identical to lag-1 in the lead-lag table.
    """
    rows = {}
    target = forward_vol_series(df, target_col, horizon)
    for col in columns:
        corrs = {}
        for window in range(1, nlags + 1):
            weights = np.arange(1, window + 1, dtype=float)  # oldest=1 ... newest=window
            weights /= weights.sum()
            lwma = (
                df[col]
                .rolling(window)
                .apply(lambda x, w=weights: np.dot(x, w), raw=True)
                .shift(1)
            )
            tmp = pd.concat([target, lwma], axis=1).dropna()
            corrs[str(window)] = tmp.iloc[:, 0].corr(tmp.iloc[:, 1])
        rows[clean_vol_label(col)] = corrs
    out = pd.DataFrame.from_dict(rows, orient="index")
    out.columns.name = "window"
    return out

def print_results(
    daily: pd.DataFrame,
    vol_cols: list[str],
    nlags: int,
    head: int,
    print_daily_measures: bool,
    lag_target_cols: list[str] | None = None,
    forward_horizons: list[int] | None = None,
    print_vol_lead_lag=True
) -> None:
    """Print adjusted results tables."""
    if lag_target_cols is None:
        lag_target_cols = ["vol_cc_ann"]
    if forward_horizons is None:
        forward_horizons = [1]

    stats_df = summary_stats(daily, vol_cols)
    acf_df = acf_table(daily, vol_cols, nlags=nlags)
    corr_df = open_close_correlation_table(daily)
    vol_corr_df = vol_correlation_matrix(daily, vol_cols)

    if print_daily_measures:
        print("\nAdjusted daily volatility measures (first rows)\n")
        print(
            daily[
                vol_cols + ["n_bars", "dividends", "capital_gains", "split_ratio"]
            ].head(head).to_string(float_format=lambda x: f"{x:.3f}")
        )

    print("\nAdjusted summary statistics of annualized volatility measures\n")
    print(stats_df.to_string(float_format=lambda x: f"{x:.3f}"))

    print("\nAdjusted daily autocorrelations of annualized volatility measures\n")
    print(acf_df.to_string(float_format=lambda x: f"{x:.3f}"))

    print("\nAdjusted lead-lag correlations of ret_oc with ret_co\n")
    print(corr_df.to_string(float_format=lambda x: f"{x:.3f}"))

    print("\nCorrelation matrix of daily volatility measures\n")
    print(vol_corr_df.to_string(float_format=lambda x: f"{x:.3f}"))

    if print_vol_lead_lag:
        for target in lag_target_cols:
            label = clean_vol_label(target)
            for h in forward_horizons:
                hdr = f"{label} {h}-day-ahead vol"
                vol_lead_lag_df = vol_lead_lag_correlation_table(daily, target, vol_cols, nlags=nlags, horizon=h)
                print(f"\n{hdr} correlated with lagged volatility measures\n")
                print(vol_lead_lag_df.to_string(float_format=lambda x: f"{x:.3f}"))

                vol_ma_df = vol_ma_correlation_table(daily, target, vol_cols, nlags=nlags, horizon=h)
                print(f"\n{hdr} correlated with moving averages of volatility measures\n")
                print(vol_ma_df.to_string(float_format=lambda x: f"{x:.3f}"))

                vol_lwma_df = vol_lwma_correlation_table(daily, target, vol_cols, nlags=nlags, horizon=h)
                print(f"\n{hdr} correlated with linearly weighted moving averages of volatility measures\n")
                print(vol_lwma_df.to_string(float_format=lambda x: f"{x:.3f}"))

def main() -> None:
    """Parse arguments and run the analysis."""
    parser = argparse.ArgumentParser(
        description=(
            "Compute adjusted scaled daily and intraday annualized volatility measures from intraday prices."
        )
    )
    parser.add_argument("csv_file", help="path to intraday csv file")
    parser.add_argument("--nlags", type=int, default=20, help="number of autocorrelation lags")
    parser.add_argument(
        "--head", type=int, default=10, help="number of daily rows to print from the measures table"
    )
    parser.add_argument(
        "--print-daily",
        action="store_true",
        help="print the daily volatility measures table",
    )
    args = parser.parse_args()

    daily0 = add_adjustment_factors(build_daily_table(read_intraday_csv(args.csv_file)))
    print("data file:", args.csv_file)
    daily = compute_daily_vol_measures(daily0)
    daily_out_file = "daily_prices_vols.csv" # set to None not to write
    vol_cols = [
        "vol_cc_ann",
        "vol_cc_pos_ann",
        "vol_cc_neg_ann",
        "vol_co_oc_ann",
        "vol_co_ann",
        "vol_oc_ann",
        "vol_intraday_rv_ann",
        "vol_rsv_pos_ann",
        "vol_rsv_neg_ann",
        "vol_bv_ann",
        "vol_jump_ann",
        "vol_medrv_ann",
        "vol_realized_range_ann",
        "vol_subsampled_rv_ann",
        "vol_tsrv_ann",
        "vol_rk_ann",
        "vol_parkinson_ann",
        "vol_gk_ann",
        "vol_rs_ann",
        "vol_yz_ann",
    ]

    pd.set_option("display.width", 200)
    pd.set_option("display.max_columns", 40)
    pd.set_option("display.float_format", "{:.6f}".format)
    lag_target_cols = ["vol_cc_ann", "vol_intraday_rv_ann"] # vol measures for which lagged correlations printed
    forward_horizons = np.array([1, 5, 21])  # forward vol horizons in days

    if daily_out_file is not None:
        daily.to_csv(daily_out_file)
        print("wrote daily prices and vols to", daily_out_file)
    print_results(
        daily,
        vol_cols,
        nlags=args.nlags,
        head=args.head,
        print_daily_measures=args.print_daily,
        lag_target_cols=lag_target_cols,
        forward_horizons=forward_horizons,
    )

if __name__ == "__main__":
    import time
    t0 = time.time()
    main()
    print("\ntime elapsed (s):", "%0.2f"%(time.time() - t0))