---

Description

Dans ce tuto, nous vous proposons de créer et d'afficher des moyennes mobiles de signaux aléatoires.

Plusieurs moyennes SMA d'un signal aléatoire (en rouge).

---

1. Préparation du PC (Poste de contrôle) et de matplotlib.
2. Fichier d'indicateurs techniques.
3. Coding du node aleatoire.
4. Coding du node moy_mobile.

---

Retour au plan

Poste de contrôle (PC) :

• ☐ Créer un graphe nommé MM.
• Ce graphe contient 3 nodes, comme sur l'image ci-dessus :
  • Aléatoire-2 → MM-1 → Plots-0.
• Avant de coder la moyenne mobile, nous devons coder le node Aléatoire :
  • ☐ Court-circuiter la MM en secouant le node.
• Le graphe n'est donc composé que du node Aléatoire, relié au node Plots.

---

show_matplotlib.py :

☐ Modifier le code de lancement, tout à la fin de show_matplotlib.py (Seul le nom du graphe a changé) :
if __name__ == '__main__': :

if __name__ == '__main__':
    """ Les 3 premières lignes permettent un fonctionnement autonome, en phase de mise au point.
        - Elles simulent l'ajout d'arguments en ligne de commande.
        - Elles définissent le nom du graphe ainsi que l'id du node 'Plots'.
        - Elles pourront être commentées en phase de production.
    """
    sys.argv.append('MM')  # Nom du graphe.
    sys.argv.append('0')  # Plots id. (La ligne de commande ne doit contenir que du str).
    sys.argv.append('-1')  # Gestion de la fermeture automatique. Poste de contrôle PID : -1 en local.

    """ Lancement de l'appli. """
    mpl = ShowMatPlotLib()

 

---

Retour au plan

• Les nodes Signaux, Aléatoire, ainsi que la plupart de ceux qui vont suivre, ont besoin de fonctions de base. Pour éviter les répétitions de code, nous allons créer une collection de fonctions mathématiques, à enrichir au fur et à mesure des besoins.
  • En trading, ces fonctions mathématiques sont nommées indicateurs techniques.
  • Le fichier que nous allons créer est indicators.py.
  • Nous le placerons dans le dossier /functions.

• ☐ /functions/indicators.py :

  # Imports externes
  import numpy as np
  import time
  
  
  def get_periodic(len_vector=100_000, signal_name='Sinus', period=100, amplitude=20):
      """
      @param len_vector:  Taille du vecteur de sortie.
      @param signal_name:  'Sinus', 'Cosinus', 'Carré', 'Triangle', 'Dent de scie montante' ou 'Dent de scie descendante'
      @param period: Période du signal.
      @param amplitude: Amplitude du signal.
      @return: vecteur numpy de taille len_vector.
      """
      nb_iter = 2 + len_vector // (2 * period)
      if signal_name == 'Sinus':
          x = np.linspace(0, 2 * np.pi, period + 1)[:-1]
          y = np.reshape(amplitude * np.sin(x), (period, 1))
          y = np.concatenate((y, y) * nb_iter)
      elif signal_name == 'Cosinus':
          x = np.linspace(0, 2 * np.pi, period + 1)[:-1]
          y = np.reshape(amplitude * np.cos(x), (period, 1))
          y = np.concatenate((y, y) * nb_iter)
      elif signal_name == 'Carré':
          y = np.ones((period, 1))
          y[:period // 2] *= amplitude
          y[period // 2:] *= - amplitude
          y = np.concatenate((y, y) * nb_iter)
      elif signal_name == 'Triangle':
          y = np.ones((period, 1))
          y[:period // 2] *= np.linspace(amplitude, -amplitude, period // 2, endpoint=False).reshape((period // 2, 1))
          y[period // 2:] *= np.linspace(-amplitude, amplitude, period // 2, endpoint=False).reshape((period // 2, 1))
          y = np.concatenate((y, y) * nb_iter)
      elif signal_name == 'Dent de scie montante':
          y = np.ones((period, 1)) * np.linspace(-amplitude, amplitude, period).reshape((period, 1))
          y = np.concatenate((y, y) * nb_iter)
      else:       # Dent de scie descendante
          y = np.ones((period, 1)) * np.linspace(amplitude, -amplitude, period).reshape((period, 1))
          y = np.concatenate((y, y) * nb_iter)
  
      return y[: len_vector]
  
  
  def get_random(len_vector=100_000, typ='Normal', seeder=0, b_sum=True, **kwargs):
      """
      @param len_vector:  Taille du vecteur de sortie.
      @param typ: Normal, poisson, binomial ou logistic.
      @param seeder: Si seeder >=0, le signal aléatoire sera toujours le même, en fonction de la valeur seeder.
                      Cela permet de répéter des expériences à l'identique.
      @param b_sum: Si Vrai, chaque valeur est cumulée, imitant un historique de trading.
      @param kwargs: Certains 'randoms' nécessitent des arguments spécifiques : nb trials, lambda, etc.
      @return: Vecteur de shape (len_vector,)
      """
      """ https://numpy.org/doc/1.16/reference/routines.random.html """
      np.random.seed(seeder) if seeder >= 0 else np.random.seed(None)
      if typ == 'Poisson':
          # à coder ...
          pass
      elif typ == 'Binomial':
          # à coder ...
          pass
      elif typ == 'Logistic':
          # à coder ...
          pass
      else:       # 'Normal'
          amplitude = kwargs.get('amplitude', 10)
          vector = np.random.standard_normal(len_vector) * amplitude / 2
      return np.cumsum(vector) if b_sum else vector
  
  
  def sma(v_signal, period):
      """ Moyenne mobile arithmétique. https://fr.wikipedia.org/wiki/Moyenne_mobile  <-- (Ctrl + Clic)
      :param v_signal: vecteur numpy.
      :param period: Longueur de la sma.
      :return: Vecteur de même shape que celui d'entrée.
          |_ Les (period-1) premières valeurs sont des np.nan.
      """
      if period <= 1:
          return v_signal                             # Exemple 'Original' a une période de 0.
      _mm_control(v_signal, period, 'sma')            # Contrôle : fin programme en cas d'erreur.
  
      """ ****************** Code provisoire : non performant à cause de la boucle. ****************** """
      t_start = time.time()                           # Supprimer après la MAP.
      np_sma = np.full(v_signal.shape, np.nan)
      """ Boucle d'exécution : A chaque tour, la moyenne est calculée et placée dans np_sma. """
      for i in range(period, v_signal.shape[0]):
          m = v_signal[i-period: i]
          _sma = np.mean(m)          # Moyenne de {period} derniers points.
          np_sma[i] = _sma
      print('Durée :', time.time() - t_start, 's.')   # Supprimer après la MAP.
      """ ****************** Code provisoire : non performant à cause de la boucle. ****************** """
      return np_sma
  
  
  def ema(v_signal, period):
      """ Moyenne Mobile Exponentielle. https://fr.wikipedia.org/wiki/Moyenne_mobile
      :param v_signal: vecteur numpy.
      :param period: Longueur de la ema.
      :return: Vecteur de même shape que celui d'entrée.
          |_ Les (period-1) premières valeurs sont des np.nan.
      """
      # à coder ...
      return np.empty((100_000,))     # <-- à supprimer après coding
  
  
  def smma(v_signal, period, nb_last=2):
      """ Moyenne mobile lissée.
      https://admiralmarkets.com/fr/formation/articles/indicateurs-forex/indicateur-moyenne-mobile-mt4
      :param v_signal: vecteur numpy.
      :param period: Longueur de la smma.
      :param nb_last: nb derniers.
      :return: Vecteur de même shape que celui d'entrée.
          |_ Les (period-1) premières valeurs sont des np.nan.
      """
      # à coder ...
      return np.empty((100_000,))     # <-- à supprimer après coding
  
  
  def lwma(v_signal, period, ratio=1.2):
      """ Moyenne mobile linéaire pondérée  (Linear Weighted Moving Average, LWMA).
      https://www.instaforex.eu/fr/forex_technical_indicators/moving_average
      :param v_signal: vecteur numpy.
      :param period: Longueur de la lwma.
      :param ratio: Raison géométrique de la pondération.
      :return: Vecteur de même shape que celui d'entrée.
          |_ Les (period-1) premières valeurs sont des np.nan.
      """
      # à coder ...
      return np.empty((100_000,))     # <-- à supprimer après coding
  
  
  def get_mm(v_signal, typ='SMA', period=14, **kwargs):
      """
      @param v_signal: Vecteur sur lequel la moyenne doit être calculée.
      @param typ: 'SMA', 'EMA', 'SMMA', 'LWMA'
      @param period: Longueur de la moyenne.
      @return:
      """
      if period > 0:
          if typ == 'SMA':                 # Moyenne Mobile Arithmétique.
              return sma(v_signal, period)
          elif typ == 'EMA':               # Moyenne Mobile Exponentielle.
              return ema(v_signal, period)
          elif typ == 'SMMA':              # Moyenne Mobile Lissée.
              nb_last = kwargs.get('nb_last', 4)
              return smma(v_signal, period, nb_last=nb_last)
          elif typ == 'LWMA':             # Moyenne Mobile Linéaire Pondérée.
              ratio = kwargs.get('ratio', 1.6)
              return lwma(v_signal, period, ratio=ratio)
  
      """ Si le type est inconnu ou la période < 1, on retourne le signal d'entrée. """
      return v_signal
  
  
  def _mm_control(v_signal, period, indic):
      if len(v_signal) < period:
          raise SystemExit(f"Erreur {indic} :\nLa période ({period}) est supérieure"
                           f" au nombre de points à traiter ({len(v_signal)})")
  
  
  if __name__ == '__main__':
      """ Lancement autonome, nécessaire en phase de développement pour la MAP (mise au point des fonctions). """
      pass
      # Coder ici vos tests ...
  

• Remarquez la première fonction, elle concerne les signaux périodiques : sinus, triangle, etc.
• Cela présente 2 avantages :
  • Le code du node Signaux va pouvoir être simplifié.
  • Cette fonction get_periodic(), ainsi que toutes les autres, vont pouvoir être utilisées partout dans le projet.
• Simplification du code de la classe Calcul du node Signaux.
  /nodes/generateurs/signaux/signaux.py > classe Calcul :

  class Calcul(CtrlCalcul):
      """ ********** Le code ci-dessous ne concerne pas le poste de contrôle, mais seulement les calculs. ********** """
      def __init__(self):
          super().__init__()
          self.o_yaml = YamlParams(self)  # Classe spécifique à 'Signaux'.
  
      def descr_signal(self, od_descr, val, root_key):
          """ Voir docstring dans la classe-mère. """
          """ Boucle sur les signaux en sortie. """
          for key in ['Sinus', 'Cosinus', 'Carré', 'Triangle', 'Dent de scie montante', 'Dent de scie descendante']:
              """ 'key_dock' doit avoir LA MÊME ORTHOGRAPHE que celle affichée dans le dockable de ses nodes-clients. """
              key_dock = f'{self.s_id}-{key}'     # Ex : Signaux2-Carré
              per_amp = self.o_yaml.get_params(key)
              od_descr.write([key_dock, 'actif'], self.od_mydic.read(key, False))    # True ou False.
              od_descr.write([key_dock, 'per_amp'], per_amp)
  
      def process(self):
          """ Voir docstring dans la classe-mère. """
          """ Traitement des signaux. """
          l_signals = ['Sinus', 'Cosinus', 'Carré', 'Triangle', 'Dent de scie montante', 'Dent de scie descendante']
          for signal_name in l_signals:
              key_dock = self.s_id + '-' + signal_name
              b_signal_actif = self.od_descr.read([key_dock, 'actif'], False)
              if b_signal_actif:
                  period, amplitude = self.o_yaml.get_params(signal_name)
                  vector = indic.get_periodic(self.len_buffer, signal_name, period, amplitude)
                  num_col = self.od_descr.read([key_dock, 'num_col'], -1)
                  self.add_vector(key_dock, num_col, vector)
  

   

• Ne pas oublier d'importer les indicateurs dans Signaux.
  Imports de Signaux :

  # Imports internes
  import functions.indicators as indic
  from pc.ctrl_node import CtrlNode, CtrlCalcul
  from nodes.generateurs.signaux.signaux_yaml import YamlParams
  

   

• Vérification :
  • Lancer un affichage contenant des signaux : Sinus, Cosinus, etc.
  • Apporter des modifications depuis le PC ou depuis les paramètres avancés yaml.
  • Leurs répercussions sont bien prises en compte, en live.

---

Retour au plan

• Ce node est un générateur, tout comme celui que nous avons déjà codé : Signaux.
• Sa présentation dans le PC est légèrement modifiée : ajout d'une case à cocher : Émulation histos.
• Nous allons suivre les mêmes étapes :
  1. Création du fichier aleatoire_seeder.yaml.
  2. Création du fichier aleatoire_yaml.py.
  3. Création, dans aleatoire.py de la classe Calcul, contenant :
     • Les 3 méthodes obligatoires : __init__(), descr_signal() et process().
     • Autant de méthodes spécifiques que nécessaire : normal(), poisson(), binomial(), etc.
• ☐ (1) - /nodes/generateurs/aleatoire/aleatoire_seeder.yaml :

  # Doc :
  # Si Graine est >= 0, la séquence aléatoire reste la même à chaque lancement.
  
  Normal:
    amplitude: 120
  
  Poisson:
    lambda: 20

   

• ☐ (2) - /nodes/generateurs/aleatoire/aleatoire_yaml.py :

  # Imports internes.
  from nodes.yaml_parent import YamlParent
  
  
  class YamlParams(YamlParent):
      def __init__(self, o_calc):
          super().__init__(o_calc)
  

   

• ☐ Ajout d'une case à cocher dans le dockable des paramètres : Émulation histos. Méthode fixed_params().
  /nodes/generateurs/aleatoire/aleatoire.py > Node.fixed_params() :

      def fixed_params(self):
          return {        # 'Émulation histos'
              'Type': ['Normal', {'values': ['Normal', 'Poisson', 'Binomial', 'Logistic']}],
              'Émulation histos': True,     # <-- Case à cocher.
              'Graine (<0=None)': [-1, {'tip': "Une valeur ≥ 0 crée une série pseudo-aléatoire,"
                                               "\nUne valeur < 0 crée une série aléatoire."}],
          }
  

   

• ☐ (3) - La classe Calcul.
  À la fin du fichier /nodes/generateurs/aleatoire/aleatoire.py :

  class Calcul(CtrlCalcul):
      """ ********** Le code ci-dessous ne concerne pas le poste de contrôle, mais seulement les calculs. ********** """
  
      def __init__(self):
          super().__init__()
          self.o_yaml = YamlParams(self)
  
      def descr_signal(self, od_descr, val, root_key):
          """ Création du super-dictionnaire de description. Voir docstring dans la classe-mère. """
          """ key_dock doit avoir EXACTEMENT la même orthographe que la clé affichée
          dans le dockable des paramètres du node CLIENT. """
          typ = self.od_mydic.read('Type')
          key_dock = f"{self.s_id}-{typ}"     # Ex : Aléatoire3-Logistic
  
          od_descr.write([key_dock, 'actif'], True)
          od_descr.write('type', typ)
          od_descr.write('emul', self.od_mydic.read('Émulation histos'))
          od_descr.write('seed', self.od_mydic.read('Graine (<0=None)'))
  
      def process(self):
          """ Voir docstring dans la classe-mère. """
          typ = self.od_descr.read('type')
          b_emul = self.od_descr.read('emul')
          seed = self.od_descr.read('seed')
          key_dock = f"{self.s_id}-{typ}"     # Ex : Aléatoire3-Logistic
          num_col_out = self.od_descr.read('num_col')
          kwargs = {
              'amplitude': self.o_yaml.od_yaml.read(['Normal', 'amplitude'], 10),
              'lambda': abs(self.o_yaml.od_yaml.read(['Poisson', 'lambda'], 15))    # Paramètre lambda.
          }
          vector = indic.get_random(self.len_buffer, typ, seed, b_emul, **kwargs)
          self.add_vector(key_dock, num_col=num_col_out, vector=vector)
  

  Importer CtrlCalcul, YamlParams et numpy (as np).

• Notez à l'avant-dernière ligne l'appel à la méthode get_random().

  • Il y a plusieurs types d'aléas. Seul le type Normal est codé dans indicators.get_random().

  • ☐ Il vous appartient de coder les autres : Poisson, Binomial, Logistic .. vous pouvez en ajouter d'autres.

• Vérification :
  • ☐ Lancer le PC, puis Matplotlib.

Adapter la présentation à votre convenance.

---

Retour au plan

• ☐ Rétablir le graphe initial : insérer le node MM-1 entre Aléatoire-2 et Plots-0.
• Ce node est à la fois client et serveur de signaux.
  • Client du node en amont (un seul) : ici, Aléatoire-2.
  • Serveur des nodes en aval (il peut y en avoir plusisurs) : ici, Plots-0.
• Sa création suit la règle :
  1. Création du fichier moy_mobile_seeder.yaml.
  2. Création du fichier moy_mobile_yaml.py.
  3. Création, dans moy_mobile.py de la classe Calcul, contenant :
     • Les 3 méthodes obligatoires: __init__(), descr_signal() et process().
     • Autant de méthodes spécifiques que nécessaire : ici, aucune.

---

• ☐ (1) /nodes/indicateurs/moy_mobile/moy_mobile_seeder.yaml :

  SMMA:             # Moyenne mobile lissée.
    nb_last: 4      # Nombre de valeurs actives
  
  LWMA:             # Moyenne mobile linéaire pondérée  (Linear Weighted Moving Average, LWMA).
    Ratio: 1.6      # Raison géométrique de la pondération.
  

  
   
• ☐ (2) /nodes/indicateurs/moy_mobile/moy_mobile_yaml.py :

  from nodes.yaml_parent import YamlParent
  
  
  class YamlParams(YamlParent):
      def __init__(self, o_calc):
          super().__init__(o_calc)
  
      @property
      def smma_nb_last(self):
          return max(1, int(self.od_yaml.read(['SMMA', 'nb_last'], 2)))
  
      @property
      def lwma_ratio(self):
          return max(.1, self.od_yaml.read(['LWMA', 'Ratio'], 1.2))
  

   

• Ajout d'une case à cocher dans le dockable des paramètres de moy_mobile : Insérer le signal original à la liste des signaux de sortie.
  ☐ /nodes/indicateurs/moy_mobile/moy_mobile.py > Node.my_params() :

      def my_params(self, context):
          return {
              'Original': True,    # <-- Case à cocher.
              'Type de MM': ['SMA', {'values': ['SMA', 'EMA', 'SMMA', 'LWMA']}],
              "Périodes (sep=',')": '14',
          }
  

  
  ☐ Ajout de la prise en compte de la case à cocher Original ajoutée ci-dessus.
   /nodes/indicateurs/moy_mobile/moy_mobile.py > Node.my_signals() :

      def my_signals(self, l_signals_in, num_socket_out):
          """ Faisceau de signaux (l_signals) délivré à la sortie. """
          l_signals = list()      # Liste de 4-tuples (tuples de 4 valeurs).
          for signals_in in l_signals_in[0]:
              typ_id_from, signal_ante_from, signal_now_from, signal_source_from, \
               _, signal_title, typ_id, signal_ante, signal_source = signals_in[:9]
              try:
                  periods = [int(p.strip()) for p in self.get_param([signal_title, "Périodes (sep=',')"]).split(',')]
              except (Exception,):
                  periods = [14]
              if self.get_param([signal_title, 'Original'], True):
                  l_signals.append((typ_id, signal_ante, 'Original', signal_source))
              for num, period in enumerate(periods):
                  ident = f"{self.get_param([signal_title, 'Type de MM'], 'SMA')}{period}"
                  signal_now = f'sign{num}'
                  signal_source = self.join(signal_source_from, self.join(typ_id_from, signal_now_from), sep='\n')
                  typ_id, signal_ante = f'{self.type}{self.id}', self.join(signal_ante_from, signal_now_from)
                  l_signals.append((typ_id, signal_ante, signal_now, signal_source, ident))
  
          return l_signals
  

   

• ☐ Imports, au début de /nodes/indicateurs/moy_mobile/moy_mobile.py :

  # Imports internes
  import functions.indicators as indic
  from nodes.indicateurs.moy_mobile.moy_mobile_yaml import YamlParams
  from pc.ctrl_node import CtrlNode, CtrlCalcul
  

   

• ☐ (3) Classe Calcul, à la fin de /nodes/indicateurs/moy_mobile/moy_mobile.py :

  class Calcul(CtrlCalcul):
      """ ********** Le code ci-dessous ne concerne pas le poste de contrôle, mais seulement les calculs. ********** """
      def __init__(self):
          super().__init__()
          self.o_yaml = YamlParams(self)
  
      def descr_signal(self, od_descr, val, root_key):
          """ Voir docstring dans la classe-mère. """
          typ = val['Type de MM']
          l_orig = [0] if ('Original' in val and val['Original']) else []     # [0] Si l'original est demandé.
          """ Chaque signal peut avoir plusieurs moyennes. """
          l_periods = [int(p.strip()) for p in val["Périodes (sep=',')"].split(',')] + l_orig
          for num, period in enumerate(l_periods):
              """ key_dock doit avoir EXACTEMENT la même orthographe que la clé affichée
              dans le dockable des paramètres du node CLIENT. """
              key_dock = f'{root_key}-sign{num}' if period > 0 else f"{root_key}-Original"
              od_descr.write([key_dock, 'typ'], typ)
              od_descr.write([key_dock, 'period'], period)
  
      def process(self):
          """ Voir docstring dans la classe-mère. """
  
          """ len_array = Nombre de lignes dans le tableau numpy du node-serveur. """
          o_server = self.dt_servers_in['e0'][0]
          """ Parcours du dictionnaire de description des traitements. """
          for key_dock, val in self.od_descr.items():
              if key_dock.startswith(self.s_id + '-'):
                  if not val['actif']:
                      continue
                  """ Entrée MM : vecteur numpy d'entrée = f(key_dock). """
                  vector_in = o_server.get_vector_in(key_dock)
                  if vector_in is None:
                      continue
                  """ Sortie MM : chaque signal d'entrée peut avoir plusieurs moyennes. """
                  period = val['period']
                  typ = val['typ']
                  num_col_out = val['num_col']
  
                  kwargs = {
                      'nb_last': min(period, self.o_yaml.smma_nb_last),
                      'ratio': self.o_yaml.lwma_ratio
                  }
                  vector_out = indic.get_mm(vector_in, typ=typ, period=period, **kwargs)
                  self.add_vector(key_dock, num_col_out, vector_out)
  

   

• Oui mais... la MM est cliente de Aléatoire, elle a besoin de récupérer ses signaux. La méthode get_vector_in() est chargée de cela.
  ☐ Décommenter la méthode CtrlCalculs.get_vector_in().

---

Vérification :

• ☐ Paramétrer la moyenne mobile comme ceci, pour obtenir 3 signaux : l'original,  une SMA de longueur 14, une autre de longueur 19.

• ☐ Lancer l'affichage :

---

Vérification

---

Snippets

• Moyennes mobiles

---

Bonjour les codeurs !

---