Article.DataLibrary.Python.ReconstitutionEffect/index_rebalance_utils.py at main · LSEG-API-Samples/Article.DataLibrary.Python.ReconstitutionEffect · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
"""
index_rebalance_utils.py

Event study functions for measuring abnormal returns and volume around
dated corporate/index events.
"""

import pandas as pd
import numpy as np


def compute_market_adjusted_car(stock_returns, benchmark_returns, event_date, event_window=(-30, 30)):
    """Market-adjusted CAR: abnormal return on each day = stock return minus market return."""
    if event_date not in stock_returns.index:
        candidates = stock_returns.index[stock_returns.index >= event_date]
        if len(candidates) == 0:
            return None
        event_date = candidates[0]

    event_loc = stock_returns.index.get_loc(event_date)
    start_loc = event_loc + event_window[0]
    end_loc = event_loc + event_window[1] + 1

    if start_loc < 0 or end_loc > len(stock_returns):
        return None

    window_dates = stock_returns.index[start_loc:end_loc]
    stock_w = stock_returns.loc[window_dates]
    bench_w = benchmark_returns.reindex(window_dates)

    valid = stock_w.notna() & bench_w.notna()
    if valid.sum() < 30:
        return None

    stock_w = stock_w[valid]
    bench_w = bench_w[valid]

    ar = stock_w.values - bench_w.values
    car = np.cumsum(ar)
    relative_days = np.arange(event_window[0], event_window[0] + len(ar))

    return pd.DataFrame({"relative_day": relative_days, "AR": ar, "CAR": car})


def compute_abnormal_volume(volume, event_date, event_window=(-30, 30), baseline_window=(-150, -31)):
    """Compute volume as a multiple of its pre-event baseline average."""
    if event_date not in volume.index:
        candidates = volume.index[volume.index >= event_date]
        if len(candidates) == 0:
            return None
        event_date = candidates[0]

    event_loc = volume.index.get_loc(event_date)

    bl_start = event_loc + baseline_window[0]
    bl_end = event_loc + baseline_window[1] + 1
    if bl_start < 0:
        return None

    baseline = volume.iloc[bl_start:bl_end].dropna()
    if len(baseline) < 20:
        return None
    baseline_mean = baseline.mean()
    if baseline_mean == 0:
        return None

    ev_start = event_loc + event_window[0]
    ev_end = event_loc + event_window[1] + 1
    if ev_start < 0 or ev_end > len(volume):
        return None

    window = volume.iloc[ev_start:ev_end].dropna()
    if len(window) < 30:
        return None

    abnormal = window / baseline_mean
    relative_days = np.arange(event_window[0], event_window[0] + len(abnormal))

    return pd.DataFrame({"relative_day": relative_days, "abnormal_volume": abnormal.values})


def aggregate_events(event_results, value_col):
    """Average individual event results by relative day, with confidence bands."""
    if not event_results:
        return pd.DataFrame()

    all_data = pd.concat(event_results, ignore_index=True)
    grouped = all_data.groupby("relative_day")[value_col]

    agg = pd.DataFrame({
        "relative_day": grouped.mean().index,
        "mean": grouped.mean().values,
        "median": grouped.median().values,
        "std": grouped.std().values,
        "count": grouped.count().values,
        "q25": grouped.quantile(0.25).values,
        "q75": grouped.quantile(0.75).values,
    })

    agg["ci_upper"] = agg["mean"] + 1.96 * agg["std"] / np.sqrt(agg["count"])
    agg["ci_lower"] = agg["mean"] - 1.96 * agg["std"] / np.sqrt(agg["count"])

    return agg.sort_values("relative_day").reset_index(drop=True)