DeepProbLog (Neuro-Symbolic)
DeepProbLog è un'estensione neuro-simbolica di ProbLog che integra reti neurali e ragionamento logico-probabilistico in modo da avere il meglio di entrambi i mondi: la capacità di apprendimento dai dati delle reti neurali e il ragionamento strutturato della programmazione logica. Mentre DTProbLog aggiunge la teoria delle decisioni — per effettuare le scelte che massimizzano l'utilità attesa —, DeepProbLog aggiunge il neuro-simbolico — che permette a una rete neurale di determinare le probabilità di un fatto probabilistico, invece di assegnarle come costante.
L'idea centrale, semplice ma potente, è che le reti neurali vengono "incapsulate" in predicati neurali usati all'interno di un programma logico esattamente come i normali predicati.
Un predicato neurale in DeepProbLog è un fatto probabilistico la cui probabilità è parametrizzata da una rete neurale invece che da un numero costante o da una distribuzione.
Ad esempio una dichiarazione come
nn(mnist_net, [X], N, [0,1,2,3,4,5,6,7,8,9]) :: digit(X,N).
dove nn, contrazione di "Neural Network", è una parola chiave del linguaggio DeepProbLog, dice che: "La rete neurale denominata mnist_net prende in input l'immagine X di una cifra scritta a mano e produce una distribuzione di probabilità per la variabile aleatoria N i cui valori sono le cifre da 0 a 9; il suo output attribuisce la probabilità del predicato neurale digit(X,N).In generale occorrerà:
- definire la rete nello script Python avvalendosi della libreria PyTorch, rete di una qualsiasi architettura standard come le Convolutional Neural Networks (CNN) o i Multi-Layer Perceptron (MLP);
- creare un oggetto istanza di quella rete, denominandolo ad esempio
mnist_net; - dichiarare nel programma ProbLog l'astrazione dichiarativa, ovvero l'interfaccia logica, cioè il predicato neurale usando la parola chiave
nn(...)nonché le regole logiche che usano il predicato neurale; - creare un oggetto Model di DeepProbLog che colleghi il programma ProbLog e l'oggetto rete neurale creato in Python;
- considerare la funzione neural_predicate, il codice che gestisce il passaggio di informazioni dal programma deepProbLog alla rete neurale e viceversa, che DeepProbLog fornisce automaticamente;
- addestrare il modello con il dataset di esempi relativo al predicato neurale permettendo a DeepProbLog di aggiornare i pesi della rete tramite il gradiente che fluisce dall'errore (loss) finale attraverso il circuito aritmetico in un ciclo iterativo che permette appunto alla rete di apprendere dai dati, minimizzando progressivamente l'errore di previsione fino a raggiungere una convergenza ottimale.
L'inferenza in DeepProbLog segue un processo che combina ragionamento logico e calcolo di probabilità.
- Nella fase Grounding il programma logico viene riscritto rispetto alla query specifica, trasformando le regole con variabili in regole ground, cioè istanziate con valori concreti;
- nella Compilazione in Circuito Aritmetico (AC) il programma ground viene compilato in un circuito aritmetico composto da operazioni di addizione e moltiplicazione che corrispondono a OR e AND logici;
- nella Valutazione del Circuito i predicati neurali vengono valutati: i dati di input vengono passati alla rete che poi restituisce le probabilità per i predicati ground; queste probabilità vengono poi combinate attraverso il circuito aritmetico per calcolare la probabilità della query.
In un programma come
% Predicato neurale: mappa un'immagine MNIST X alla sua cifra Y
nn(mnist_net, [X], N, [0,1,2,3,4,5,6,7,8,9]) :: digit(X,N).
% Regola logica per la somma
addition(X, Y, Z) :- digit(X, N1), digit(Y, N2), Z is N1 + N2.
la rete neurale mnist_net, che richiama l'ampio database di cifre scritte a mano MNIST elemento anche del Set di dati TensorFlow, è addestrata con esempi di addition(X, Y, Z) dove X e Y sono immagini e Z è il risultato della somma, impara a riconoscere le singole cifre senza essere mai stata esplicitamente addestrata per farlo ma grazie al ragionamento logico e alla retropropagazione del gradiente attraverso l'intero sistema
Mentre una CNN "vanilla" per riconoscere cifre ha bisogno chea ogni immagine sia etichettata con la cifra corrispondente, qui la rete non vede mai l'etichetta della singola cifra ma solo se la somma finale era giusta o sbagliata. Questo è un segnale di supervisione molto più povero, di un solo bit "giusto/sbagliato" invece di un'etichetta su 10 classi, eppure la rete riesce comunque a imparare a riconoscere le cifre correttamente, perché il gradiente si propaga all'indietro attraverso la regola logica fino ai pixel. È il motivo per cui negli esperimenti del paper originale (Manhaeve et al. 2018) il modello arriva a un'accuratezza sulla singola cifra comparabile a un classificatore supervisionato standard, partendo solo da etichette di somma.
Come esempio didattico possiamo considerare una rete neurale che stima la probabilità di regime di mercato (rialzista/ribassista) da un vettore di feature, e delle regole logiche sopra che decidono un'azione di trading.
Acquisiamo innanzitutto i dati sintetici in un caso reale: es. log-rendimenti e volatilità realizzata, etichette di regime bull/bear.
import yfinance as yf
from datetime import datetime, timedelta
def regime_switching_prices(ticker, n_days=2000, p_stay=0.98, seed=0):
end_date = datetime.now()
start_date = end_date - timedelta(days=n_days)
prices = yf.download(ticker, start_date, end_date).Close.squeeze()
log_returns = np.diff(np.log(prices))
WINDOW = 30
realized_vol_30 = np.array([
np.std(log_returns[max(0, i - WINDOW + 1): i + 1]) * np.sqrt(252)
for i in range(len(log_returns))
])
log_price_all = np.log(prices[1:])
HORIZON = 10
fwd_log_return = log_price_all[HORIZON:] - log_price_all[:-HORIZON].values # .values to align indices
labels_all = np.where(fwd_log_return > 0, "bull", "bear")
# le feature vanno tagliate agli stessi indici delle etichette:
# scartiamo i primi WINDOW giorni (vol non affidabile) e gli ultimi
# HORIZON giorni (non hanno un'etichetta futura disponibile)
start = WINDOW
end = len(log_returns) - HORIZON
features = np.stack([log_returns, realized_vol_30], axis=1)[start:end]
labels = labels_all[start:] # fwd_log_return è già "accorciato" di HORIZON in coda
return features, labels
Occorre installare deepproblog e PySDD, pacchetto per costruire, manipolare e ottimizzare i diagrammi decisionali sentenziali (SDD), “struttura dati” per la rappresentazione delle funzioni booleane che supportano una serie di operazioni efficienti per la loro costruzione e manipolazione.
!pip install deepproblog
!pip install PySDD
A questo punto occorre importare le librerie Python tra le quali in particolare PyTorch, la tensor library ottimizzata per il deep learning che utilizza GPU e CPU.
import random
import torch
import torch.nn as nn
from problog.logic import Term, Constant
from deepproblog.dataset import Dataset, DataLoader
from deepproblog.engines import ExactEngine
from deepproblog.model import Model
from deepproblog.network import Network
from deepproblog.query import Query
from deepproblog.train import train_model
torch.manual_seed(0)
random.seed(0)
features, labels = regime_switching_prices('FTSEMIB.MI',n_days=2000)
split = int(len(labels) * 0.8)
# standardizzazione: rendimento (~0.001) e volatilità (~0.15-0.30) hanno
# scale molto diverse; senza normalizzare la rete fatica a pesarle bene
mean, std = features[:split].mean(axis=0), features[:split].std(axis=0)
features_norm = (features - mean) / std
train_features = torch.tensor(features_norm[:split], dtype=torch.float32)
train_labels = labels[:split]
test_features = torch.tensor(features_norm[split:], dtype=torch.float32)
test_labels = labels[split:]
print(f"Esempi train: {len(train_labels)}, test: {len(test_labels)}")
# ---------------------------------------------------------------
# Tensor source e Dataset
# ---------------------------------------------------------------
class MarketFeatures:
def __init__(self, features_tensor):
self.features_tensor = features_tensor
def __getitem__(self, item):
return self.features_tensor[int(item[0])]
class RegimeDataset(Dataset):
def __init__(self, source_name, labels):
self.source_name = source_name
self.labels = labels
def __len__(self):
return len(self.labels)
def to_query(self, i):
return Query(
Term("regime", Term("tensor", Term(self.source_name, Term("a"))), Term(self.labels[i])),
substitution={Term("a"): Constant(i)},
)
train_set = RegimeDataset("train", train_labels)
test_set = RegimeDataset("test", test_labels)
# ---------------------------------------------------------------
# Rete: (log_return, realized_vol_30) -> distribuzione su [bull, bear]
# ---------------------------------------------------------------
class RegimeNet(nn.Module):
def __init__(self):
super().__init__()
self.net = nn.Sequential(
nn.Linear(2, 16), nn.ReLU(),
nn.Linear(16, 2), nn.Softmax(dim=-1),
)
def forward(self, x):
return self.net(x)
net_module = RegimeNet()
network = Network(net_module, "regime_net", batching=True)
network.optimizer = torch.optim.Adam(net_module.parameters(), lr=1e-2)
program = """
nn(regime_net, [X], Y, [bull, bear]) :: regime(X, Y).
long_signal(X) :- regime(X, bull).
"""
with open("regime_finance.pl", "w") as f:
f.write(program)
model = Model("regime_finance.pl", [network])
model.set_engine(ExactEngine(model), cache=True)
model.add_tensor_source("train", MarketFeatures(train_features))
model.add_tensor_source("test", MarketFeatures(test_features))
# ---------------------------------------------------------------
# Training
# ---------------------------------------------------------------
loader = DataLoader(train_set, 16, True)
train_model(model, loader, 8, log_iter=50, profile=0)
# ---------------------------------------------------------------
# Valutazione
# ---------------------------------------------------------------
correct = 0
for i in range(len(test_set)):
bull_query = Query(
Term("regime", Term("tensor", Term("test", Term("a"))), Term("bull")),
substitution={Term("a"): Constant(i)},
).substitute()
result = model.solve([bull_query])[0]
prob_bull = float(result.result[bull_query.query])
predicted = "bull" if prob_bull > 0.5 else "bear"
if predicted == test_labels[i]:
correct += 1
print(f"Accuratezza sul test set: {correct}/{len(test_set)} = {correct/len(test_set):.2%}")
/tmp/ipykernel_3903/427398257.py:7: FutureWarning: YF.download() has changed argument auto_adjust default to True
prices = yf.download(ticker, start_date, end_date).Close.squeeze()
[*********************100%***********************] 1 of 1 completed
/tmp/ipykernel_3903/427398257.py:19: FutureWarning: Series.__getitem__ treating keys as positions is deprecated. In a future version, integer keys will always be treated as labels (consistent with DataFrame behavior). To access a value by position, use `ser.iloc[pos]`
log_price_all[i] - log_price_all[max(0, i - MOM_WINDOW)]
Esempi train: 1081, test: 271
Caching ACs
Training for 8 epoch(s)
Epoch 1
Iteration: 50 s:0.7297 Average Loss: 0.6911364194005728
Epoch time: 1.0200934410095215
Epoch 2
Iteration: 100 s:0.7664 Average Loss: 0.6717155531048775
Epoch time: 0.9903242588043213
Epoch 3
Iteration: 150 s:0.6978 Average Loss: 0.6901241899840533
Iteration: 200 s:0.7266 Average Loss: 0.6837622175831348
Epoch time: 0.9733421802520752
Epoch 4
Iteration: 250 s:0.7054 Average Loss: 0.6763121287897229
Epoch time: 0.9546334743499756
Epoch 5
Iteration: 300 s:0.6879 Average Loss: 0.6798868580441922
Epoch time: 0.9578890800476074
Epoch 6
Iteration: 350 s:0.7068 Average Loss: 0.6737666217330843
Iteration: 400 s:0.6875 Average Loss: 0.6792146127764136
Epoch time: 0.9300782680511475
Epoch 7
Iteration: 450 s:0.6716 Average Loss: 0.6769742991216481
Epoch time: 0.9166624546051025
Epoch 8
Iteration: 500 s:0.6777 Average Loss: 0.6680871949857101
Epoch time: 0.9333934783935547
Accuratezza sul test set: 174/271 = 64.21%