DEV Community: 1N0T

ollama

1N0T — Sat, 10 May 2025 21:00:12 +0000

Python y firma XML

1N0T — Fri, 05 May 2023 21:36:30 +0000

Vamos a ver, cómo firmar documentos XML y, cómo validar si el contenido de un documento firmado, se mantiene inalterado o, por el contrario, ha sufrido alguna modificación.

Como no puede ser de otra manera, lo primero que vamos a hacer es crear un nuevo entorno virtual y a activarlo.

mkdir firmaXML
cd firmaXML
python3 -m venv venv
source venv/bin/activate

Ahora instalamos las librerías que vamos a necesitar.

pip install pip --update
pip install lxml
pip install signxml

Para poder proceder a firmar documentos, deberemos disponer de un certificado x509, aunque podemos crear uno autofirmado con el siguiente comando:

openssl req -x509 -nodes \
        -subj "/CN=Certificado autofirmado" -days 365 \
        -newkey rsa \
        -keyout clave_privada.key \
        -out clave_publica.crt

Ya estamos en disposición de crear un ejemplo de como firmar un XML.

from lxml import etree
from signxml import XMLSigner, XMLVerifier 
from signxml.exceptions import InvalidSignature

# Nuestro contenido XML
XML = """
<test>
   Contenido importante
</test>
"""

# Recuperamos nuestras claves de certificado.Para la firma, se necesitan tanto la
# clave pública como la privada. Para la validación, sólo se requiere la pública.
clave_publica = open("clave_publica.crt", "r").read()
clave_privada = open("clave_privada.key", "r").read()

# Construimos un objeto XNL a partir de string con el contenido
nodo_raiz = etree.fromstring(XML)

# Creamos objeto XML firmado
nodo_raiz_firmado = XMLSigner().sign( 
    data=nodo_raiz,
    key=clave_privada,
    cert=clave_publica 
)

# Convertimos objeto XML a formato string y lo guardamos en fichero para comprobar después
# que se produce un error si manipulamos su contenido
XML_firmado = etree.tostring(nodo_raiz_firmado, pretty_print=True).decode()
with open("XML_firmado.xml", "w") as f:
    f.write(XML_firmado)


# Validamos integridad del contenido XML
try:
    XML_validado = XMLVerifier().verify(
        data=etree.fromstring(XML_firmado),
        x509_cert=clave_publica
    ).signed_xml
except InvalidSignature as err:
    print("ERROR validando integridad XML")
    exit()

print("=== XML original =============================================================")
print(XML)

print("=== XML firmado ==============================================================")
print(XML_firmado)

print("=== XML validado =============================================================")
print(etree.tostring(XML_validado, pretty_print=True).decode())

Creo que el código es bastante autoexplicativo y no necesita de una mayor esplicación. Tras su ejecución, dispondremos de un fichero llamado XML_firmado.xml. Por lo que ahora, ya podemos proceder a comprobar la integridad del fichero.

from lxml import etree
from signxml import XMLSigner, XMLVerifier 
from signxml.exceptions import InvalidSignature

# Recuperamos contenido fichero XML
XML = open("XML_firmado.xml", "r").read()

# Para la validación, sólo se requiere la clave pública.
clave_publica = open("clave_publica.crt", "r").read()

# Construimos un objeto XNL a partir de string con el contenido
nodo_raiz = etree.fromstring(XML)

# Validamos integridad del contenido XML
try:
    XML_validado = XMLVerifier().verify(
        data=etree.fromstring(XML),
        x509_cert=clave_publica
    ).signed_xml
except InvalidSignature as err:
    print("ERROR validando integridad XML")
    exit()

print("=== XML original =============================================================")
print(XML)

print("=== XML validado =============================================================")
print(etree.tostring(XML_validado, pretty_print=True).decode())

Si lo ejecutamos, sin haber modificado el fichero, y después de haber realizado alguna modificación del contenido, podremos observar la diferencia.

Como puedes comprobar, la cosa es bastante simple ¿no?

WSGI Hello world.

1N0T — Sun, 02 Apr 2023 16:23:15 +0000

Si te interesa el desarrollo de aplicaciones web en python, es más que probable que te suenen estas siglas, o incluso, que conceptualmente tengas claro de qué se trata.

WSGI (Web Server Gateway Interface) es simplemente una convención, que define cómo debe ser la comunicación entre un servidor web y una aplicación, que se ajusten a este patrón. Esta definición, se encuentra regogida en el PEP 3333. Bueno, en realidad, ésta es de aplicación para las versiones 3.x de python (la definición inicial para versiones 2.x es la PEP 333.

Entre los paquetes incluidos en python, podemos encontrar un servidor web WSGI de referencia (dentro de wsgiref.simple_server) y una aplicación de demo.

Una vez hechas las presentaciones, vamos a ponernos manos a la obra e implementar nuestra primera versión.

from wsgiref.simple_server import make_server, demo_app

IP = "0.0.0.0"
PUERTO = 8080
APP = demo_app

print(f"Iniciando aplicación APP en servidor WSGI en { IP }: { PUERTO }")
with make_server(IP, PUERTO, APP) as servidor_wsgi:
    print("Puedes finalizar con CTRL+C")
    servidor_wsgi.serve_forever()

Si ejecutamos el código que acabamos de escribir, habremos levantado un servidor web WSGI, esperando peticiones HTTP al puerto 8080.

Ahora podemos abrir nuestro navegador favorito y solicitar la siguiente URL (http://localhost:8080). Y, ¡oh sorpresa!, aparece una página con, la tan manida frase en el mundo de la programación Hello world!, seguida de una probablemente larga lista de VARIABLE = 'valor'.

Llegados a este punto, ye está conseguido el objetivo indicado en el titular, por lo que podríamos ya dar por concluido el artículo. Pero no te preocupes, aunque no mucho, vamos a profundizar un poco más en el tema.

Resumiendo al extremo, una aplicación WSGI, tiene que ser un callable (una función, o un objeto que implemente el método __call__) que recibirá 2 parámetros. El primero es el entorno de ejecución y, el segundo, un callback, que debe ser llamado para construir la respuesta.

Nuestra aplicación de demo, nos muestra el contenido del entorno de ejecución. Un diccionario con los valores de nuestras variables de entorno, a las que se han añadido algunos valores adicionales incluidos por nuestro servidor WEB (como PATH_INFO = '/' que contiene ruta del recurso que hemos solicitado anteriroremnte al servidor). Por otra parte, se espara que nuestra aplicación devuelva la respuesta a entregar al solicitante.

Vamos pues, a crear nuestra primera aplicación WSGI, dejando nuestro código, con algo parecido a:

from wsgiref.simple_server import make_server


#=============================================================================
# Esta función es nuestra aplicación WSGI y, como tal, al ser llamada
# recibirá los 2 parámetros esperados:
#   * diccionario con el entorno de ejecución.
#   * callback para crear respuesta.
#=============================================================================
def mi_app_wsgi(entorno, responder):

    # Configuramos opciones de respuesta por defecto
    cabeceras = [('Content-type', 'text/plain; charset=utf-8')]

    # Iniciamos respuesta
    responder('200 OK', cabeceras)    
    respuesta = "Contenido devuelto por mi_app_wsgi."

    # El cuerpo de la respuesta no puede ser un string.
    # Debe ser una lista de bytes.
    return [respuesta.encode("utf-8")]


IP = "0.0.0.0"
PUERTO = 8080
APP = mi_app_wsgi


print(f"Iniciando aplicación APP en servidor WSGI en { IP }: { PUERTO }")
with make_server(IP, PUERTO, APP) as servidor_wsgi:
    print("Puedes finalizar con CTRL+C")
    servidor_wsgi.serve_forever()

La nueva versión, no ha cambiado mucho. Hemos añadido la función mi_app_wsgi, que recibirá los 2 parámetros esperados y, nota importante, devolverá una lista de bytes (si devolvemos un string saltará una excepción, ya que no se corresponde con lo esperado).

También es necesario que llamemos a la función de calback (la hemos llamado responder, en nuestro caso concreto). Por lo demás, creo que el código es lo suficientemente explícito, como para no requerir de ninguna explicación adicional.

Si ejecutamos nuestra nueva versión y, volvemos a solicitar la URL http://localhost:8080 creo que es bastante previsible cual será el resultado que obtendremos.

También lo podríamos dejar aquí pero, ya que estamos puestos, vamos ha hacer algo un poquito más elaborado.

from wsgiref.simple_server import make_server
from base64 import b64decode



#=============================================================================
# Esta función es una aplicación WSGI y, como tal, al ser llamada
# recibirá los 2 parámetros esperados:
#   * diccionario con el entorno de ejecución.
#   * callback para crear respuesta.
#=============================================================================
def mi_app_wsgi(entorno, responder):

    # Configuramos opciones de respuesta por defecto
    cabeceras = [('Content-type', 'text/plain; charset=utf-8')]

    # Iniciamos respuesta
    responder('200 OK', cabeceras)    
    respuesta = "Contenido devuelto por mi_app_wsgi."

    # El cuerpo de la respuesta no puede ser un string.
    # Debe ser una lista de bytes.
    return [respuesta.encode("utf-8")]



#=============================================================================
# Esta clase es nuestra aplicación WSGI y, como tal, al ser llamada
# recibirá los 2 parámetros esperados:
#   * diccionario con el entorno de ejecución.
#   * callback para crear respuesta.
#=============================================================================
class AutenticacionBasica():

    USUARIO = "mi_usuario"
    PASSWORD = "mi contraseña"


    def __init__(self, app):
        self._app = app


    def __call__(self, entorno, responder):
        # Comprobamos si la solicitud tiene una cabecera de autenticación válida.
        # Si esto es así, llamamos a nuestra otra aplicación WSGI y devolveremos
        # lo que ésta nos devuelva.
        if self._autenticado(entorno.get('HTTP_AUTHORIZATION')):
            return self._app(entorno, responder)

        # Solicitamos autenticación para peticiones no autenticadas-
        return self._no_autenticado(entorno, responder)


    def _autenticado(self, cabecera_autenticacion):
        # Comprobamos si hemos recibido cabedera de autenticación
        if not cabecera_autenticacion:
            return False

        # Si hemos recibido cabedera de autenticación, decodificamos contenido y
        _, valor_codificado = cabecera_autenticacion.split(None, 1)
        valor_decodificado = b64decode(valor_codificado).decode('utf-8')
        username, password = valor_decodificado.split(':', 1)

        # comprobamos si usuario y contraseña se corresponden con valores aceptados
        # como válidos.
        return username == self.USUARIO and password == self.PASSWORD


    def _no_autenticado(self, entorno, responder):
        # Configuramos cabecras por defecto de respuesta
        cabeceras = [
            ('Content-type', 'text/plain; charset=utf-8'),
            ('WWW-Authenticate', 'Basic realm="Login"')
        ]

        # Iniicamos respuesta indicando que se requiere autenticación previa.
        responder('401 Authentication Required', cabeceras)
        respuesta = "Autenticación de usuario es requerida."

        return [respuesta.encode("utf-8")]



IP = "0.0.0.0"
PUERTO = 8080
APP = AutenticacionBasica(mi_app_wsgi)


print(f"Iniciando aplicación APP en servidor WSGI en { IP }: { PUERTO }")
with make_server(IP, PUERTO, APP) as servidor_wsgi:
    print("Puedes finalizar con CTRL+C")
    servidor_wsgi.serve_forever()

En este caso, hemos añadido una clase callable como aplicación WSGI. Su método __call__ recibirá los parámetros esperados según lo especificado en WSGI y, adicionalmente (durante su instanciación), la misma aplicación WSGI del ejemplo anterior sin modificaciones.

Con este ejemplo, vemos como podemos interponer un midleware que, en nuestro caso, forzará la autenticación previa del usuario, antes de poder a nuestra aplicación inicial (sin tener que modificar ésta). Por cuestión de simplicidad del código, hemos mantenido todo en un único fuente, pero, mi_app_wsgi, en lugar de una simple función, podría ser una compleja aplicación Flask p.ej. y la clase AutenticacionBasica podría implementar una elaborada validación de tokens JWT.

Creo que, ahora sí, es un buen momento para concluir el artículo. Espero haber aportado algo que te haya resultado de utilidad.

Bottle + https

1N0T — Fri, 10 Mar 2023 20:37:56 +0000

Bottle, es un framework web, escrito en python, que por su tamaño (un único fichero de 149K), entraría dentro de la categoría de microframework, aunque puede cubrir la mayoría de nuestras necesidades.

Una de sus limitaciones, es que no soporta de base HTTPs, pero como enseguida veremos, esta limitación es fácilmente subsanable.

Naturalmente, lo primero que necesitaremos, es un certificado. Si disponemos de uno firmado por una CA será perfecto. En caso contrario, no importa, vamos a crear nuestro propio certificado autofirmado.

Naturalmente, como no puede ser de otra manera, en este último caso, recibiremos una advertencia del navegador indicanco que el certificado no es confiable.

Aquí tienes un pequeño script en bash que creará las claves pública y privada del certificado. Naturalmente, deberás cambiar la variable DOMINIO, por el valor que te interese.

#!/bin/bash

# Este script crea un certificado autofirmado.
#    Clave privada: mi.dominio.net.key
#    Clave pública: mi.dominio.net.crt

DOMINIO=mi.dominio.net
DIASVALIDEZ=3650
CLAVEPRIVADA=$DOMINIO.key
CLAVEPUBLICA=$DOMINIO.crt

openssl req -x509 -nodes -days $DIASVALIDEZ -newkey rsa:2048 \
       -subj /CN=$DOMINIO                                    \
       -keyout $CLAVEPRIVADA                                 \
       -out $CLAVEPUBLICA

Como imagino que harás las pruebas en tu equipo local, no te olvides de hacer una entrada en tu fichero /etc/hosts para que se pueda resolver la IP de tu dominio ficticio.

127.0.0.1 mi.dominio.net

Ya tenemos nuestro certificado, así que vamos a crear nuestro entorno virtual y ha instalar lo que necesitamos. Bottle y gunicorn (que será el encargado de gestionar la comunicación https).

python3 -m venv venv
source venv/bin/activate
pip install bottle
pip install gunicorn

Ya sólo nos queda crear nuestro maravilloso sitio https.

import os
from bottle import (route, run, request, response, abort, error, static_file) 


DOMINIO = "mi.dominio.net"


#===============================================================================
# Configuramos directorios para recursos estáticos.
#===============================================================================
BASE_DIR   = os.path.dirname(os.path.abspath(__file__))
STATIC_DIR = os.path.join(BASE_DIR, 'static')


@route('/', ['GET'])
def home():
   return "Sitio seguro !!!"


#===============================================================================
# Iniciamos el servidor web que expone las funciones en todas sus direcciones
# de red.
# 
# En lugar de 443 para https, para no requerir permisos de root, usamos el 8443.
#===============================================================================
run(
    reloader = True, 
    debug = True, 
    host = "0.0.0.0", 
    server='gunicorn',
    keyfile=f"{ DOMINIO }.key",
    certfile=f"{ DOMINIO }.crt",
    port = 8443
)

El secreto, bottle cumple con la espacificación de WSGI (Web Server Gateway Interface) y, nos permite ejecutarlo como servidor web autónomo, o delegar en un servidor que cumpla con la especificación. En nuestro caso, gunicorn que es quien nos dará el soporte para https.

Ya solo nos queda ejecutar nuestro servidor, con algo parecido a:

python3 main.py

y comprobar el resultado de nuesto magnífico trabajo

https://mi.dominio.net:8443

Fácil ¿no?

Alfresco en Rocky Linux

1N0T — Sat, 04 Mar 2023 20:14:45 +0000

Recientemente, he tenido que instalar un Alfresco 7.2 en un servidor con Rocky Linux. Seguramente, la calidad de este post no merezca miles de likes, pero como tuve que dedicar cierto tiempo para conseguir que todo funcionara correctamente, he decidido compartirlo por si a laguien le pudiera resultar de utilidad.

El proceso de instalación, está perfectamente detallado aquí, por lo que no voy a repetirlo pero, si te voy a indicar aquellos pasos que han requerido alguna acción adicional.

Yo opté por la instalación con ansible en un servidor único.

En un servidor con una instalación convencional de Rocky Linux 8, segui las intruciones de la documentación que en resumidas cuentas sería algo así:

yum update
yum install unzip https://dl.fedoraproject.org/pub/epel/epel-release-latest-7.noarch.rpm
yum update --nobest
yum install ansible
mkdir alfresco_tmp
cd alfresco_tmp
wget https://nexus.alfresco.com/nexus/service/local/repositories/releases/content/org/alfresco/alfresco-ansible-deployment/1.1.1/alfresco-ansible-deployment-1.1.1.zip
unzip alfresco-ansible-deployment-1.1.1.zip
cd alfresco-ansible-deployment-1.1.1

Hasta aquí ningún problema, pero al ejecutar el playbook de instalación, me encuentré con la sorpresa de que la cosa no fue demasiado bien. Así que te aconsejo que te detengas aquí un momento y, no sigas con la instalación.

Tras algo de troubleshooting descubro que, el origen de todo, está en que hay una serie de variables que no están definidas en caso de que la instalación se ejecute sobre Rocky Linux. Una vez descubierto esto, la solución es fácil, crear las variables a partir de las de sus distros gemelas.

cp ./roles/sfs/vars/RedHat.yml ./roles/sfs/vars/Rocky8.yml
cp ./roles/repository/vars/RedHat.yml ./roles/repository/vars/Rocky8.yml
cp ./roles/search/vars/RedHat.yml ./roles/search/vars/Rocky8.yml
cp ./roles/trouter/vars/RedHat.yml ./roles/trouter/vars/Rocky8.yml
cp ./roles/sync/vars/RedHat.yml ./roles/sync/vars/Rocky8.yml
cp ./roles/nginx/vars/RedHat.yml ./roles/nginx/vars/Rocky8.yml
cp ./roles/adw/vars/RedHat.yml ./roles/adw/vars/Rocky8.yml
cp ./roles/common/vars/CentOS8.yml ./roles/common/vars/Rocky8.yml
cp ./roles/postgres/vars/CentOS8.yml ./roles/postgres/vars/Rocky8.yml
cp ./roles/transformers/vars/CentOS8.yml ./roles/transformers/vars/Rocky8.yml

Ahota sí, ya podemos ejecutar el playbook

ansible-playbook playbooks/acs.yml -i inventory_local.yml -e "@community-extra-vars.yml"

Esto tardará unos minutos pero, tenemos que agradecer a los autores que nos hayan puesto tan fácil la instlación. Al finalizar, tendremos un alfresco completamente funcional, eso sí, si tienes activado el firewall, asegurate de que permites las conexiones entrantes.

firewall-cmd --permanent --add-service=http
firewall-cmd --permanent --add-service=https
firewall-cmd --reload

Llegados a este punto, ya podemos entra en http://mi-servidor/share con el usuario admin y la contraseña admin. Supongo que no hace falta que recomiende el cambio de contraseña como la primera acción a realizar tras comprobar que la instlación ha funcionado correctamente.

Ya podemos utilzar usuarios locales, pero lo que me interesaba era permitir el acceso a los usuarios de nuestro Active Directory. La ubicación de los ficheros de configuración ha cambiado respecto a versiones anteriores y, tuve que realizar varias pruebas hasta encontrar la ubicación correcta.

nano /etc/opt/alfresco/content-services/classpath/alfresco-global.properties

Al final del fichero, tenemos que añadir algo parecido a esto:

authentication.chain=alfrescoNtlm1:alfrescoNtlm,ldap1:ldap
synchronization.syncWhenMissingPeopleLogIn=true
synchronization.import.cron=0 0/10 * * * ?

ldap.authentication.active=true
ldap.authentication.allowGuestLogin=true
ldap.authentication.userNameFormat=%s@tudominio.net
ldap.authentication.java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory
ldap.authentication.java.naming.provider.url=ldap://tudominio.net:389
ldap.authentication.java.naming.security.authentication=simple
ldap.authentication.escapeCommasInBind=false
ldap.authentication.escapeCommasInUid=false
ldap.authentication.defaultAdministratorUserNames=Administrator,otrousuario,tuusuario

ldap.synchronization.active=false
ldap.synchronization.java.naming.security.authentication=simple
ldap.synchronization.java.naming.security.principal=cn\=UsuarioAd,OU\=IT,OU\=OtraOU,DC\=tudominio,DC\=net
ldap.synchronization.java.naming.security.credentials=ContraseñaDeUsuarioConexionAD
ldap.synchronization.queryBatchSize=1000
ldap.synchronization.attributeBatchSize=1000
ldap.synchronization.groupQuery=(objectclass\=group)
ldap.synchronization.groupDifferentialQuery=(&(objectclass\=group)(!(whenChanged<\={0})))
ldap.synchronization.personQuery=(&(objectclass\=user)(userAccountControl\:1.2.840.113556.1.4.803\:\=512))
ldap.synchronization.personDifferentialQuery=(&(objectclass\=user)(userAccountControl\:1.2.840.113556.1.4.803\:\=512)(!(whenChanged<\={0})))
ldap.synchronization.groupSearchBase=DC\=tudominio,DC\=net
ldap.synchronization.userSearchBase=DC\=tudominio,DC\=net
ldap.synchronization.modifyTimestampAttributeName=whenChanged
ldap.synchronization.timestampFormat=yyyyMMddHHmmss'.0Z'
ldap.synchronization.userIdAttributeName=sAMAccountName
ldap.synchronization.userFirstNameAttributeName=givenName
ldap.synchronization.userLastNameAttributeName=sn
ldap.synchronization.userEmailAttributeName=mail
ldap.synchronization.userOrganizationalIdAttributeName=company
ldap.synchronization.defaultHomeFolderProvider=largeHomeFolderProvider
ldap.synchronization.groupIdAttributeName=cn
ldap.synchronization.groupDisplayNameAttributeName=displayName
ldap.synchronization.groupType=group
ldap.synchronization.personType=user
ldap.synchronization.groupMemberAttributeName=member
ldap.synchronization.enableProgressEstimation=true
ldap.authentication.java.naming.read.timeout=0

Naturalmente, si quieres que esto funcione con tu Active Directory, tendrás que realizar las adaptaciones pertinentes para que se adapte a tu configuración particular.

Y esto es todo, espero que a alguien le resulte de utilidad.

matplotlib heatmap calendar.

1N0T — Sun, 10 Oct 2021 16:57:40 +0000

Aunque la expresión, no hay que reiventar al rueda, tiene todo el sentido del mundo, también es cierto que si te dedicas a proveer de ruedas a los demás. no viene mal tener una idea clara de como se fabrican.

Nuestro objetivo de hoy, es desarrollar una función que reciba una lista de valores diarios, de algún hecho concreto y, construya un calendario clásico donde se represente las ocurrencias de cada día usando una escala de colores.

Aunque ya existen librerías como July que nos permitirían hacerlo, vamos a explicar desde cero como conseguir un resultado como el siguiente, usando python y matplotlib.

Para no alargarnos demasiado, en esta entrega, nos limitaremos a la salida de la izquierda. Y si ésta despierta interés, mostraré las otras en alguna publicación posterior.

Como hemos comentado, necesitaremos matplotlib, así que, para no interferir con lo que tengamos instalado en nuestro equipo, crearemos un entorno virtual.

mkdir heatmap_calendar
cd heatmap_calendar
python3 -m venv venv
source venv/bin/activate
pip3 install matplotlib

Abrimos nuestro editor favorito, pero antes de empezar, un pequeño recordatorio. Cualquier imagen, en realidad es una matiz de puntos (pixels), cada uno de los cuales acostumbra a estar representado por 3 valores (entre el 0 y el 255), que representan la cantidad de rojo, verde y azul que lo componen y que determinan su color (por eso se dice que es una imagen RGB que se corresponde con las siglas en inglés de los colores mencionados).

Un pixel rojo sería (255, 0, 0), uno verde (0, 255, 0) y, un punto azul se representaría con (0, 0, 255).

Una vez visto esto, en teoría, podríamos definir una imagen con un código parecedo al siguiente:

rojo  = (255, 0, 0)
verde = (0, 255, 0)
azul  = (0, 0, 255)
valores = [
    [rojo,  verde, azul], 
    [verde, azul,  rojo], 
    [azul,  rojo,  verde], 
]

Esto podría representar una imagen de 3x3 pixels rojos, verdes y azules, colocados en diferente posición.

matplotlib dispone de la función imshow que nos permite mostrar una imagen. Vamos a escribir algo de código y ver que pasa:

import matplotlib.pyplot as plt

rojo  = (255, 0, 0)
verde = (0, 255, 0)
azul  = (0, 0, 255)

valores = [
    [rojo,  verde, azul], 
    [verde, azul,  rojo], 
    [azul,  rojo,  verde], 
]

# Creamos el lienzo sobre el que dibujar las gráficas
fig, ax = plt.subplots()

# Mostramos la imagen que hemos definido
ax.imshow(valores)
plt.show()

Voilà !!! tenemos lo que queríamos, una imagen de 3x3 pixels (bastante gordos, por cierto).

Ya disponemos de una forma fácil de dibujar una cuadrícula con idferentes colores (que es una de las cosas que necesitaremos para construir nuestro calendario con mapa de colores).

Hasta el momento, hemos trabajado con una imagen RGB de color, en el que el color de cada pixel (cuadradito) está constituido de una lista de 3 valores, que indican la proporción de rojo, verde y azul que lo componen (todos los colores se pueden conseguir sumando estos tres en su proporción adecuada).

Pero antes de que las imágenes a color fueran una realidad, las imágenes eran en blanco y negro y se representaban con una escala de grises, por lo que para cada punto, en lugar de tres, sólo era necesario un valor (también entre 0 y 255), para indicar la intensidad de gris ha representar.

Vamos a modificar nuestro código, para que viaje atrás en el tiempo, y vamos a representar cada uno de nuestros cuadraditos con un único valor.

import matplotlib.pyplot as plt

valores = [
    [0, 175, 255], 
    [175, 255, 0], 
    [255, 0, 175], 
]

fig, ax = plt.subplots()

# Indicamos a la función que la imagen es de escala de grises en lugar RGB (cmap='gray')
ax.imshow(valores, cmap='gray')
plt.show()

Nuestro código sigue funcionando y, ya tenemos otra de las cosas que necesitamos, representar una cuadrícula, en la que el color de cada elemento representa un valor concreto.

Si en lugar de 3x3, la ampliamos a 7x6, podríamos representar con colores los valores de un mes (7 para los días de la semana y 6 para el máximo de semanas que pueden formar parte de un mismo mes).

Pero seguimos teniendo un par de problemas. Estéticamente, una imagen de grises acostumbra a no ser demasiado agradable y, cada cuadrado sólo permite representar valores entre 0 y 255 (que són los permitidos para una imagen).

Afortunadamente, imshow dispone del parámetro cmap que nos permite especificar una paleta de colores diferentes y, los parámetros vmin y vmax que, como probalemente ya habrás deducido, nos permiten especificar el rango de valores a representar.

import matplotlib.pyplot as plt

valores = [
    [10, 20, 30, 40, 50, 60, 70], 
    [20, 30, 40, 50, 60, 70, 10], 
    [30, 40, 50, 60, 70, 10, 20], 
    [40, 50, 60, 70, 10, 20, 30], 
    [50, 60, 70, 10, 20, 30, 40], 
    [60, 70, 10, 20, 30, 40, 50], 
]

fig, ax = plt.subplots()

# Especificamos paleta de colores a usar y rango de valores a representar.
ax.imshow(valores, cmap='YlGn', vmin=0, vmax=100)
plt.show()

Esto ya empieza a tener cierto parecido con las imágenes de ejemplo mostradas al inicio. Así que, vamos colocarle el nombre de un mes, etiquetas para los días de la semana y a eliminar los números de la izquierda, que no los necesitamos para nada.

import matplotlib.pyplot as plt

# Lista de etiquetas para representar los días de la semana.
dias_semana = ['L', 'M', 'M', 'J', 'V', 'S', 'D']
valores = [
    [10, 20, 30, 40, 50, 60, 70], 
    [20, 30, 40, 50, 60, 70, 10], 
    [30, 40, 50, 60, 70, 10, 20], 
    [40, 50, 60, 70, 10, 20, 30], 
    [50, 60, 70, 10, 20, 30, 40], 
    [60, 70, 10, 20, 30, 40, 50], 
]

fig, ax = plt.subplots()

ax.imshow(valores, cmap='YlGn', vmin=0, vmax=100)
# Colocamos el título
ax.set_title("Enero\n")

# Borramos el contenido de las etiquetas del eje Y 
ax.set_yticklabels([])

# Indicamos las posiciones posiciones del ejs X con las que secorresponden las etiquetas.
ax.set_xticks([0, 1, 2, 3, 4, 5, 6])

# Especificamos las etiquetas para las posiciones especificadas anteriormente.
ax.set_xticklabels(dias_semana)
plt.show()

Ya estamos más cerca, pero tenemos que mover las etiquetas de los días a la parte superior y, elimicar el marco de alrededor que quedará un poco raro cuando desaparezcan los cuadrados sobrantes de inicio y final de mes (la mayoría de las veces, los meses no empiezan en lunes y acaban en domingo).

import matplotlib.pyplot as plt

dias_semana = ['L', 'M', 'M', 'J', 'V', 'S', 'D']
valores = [
    [10, 20, 30, 40, 50, 60, 70], 
    [20, 30, 40, 50, 60, 70, 10], 
    [30, 40, 50, 60, 70, 10, 20], 
    [40, 50, 60, 70, 10, 20, 30], 
    [50, 60, 70, 10, 20, 30, 40], 
    [60, 70, 10, 20, 30, 40, 50], 
]

fig, ax = plt.subplots()

ax.imshow(valores, cmap='YlGn', vmin=0, vmax=100)
ax.set_title("Enero\n")
ax.set_yticklabels([])
ax.set_xticks([0, 1, 2, 3, 4, 5, 6])
ax.set_xticklabels(dias_semana)

# Colocamos las etiquetas del eje X en la parte superior
ax.xaxis.tick_top()

# Ocultamo cada uno de lso dados que circundan el gráfico
for lado in ['left', 'right', 'bottom', 'top']:
    ax.spines[lado].set_visible(False)

plt.show()

Ahora, ya sólo nos falta eliminar las casillas sobrantes y deshacernos de las marquitas que han quedado en los ejes.

import numpy as np
import matplotlib.pyplot as plt

# Colocaremos el valor especial de Not a Number a las cassila que sobran y 
# conseguiremos que maptlotlib no las muestre
NA = np.nan

dias_semana = ['L', 'M', 'M', 'J', 'V', 'S', 'D']
valores = [
    [NA, NA, NA, NA, NA, 60, 70], 
    [20, 30, 40, 50, 60, 70, 10], 
    [30, 40, 50, 60, 70, 10, 20], 
    [40, 50, 60, 70, 10, 20, 30], 
    [50, 60, 70, 10, 20, 30, 40], 
    [60, NA, NA, NA, NA, NA, NA], 
]

fig, ax = plt.subplots()

ax.imshow(valores, cmap='YlGn', vmin=0, vmax=100)
ax.set_title("Enero\n")
ax.set_yticklabels([])
ax.set_xticks([0, 1, 2, 3, 4, 5, 6])
ax.set_xticklabels(dias_semana)
ax.xaxis.tick_top()

# Reducimos la longitud de las marcas a 0 para que no sean visibles
ax.tick_params(axis=u'both', which=u'both', length=0)
for lado in ['left', 'right', 'bottom', 'top']:
    ax.spines[lado].set_visible(False)

plt.show()

Y para acabar, añadimos un pequeño detalle estético.

import numpy as np
import matplotlib.pyplot as plt

NA = np.nan

dias_semana = ['L', 'M', 'M', 'J', 'V', 'S', 'D']
valores = [
    [NA, NA, NA, NA, NA, 60, 70], 
    [20, 30, 40, 50, 60, 70, 10], 
    [30, 40, 50, 60, 70, 10, 20], 
    [40, 50, 60, 70, 10, 20, 30], 
    [50, 60, 70, 10, 20, 30, 40], 
    [60, NA, NA, NA, NA, NA, NA], 
]

fig, ax = plt.subplots()

ax.imshow(valores, cmap='YlGn', vmin=0, vmax=100)
ax.set_title("Enero\n")
ax.set_yticklabels([])
ax.set_xticks([0, 1, 2, 3, 4, 5, 6])
ax.set_xticklabels(dias_semana)
ax.xaxis.tick_top()

ax.tick_params(axis=u'both', which=u'both', length=0)
for lado in ['left', 'right', 'bottom', 'top']:
    ax.spines[lado].set_visible(False)

# Indicamos las posiciones donde dibujaremos la rejilla    
ax.set_xticks([-0,5, 0.5, 1.5, 2.5, 3.5, 4.5, 5.5, 6.5], minor=True)
ax.set_yticks([-0,5, 0.5, 1.5, 2.5, 3.5, 4.5, 5,5], minor=True)
# Dibujamos la rejilla de color blanco para que actue como separador.
ax.grid(which='minor', color='w', linestyle='-', linewidth=2)

plt.show()

Y ya lo tenemos. Hemos conseguido representar en un calendario de un mes, una serie de valores utilizando un código de colores.

Ya hemos explicado lo que podría resultar menos evidente. Lo que queda, es repetir el proceso para los 12 meses y repartir nuestra lista de valores para que encaje, de forma adecuada, con el mes correspondiente.

import calendar
import numpy as np
import matplotlib.pyplot as plt

def heatmap_calendar(year, valores):

    dias_semana = ['L', 'M', 'M', 'J', 'V', 'S', 'D']
    nombres_meses = ['Enero', 'Febrero', 'Marzo', 'Abril', 'Mayo', 'Junio', 'Julio', 'Agosto', 'Septiembre', 'Octubre', 'Noviembre', 'Diciembre']
    maximo_semanas_mes = 6
    meses = 12
    columnas = 3
    filas = 4

    # Creamos un gráfico para cada més del año
    fig, ax = plt.subplots(filas, columnas, figsize=(3.5*columnas, 3.5*filas))

    # Para facilitar la iteración convertimos la tabla 2D en una lista 1D
    axs = np.array(ax).reshape(-1)

    numero_valores = len(valores)
    maximo = max(valores)
    puntero_dia_actual = 0

    # Repetimos el proceso para cada mes
    for mes in range(meses):
        # Averiguamos que día de la semana empieza el mes y cuantos días tiene
        dia_semana, dias_mes = calendar.monthrange(year, mes + 1)

        # Creamos un array nullo como plantilla para ir colocando los valores del mes
        plantilla_mes = np.empty((maximo_semanas_mes, len(dias_semana)))
        plantilla_mes[:] = np.nan

        # Mostramos nombre del mes
        axs[mes].set_title(nombres_meses[mes]+'\n', fontsize=12)

        # Colocamos etiqueta de días de la semana.
        axs[mes].set_xticks(np.arange(len(dias_semana)))
        axs[mes].set_xticklabels(dias_semana, fontsize=10, fontweight='bold', color='#555555')
        axs[mes].set_yticklabels([])
        axs[mes].xaxis.tick_top()

        # Ocultamos las marcas de los ejes
        axs[mes].tick_params(axis=u'both', which=u'both', length=0)  # remove tick marks

        # Dibujamos rejilla blanca para separar casillas
        axs[mes].set_xticks(np.arange(-.5, len(dias_semana), 1), minor=True)
        axs[mes].set_yticks(np.arange(-.5, maximo_semanas_mes, 1), minor=True)
        axs[mes].grid(which='minor', color='w', linestyle='-', linewidth=2)

        # Ocultamos cada uno de los lados del marco que rodea la cuadrícula.
        for lado in ['left', 'right', 'bottom', 'top']:
            axs[mes].spines[lado].set_visible(False)

        # Empezamos a rellenar cada mes desde la primera fila y desde el día de la
        # semana en la que empieza el mes        
        fila = 0
        columna = dia_semana

        # Copiamos tantos valores de nuestra lista como días tiene el mes
        for n in range(dias_mes):
            if puntero_dia_actual < numero_valores:
                plantilla_mes[fila][columna] = valores[puntero_dia_actual]

            # Conservamos cual es el próximo valor de nuestra lista a procesar
            puntero_dia_actual += 1

            # Controlamos cuando se produce cambio de semana
            if columna == 6:
                columna = 0
                fila +=1
            else:
                columna += 1

        # Mostramos los valores del mes        
        axs[mes].imshow(plantilla_mes, cmap='YlGn', vmin=1, vmax=maximo)

    # Colocamos el año del calendario y reajustamos el conjunto para dejas
    # un pequeño margen alrededor        
    fig.suptitle("\n" + str(year), fontsize=16)
    plt.subplots_adjust(left=0.04, right=0.96, top=0.88, bottom=0.04)

    # La visualización online es correcta en el caso de jupyter notebook, pero hay solapamientos
    # incorrectos si lo vemos directamente con el visor de matplotlib. Salvamos en un documento PDF
    # para tener la salida correcta
    plt.savefig('heatmap_calendar.pdf')
    plt.show()

if __name__ == '__main__':
    # Generamos valores aleatorios para probar la función
    valores = np.random.randint(500, size=365)
    heatmap_calendar(2021, valores=valores)

La programación, a veces, tiene bastantes similitudes con la magia. Lo que ves parece imposible, hasta que alguien te explica el truco y piensas, "pues vaya tontería".

Hemos mostrado la versión más básica del truco, pero seguro que se te ocurren multitud de variantes para enrriquecerlo.

Timeline con matplotlib.

1N0T — Sun, 03 Oct 2021 13:54:48 +0000

Si estás aquí, lo más probable es que ya conozcas matplotlib. Y si es así, casi seguro que te habrá sorprendido la elección de esta librería para representar un timeline.

Lo cierto es que el procedimiento para obtener un resultado decente es algo engorroso. Pero resolver este tipo de circunstancias es una de las cosas que justifica nuestro trabajo como desarrolladores. Así que, ¿te interesa una función que genere un gráfico como el siguiente?

Bueno, ha llegado la hora de poner manos a la obra, así que vamos a crear un entorno virtual e instalar las librerías necesarias.

mkdir matplotlib_timeline
cd matplotlib_timeline
python3 -m venv venv
source venv/bin/activate
pip3 install matplotlib

Ahora, abrimos nuetro editor favorito y tecleamos el siguiente código:

import matplotlib.pylab as plt

def timeline(titulo, eventos_fechas, eventos_textos, eventos_delta_y):

    # Representaremos los años dentro de una línea de 10 unidades
    ESCALA = 10.0

    # Dejaremos un margen e una unidad, por delante y por detrás en el eje X, por lo que el gráfico
    # contendrá 12 unidades (las 10 de los años + 2 de los márgenes).
    MARGEN = 1
    MARGEN_DELTA = 1.0 / (ESCALA + MARGEN * 2) 

    # Averiguamos la distancia en años entre el primer acontecimiento y el último, para calcular
    # a cuantas unidades de nuestro eje X se corresponde un año. 
    fecha_minimo = min(eventos_fechas)
    fecha_maximo = max(eventos_fechas)
    fecha_rango = fecha_maximo - fecha_minimo
    fecha_posicion = ESCALA / fecha_rango

    # Generamos una lista con la posición relativa que cada año tiene que ocupar dentro del eje X
    # Y por comodidad, una lista de posición en el eje Y, para los puntos y las marcas de texto
    fechas_x = [(fecha - fecha_minimo) * fecha_posicion + MARGEN for fecha in eventos_fechas]
    fechas_y = [0 for fecha in eventos_fechas]
    lineas_y = [1 for fecha in eventos_fechas]

    # Creamos el gŕafico y establecemos los límites que nos interesan
    fig, ax = plt.subplots(figsize=(15, 8),  constrained_layout=True)
    ax.set_ylim(-2, 2)
    ax.set_xlim(0, ESCALA + MARGEN * 2)
    ax.set_title(titulo, fontweight="bold", fontfamily='sans', fontsize=22, color='#4a4a4a')

    # Dibujamos la línea sobre la que representar los puntos del timeline
    ax.axhline(0, xmin=MARGEN_DELTA, xmax=1-MARGEN_DELTA, c='#4a4a4a', zorder=1)

    # Dibujamos sobre la línea los puntos correspondientes a cada año de la lista de acontecimientos.
    # Primero un círculo grande y después otro más pequeño encima, para dar la sensación de una
    # circunferencia con borde y rellena de color.
    ax.scatter(fechas_x, fechas_y, s=120, c='#4a4a4a', zorder=2)
    ax.scatter(fechas_x, fechas_y, s=30, c='#faff00', zorder=3)

    # Establecemos el tamaño de los texto, así como la separación vertical respecto a la
    # linea vertical de marca. 
    FONT_SIZE = 12
    DELTA_TEXTO = 0.1

    # Determinamos la posición que debe ocupara cada texto dentro del gráfico
    # En el eje X coincide con la posición del año sobre el eje.
    # En el eje Y lo determina el desplazamiento recibido como parámetro, más la separación
    # adicional estipulada. 
    for x, fecha, texto, delta_y  in zip(fechas_x, eventos_fechas, eventos_textos, eventos_delta_y ):

        # El cálculo es diferente para valores positivos y negativos
        DELTA_TEXTO_y = delta_y + DELTA_TEXTO if delta_y > 0 else delta_y - DELTA_TEXTO

        # Cambiamos la alineación vertical en función de si están por enciama, o por debajo
        # del eje. Así nos aseguramos una disposición consistente en el caso de textos con
        # más de una línea (sin necesidad de tener que ajustar la posición en función
        # del tamaño real del texto)
        va = 'bottom' if delta_y > 0 else 'top'

        # Para mayor claridad, anteponemos al texto un línea con el año al que corresponde  
        ax.text(
            x, DELTA_TEXTO_y, str(fecha) + "\n" + texto, 
            ha='center', va=va, 
            fontfamily='sans', fontweight='bold', fontsize=FONT_SIZE,
            color='#4a4a4a'
        )

    # Establecemos márcas verticales para cada acontecimiento. La posición en el eje X,
    # nuevamente, la determina el año, y la longitud vertical la que hemos recibido como
    # parámetro (con la que hemos ajustado la distancia vertical del texto)
    markerline, stemline, baseline = ax.stem(fechas_x, eventos_delta_y, use_line_collection=True)

    # Damos estilo a las marcas verticales    
    plt.setp(baseline, zorder=0)
    plt.setp(markerline, marker='', color='#4a4a4a')
    plt.setp(stemline, color='#4a4a4a')


    # Para que el gŕafico quede más limpio, eliminamos las marcas de los ejes ejes
    ax.set_xticks([]) 
    ax.set_yticks([]) 

    # También eliminamos líneas del marco
    for spine in ["left", "top", "right", "bottom"]:
        ax.spines[spine].set_visible(False)

    # Mostramos resultado    
    plt.show()


# Título del timeline
titulo = "\nNuevos antibióticos desde el año 2000"

# Relación de años a mostrar en el timeline
eventos_fechas = [
    2000,
    2001,
    2003,
    2005,
    2005,
    2009,
    2010,
    2011,
    2012,
    2013,
    2014,
    2014,
    2015,
    2017,
    2019,
    2019,
]   

# Evento ocurrido en el año de la relación anterior
eventos_textos = [
    "linezolid",
    "elithromycin",
    "daptomycin",
    "tigecycline",
    "doripenem",
    "telavancin",
    "ceftaroline",
    "fidaxomicin",
    "bedaquiline",
    "telavancin",
    "tedizolid",
    "dalbavancin\nceftolozane\ntazobactam",
    "ceftazidime\navibactam",
    "meropenem\nvaborbactam",
    "imipenem\ncilastatin\nrelebactam",
    "cefiderocol",
]

# Dado que dependiendo de la longitud de los texto, es bastante probable el solapamiento
# de los mismmos, para cada uno de ellos indicamos el desplazamiento respecto al eje central.
eventos_delta_y = [
    0.5,
   -0.5,
    0.5,
   -0.5,
    0.5,
   -0.5,
    0.5,
   -0.5,
    0.5,
   -0.5,
    0.5,
   -0.8,
    0.8,
   -0.5,
    0.5,
   -0.5,
]

# Llamamos a la función que genera el gráfico del timeline
timeline(titulo, eventos_fechas, eventos_textos, eventos_delta_y)

Como puedes comprobar, he añadido comentarios a casi todas las líneas, por lo que no creo que sea necesario hacer explicaciones adicionales al còdigo.

La función espera recibir el título, y 3 listas con los datos a representar, después se encarga de hacer el resto del trabajo sucio.

Lista de años para cada evento a representar.
Lista con los textos descriptivos del evento.
Lista con desplazamiento vertical de los textos, respecto al eje central. Jugando con estos valores, podemos lidiar con, el más que probable solapamiento de textos, entre eventos próximos en el tiempo.

Al final, creo que noa ha quedado una función que permite obtener un resultado decente, con muy poco esfuerzo.

Espero que la disfrutes.

Bollinger bands ®

1N0T — Sun, 12 Sep 2021 18:08:03 +0000

Si lo que buscas es una explicación de la mano de un experto en la materia, te aconsejo que continues con tu búsqueda, mis conocimientos financieros se podría calificar como prácticamente nulos. No obstante, últimamaente he estado leyendo publicaciones relacionada con el mundo financiero y éste, me ha parecido un concepto interesante.

Si estás leyendo esto, imagino que ya sabes en que consisten, por lo que no te voy a aburrir con los detalles (de no ser así encontrarás infinidad de entradas si haces una búsqueda del término). Pero creo que si que te puede resultar de insterés el código de una función que construya una representación gráfica de las mismas para un valor determinado.

Para ello, vamos a utilizar el API de Yahoo finance, para recuperar la información de una acción determinada, python como lenguaje de programación, pandas para realizar los cálculos necesarios y matplotlib para crear el gráfico deseado y que tendrá el siguiente aspecto.

Bueno, ha llegado la hora de poner manos a la obra, así que vamos a crear un entorno virtual e instalar las librerías necesarias.

mkdir bollinger_bands
cd bollinger_bands
python3 -m venv venv
source venv/bin/activate
pip3 install yfinance pandas matplotlib

Ahora, abrimos nuetro editor favorito y tecleamos el siguiente código:

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import yfinance as yf
from datetime import datetime, timedelta

def bollinger_band(stock_name, desde=datetime.today() - timedelta(days=365) , window=20):
    # La función espera recibir los siguientes parámetros:
    #   stock_name: ID que identifica a la acción p.ej "GOOGL" en el caso de Google
    #   desde: Fecha desde la cual se desea iniciar la representación. Si no se especifica
    #          nada, por defecto, se tomará un periodo de 1 año contando a partir de hoy
    #   windows: Número de días a considerar para los cálculos de la media móvil simple y
    #            de las desviaciones standar. Si no se especifica, se tomarán 20 días.
    #
    # Documentación Yahoo finance en https://pypi.org/project/yfinance/ 

    # Utilizamos la API de Yahoo finance para recuperar la 
    # información histórica de las cotizaciones en bolsa para
    # las acciones de la empresa seleccionada.
    stock_data = yf.Ticker(stock_name)

    # Guardamos todos los datos recuperados en un DataFrame de pandas.
    # Aunque podríamos limitar el periodo deseado en la descarga, dado que el 
    # volumen de información es reducido, descargamos toda la información disponible
    # y realizaremos el filtrado posteriormente. 
    stock_hist = stock_data.history(period="max", auto_adjust=True)

    # Añadimos la columna "mean" al DataFrame con el valor de la SMA (media móvil simple)
    # tomado en consideración la cantidad de días deseados (20 por defecto)
    stock_hist['mean'] = stock_hist["Close"].rolling(window=window).mean()

    # Guardamos desviación standar para el mismo periodo considerado para la media.
    stock_hist['std'] = stock_hist["Close"].rolling(window=window).std()

    # Creamos columna con el valor de la banda superior 
    # (considerando 2 desviaciones standar)
    stock_hist['BU'] = stock_hist["mean"] + 2 * stock_hist["std"]

    # Creamos columna con el valor de la banda inferior 
    # (considerando 2 desviaciones standar)
    stock_hist['BD'] = stock_hist["mean"] - 2 * stock_hist["std"]

    # Añadimos columna que contendrá los valores que superan la banda superior.
    # Se supone que deberíamos considerar la venta de acciones por estár sobrevaloradas.
    stock_hist['BU point'] = stock_hist['Close']
    condicion = (stock_hist['BU point'] <= stock_hist['BU'])
    stock_hist.loc[condicion,'BU point'] = np.nan

    # Añadimos columna que contendrá los valores que sean menores a la banda inferior.
    # Se supone que deberíamos considerar la compra de acciones por estár infravaloradas.
    stock_hist['BD point'] = stock_hist['Close']
    condicion = (stock_hist['BD point'] >= stock_hist['BD'])
    stock_hist.loc[condicion,'BD point'] = np.nan

    # Creamos el lienzo para dibujar la imagen
    plt.figure(figsize=(15, 6))

    # Calculamos la fecha desde la cual queremos iniciar la representación.
    inicio = pd.to_datetime(desde)

    # Dibujoas los valores de cierre de la acción.
    stock_hist.loc[inicio:,"Close"].plot(alpha=0.5)

    # Para no sobrecargar el gráfico, no mostraremos las líneas que representan las bandas.
    # Pero si deseas añadirlas, descomenta las líneas siguientes.
    # stock_hist.loc[inicio:,"BU"].plot()
    # stock_hist.loc[inicio:,"BD"].plot()

    # Mostramos en rojo las supuestas oportunidades de venta. Donde el valor de cierre
    # supera la banda superior.
    stock_hist.loc[inicio:,"BU point"].plot(lw=3, color={"BU point": "red"})

    # Mostramos en verde las supuestas oportunidades de compra. Donde el valor de cierre
    # es inferiro la banda inferior.
    stock_hist.loc[inicio:,"BD point"].plot(lw=3, color={"BD point": "green"})

    # Como no hemos representado las bandas, sombreamos la zona entre ambas pra que el
    # rango sea visible.
    plt.fill_between(stock_hist[inicio:].index, stock_hist.loc[inicio:,'BD'], stock_hist.loc[inicio:,'BU'], facecolor='orange', alpha=0.25)

    # Para que sea más fácil ver su valor, dibujamos un línea horizontal con el
    # último valor de cierre conocido para la acción deseada.
    plt.axhline(stock_hist.loc[stock_hist.index[-1], 'Close'], color='k', linestyle='--')

    # Colocamos título y etiquetas.
    plt.title(f"Bollinger bands® {stock_name} ventana={window}")
    plt.xlabel("Fecha")
    plt.ylabel("Precio de cierre")
    plt.ylim(0)
    plt.show()


# Seleccionamos el identificador de la empresa que nos interese y llamamos a la función
# que creará el gráfico. En este caso, he usaso el Banco Santander que muestra toda 
# la casuística.
STOCK = "SAN"
bollinger_band(STOCK)

Y ya tenemos lo que, al parecer, es una de las herramientas principales para el análisis tècnico de los mercados de valores.

He añadido comentarios a casi todas las líneas, por lo que no creo que sea necesario hacer explicaciones adicionales al còdigo.

Ya para finalizar, he de decir que, aunque el concepto me pareció interesante, por lo que he podido ver al analizar diferetnes ejemplos, no creo que su aplicación estricta sea la panacea para aumentar espectacularmente nuestros beneficios. No obstante, como comenté al principio, no soy ningún especialista en la materia, por lo que si tiene tanta popularidad, digo yo que por algo será...

Yahoo finance + Facebook prophet.

1N0T — Sun, 29 Aug 2021 18:28:11 +0000

Quien más, quien menos, ha deseado en algún momento ser lo suficientemente rico como para poder dejar de trabajar, sin tener que preocuparse por los recursos económicos disponibles.

Hay quien dice que esto se puede conseguir trabajando duro. Pero os he de decir que yo llevo muchos años intentandolo, sin demasiado éxito.

Otra alternativa sería dar el golpe del siglo. He revisado toda la filmografía sobre el tema, y no he encontrado ningún plan que pueda funcionar en la vida real.

Ya sólo me queda empezar a especular en bolsa, pero como no tengo ni idea sobre la materia, tendré que suplirlo tirando de mis habilidades en programación.

Vamos a utilizar el ya omnipresente python (a estas alturas, ya no hay discusión sobre si debe ser la versión 3). Así que, siguiendo las buenas prácticas, vamos a crear nuestro entorno virtual, donde instalaremos las librerías que vamos a necesitar.

mkdir prophetStockPrice
cd prophetStockPrice/
python3 -m venv venv
source venv/bin/activate

Nuestro entorno, ya está preparado para empezar a instalar las librerías que necesitaremos.

pip3 install pandas matplotlib yfinance fbprophet

Ya tenemos todo lo necesario, así que abrimos nuestro editor preferido y, escribimos algo parecido a lo siguiente:

import yfinance as yf
import pandas as pd
import matplotlib.pyplot as plt
from fbprophet import Prophet

def predict_stock(stock_name): 
    # Utilizamos la API de Yahoo finance para recuperar la 
    # información histórica de las cotizaciones en bolsa para
    # las acciones de la empresa seleccionada.
    stock_data = yf.Ticker(stock_name)

    # Guardamos todos los datos recuperados en un DataFrame
    # de pandas.
    stock_hist = stock_data.history(period="max", auto_adjust=True)

    # Creamos un DataFrame vacío, al que añadiremos la 
    # información que necesita la librería prophet de
    # Facebook.
    stock_df = pd.DataFrame()

    # Le añadimos la columna con la serie de fechas (se
    # tiene que llamar 'ds').
    stock_df['ds'] = stock_hist.index

    # Copiamos los valores de las cotizaciones en el momento
    # del cierre de la sesión de bolsa (se tiene que llamar
    # 'y').
    stock_df['y'] = stock_hist['Close'].values

    # Ya disponemos de la información en el formato 
    # requerido.
    # Creamos una instancia del predictor (presuponiendo que
    # nuestros datos no están sujetos a ciclos diarios).
    m = Prophet(daily_seasonality=False)

    # Entrenamos nuestro modelo predictor con los datos
    # diponibles, para que encuentre posibles patrones
    # predicores.
    m.fit(stock_df)

    # Creamos una secuencia de días futuros vacios (de un año
    # en este caso).
    future = m.make_future_dataframe(365 * 1, freq='D')

    # Predecimos los posibles valores para los días que
    # acabamos de añadir (aplicará las reglas encontradas
    # en el paso anterior).
    forecast = m.predict(future)

    # Mostramos la gráfica con los resultados de la
    # predicción.
    m.plot(forecast)
    plt.show()

# Informamos las iniciales que identifican a la empresa
# que hemos seleccionado.
STOCK = "FB"

# Llamamos a la función que realizará todo el trabajo
# por nosotros.
predict_stock(STOCK)

Y he aquí el resultado obtenido.

Según las predicciones del propio Facebook, invertir en sus acciones, va a resultar un negocio rentable durante el próximo año. Es posible que esto pueda ser cierto, pero antes de ir corriendo a decirle a tu agente en bolsa que compre todas las acciones disponibles de la compañía, ten presente que, si bien la bolsa parece tener algún tipo de ciclos temporales, hay todo tipo de variables que la pueden infuir de forma considerable.

Imagina que pasaría si, al día siguiente a tu compra, se hace público que su seguridad ha sido comprometida y han sido robados todos los datos de sus usuarios ...

Es posible incluso que se pudiera producir un caso de profecía autocumplida. Imagina que todo el mundo utilizara prophet para predecir las cotizaciones en bolsa. Cuando arrojara un resultado positivo, todos a comprar como locos (haciendo que suba la cotización). Cuando la predicción fuera negativa, todos a vender para no perder dinero (lo que provocaría la pérdida de valor en bolsa).

Lo más probable es que esto no te sirva para convertirte en millonario (no era esa mi intención), pero espero que te pueda resultar de utilidad en algún caso.

Gestiona Oracle cloud con ansible.

1N0T — Thu, 08 Oct 2020 19:05:02 +0000

Aunque, al tratarse de uno de los úlltimos en llegar, no están entre los proveedores de servicios cloud más populares; lo cierto es que durante el último año, al menos según mi impresión particular, ha mejorado bastante y ya da la imagen de una plataforma madura a tener en cuenta.

El motivo de este post es porque una de las opciones disponibles, es la administración de los componentes principales, utilizando ansible. Para ello, se requiere instalar previamente el cliente oci y el SDK de python.

El procedimiento, está perfectamente explicado en la propia documentación de OCI pero, como está principalmente orientado a servidores linux de la familia redhat y, me encontré alguna dificultad inicial al realizarla sobre una distribución de la familia Debian (Ubuntu 18.04 en mi caso concreto), he creido conveniente compartir los pasos que he seguido para obtener un resultado satisfactorio.

Lo primero que haremos, para evitar interferencias con nuestra instalación actual, es crear un entorno virtual de python, dentro del cual haremos toda la instalación de todos los elementos.

mkdir ansibleOCI
cd ansibleOCI/
mkdir configuracion
source venv/bin/activate
pip3 install oci 
pip3 install oci-cli

A continuación, deberemos configurar el entorno para que disponga de todos los datos requeridos para la conexión con nuestro tenant de oci.

Antes de empezar, necesitamos recuperar cierta información que vamos a necesitar durante el proceso de configuración.

Naturalmente, debemos disponer de una cuenta oci con un usuario que disponga de los permisos necesarios para realizar las acciones que nos interesa. Con éste, nos logaremos a nuestro tenant y recuperaremos el tenant OCID (menú Administration > Tenancy Details). También necesitaremos el OCID del usuario (menú Identity > Users > User Details) y conocer al región del mismo (eu-frankfurt-1 en mi caso).

Una vez recopilados estos datos, podemos proceder a configurar el entorno ejecutando el siguiente comando e informando los datos solicitados.

oci setup config

    This command provides a walkthrough of creating a valid CLI config file.

    The following links explain where to find the information required by this
    script:

    User API Signing Key, OCID and Tenancy OCID:
        https://docs.cloud.oracle.com/Content/API/Concepts/apisigningkey.htm#Other

    Region:
        https://docs.cloud.oracle.com/Content/General/Concepts/regions.htm

    General config documentation:
        https://docs.cloud.oracle.com/Content/API/Concepts/sdkconfig.htm


Enter a location for your config [/home/miusuario/.oci/config]: /home/miusuario/ansibleOCI/configuracion/config
Enter a user OCID: ocid1.user.oc1..aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
Enter a tenancy OCID: ocid1.tenancy.oc1..aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
Enter a region (e.g. ap-chuncheon-1, ap-hyderabad-1, ap-melbourne-1, ap-mumbai-1, ap-osaka-1, ap-seoul-1, ap-sydney-1, ap-tokyo-1, ca-montreal-1, ca-toronto-1, eu-amsterdam-1, eu-frankfurt-1, eu-zurich-1, me-jeddah-1, sa-saopaulo-1, uk-gov-cardiff-1, uk-gov-london-1, uk-london-1, us-ashburn-1, us-gov-ashburn-1, us-gov-chicago-1, us-gov-phoenix-1, us-langley-1, us-luke-1, us-phoenix-1, us-sanjose-1): eu-frankfurt-1
Do you want to generate a new API Signing RSA key pair? (If you decline you will be asked to supply the path to an existing key.) [Y/n]: y
Enter a directory for your keys to be created [/home/miusuario/.oci]: /home/miusuario/ansibleOCI/configuracion
Enter a name for your key [oci_api_key]: 
Public key written to: /home/miusuario/ansibleOCI/configuracion/oci_api_key_public.pem
Enter a passphrase for your private key (empty for no passphrase): 
Private key written to: /home/miusuario/ansibleOCI/configuracion/oci_api_key.pem
Fingerprint: 94:4a:85:a6:57:a9:a5:80:76:9a:04:7a:0a:6a:48:59
Config written to /home/miusuario/ansibleOCI/configuracion/config

    If you haven't already uploaded your API Signing public key through the
    console, follow the instructions on the page linked below in the section
    'How to upload the public key':
        https://docs.cloud.oracle.com/Content/API/Concepts/apisigningkey.htm#How2

Los pasos anterirores, generarán un par de claves (pública y privada) y, un fichero de configuración de la conexión, en el directorio que hayamos especificado. Si hemos dejado las ubicaciones por defecto, ambos se situarán dentro de $HOME/.oci/, que es donde lo esperan encontrar, tanto el Python SDK como los módulos ansible. Pero, para completar el ejemplo, he modificado la ubicación por defecto. De esta forma podremos ver las implicaciones que tienene.

Antes de poder probar la conexión, tenemos que subir la clave pública que hemos generado anteriormrnte al perfil del usuario (menú Identity > Users > User Details > API Keys).

Si hemos mantenido el destino por defecto, podríamos probar la configuración de la conexión con el siguiente comando.

oci os ns get

Pero, como hemos decidido complicarnos un poco la vida personalizando el destino, deberemos asumir las consecuencias de nuestra decición y ejecutar el comando de la siguiente forma.

oci os ns get --config-file /home/miusuario/ansibleOCI/configuracion/config

Una forma de solventar en parte esta incomodidad, es definir una variable de entorno OCI_CONFIG_FILE

export OCI_CONFIG_FILE="/home/miusuario/ansibleOCI/configuracion/config"

En caso de no definir la variable de entorno, posteriormente, nos veremos obligados a informar el parámetro config_file_location siempre que utilicemos los módulos ansible para oci.

Sea como fuere, si todo ha ido bien, obtendremos una respuesta parecida a la siguiente:

{
  "data": "frdjxxxxxjgwk"
}

Una vez confirmado que podemos conectar a nuestra infraestructura cloud, procederemos a instalar ansible, y los módulos oci, dentro de nuestro entorno virtual.

pip3 install ansible
git clone https://github.com/oracle/oci-ansible-modules.git
cd oci-ansible-modules/
python3 ./install.py

A partir de este momento, ya deberíamos poder utilizar ansible para gestionar nuestra infraestructura oci.

systemd, configurar un script como servicio.

1N0T — Sat, 12 Sep 2020 23:27:53 +0000

Ya hace algún tiempo que systemd está entre nosotros y, parece que no tiene intención de desaparecer, ya que ha sido adoptado por la mayoria de distribuciones linux. Así que será mejor que nos hagamos a la idea y empecemos a aceptarlo como un compañero más que estará entre nosotros.

Para romper el hielo, nada mejor que hacer una prueba de concepto y definir nuestro propio servicio.

Procederemos a crear el script que queremos ejecutar como servicio, que aunque con poca utilidad, nos permitirá comprobar el correcto funcionamiento de nuestra configuración.

#!/bin/bash

# La única finalidad de este script es probar el proceso
# de configuración de un servicio en systemd.

# Enviamos mensaje inforamtivo del inicio de nuestro servicio.
echo "mi-servicio.service   *** iniciando ***" | systemd-cat -p info

# realizamos un loop infinito que mantendrá nuestro escript en ejecución.
# Grabaremos la hora cada minuto en el log del sistema.
while true; do
   ahora=$(date '+%Y-%m-%d %H:%M:%S') 
   echo "mi-servicio: $ahora" | systemd-cat -p info
   sleep 60
done

Aunque no es requisito que se encuentre en alguna ubicación concreta, lo copiaremos a /usr/local/bin que es el sitio recomendado para la instalación de nuestros programas.

sudo cp mi-servicio.sh /usr/local/bin

Naturalmente, tenemos que asegurarnos que sea ejecutable.

sudo chmod +x /usr/local/bin/mi-servicio.sh

Ahora, deberemos crear el fichero de configuración del servicio dentro del directorio /etc/systemd/system/ que por convención lo llamaremos mi-servicio.service, pero antes, nos aseguraremos que no se esté utilizando ya este nombre.

sudo systemctl list-unit-files --type=service

Como seguramente la lista será larga, podemos optar por filtrar la salida del comando.

sudo systemctl list-unit-files --type=service | grep mi-servicio

Si no aparece nada, podremos utilizar el nombre elegido. Así que, seleccionamos nuestro editor favorito y añadimos un contenido parecido al siguiente:

[Unit]
Description=Mi servicio super-útil

[Service]
ExecStart=/usr/local/bin/mi-servicio.sh

[Install]
WantedBy=multi-user.target

Aunque existen muchos más parámetros disponibles, nos hemos limitado a especificar el mínimo indispensable, que en este caso proporcionan una descripción, indican el programa a ejecutar y que esté disponible cuando el sistema se inicie en modo multiusuario con red (nuestro antiguo runlevel 3).

Como ya hemos convenido, lo salvaremos con el nombre /etc/systemd/system/mi-servicio.service

No se requiere que el fichero sea ejecutable, pero si es muy conveniente modificar los permisos.

sudo chmod 640 /etc/systemd/system/mi-servicio.service

Vamos a ver que tal pinta tiene lo realizado

sudo systemctl status mi-servicio

Si obtenemos algo parecido a lo siguiente, será indicativo de que el fichero de configuración es correcto.

● mi-servicio.service - Mi servicio super-útil
   Loaded: loaded (/etc/systemd/system/mi-servicio.service; disabled; vendor preset: enabled)
   Active: inactive (dead)

Si es así, ya tenemos nuestro servicio, que podremos administrar como cualquier otro. Procedamos pues a iniciarlo.

sudo systemctl start mi-servicio

Si todo ha ido bien, cuando miremos su status, tendrá un aspecto similar a la siguiente:

sudo systemctl status mi-servicio

● mi-servicio.service - Mi servicio super-útil
   Loaded: loaded (/etc/systemd/system/mi-servicio.service; disabled; vendor preset: enabled)
   Active: active (running) since Sun 2020-09-13 01:12:27 CEST; 25s ago
 Main PID: 17095 (mi-servicio.sh)
    Tasks: 2 (limit: 4915)
   CGroup: /system.slice/mi-servicio.service
           ├─17095 /bin/bash /usr/local/bin/mi-servicio.sh
           └─17117 sleep 60

sep 13 01:12:27 P1912 systemd[1]: Started Mi servicio super-útil.
sep 13 01:12:27 P1912 cat[17116]: mi-servicio: 2020-09-13 01:12:27

Eureka !!! ya tenemos nuestro servicio funcionando.

Ansible, encriptar el valor de una variable.

1N0T — Fri, 11 Sep 2020 20:03:39 +0000

Antes o después, acabaremos poniendo en nuestro playbook (de forma provisional, naturalmente) una contraseña de usuario requerida para alguna instalación. Pero llega una edad en que la memoria flaquea y, con la complicidad de la larga lista de TODOs, la provisionalidad deja de ser tal (voluntaria o involuntariamente) y el valor en claro se perpetua en el playbook definitivo (o en alguno de nuestros commits intermedios).

Para que esto no ocurra, nada mejor que colocar desde el inicio el valor encriptado, después de todo tampoco cuesta tanto.

Imaginemos que quremos definir el equvalente a:

---
  - hosts: localhost

    vars:
      mi_password_secreto: "NoLoConoceNadie"

Deberemos ejecutar el siguiente comando:

echo -n "NoLoConoceNadie" | ansible-vault encrypt_string --stdin-name "mi_password_secreto"

Después de introducir la contraseña para la encriptación que se nos solicitará, obtendremos una salida parecida a la siguiente:

New Vault password: 
Confirm New Vault password: 
Reading plaintext input from stdin. (ctrl-d to end input)
mi_password_secreto: !vault |
          $ANSIBLE_VAULT;1.1;AES256
          62626464613161396533623034613237363839353836353334333739373463326437313163353263
          3661346139613838336336336232323830326237323364370a316464363563653039353061303338
          63613833303461356265623232376563373839343432643262326237343665313530346262336230
          3332616365383161340a353564623333303735623131326539343435626638326333376639633337
          3262
Encryption successful

Sólo hay que tener en cuenta un par de consideraciones:

No olvidar el parámetro -n del comando echo, para que no nos añada un salto de línea \n al final del string, cosa que con toda probabilidad no deseamos.
Si vamos a crear más de una variable, debemos utilizar la misma contraseña para encriptar todos los valores del mismo playbook.

Ya sólo nos queda copiar el resultado obtenido para crear algo parecido a:

---
- hosts: localhost

  vars:
    mi_password_secreto: !vault |
        $ANSIBLE_VAULT;1.1;AES256
        62626464613161396533623034613237363839353836353334333739373463326437313163353263
        3661346139613838336336336232323830326237323364370a316464363563653039353061303338
        63613833303461356265623232376563373839343432643262326237343665313530346262336230
        3332616365383161340a353564623333303735623131326539343435626638326333376639633337
        3262
    desencriptada: "Mi contraseña secreta es: [{{ mi_password_secreto }}]"

  tasks:
    - name: Mostramos valor desencriptado
      debug: var=desencriptada

Para ejecutar nuestro playbook, deberemos hacer lo que se muestra a continuación:

ansible-playbook playbook.yml --ask-vault-pass

Tras introducir la contraseña que nos solicita, que es la utilizada para la encriptación, obtendremos la siguiente salida:

Vault password: 

PLAY [localhost] **********************************************************

TASK [Gathering Facts] ****************************************************
ok: [localhost]

TASK [Mostramos valor desencriptado] **************************************
ok: [localhost] => {
    "desencriptada": "Mi contraseña secreta es: [NoLoConoceNadie]"
}

PLAY RECAP ****************************************************************
localhost                  : ok=2    changed=0    unreachable=0    failed=0

Como se puede comprobar, una vez proporcionada la contraseña de encriptación, la variable se comporta internamente como si se hubiera especificado el valor en claro.

Aquellos que no conozcan la contraseña de encriptación, obtendrán un resultado como este.

Vault password: 

PLAY [localhost] **********************************************************************

TASK [Gathering Facts] ****************************************************************
ok: [localhost]

TASK [Mostramos valor desencriptado] **************************************************
fatal: [localhost]: FAILED! => {"msg": "An unhandled exception occurred while templating 'Mi contraseña secreta es: [{{ mi_password_secreto }}]'. Error was a <class 'ansible.parsing.vault.AnsibleVaultError'>, original message: Decryption failed (no vault secrets were found that could decrypt)"}
    to retry, use: --limit @/home/usuario/sources/ansible/ansible-vault/playbook.retry

PLAY RECAP ****************************************************************************
localhost                  : ok=1    changed=0    unreachable=0    failed=1