Pipelines do scikit-learn para retornos

Aprendizado de máquina sobre retornos

O aprendizado supervisionado ajusta uma função das variáveis X ao alvo y. Em finanças o alvo costuma ser um retorno futuro e as variáveis são coisas conhecidas de antemão: retornos passados, volatilidade, volume. O scikit-learn oferece uma interface uniforme: todo modelo tem fit(X, y) e predict(X). Uma diferença-chave dos problemas de livro-texto é a relação sinal/ruído: retornos são quase puro ruído, então mesmo um bom modelo explica uma fração mínima, e uma configuração descuidada encontra padrões que não existem.

import numpy as np

returns = np.array([0.01, -0.02, 0.015, 0.005, -0.01, 0.02])
n_lags = 2

# Row i holds the two returns before the target; y[i] is the return that follows them
windows = np.lib.stride_tricks.sliding_window_view(returns, n_lags)
X, y = windows[:-1], returns[n_lags:]
print(X)
print(y)                                             # the targets are one step ahead of the features

Variáveis a partir de defasagens

As variáveis mais simples de um retorno são os retornos anteriores. Construa uma matriz de defasagens: a linha i contém returns[i : i + n_lags], e o alvo é returns[i + n_lags], o valor seguinte. Uma visão de janela deslizante a dá sem laço. Tudo numa linha deve ser conhecido no momento de prever, a regra de antecipação aplicada a um modelo: uma variável deslocada uma posição no sentido errado deixaria o modelo ver a resposta.

import numpy as np
from sklearn.linear_model import Ridge
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

# Pipeline: scale each feature, then a ridge regression (a linear model with a penalty on large coefficients)
pipe = Pipeline([("scale", StandardScaler()), ("model", Ridge(alpha=10.0))])
print([name for name, _ in pipe.steps])

rng = np.random.default_rng(1)
returns = np.zeros(400)
noise = rng.normal(0, 0.01, size=400)
for t in range(1, 400):
    returns[t] = 0.3 * returns[t - 1] + noise[t]                    # a series with a little memory

windows = np.lib.stride_tricks.sliding_window_view(returns, 3)
X, y = windows[:-1], returns[3:]
pipe.fit(X[:300], y[:300])                                          # fit on the first 300 rows
print(np.round(pipe.named_steps["model"].coef_, 4))                 # the last lag weighs most

Pipelines e vazamento

Um Pipeline reúne o escalonador e o modelo de modo que fit aprende as estatísticas de escala (média e desvio de cada variável) só com as linhas de treino e predict as aplica sem alteração a linhas novas. Escalar o conjunto inteiro antes de dividir vazaria a média e a dispersão do período de teste para o treino: o mesmo viés de antecipação, num lugar novo. Numa série temporal a divisão também é cronológica: treinar nas linhas iniciais, testar nas posteriores, nunca um embaralhamento aleatório.

import numpy as np
from sklearn.linear_model import Ridge
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

def build_pipeline(alpha):
    return Pipeline([("scale", StandardScaler()), ("model", Ridge(alpha=alpha))])

rng = np.random.default_rng(1)
returns = np.zeros(600)
noise = rng.normal(0, 0.01, size=600)
for t in range(1, 600):
    returns[t] = 0.3 * returns[t - 1] + noise[t]

windows = np.lib.stride_tricks.sliding_window_view(returns, 3)
X, y = windows[:-1], returns[3:]
model = build_pipeline(10.0).fit(X[:400], y[:400])                  # chronological split: early rows to train
pred = model.predict(X[400:])                                       # later rows to test
print(round(float(np.mean(np.sign(y[400:]) == np.sign(pred))), 3))  # direction accuracy on unseen data

Julgar o resultado com honestidade

Para um sinal de trading, a direção costuma ser o que importa, então uma pontuação simples é a acurácia direcional: a fração de dias em que o retorno previsto e o real têm o mesmo sinal. 50% é cara ou coroa. Preditores reais de retornos chegam a 51 a 53%, e 60% em dados de teste é mais provavelmente um vazamento que uma descoberta. Compare com as referências triviais (prever sempre alta, prever zero) antes de acreditar em qualquer coisa.

import numpy as np

y_true = np.array([0.01, -0.02, 0.015, -0.005, 0.02, -0.01])
y_pred = np.array([0.004, 0.001, 0.002, -0.001, 0.003, 0.002])        # the model is right on 4 of 6 signs

accuracy = float(np.mean(np.sign(y_true) == np.sign(y_pred)))
always_up = float(np.mean(np.sign(y_true) == 1))                      # the trivial baseline: always predict a rise
print(round(accuracy, 3), round(always_up, 3))                        # compare the model with the baseline

• Uma matriz de defasagens (sliding_window_view) dá variáveis a partir dos retornos antes de cada alvo; tudo numa linha deve ser conhecido no momento da previsão.
• Um Pipeline([("scale", StandardScaler()), ("model", Ridge(alpha=alpha))]) aprende a escala só nas linhas de treino; divida cronologicamente, nunca ao acaso.
• Pontue a direção (fração de sinais iguais) contra referências triviais; 51 a 53% é realista, 60% sugere vazamento; modelos lineares regularizados servem a dados ruidosos.

A seguir: a engenharia de variáveis para dados financeiros, construir entradas melhores que retornos defasados brutos.

Lab: o que é avaliado

• Transformar uma série em variáveis com lag_matrix25 pts
• Encadear as etapas com build_pipeline25 pts
• Treinar no passado e prever o resto com fit_predict25 pts
• Pontuar as direções com direction_accuracy25 pts