Chapter 53: Financial Competition Patterns

import numpy as np
from sklearn.linear_model import Ridge

from kaggle_companion.activities._common import clean

SEED = 53
REPLICATES = 8                  # independent constructed markets; every estimate is their mean
ASSETS, TRAIN_DATES, TEST_DATES = 120, 200, 200
LOAD = np.array([0.8, 0.6, 0.5, 0.4, 0.3])      # how strongly each feature carries the market exposure (beta)
SIGNAL = np.array([0.5, 0.4, 0.4, 0.3, 0.3])    # how strongly each feature carries the true alpha signal
TRAIN_PREMIUM = 0.5             # mean daily factor return while the model is trained
FRACTIONS = [0.0, 0.5, 1.0]     # share of the exposure removed from each feature


def make_market(rng, beta, dates, premium):
    """Features, exposures and next-day returns for `dates` dates. A common factor return moves every asset by its beta."""
    alpha = rng.normal(size=(dates, ASSETS))                       # hidden idiosyncratic signal
    exposure = np.tile(beta, (dates, 1))
    X = np.stack([SIGNAL[k] * alpha + LOAD[k] * exposure + rng.normal(size=(dates, ASSETS)) for k in range(len(LOAD))], axis=-1)
    factor = rng.normal(premium, 0.5, size=(dates, 1))             # the factor premium: mean `premium`, daily noise 0.5
    ret = 0.06 * alpha + 0.5 * exposure * factor + 0.5 * rng.normal(size=(dates, ASSETS))
    return X, exposure, ret


def neutralize(X, exposure, fraction):
    """Cross-sectional regression per date: subtract `fraction` of each feature's projection on the exposure."""
    b = exposure - exposure.mean(axis=1, keepdims=True)
    slope = ((X - X.mean(axis=1, keepdims=True)) * b[..., None]).sum(axis=1) / (b ** 2).sum(axis=1, keepdims=True)
    return X - fraction * b[..., None] * slope[:, None, :]


def factor_neutral(ret, exposure):
    """The return left after the day's regression on exposure: what a factor-neutral metric scores."""
    b = exposure - exposure.mean(axis=1, keepdims=True)
    centered = ret - ret.mean(axis=1, keepdims=True)
    return centered - b * ((b * centered).sum(axis=1, keepdims=True) / (b ** 2).sum(axis=1, keepdims=True))


def ranks(a):
    """Per-date ranks scaled to [-0.5, 0.5]."""
    return (a.argsort(axis=1).argsort(axis=1) + 0.5) / a.shape[1] - 0.5


def date_correlations(pred, target):
    """Spearman correlation between prediction and target on each date."""
    p, t = ranks(pred), ranks(target)
    return (p * t).sum(axis=1) / np.sqrt((p ** 2).sum(axis=1) * (t ** 2).sum(axis=1))


def one_market(test_premium, replicate):
    rng = np.random.default_rng(SEED * 100 + replicate)
    beta = rng.normal(size=ASSETS)
    X_train, e_train, r_train = make_market(rng, beta, TRAIN_DATES, TRAIN_PREMIUM)
    X_test, e_test, r_test = make_market(rng, beta, TEST_DATES, test_premium)
    targets = {"raw": r_test, "neutral": factor_neutral(r_test, e_test)}
    out = {}
    for fraction in FRACTIONS:
        # Features are neutralized with the same rule in training and in the later period; the target is never altered.
        model = Ridge(alpha=100).fit(neutralize(X_train, e_train, fraction).reshape(-1, len(LOAD)), ranks(r_train).reshape(-1))
        pred = model.predict(neutralize(X_test, e_test, fraction).reshape(-1, len(LOAD))).reshape(TEST_DATES, ASSETS)
        for name, target in targets.items():
            c = date_correlations(pred, target)
            out[(fraction, name)] = (float(c.mean()), float(c.std()))
    return out


def run(test_premium):
    markets = [one_market(test_premium, i) for i in range(REPLICATES)]
    score = {}
    for name in ("raw", "neutral"):
        score[name] = {str(f): {"mean": float(np.mean([m[(f, name)][0] for m in markets])),
                               "daily_sd": float(np.mean([m[(f, name)][1] for m in markets])),
                               "per_market": [m[(f, name)][0] for m in markets]} for f in FRACTIONS}
    return clean({"test_premium": test_premium, "score": score, "train_premium": TRAIN_PREMIUM,
                  "replicates": REPLICATES, "assets": ASSETS, "test_dates": TEST_DATES})

import pandas as pd
from kaggle_companion.activities.ch53 import explain

VALUES = [0.5, 0.25, 0.0, -0.5]
results = {value: run(value) for value in VALUES}
table = pd.DataFrame({str(v): explain(results[v], v)['metrics'] for v in VALUES})
print(table.to_string())

import matplotlib.pyplot as plt
from kaggle_companion.activities.ch53 import draw, NCOLS, HEIGHT

DEFAULT = -0.5
fig, axes = plt.subplots(1, NCOLS, figsize=(6.6 if NCOLS == 1 else 10.4, HEIGHT), layout='constrained')
draw(axes, results[DEFAULT], DEFAULT)
plt.show()