Metodi Monte Carlo basati su Dataproc e Apache Spark

Creazione di un cluster Dataproc

Segui i passaggi per creare un cluster Dataproc dalla console Google Cloud. Le impostazioni predefinite del cluster, che includono due nodi worker, sono sufficienti per questo tutorial.

Disattivazione del logging per gli avvisi

Per impostazione predefinita, Apache Spark stampa log dettagliati nella finestra della console. Per scopo di questo tutorial, modifica il livello di logging in modo da registrare solo gli errori. Segui questi passaggi:

Usa ssh per connetterti al nodo principale del cluster Dataproc

Il nodo principale del cluster Dataproc ha il suffisso -m nel nome della VM.

1. In the Google Cloud console, go to the VM instances page.

   Go to VM instances

2. In the list of virtual machine instances, click SSH in the row of the instance that you want to connect to.

   

Si apre una finestra SSH connessa al nodo principale.

Connected, host fingerprint: ssh-rsa 2048 ...
...
user@clusterName-m:~$

Modificare l'impostazione di registrazione

1. Modifica /etc/spark/conf/log4j.properties dalla home directory del nodo principale.

   sudo nano /etc/spark/conf/log4j.properties
   

2. Imposta log4j.rootCategory uguale a ERROR.

   # Set only errors to be logged to the console
   log4j.rootCategory=ERROR, console
   log4j.appender.console=org.apache.log4j.ConsoleAppender
   log4j.appender.console.target=System.err
   log4j.appender.console.layout=org.apache.log4j.PatternLayout
   log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
   

3. Salva le modifiche ed esci dall'editor. Se vuoi riattivare il logging dettagliato, annulla la modifica ripristinando il valore originale (INFO) per .rootCategory.

Linguaggi di programmazione Spark

Spark supporta Python, Scala e Java come linguaggi di programmazione per applicazioni autonome e fornisce interpreti interattivi per Python e Scala. La lingua che scegli è una questione di preferenza personale. Questo tutorial utilizza gli interpreti interattivi perché puoi fare esperimenti modificando il codice, provando diversi valori di input e visualizzando i risultati.

Stima della crescita del portafoglio

In finanza, a volte i metodi Monte Carlo vengono utilizzati per eseguire simulazioni che tentano di predire il rendimento di un investimento. Producendo campioni casuali di risultati in un intervallo di probabili condizioni di mercato, una simulazione di Monte Carlo può rispondere a domande sul rendimento medio di un portafoglio o in scenari peggiori.

Segui questa procedura per creare una simulazione che utilizza i metodi di Monte Carlo per tentare di stimare la crescita di un investimento finanziario in base ad alcuni fattori di mercato comuni.

1. Avvia l'interprete Python dal nodo principale Dataproc.

   pyspark
   

   Attendi il prompt di Spark >>>.

2. Inserisci il seguente codice. Assicurati di mantenere l'indentazione nella definizione della funzione.

   import random
   import time
   from operator import add
   
   def grow(seed):
       random.seed(seed)
       portfolio_value = INVESTMENT_INIT
       for i in range(TERM):
           growth = random.normalvariate(MKT_AVG_RETURN, MKT_STD_DEV)
           portfolio_value += portfolio_value * growth + INVESTMENT_ANN
       return portfolio_value
   

3. Premi return finché non visualizzi di nuovo il prompt di Spark.

   Il codice precedente definisce una funzione che modella ciò che potrebbe accadere quando un investitore ha già un conto pensionistico investito nel mercato azionario, a cui aggiunge ogni anno altri soldi. La funzione genera un ritorno sull'investimento casuale, in percentuale, ogni anno per la durata di un periodo specificato. La funzione accetta un valore iniziale come parametro. Questo valore viene utilizzato per reimpostare il generatore di numeri casuali, in modo che la funzione non restituisca lo stesso elenco di numeri casuali ogni volta che viene eseguita. La funzione random.normalvariate garantisce che i valori random si verifichino in una distribuzione normale per la media e la deviazione standard specificate. La funzione aumenta il valore del portafoglio dell'importo della crescita, che può essere positivo o negativo, e aggiunge una somma annuale che rappresenta un ulteriore investimento.

   Definisci le costanti richieste in un passaggio successivo.

4. Crea molti semi da fornire alla funzione. Al prompt di Spark, inserisci il seguente codice, che genera 10.000 semi:

   seeds = sc.parallelize([time.time() + i for i in range(10000)])
   

   Il risultato dell'operazione parallelize è un set di dati distribuito resiliente (RDD), ovvero una raccolta di elementi ottimizzati per l'elaborazione parallela. In questo caso, l'RDD contiene semi basati sull'ora corrente del sistema.

   Durante la creazione dell'RDD, Spark suddivide i dati in base al numero di worker e core disponibili. In questo caso, Spark sceglie di utilizzare otto slice, uno per ogni core. Va bene per questa simulazione, che contiene 10.000 elementi di dati. Per le simulazioni più grandi, ogni segmento potrebbe essere più grande del limite predefinito. In questo caso, specificare un secondo parametro per parallelize puoi aumentare il numero di slice, il che può contribuire a mantenere gestibili le dimensioni di ogni slice, mentre Spark sfrutta comunque tutti e otto i core.

5. Alimenta la funzione di crescita con l'RDD contenente i semi.

   results = seeds.map(grow)
   

   Il metodo map passa ogni seme nell'RDD alla funzione grow e appende ogni risultato a un nuovo RDD, che viene archiviato in results. Tieni presente che questa operazione, che esegue una trasformazione, non produce subito i risultati. Spark non eseguirà questa operazione finché i risultati non saranno necessari. Questa valutazione lazy è il motivo per cui puoi inserire codice senza che le costanti siano definite.

6. Specifica alcuni valori per la funzione.

   INVESTMENT_INIT = 100000  # starting amount
   INVESTMENT_ANN = 10000  # yearly new investment
   TERM = 30  # number of years
   MKT_AVG_RETURN = 0.11 # percentage
   MKT_STD_DEV = 0.18  # standard deviation
   

7. Chiama reduce per aggregare i valori nell'RDD. Inserisci il seguente codice per sommare i risultati nell'RDD:

   sum = results.reduce(add)
   

8. Stimare e visualizzare il rendimento medio:

   print (sum / 10000.)
   

   Assicurati di includere il carattere punto (.) alla fine. Indica l'aritmetica con virgola mobile.

9. Ora modifica un'ipotesi e osserva come cambiano i risultati. Ad esempio, puoi inserire un nuovo valore per il rendimento medio del mercato:

   MKT_AVG_RETURN = 0.07
   

10. Esegui di nuovo la simulazione.

    print (sc.parallelize([time.time() + i for i in range(10000)]) \
            .map(grow).reduce(add)/10000.)
    

11. Al termine dell'esperimento, premi CTRL+D per uscire dall'interprete Python.

Programmazione di una simulazione Monte Carlo in Scala

Monte Carlo, ovviamente, è famosa come meta per i giochi e scommesse. In questa sezione, utilizzerai Scala per creare una simulazione che modella il vantaggio matematico di un casinò in un gioco a premi. Il "vantaggio della casa" in un casinò reale varía notevolmente da un gioco all'altro; ad esempio, può superare il 20% nel keno. Questo tutorial crea un semplice gioco in cui la casa ha solo un vantaggio del uno percento. Ecco come funziona il gioco:

• Il giocatore effettua una scommessa, composta da un numero di chip di un fondo di bankroll.
• Il giocatore lancia un dado a 100 facce (quanto sarebbe bello?).
• Se il risultato del lancio è un numero compreso tra 1 e 49, il giocatore vince.
• Per i risultati da 50 a 100, il giocatore perde la scommessa.

Puoi vedere che questo gioco crea uno svantaggio dell'1% per il giocatore: in 51 dei 100 possibili risultati per ogni lancio, il giocatore perde.

Per creare ed eseguire il gioco:

1. Avvia l'interprete Scala dal nodo principale Dataproc.

   spark-shell
   

2. Copia e incolla il seguente codice per creare il gioco. Scala non ha gli stessi requisiti di Python per quanto riguarda il rientro, quindi puoi semplicemente copiare e incollare questo codice al prompt scala>.

   val STARTING_FUND = 10
   val STAKE = 1   // the amount of the bet
   val NUMBER_OF_GAMES = 25
   
   def rollDie: Int = {
       val r = scala.util.Random
       r.nextInt(99) + 1
   }
   
   def playGame(stake: Int): (Int) = {
       val faceValue = rollDie
       if (faceValue < 50)
           (2*stake)
       else
           (0)
   }
   
   // Function to play the game multiple times
   // Returns the final fund amount
   def playSession(
      startingFund: Int = STARTING_FUND,
      stake: Int = STAKE,
      numberOfGames: Int = NUMBER_OF_GAMES):
      (Int) = {
   
       // Initialize values
       var (currentFund, currentStake, currentGame) = (startingFund, 0, 1)
   
       // Keep playing until number of games is reached or funds run out
       while (currentGame <= numberOfGames && currentFund > 0) {
   
           // Set the current bet and deduct it from the fund
           currentStake = math.min(stake, currentFund)
           currentFund -= currentStake
   
           // Play the game
           val (winnings) = playGame(currentStake)
   
           // Add any winnings
           currentFund += winnings
   
           // Increment the loop counter
           currentGame += 1
       }
       (currentFund)
   }
   

3. Premi return finché non viene visualizzato il prompt scala>.

4. Inserisci il seguente codice per giocare 25 volte, che è il valore predefinito per NUMBER_OF_GAMES.

   playSession()
   

   Il tuo bankroll ha iniziato con un valore di 10 unità. Ora è più alto o più basso?

5. Ora simula 10.000 giocatori che scommettono 100 fiches a partita. Giocare 10.000 partite in una sessione. Questa simulazione di Monte Carlo calcola la probabilità di perdere tutto il denaro prima della fine della sessione. Inserisci il seguente codice:

   (sc.parallelize(1 to 10000, 500)
     .map(i => playSession(100000, 100, 250000))
     .map(i => if (i == 0) 1 else 0)
     .reduce(_+_)/10000.0)
   

   Tieni presente che la sintassi .reduce(_+_) è un'abbreviazione in Scala per l'aggregazione mediante una funzione di somma. È funzionalmente equivalente alla sintassi.reduce(add) che hai visto nell'esempio di Python.

   Il codice precedente esegue i seguenti passaggi:

   • Crea un RDD con i risultati della riproduzione della sessione.
   • Sostituisce i risultati dei giocatori in bancarotta con il numero 1 e i risultati diversi da zero con il numero 0.
   • Sommi il numero di giocatori in bancarotta.
   • Divide il conteggio per il numero di giocatori.

   Un risultato tipico potrebbe essere:

   0.998
   

   Il che rappresenta una quasi certezza di perdere tutti i tuoi soldi, anche se il casinò aveva solo un vantaggio dell'1%.