Metodi Monte Carlo basati su Dataproc e Apache Spark

Crea 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.

Disattivare la registrazione per gli avvisi

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

Utilizza ssh per connetterti al nodo primario del cluster Dataproc

Il nodo primario 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 primario.

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

Modificare l'impostazione di logging

1. Dalla home directory del nodo primario, modifica /etc/spark/conf/log4j.properties.

   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 di .rootCategory al suo valore originale (INFO).

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 preferenze personali. Questo tutorial utilizza gli interpreti interattivi perché puoi sperimentare modificando il codice, provando diversi valori di input e visualizzando i risultati.

Stimare la crescita del portfolio

In finanza, i metodi Monte Carlo vengono a volte utilizzati per eseguire simulazioni che tentano di prevedere il rendimento di un investimento. Producendo campioni casuali di risultati in una gamma di probabili condizioni di mercato, una simulazione Monte Carlo può rispondere a domande su come potrebbe comportarsi un portafoglio in media o negli scenari peggiori.

Segui questi passaggi per creare una simulazione che utilizza i metodi 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 primario Dataproc.

   pyspark
   

   Attendi il prompt di Spark >>>.

2. Inserisci il seguente codice. Assicurati di mantenere il rientro 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 vedi di nuovo il prompt di Spark.

   Il codice precedente definisce una funzione che modella ciò che potrebbe accadere quando un investitore ha un conto pensionistico esistente investito nel mercato azionario, a cui aggiunge denaro ogni anno. La funzione genera un rendimento casuale dell'investimento, in percentuale, ogni anno per la durata di un periodo specificato. La funzione accetta un valore di inizializzazione come parametro. Questo valore viene utilizzato per reinizializzare il generatore di numeri casuali, il che garantisce che la funzione non ottenga lo stesso elenco di numeri casuali ogni volta che viene eseguita. La funzione random.normalvariate garantisce che i valori casuali 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 seed da fornire alla funzione. Al prompt di Spark, inserisci il seguente codice, che genera 10.000 seed:

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

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

   Quando crea l'RDD, Spark suddivide i dati in base al numero di worker e ai core disponibili. In questo caso, Spark sceglie di utilizzare otto segmenti, uno per ogni core. Va bene per questa simulazione, che contiene 10.000 elementi di dati. Per le simulazioni più grandi, ogni slice potrebbe essere più grande del limite predefinito. In questo caso, specificare un secondo parametro per parallelize può aumentare il numero di partizioni, il che può contribuire a mantenere gestibile la dimensione di ogni partizione, mentre Spark continua a sfruttare tutti gli otto core.

5. Inserisci l'RDD che contiene i seed nella funzione di crescita.

   results = seeds.map(grow)
   

   Il metodo map passa ogni seme nell'RDD alla funzione grow e aggiunge ogni risultato a un nuovo RDD, che viene archiviato in results. Tieni presente che questa operazione, che esegue una trasformazione, non produce risultati immediati. Spark non eseguirà questa operazione finché non saranno necessari i risultati. Questa valutazione pigra ti consente di inserire il 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 in virgola mobile.

9. Ora modifica un presupposto e vedi 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. Quando hai finito di sperimentare, premi CTRL+D per uscire dall'interprete Python.

Programmare una simulazione Monte Carlo in Scala

Monte Carlo è famosa per i suoi casinò. In questa sezione, utilizzerai Scala per creare una simulazione che modelli il vantaggio matematico di cui gode un casinò in un gioco d'azzardo. Il "vantaggio della casa" in un vero casinò varia molto da gioco a gioco; può superare il 20% nel keno, ad esempio. Questo tutorial crea un semplice gioco in cui la casa ha solo l'1% di vantaggio. Ecco come funziona il gioco:

• Il giocatore piazza una scommessa, costituita da un certo numero di chip di un fondo di bankroll.
• Il giocatore tira un dado a 100 facce (che figata!).
• Se il risultato del tiro è un numero compreso tra 1 e 49, il giocatore vince.
• Per i risultati da 50 a 100, il giocatore perde la scommessa.

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

Per creare ed eseguire il gioco:

1. Avvia l'interprete Scala dal nodo primario 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 l'indentazione, 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 vedi il prompt scala>.

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

   playSession()
   

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

5. Ora simula 10.000 giocatori che scommettono 100 chip a partita. Gioca 10.000 partite in una sessione. Questa simulazione Monte Carlo calcola la probabilità di perdere tutti i tuoi soldi 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 utilizzando una funzione di somma. È funzionalmente equivalente alla sintassi .reduce(add) che hai visto nell'esempio 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.
   • Somma il numero di giocatori in bancarotta.
   • Divide il conteggio per il numero di giocatori.

   Un risultato tipico potrebbe essere:

   0.998
   

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