""" Corporate actions as trading signals on US common stocks (FinLab research, 2026-11). Six event signals, each on a 3 x 3 grid fixed before any result was seen. The middle cell is the original setting. dividend_increase A company raises its regular dividend per share (split-adjusted, specials left out). Hold 60 / 120 / 240 trading days. Liquidity $5M / $20M / $50M. dividend_initiation A company pays a dividend after two years with none, and has at least two years of prices. Hold 60 / 120 / 240 days. Liquidity $1M / $5M / $20M. dividend_capture Buy 1 / 3 / 5 trading days before the ex-dividend date, sell at the close of the ex-dividend date. Only dividends already declared on the day the signal is read. Minimum dividend yield of that one payment 0.5% / 1% / 2%. Liquidity $20M. split_drift A forward split of at least 3-for-2 (split_factor >= 1.5). Hold 60 / 120 / 240 days. Liquidity $5M / $20M / $50M. sp500_addition A stock joins the S&P 500. Hold 20 / 60 / 120 days. At most 5% / 10% / 20% of the account in one stock. sp500_deletion A stock leaves the S&P 500. Same grid. Dates. A dividend is known on its declaration date, or on its ex-dividend date when the source has no declaration date. A split is known on its split date. Index membership is known on the day the snapshot was taken: before November 2022 snapshots are days apart, so that date can lag the real announcement. A signal read at one day's close trades at the next day's close (finlab sim default). Universe. Common stocks, delisted ones included, one share class per company, no ETFs, funds or ADRs. Dividend and split signals also leave out companies whose profile names a country other than the US (their dividends are translated into dollars, so an unchanged payment can look like a raise). On the day of the signal a stock must close above $5, have traded more than the liquidity threshold per day on average over 60 days (raw close x volume) and have no +/-50% day in the past year. The S&P 500 signals use index membership in place of these filters. Portfolio. Equal weight across the stocks currently held, at most 5% in one stock unless the grid says otherwise, so the rest is cash when fewer than 20 stocks qualify. Prices include dividends. "Beats" means a higher daily Sharpe ratio (no risk-free rate) than QQQ with dividends over the same dates. A signal passes only if 1. at least 7 of its 9 settings beat QQQ at sim's default cost, 2. its original setting beats QQQ in both halves of the period, 3. its original setting beats QQQ at 0.1% per side, 4. its original setting beats QQQ on returns above T-bills (BIL), 5. its original setting beats QQQ without the calendar year of its largest lead. Besides the backtests, the script measures the events themselves: the average return after each event against QQQ, the size of the price drop on ex-dividend days against the dividend, and the path of S&P 500 additions and deletions around the snapshot date. Run: pip install "finlab>=2.3.8" python research.py writes metrics.json and data.csv next to this file, and the dividend capture and S&P 500 sections next to their articles (../dividend-capture-strategy-backtest, ../sp-500-additions-backtest) python research.py --events prints event counts only """ import argparse import json from functools import cache from itertools import product from pathlib import Path import numpy as np import pandas as pd from finlab import data from finlab.backtest import sim START = "2016-07-01" INITIATION_START = "2018-01-02" # dividend history starts in 2016 and the signal needs two years of it END = "2026-11-06" OUT = Path(__file__).resolve().parent CAPTURE_OUT = OUT.parent / "dividend-capture-strategy-backtest" INDEX_OUT = OUT.parent / "sp-500-additions-backtest" COST_PER_SIDE = 0.001 HORIZONS = (1, 5, 20, 60, 120, 240) REFERENCES = ("QQQ", "SPY", "RSP") YIELD_BUCKETS = ((0.0, 0.005), (0.005, 0.01), (0.01, 0.02), (0.02, 1.0)) MAX_WEIGHT = 0.05 MIN_RAISE = 1.01 # below 1%, a "raise" can be rounding SPECIAL_MULTIPLE = 2.0 # a payment of 2x the median of the last four regular ones is a special MIN_GAP_DAYS = 20 # a payment this soon after the last one is an extra MAX_GAP_DAYS = 400 # a longer gap is a resumption, not a raise NO_DIVIDEND_DAYS = 730 MIN_PRICE_HISTORY = 504 RENAME_WINDOW_DAYS = 7 RENAME_CORRELATION = 0.97 LIQ = (5e6, 20e6, 50e6) # Ticker changes and mergers where the old ticker has no price history to compare with SAME_COMPANY = {("MYL", "VTRS"), ("ANTM", "ELV"), ("PX", "LIN"), ("VIAB", "VIAC"), ("BK", "BNY")} CAPTURE_LIQUIDITY = 20e6 INDEX_SIGNALS = ("sp500_addition", "sp500_deletion") SNAPSHOT_PERIODS = {"before_daily_snapshots": (START, "2022-11-07"), "daily_snapshots": ("2022-11-08", END)} GRIDS = { "dividend_increase": {"hold": (60, 120, 240), "liquidity": LIQ}, "dividend_initiation": {"hold": (60, 120, 240), "liquidity": (1e6, 5e6, 20e6)}, "dividend_capture": {"lead": (1, 3, 5), "min_yield": (0.005, 0.01, 0.02)}, "split_drift": {"hold": (60, 120, 240), "liquidity": LIQ}, "sp500_addition": {"hold": (20, 60, 120), "max_weight": (0.05, 0.10, 0.20)}, "sp500_deletion": {"hold": (20, 60, 120), "max_weight": (0.05, 0.10, 0.20)}, } data.set_market("us") data.truncate_end = END @cache def get(key): return data.get(key) def adj(): return get("us_price:adj_close") def days(): return adj().index def next_trading_day(dates): """Position on the trading calendar of each date (the same day, or the next trading day).""" return np.searchsorted(days().values, pd.DatetimeIndex(dates).values, side="left") @cache def common_stocks(): """Common stocks (delisted ones included), one share class per company.""" p = pd.DataFrame(get("us_company_profile")).astype({"stock_id": str}).drop_duplicates("stock_id") keep = ( (p["security_type"] == "common_stock") & ~p["is_etf"].fillna(False).astype(bool) & ~p["is_fund"].fillna(False).astype(bool) & ~p["is_adr"].fillna(False).astype(bool) ) rows = p.loc[keep, ["stock_id", "company_name", "country"]].astype(str).sort_values("stock_id") rows = rows.drop_duplicates("company_name") return rows[rows["stock_id"].isin(adj().columns)] @cache def stock_ids(): return sorted(common_stocks()["stock_id"]) @cache def us_stock_ids(): """The common stocks whose profile names no country other than the US.""" rows = common_stocks() return sorted(rows.loc[rows["country"].isin(["US", "nan"]), "stock_id"]) @cache def dollar_volume(): close, volume = get("us_price:close"), get("us_price:volume") return (close * volume).rolling(60, min_periods=20).mean() @cache def price_clean(): """Close above $5 and no +/-50% day in the past year.""" daily = adj().pct_change(fill_method=None) extreme = (daily.abs() >= 0.5).astype(float).rolling(252, min_periods=1).max().shift(1).fillna(0) > 0 return (get("us_price:close") > 5) & ~extreme def tradable(liquidity): """Boolean frame on the trading calendar and all stock ids.""" mask = price_clean() & (dollar_volume() > liquidity) return pd.DataFrame(mask).reindex(index=days(), columns=stock_ids()).fillna(False).astype(bool) def blank(): return pd.DataFrame(False, index=days(), columns=stock_ids()) def mark(events, rows, symbols): """Set the (row, symbol) cells that exist.""" columns = events.columns.get_indexer(symbols) rows = np.asarray(rows) ok = (columns >= 0) & (rows >= 0) & (rows < len(events)) values = events.values.copy() values[rows[ok], columns[ok]] = True return pd.DataFrame(values, index=events.index, columns=events.columns) def hold(events, length): """True from each event day for `length` trading days.""" return events.astype(float).rolling(length, min_periods=1).max() > 0 # --------------------------------------------------------------------------- # Dividends and splits # --------------------------------------------------------------------------- @cache def actions(): return pd.DataFrame(get("us_corporate_actions")).astype({"symbol": str}) @cache def dividends(): """Every cash dividend of the US common stocks, with split-adjusted amount and known-on date.""" a = actions() d = a[(a["dividend"] > 0) & a["symbol"].isin(us_stock_ids())].sort_values(["date", "symbol"]).copy() splits = a[a["split_factor"] > 0].sort_values(["symbol", "date"]).copy() splits["cum_split"] = splits.groupby("symbol", observed=True)["split_factor"].cumprod() splits = splits.sort_values("date") d = pd.merge_asof(d, splits[["symbol", "date", "cum_split"]], on="date", by="symbol", direction="backward") d["cum_split"] = d["cum_split"].fillna(1.0) d["adj_amount"] = d["dividend"] * d["cum_split"] known = d["declaration_date"].where(d["declaration_date"].notna() & (d["declaration_date"] <= d["date"]), d["date"]) d["known"] = known return d.sort_values(["symbol", "date"]).reset_index(drop=True) @cache def classified(): """Each dividend labelled raise / initiation, specials removed from the raise baseline.""" d = dividends() raises, initiations = [], [] for symbol, group in d.groupby("symbol", observed=True): regular, last_date, previous_any = [], None, None for row in group.itertuples(): gap_any = (row.date - previous_any).days if previous_any is not None else None previous_any = row.date if gap_any is None or gap_any > NO_DIVIDEND_DAYS: initiations.append((symbol, row.known)) if gap_any is not None and gap_any < MIN_GAP_DAYS: continue if regular and row.adj_amount >= SPECIAL_MULTIPLE * float(np.median(regular[-4:])): continue if regular and (row.date - last_date).days <= MAX_GAP_DAYS and row.adj_amount > regular[-1] * MIN_RAISE: raises.append((symbol, row.known)) regular.append(row.adj_amount) last_date = row.date return raises, initiations @cache def long_history(): enough = pd.DataFrame(adj().notna().cumsum() >= MIN_PRICE_HISTORY) return enough[stock_ids()] def known_events(pairs): events = blank() if pairs: symbols, dates = zip(*pairs) events = mark(events, next_trading_day(list(dates)), list(symbols)) return events def declared_dividends(lead): """(column, ex-date row, signal row, amount) of each dividend declared and tradable on its signal day.""" d = dividends() calendar = days() columns = blank().columns.get_indexer(d["symbol"]) ok_values = tradable(CAPTURE_LIQUIDITY).values for col, ex, dividend, declared in zip(columns, next_trading_day(d["date"]), d["dividend"], d["declaration_date"]): signal = ex - lead - 1 if signal < 0 or ex >= len(calendar): continue if pd.isna(declared) or declared > calendar[signal] or not ok_values[signal, col]: continue yield col, ex, signal, dividend def capture_position(lead, min_yield): """Signal days ex - lead - 1 ... ex - 2: bought at the close of ex - lead, sold at the close of ex.""" position = blank() close_values = pd.DataFrame(get("us_price:close")).reindex(index=days(), columns=position.columns).values out = np.zeros(position.shape, dtype=bool) for col, ex, signal, dividend in declared_dividends(lead): if dividend / close_values[signal, col] >= min_yield: out[signal:ex - 1, col] = True return pd.DataFrame(out, index=position.index, columns=position.columns) # --------------------------------------------------------------------------- # S&P 500 additions and deletions # --------------------------------------------------------------------------- @cache def members(): return get("us_index_constituents:sp500").fillna(False).astype(bool) @cache def list_changes(): """(date, symbol) of every addition to and deletion from the S&P 500 list.""" change = members().astype(int).diff().loc[START:END] adds = [(date, s) for date in change.index for s in change.columns[change.loc[date] == 1]] drops = [(date, s) for date in change.index for s in change.columns[change.loc[date] == -1]] return adds, drops @cache def index_changes(): """Additions and deletions, without ticker changes and same-company reorganisations.""" adds, drops = list_changes() returns = adj().pct_change(fill_method=None) near = pd.Timedelta(days=RENAME_WINDOW_DAYS) fake_adds, fake_drops, renamed = set(), set(), [] for add_date, added in adds: for drop_date, dropped in drops: if abs(add_date - drop_date) > near: continue same = added == dropped or (dropped, added) in SAME_COMPANY if not same and added in returns and dropped in returns: before = returns.loc[:min(add_date, drop_date)].iloc[-60:] pair = before[[added, dropped]].dropna() same = len(pair) >= 20 and pair[added].corr(pair[dropped]) >= RENAME_CORRELATION if same: fake_adds.add((add_date, added)) fake_drops.add((drop_date, dropped)) renamed.append(f"{dropped}->{added}") return [e for e in adds if e not in fake_adds], [e for e in drops if e not in fake_drops], renamed def index_events(pairs): """Trading days of the given list changes, on the stocks with a price that day.""" dates, symbols = zip(*pairs) columns = sorted(set(adj().columns).intersection(symbols)) events = pd.DataFrame(False, index=days(), columns=columns) events = mark(events, next_trading_day(list(dates)), list(symbols)) return events & pd.DataFrame(adj()[columns].notna()) # --------------------------------------------------------------------------- # Backtest and checks # --------------------------------------------------------------------------- def signal_events(signal, liquidity=None): """Days a signal fires, before the holding period. The S&P 500 signals ignore `liquidity`.""" if signal == "dividend_increase": raises, _ = classified() return known_events(raises) & tradable(liquidity) if signal == "dividend_initiation": _, initiations = classified() events = known_events(initiations) start = np.searchsorted(days().values, np.datetime64(INITIATION_START)) events.iloc[:start] = False return events & tradable(liquidity) & long_history() if signal == "split_drift": a = actions() forward = a[(a["split_factor"] >= 1.5) & a["symbol"].isin(us_stock_ids())] return mark(blank(), next_trading_day(forward["date"]), forward["symbol"].tolist()) & tradable(liquidity) adds, drops, _ = index_changes() return index_events({"sp500_addition": adds, "sp500_deletion": drops}[signal]) def position_for(signal, settings): """Daily holdings of one grid cell and its weight limit.""" first, second = settings if signal == "dividend_capture": return capture_position(first, second), MAX_WEIGHT if signal in INDEX_SIGNALS: return hold(signal_events(signal), first), second return hold(signal_events(signal, second), first), MAX_WEIGHT def start_of(signal): return INITIATION_START if signal == "dividend_initiation" else START def original_settings(signal): """The middle cell of a signal's grid, by setting name.""" return {name: values[1] for name, values in GRIDS[signal].items()} def study_window(signal, position): """The position inside the study period, without the stocks it never holds.""" position = position.loc[start_of(signal):END] return position.loc[:, position.any()] def backtest(position, limit, fee_ratio=None): return sim(position, resample=None, position_limit=limit, upload=False, fee_ratio=fee_ratio) def equity(report): return report.creturn.loc[:END].dropna() @cache def fund(symbol): return get("us_fund_price:adj_close")[symbol].dropna() def on(series, index): return series.reindex(series.index.union(index)).ffill().reindex(index) def sharpe(returns): returns = returns.dropna() return float(returns.mean() / returns.std() * 252 ** 0.5) if returns.std() > 0 else float("nan") def curve_sharpe(curve): return sharpe(curve.pct_change()) def stats(curve): years = (curve.index[-1] - curve.index[0]).days / 365.25 daily = curve.pct_change() return { "start": str(curve.index[0].date()), "end": str(curve.index[-1].date()), "cagr": round(float((curve.iloc[-1] / curve.iloc[0]) ** (1 / years) - 1), 6), "sharpe": round(curve_sharpe(curve), 6), "mdd": round(float((curve / curve.cummax() - 1).min()), 6), "volatility": round(float(daily.std() * 252 ** 0.5), 6), } def halves(curve): middle = curve.index[len(curve) // 2] return curve.loc[:middle], curve.loc[middle:] def yearly(curve): ends = curve.resample("YE").last() return ends / ends.shift(1, fill_value=curve.iloc[0]) - 1 def without_best_year(curve, bench): lead = yearly(curve) - yearly(bench) best = int(lead.idxmax().year) keep = curve.index.year != best return {"year": best, "lead": round(float(lead.max()), 6), "sharpe": round(sharpe(curve.pct_change()[keep]), 6), "bench_sharpe": round(sharpe(bench.pct_change()[keep]), 6)} def turnover(position, limit): """One-way turnover of target weights per year: the account share bought per year.""" weights = position.astype(float).div(position.sum(axis=1).clip(lower=1 / limit), axis=0) years = (position.index[-1] - position.index[0]).days / 365.25 return float(weights.diff().abs().sum(axis=1).sum() / 2 / years) def study_signal(signal): original = original_settings(signal) cells, curves = {}, {} for settings in product(*GRIDS[signal].values()): position, limit = position_for(signal, settings) position = study_window(signal, position) report = backtest(position, limit) curve = equity(report) with_cost = equity(backtest(position, limit, COST_PER_SIDE)) curves[settings] = (curve, with_cost) same_bench = on(fund("QQQ"), curve.index) trades = pd.DataFrame(report.trades) held = position.sum(axis=1) cells["/".join(map(str, settings))] = { **stats(curve), "sharpe_at_cost": round(curve_sharpe(with_cost), 6), "cagr_at_cost": stats(with_cost)["cagr"], "bench_sharpe": round(curve_sharpe(same_bench), 6), "beats": bool(curve_sharpe(curve) > curve_sharpe(same_bench)), "invested": round(float(held.gt(0).mean()), 6), "avg_names": round(float(held.mean()), 2), "exposure": round(float(np.minimum(held * limit, 1).mean()), 6), "trades": int(len(trades)), "win_rate": round(float((trades["return"] > 0).mean()), 6), "mean_trade": round(float(trades["return"].mean()), 6), "median_trade": round(float(trades["return"].median()), 6), "turnover": round(turnover(position, limit), 3), } curve, cost_curve = curves[tuple(original.values())] index = curve.index bench = on(fund("QQQ"), index) first, second = halves(curve) first_bench, second_bench = halves(bench) half_sharpes = { "split": str(curve.index[len(curve) // 2].date()), "first": curve_sharpe(first), "second": curve_sharpe(second), "bench_first": curve_sharpe(first_bench), "bench_second": curve_sharpe(second_bench), } cost_01 = {"sharpe": curve_sharpe(cost_curve), "bench_sharpe": curve_sharpe(on(fund("QQQ"), cost_curve.index)), "cagr": stats(cost_curve)["cagr"]} bil = on(fund("BIL"), index).pct_change() above_bil = {"sharpe": sharpe(curve.pct_change() - bil), "bench_sharpe": sharpe(bench.pct_change() - bil)} wby = without_best_year(curve, bench) beat_count = sum(c["beats"] for c in cells.values()) checks = { "seven_of_nine": beat_count >= 7, "both_halves": half_sharpes["first"] > half_sharpes["bench_first"] and half_sharpes["second"] > half_sharpes["bench_second"], "cost_01": cost_01["sharpe"] > cost_01["bench_sharpe"], "above_bil": above_bil["sharpe"] > above_bil["bench_sharpe"], "without_best_year": wby["sharpe"] > wby["bench_sharpe"], } result = { "names": list(GRIDS[signal]), "grid": {k: list(v) for k, v in GRIDS[signal].items()}, "original": original, "cells": cells, "beat_count": beat_count, "original_stats": stats(curve), "bench_stats": stats(bench), "halves": half_sharpes, "cost_01": cost_01, "above_bil": above_bil, "without_best_year": wby, "reference": {sym: stats(on(fund(sym), index)) for sym in REFERENCES}, "yearly": {str(k.year): round(float(v), 6) for k, v in yearly(curve).items()}, "bench_yearly": {str(k.year): round(float(v), 6) for k, v in yearly(bench).items()}, "checks": checks, "passed": all(checks.values()), } return result, curve, bench # --------------------------------------------------------------------------- # What happens after the events # --------------------------------------------------------------------------- def filled_prices(columns): return pd.DataFrame(adj())[columns].ffill().values def event_cells(signal): """Trading-day row and stock column of each event at the original setting's universe.""" events = signal_events(signal, original_settings(signal).get("liquidity")).loc[start_of(signal):END] rows, cols = np.nonzero(events.values) return rows + days().get_loc(events.index[0]), cols, list(events.columns) def t_by_date(values, rows): """t-statistic of the mean, with events on the same date averaged first.""" by_date = pd.Series(values).groupby(rows).mean() return round(float(by_date.mean() / by_date.std() * len(by_date) ** 0.5), 3) if len(by_date) > 2 else None def after_event(signal): """Average return from the trade date (the day after the signal) to h days later, against QQQ.""" rows, cols, columns = event_cells(signal) prices = filled_prices(columns) qqq = on(fund("QQQ"), days()).values years = pd.Series(days()[rows].year).value_counts().sort_index() out = {"events": int(len(rows)), "per_year": {str(k): int(v) for k, v in years.items()}} for h in HORIZONS: entry, exit_ = rows + 1, rows + 1 + h ok = exit_ < len(prices) stock = prices[exit_[ok], cols[ok]] / prices[entry[ok], cols[ok]] - 1 bench = qqq[exit_[ok]] / qqq[entry[ok]] - 1 excess = pd.Series(stock - bench).dropna() out[str(h)] = { "n": int(len(excess)), "mean_excess": round(float(excess.mean()), 6), "median_excess": round(float(excess.median()), 6), "share_above_qqq": round(float((excess > 0).mean()), 6), "t_by_date": t_by_date(excess.values, rows[ok][excess.index]), } return out def capture_study(): """Ex-dividend days of dividends declared at least 4 trading days ahead, $20M liquidity.""" close = pd.DataFrame(get("us_price:close")).reindex(index=days(), columns=stock_ids()).ffill().values prices = filled_prices(stock_ids()) qqq = on(fund("QQQ"), days()).values rows = [] calendar = days() span = range(-10, 11) for col, ex, signal, dividend in declared_dividends(original_settings("dividend_capture")["lead"]): if signal < 10 or ex + 10 >= len(calendar) or calendar[ex] < pd.Timestamp(START): continue before, on_ex = close[ex - 1, col], close[ex, col] if not (before > 0 and on_ex > 0): continue one_day = prices[ex, col] / prices[ex - 1, col] - 1 path = [prices[ex + k, col] / prices[ex - 11, col] - qqq[ex + k] / qqq[ex - 11] for k in span] rows.append((ex, dividend / before, (before - on_ex) / dividend, one_day, one_day - (qqq[ex] / qqq[ex - 1] - 1), path)) frame = pd.DataFrame(rows, columns=["ex", "yield", "drop_ratio", "one_day", "one_day_excess", "path"]) out = {"events": int(len(frame)), "buckets": {}, "days": list(span)} for low, high in YIELD_BUCKETS: part = frame[(frame["yield"] >= low) & (frame["yield"] < high)] name = f"{low:g}-{high:g}" out["buckets"][name] = { "n": int(len(part)), "mean_yield": round(float(part["yield"].mean()), 6), "median_drop_ratio": round(float(part["drop_ratio"].median()), 6), "mean_drop_ratio": round(float(part["drop_ratio"].clip(-5, 5).mean()), 6), "mean_one_day": round(float(part["one_day"].mean()), 6), "mean_one_day_excess": round(float(part["one_day_excess"].mean()), 6), "t_one_day": t_by_date(part["one_day"].values, part["ex"].values), "t_one_day_excess": t_by_date(part["one_day_excess"].values, part["ex"].values), "break_even_cost_per_side": round(float(part["one_day"].mean() / 2), 6), "path": [round(float(np.mean([p[i] for p in part["path"]])), 6) for i in range(len(span))], } return out def index_paths(): """Average cumulative return around the snapshot date, in excess of QQQ and of RSP.""" out = {} qqq, rsp = on(fund("QQQ"), days()).values, on(fund("RSP"), days()).values span = np.arange(-20, 121) cuts = {"all": (START, END), **SNAPSHOT_PERIODS} for signal in INDEX_SIGNALS: rows, cols, columns = event_cells(signal) prices = filled_prices(columns) out[signal] = {} for name, (first, last) in cuts.items(): mask = (days()[rows] >= pd.Timestamp(first)) & (days()[rows] <= pd.Timestamp(last)) paths = {"qqq": [], "rsp": []} for i, c in zip(rows[mask], cols[mask]): if i + span[0] < 0 or i + span[-1] >= len(prices) or not np.isfinite(prices[i + span[0], c]): continue stock = prices[i + span, c] / prices[i + span[0], c] paths["qqq"].append(stock - qqq[i + span] / qqq[i + span[0]]) paths["rsp"].append(stock - rsp[i + span] / rsp[i + span[0]]) out[signal][name] = { "n": len(paths["qqq"]), "days": span.tolist(), "vs_qqq": np.round(np.mean(paths["qqq"], axis=0), 6).tolist(), "vs_rsp": np.round(np.mean(paths["rsp"], axis=0), 6).tolist(), "median_vs_qqq": np.round(np.median(paths["qqq"], axis=0), 6).tolist(), "share_above_qqq": np.round(np.mean(np.array(paths["qqq"]) > 0, axis=0), 6).tolist(), } return out def additions_as_listed(): """The addition rule at its original setting with every list change kept, as in the article's short code.""" signal = "sp500_addition" settings = original_settings(signal) adds, _ = list_changes() position = study_window(signal, hold(index_events(adds), settings["hold"])) curve = backtest(position, settings["max_weight"]).creturn return {"sharpe": round(curve_sharpe(curve), 6), "annual_return": round(float(curve.iloc[-1] ** (252 / len(curve)) - 1), 6)} def data_notes(): """Facts about the source data that the articles quote.""" a = actions() paid = a[(a["dividend"] > 0) & (a["date"] >= "2016-01-01") & (a["date"] <= END)] declared = paid["declaration_date"].notna() key_date = pd.to_datetime(a["key_date"], errors="coerce") # On which trading day does the adjustment factor move, relative to the ex-dividend date? factor = adj() / get("us_price:close") sample = dividends() sample = sample[sample["date"] >= "2017-01-01"].sample(4000, random_state=1) moves = {-2: 0, -1: 0, 0: 0, 1: 0} for symbol, i in zip(sample["symbol"], next_trading_day(sample["date"])): if symbol not in factor or i < 3 or i >= len(days()) - 3: continue f = factor[symbol].values if not np.isfinite(f[i - 3:i + 3]).all(): continue jumps = np.abs(f[i - 2:i + 2] / f[i - 3:i + 1] - 1) if jumps.max() >= 1e-5: moves[int(np.argmax(jumps)) - 2] += 1 # Why a list change could not be traded adds, drops, _ = index_changes() reasons = {} for name, pairs in (("additions", adds), ("deletions", drops)): count = {"tradable": 0, "no_price_history": 0, "prices_ended_before_list_date": 0, "other": 0} for date, symbol in pairs: row = next_trading_day([date])[0] if symbol not in adj(): count["no_price_history"] += 1 continue series = adj()[symbol] if row < len(series) and pd.notna(series.iloc[row]): count["tradable"] += 1 elif series.last_valid_index() is not None and series.last_valid_index() < days()[min(row, len(days()) - 1)]: count["prices_ended_before_list_date"] += 1 else: count["other"] += 1 reasons[name] = count tesla = members()["TSLA"].astype(int).diff() return { "dividends_2016_on": int(len(paid)), "share_with_declaration_date": round(float(declared.mean()), 6), "share_with_declaration_date_by_year": {str(k): round(float(v), 6) for k, v in declared.groupby(paid["date"].dt.year).mean().items()}, "key_date_range": [str(key_date.min().date()), str(key_date.max().date())], "adjustment_factor_move_day_vs_ex_date": {str(k): v for k, v in moves.items()}, "list_changes_by_tradability": reasons, "tesla_first_on_list": str(tesla[tesla == 1].index[0].date()), "sp500_list_source": "FinLab dataset description: archived Wikipedia snapshots carried forward before 2022-11-08 " "(spacing median 8 days, maximum 36 days), daily Wikipedia snapshots after that; dates are not official effective dates.", "addition_rule_every_list_change_kept": additions_as_listed(), } def count_events(): out = {} raises, initiations = classified() out["dividend rows"] = len(dividends()) out["raises"] = len(raises) out["initiations"] = len(initiations) for liquidity in LIQ: out[f"raises in liquid {liquidity:.0f}"] = int(signal_events("dividend_increase", liquidity).loc[START:].sum().sum()) for liquidity in GRIDS["dividend_initiation"]["liquidity"]: ev = known_events(initiations) & tradable(liquidity) & long_history() out[f"initiations in liquid {liquidity:.0f}"] = int(ev.loc[INITIATION_START:].sum().sum()) out[f"avg names 120d initiation {liquidity:.0f}"] = round(float(hold(ev, 120).loc[INITIATION_START:].sum(axis=1).mean()), 1) for liquidity in LIQ: for signal, label in (("dividend_increase", "raise"), ("split_drift", "split")): out[f"avg names 120d {label} {liquidity:.0f}"] = round(float(hold(signal_events(signal, liquidity), 120).loc[START:].sum(axis=1).mean()), 1) for lead, floor in product(*GRIDS["dividend_capture"].values()): p = capture_position(lead, floor).loc[START:] out[f"capture names lead {lead} floor {floor}"] = (round(float(p.sum(axis=1).mean()), 1), int((p.sum(axis=1) > 0).sum()), int(p.astype(int).diff().clip(lower=0).sum().sum())) adds, drops, renamed = index_changes() out["sp500 adds"] = len(adds) out["sp500 drops"] = len(drops) out["renamed"] = renamed for s in INDEX_SIGNALS: for h in GRIDS[s]["hold"]: out[f"{s} avg names {h}"] = round(float(hold(signal_events(s), h).loc[START:].sum(axis=1).mean()), 1) return out def write_outputs(folder, metrics, frame): folder.mkdir(exist_ok=True) (folder / "metrics.json").write_text(json.dumps(metrics, indent=1, default=float)) frame.round(6).to_csv(folder / "data.csv", date_format="%Y-%m-%d") def main(): parser = argparse.ArgumentParser() parser.add_argument("--events", action="store_true") args = parser.parse_args() if args.events: for key, value in count_events().items(): print(key, value) return metrics = {"start": START, "end": END, "signals": {}} daily = {} for signal in GRIDS: result, curve, bench = study_signal(signal) metrics["signals"][signal] = result daily[signal] = curve daily[f"qqq_{signal}"] = bench print(signal, "beat", result["beat_count"], "of 9", result["checks"], flush=True) metrics["passing"] = [s for s, r in metrics["signals"].items() if r["passed"]] metrics["after_event"] = {s: after_event(s) for s in GRIDS if s != "dividend_capture"} adds, drops, renamed = index_changes() metrics["sp500_changes"] = {"additions": len(adds), "deletions": len(drops), "same_company_pairs_removed": renamed} notes = data_notes() metrics["data_notes"] = notes index_periods = { signal: { name: {"rule": stats(daily[signal].loc[first:last].dropna()), "qqq": stats(daily[f"qqq_{signal}"].loc[first:last].dropna())} for name, (first, last) in SNAPSHOT_PERIODS.items() } for signal in INDEX_SIGNALS } frame = pd.DataFrame(daily) for symbol in REFERENCES: frame[symbol] = on(fund(symbol), frame.index) write_outputs(OUT, metrics, frame) capture_metrics = { "start": START, "end": END, "signal": metrics["signals"]["dividend_capture"], "ex_dividend_days": capture_study(), "data_notes": notes, } write_outputs(CAPTURE_OUT, capture_metrics, frame[["dividend_capture", "qqq_dividend_capture", *REFERENCES]].dropna(subset=["dividend_capture"])) index_metrics = { "start": START, "end": END, "signals": {k: metrics["signals"][k] for k in INDEX_SIGNALS}, "after_event": {k: metrics["after_event"][k] for k in INDEX_SIGNALS}, "sp500_changes": metrics["sp500_changes"], "paths": index_paths(), "periods": index_periods, "data_notes": notes, } write_outputs(INDEX_OUT, index_metrics, frame[[*INDEX_SIGNALS, *REFERENCES]].dropna(subset=["sp500_addition"])) print("passing:", metrics["passing"]) if __name__ == "__main__": main()