Forem: Pablo

Scrappeando propiedades con Python

Pablo — Wed, 04 Mar 2020 14:16:28 +0000

Hace un tiempo estaba buscando departamento para alquilar y me frustraba entrar periódicamente a las páginas de propiedades sólo para encontrarme una y otra vez con los mismos inmuebles.

Buscar un departamento es una tarea tediosa y estresante, por eso, se me ocurrió dar vuelta el problema y transformarlo en algo divertido: crear un bot que busque las propiedades y avise vía chat cuando aparece una nueva.

El código es relativamente sencillo, menos de 100 líneas de python, pero de todos modos vamos a seguirlo paso por paso para ver qué es lo que hace.

URLS y Main

Primero notamos que tenemos un set con las URLs de búsqueda:


urls = {
    "https://www.argenprop.com/departamento-alquiler-barrio-chacarita-hasta-25000-pesos-orden-masnuevos",
    "https://www.zonaprop.com.ar/departamento-alquiler-villa-crespo-con-terraza-orden-publicado-descendente.html",
}

Estas son simplemente URLs con resultados de búsqueda de los sitios soportados por nuestro script. En este caso departamentos en Chacarita hasta 25.000 pesos y en Villa Crespo con terraza (sin límite de precio). Ambas en distintos portales de búsqueda. Es importante que los resultados esten ordenados por "más nuevos"

La función principal (main) define los pasos a seguir:


for url in urls: # 1
  res = requests.get(url) # 2
  ads = list(extract_ads(url, res.text)) # 3
  seen, unseen = split_seen_and_unseen(ads) # 4

  print("{} seen, {} unseen".format(len(seen), len(unseen)))

  for u in unseen: # 5
    notify(u)

  mark_as_seen(unseen) # 6

Para cada una de nuestras URLs definidas previamente
Ir a buscar los resultados
Extraer los links de los ads
Dividir links en vistos y no vistos
Para los no vistos, notificar por mensaje
Marcar los no vistos como vistos

De acá nos enfocamos en las partes más interesantes:

Extraer links de los ads

Cada página tiene su estructura propia, para eso definimos la dataclass Parser:

@dataclass
class Parser:
    website: str
    link_regex: str

    def extract_links(self, contents: str):
        soup = BeautifulSoup(contents, "lxml")
        ads = soup.select(self.link_regex)
        for ad in ads:
            href = ad["href"]
            _id = sha1(href.encode("utf-8")).hexdigest()
            yield {"id": _id, "url": "{}{}".format(self.website, href)}


parsers = [
    Parser(website="https://www.zonaprop.com.ar", link_regex="a.go-to-posting"),
    Parser(website="https://www.argenprop.com", link_regex="div.listing__items div.listing__item a"),
    Parser(website="https://inmuebles.mercadolibre.com.ar", link_regex="li.results-item .rowItem.item a"),
]

Como vemos, cada Parser actúa sobre un dominio en particular de los que soportamos, porque cada página tiene su estructura de HTML propia.

Con beautifulsoup podemos usar CSS queries para movernos por el markup y obtener los links de los avisos que nos interesan.

Una vez extraídos los devolvemos junto con un id que generamos hasheando la URL del aviso (esto será útil para marcar los ya vistos).

Recordar Links vistos

El bot no sería muy útil si reportara todo el tiempo lo mismo, deberíamos buscar una forma de "recordar" qué avisos ya envíamos.

Una forma sencilla es un archivo de texto seen.txt con los ids de los avisos reportados:

def split_seen_and_unseen(ads):
    history = get_history()
    seen = [a for a in ads if a["id"] in history]
    unseen = [a for a in ads if a["id"] not in history]
    return seen, unseen


def get_history():
    try:
        with open("seen.txt", "r") as f:
            return {l.rstrip() for l in f.readlines()}
    except:
        return set()

También agregamos una función que nos permita distinguir los que ya mandamos (seen) de los nuevos (unseen).

Notificaciones

def notify(ad):
    bot = "YOUR_BOT_ID"
    room = "YOUR_CHAT_ROOM_ID"
    url = "https://api.telegram.org/bot{}/sendMessage?chat_id={}&text={}".format(bot, room, ad["url"])
    r = requests.get(url)

Luego, usando un simple request de HTTP posteamos la url del aviso a un grupo de Telegram.

Gracias a la función de "preview" de Telegram, los ads se muestran con una imagen de la propiedad:

Por último sólo nos queda agregar los avisos notificados a seen.txt y listo.

Hosting

Ahora sólo necesitamos una máquina donde correrlo periódicamente. Por suerte AWS tiene un free tier que nos brinda una máquina básica por un mes, gratis.

Si bien la virtual no es muy poderosa, para este simple script basta y sobra. Yo lo tengo corriendo con un cronjob que lo ejecuta cada 5 minutos:

# m h  dom mon dow   command
*/5 * * * * /home/ubuntu/propfinder/search_props.sh

Conclusión

Con 100 líneas de scripting pudimos hacer un scrapper multi-sitio, que nos envía propiedades por chat como máximo a 5 minutos de ser publicadas.

Mucho más divertido (y menos frustrante) que meterse periódicamente a los distintos sitios a ver si hay alguna novedad.

Breve introducción a Algoritmos Genéticos - Parte 2

Pablo — Wed, 29 Jan 2020 20:07:12 +0000

En la primera parte vimos qué eran los algoritmos genéticos y cómo aplicarlos para encontrar el máximo (o el mínimo) de una función. Sin embargo, el ejemplo que dimos era una función continua, diferenciable y con un máximo muy claro:

y = 8x -2x^2

Veremos en este post cómo aplicar el mismo algoritmo pero para buscar el resultado óptimo de una función no continua más compleja, como ser el conjunto de ecuaciones que presentó el Ministro de Educación hace unas semanas:

Definiendo el sistema de ecuaciones

Los algoritmos genéticos intentan aproximarse a la solución de un problema dado. Hacen esto tratando de minimizar o maximizar una función. ¿Cómo podemos representar este problema como algo a minimizar?

Una forma posible es la siguiente:

def f(xyz):

  # calculator => x
  # book => y
  # computer => z

  x,y,z = xyz

  # 1
  s_1 = x + x + x
  s_2 = x + y + y
  s_3 = y - z * 2

  # 2
  err_1 = np.abs(s_1 - 30)
  err_2 = np.abs(s_2 - 18)
  err_3 = np.abs(s_3 - 2)

  # 3
  return np.sum([err_1, err_2, err_3])

escribimos las 3 ecuaciones en base a los argumentos x, y, z, que serán los que nuestro algoritmo genético produzca
calculamos el "error" de cada ecuación, es decir, qué tan lejos está la suma con los parámetros que generamos del resultado esperado
el "fitness" de nuestro x, y, z es la suma de todos los errores. Ésta es la función que intentaremos minimizar

Población inicial

Al igual que en el post anterior creamos una población inicial, pero esta vez:

los individuos de la población son tres valores en vez de uno (x, y, z)
asumimos que las soluciones están entre 1000 y -1000

np.random.seed(1234) # make results reproducible.


def _r():
  return np.random.random() * 2000 - 1000


def random_individual(): 
  return np.array([_r(), _r(), _r()])


initial_generation_size = 1000
first_generation = [random_individual() for _ in range(initial_generation_size)]
pop = Population(chromosomes=first_generation, eval_function=f, maximize=False)

Esta vez seteamos maximize=False porque queremos minimizar, o sea hacer el error lo más pequeño posible.

Definiendo la Evolución

Previamente hicimos todos los pasos de a uno para que sea más fácil entender el proceso pero la librería evol tiene una función para definir de forma más concisa el pipeline de evolución, el objeto Evolution:


def pick_parents(population):
  mom, dad = np.random.choice(population, size=2)
  return mom, dad


def make_child(mom, dad):
  mom_p = np.random.random() # percent of mom contribution
  dad_p = 1 - mom_p # percent of dad contribution
  mom_x, mom_y, mom_z = mom
  dad_x, dad_y, dad_z = dad
  x = mom_x * mom_p + dad_x * dad_p
  y = mom_y * mom_p + dad_y * dad_p
  z = mom_z * mom_p + dad_z * dad_p
  return np.array([x, y, z])


# add gaussian noise with mean zero.
def random_noise(individual, sigma):
  noise = np.random.normal(loc=0, scale=sigma, size=3)
  return individual + noise


best_of = []

def add_best_of(population):
  b = population.current_best
  best_of.append((b.chromosome, f(b.chromosome)))

evo = (Evolution()
      .survive(fraction=0.1)
      .breed(parent_picker=pick_parents, combiner=make_child)
      .mutate(mutate_function=random_noise, sigma=2)
      .callback(add_best_of))

Algunos detalles que valen la pena mencionar:

la función make_child en vez de tomar el promedio elige un valor al azar entre 0 y 1, toma ese porcentaje de la madre y el restante del padre.
usamos un array (best_of) y el hook callback del objeto Evolution para ir acumulando los mejores cromosomas de cada generación.

Con el pipeline listo, sólo es cuestión de pasar la población inicial y decirle cuántas generaciones queremos:

new_generation = pop.evolve(evolution=evo, n=25)

Listo! ya tenemos en best_of a los mejores exponentes de cada generación y su valor de "fitness", es decir, qué tanto minimizan el error de nuestra función y por lo tanto encuadran mejor en el sistema de ecuaciones.

Como la última vez, observemos cómo se reduce el error a medida que se van creando nuevas generaciones.

Tomamos entonces como solución los valores de la última generación:

x,y,z = best_of[-1][0]

10.155879542288524, 4.02024292752548, 0.92945755333642

Pero la solución no es la óptima 🤨

No, no lo es, el algoritmo puede no encontrar el mínimo (o máximo) global. En este caso, como indica el Ministro la respuesta correcta era:

x = 10, y = 4, z = 1.

Sin embargo, a diferencia de este sistema sencillo de ecuaciones, hay otros donde la respuesta 100% correcta no es obvia.

Nuestro algoritmo puede generalizar bien a estos otros problemas como veremos a continuación.

Modificando un poco la ecuación

Como muchos conocerán, este enunciado fue blanco de críticas por quienes pensaban que la última ecuación no es y - z * 2 sino y - z^z:

Para resolver esta variante sólo tenemos que cambiar la función a minimizar y correr el programa otra vez:

def f2(xyz):

  # calculator => x
  # book => y
  # computer => z

  x,y,z = xyz

  # 1
  s_1 = x + x + x
  s_2 = x + y + y
  s_3 = y - z ** z # no longer y - 2 * z

  # 2
  err_1 = np.abs(s_1 - 30)
  err_2 = np.abs(s_2 - 18)
  err_3 = np.abs(s_3 - 2)

  # 3
  return np.sum([err_1, err_2, err_3])

initial_generation_size = 1000
first_generation = [random_individual() for _ in range(initial_generation_size)]
pop2 = Population(chromosomes=first_generation, eval_function=f2, maximize=False)

new_generation = pop2.evolve(evolution=evo, n=100)

Analizamos los errores:

Como vemos, los errores no convergen esta vez de forma tan directa. Sin embargo muchas observaciones están con un error bajo. Nos quedamos con los valores que tienen el mejor fit:

fits = [b[1] for b in best_of]
best_fit = np.argmin(fits)
x,y,z = best_of[best_fit][0]

x = 10.60, y = 9.13, z = 2.27
error: 15.27

Un error de "solamente" 15 puntos.

Conclusiones

Pudimos adaptar nuestro algoritmo de optimización a una nueva función de manera trivial
Estos algoritmos no llegan siempre al óptimo global sino que intentan aproximarse a una solución "good enough"
Es posible iterar la solución. En ambos casos al ver los resultados nos damos cuenta de que nuestra intuición general del rango de valores óptimos (-1.000, 1.000) era exagerada. Podríamos volver a intentar con un rango más razonable, de hecho es posible llegar a un error muy bajo, de casi 1 punto, tomando un rango más acotado (esto queda como ejercicio para el lector)

Introduction to genetic algorithms in Python

Pablo — Tue, 28 Jan 2020 13:52:59 +0000

What is a genetic algorithm?

A genetic algorithm is a method to solve a problem inspired in the natural selection process popularised by Charles Darwin:

As many more individuals of each species are born than can possibly survive; and as, consequently, there is a frequently recurring struggle for existence, it follows that any being, if it vary however slightly in any manner profitable to itself, under the complex and sometimes varying conditions of life, will have a better chance of surviving, and thus be naturally selected. From the strong principle of inheritance, any selected variety will tend to propagate its new and modified form

There are many variants of these algorithms (they are actually an algorithm family) but all follow roughly the same steps:

Generate a "population" of individuals (or chromosomes in genetic algorithm parlance)
Evaluate these individuals by their "fitness"
Keep N "fittest" subjects
Make these subjects breed and build the "next generation"
Go back to step 2 and repeat this process

What type of problems can you solve?

You can solve any problem that involves optimizing some function, either maximizing or minimizing it.

For example, given:

y = 8x - 2x^2

def f(x): return 8 * x - 2 * x**2

We'd like to find the value of x that maximizes y, and we are going to do it with a tiny genetic library called evol

First Generation

For this process to begin we need to build a "first generation" from scratch. We'll do this picking random numbers for the chromosomes or x values.

We assume the value of x that yields the optimal value of y is unknown (although we can see in the plot pretty much where it is) but we have a prior belief that it ranges from -1,000,000 to 1,000,000.

Our first generation will pick values on that range:

np.random.seed(1234) # make results reproducible.


def random_guy(): return np.random.randint(low=-1_000_000, high=1_000_000)


initial_generation_size = 100
first_generation = [random_guy() for _ in range(initial_generation_size)]

Now we can build or Population object. We also provide the function that requires optimizing and we specify this is a maximization problem.

pop = Population(chromosomes=first_generation, eval_function=f, maximize=True)

Survival of the fittest

We'll keep 10% of the population members using the survive function:

survivors = pop.survive(fraction=0.1)

Breeding and mutations

These survivors will create a new generation, by breeding and mutations

For this step we need to define:

A function for picking two parents from the survivors (we'll pick any two)
Another function to specify how these two individuals (parents) generate an offspring (breeding)
A last one to add some random mutations to the offspring (this is usually done so you don't get stuck in local maxima)

def pick_parents(population):
    mom, dad = np.random.choice(population, size=2)
    return mom, dad


# the child is the mean between two parents.
def make_child(mom, dad):
    return (mom + dad) / 2


# add gaussian noise with mean zero.
def random_noise(individual, sigma):
    noise = np.random.normal(loc=0, scale=10) * sigma
    return individual + noise


new_generation = survivors.breed(parent_picker=pick_parents, combiner=make_child).mutate(random_noise, sigma=1)

It's evolution, baby

We now have a new_generation with (hopefully) fitter individuals.We'll repeat this process a few more times, keeping the fittest individual of each generation.

def fittest(generation):
    fit = sorted(generation.evaluate(), key=lambda x: x.fitness)[-1]
    return fit.chromosome


generations = 120
fittest_of_each_gen = []

gen = new_generation
for _ in range(generations):
    survivors = gen.survive(fraction=0.1)
    new_gen = survivors.breed(parent_picker=pick_parents, combiner=make_child).mutate(random_noise, sigma=1)
    fittest_of_each_gen.append(fittest(new_gen))
    gen = new_gen

Results

Let's plot the original function again, this time with the fittest individuals of each generation on top.

Note individuals go chronologically from blue (cool) to red (warm). It's also useful to include the error of each approximation.

About generation #100 the output of our algorithm starts to converge with the maximum y and error goes down to zero.

Closing remarks

Does this technique make sense? after all we know we can find the function maximum analytically with it's derivative:

f(x) = 8x - 2x^2

f'(x) = 8 - 4x

f'(x) = 8 - 4 * 2 = 0

Though the process is indeed an overkill for cases like this, the algorithm generalizes to all kinds of problems and functions, not only continuous and differentiable.

For more information on genetic algorithms and the evol library make sure to check it's GitHub repository.

Hope you enjoyed this! If you did let me know on Twitter

Breve introducción a Algoritmos Genéticos

Pablo — Fri, 24 Jan 2020 17:20:54 +0000

(el contenido también se encuentra disponible en formato de IPython notebook)

¿Qué es un algoritmo genético?

Un algoritmo genético es un método para resolver un problema inspirado en el proceso de selección natural propuesto por Charles Darwin:

As many more individuals of each species are born than can possibly survive; and as, consequently, there is a frequently recurring struggle for existence, it follows that any being, if it vary however slightly in any manner profitable to itself, under the complex and sometimes varying conditions of life, will have a better chance of surviving, and thus be naturally selected. From the strong principle of inheritance, any selected variety will tend to propagate its new and modified form

Hay muchas variaciones de dichos algoritmos pero todos siguen a grandes rasgos este patrón:

Generar una población de individuos (o cromosomas en el léxico de los algoritmos genéticos)
Evaluar a los individuos de la población por una métrica de aptitud (conocido como fitness)
Sólo quedarse con los N sujetos de la población más aptos
Crear una nueva generación de individuos a partir de los más aptos a través de:
1. cross breeding: "cruza" entre individuos
2. mutations: mutaciones aleatorias
Volver a paso 2 y repetir

¿Qué problemas se pueden resolver?

Los problemas atacados por estos algoritmos son en general de optimización de funciones (más adelante veremos como esto se puede generalizar a soluciones más interesantes), por ejemplo, dada una función:

y = 8x - 2x^2

def f(x): return 8 * x - 2 * x**2

Nos interesaría encontrar el punto máximo, es decir el valor de x que maximiza y.

Vamos a usar un algoritmo genético para esto, apoyándonos en la librería evol de python

Paso 1: Primera Generación

Para comenzar tenemos que armar una primera generación de individuos o cromosomas, en este caso los individuos serían distintos valores de x.

Imaginemos que no sabemos dónde está el máximo de la función pero tenemos una vaga idea de en qué rango puede estar. Para este ejemplo vamos a suponer que el valor está entre -1.000.000 (menos un millón) y 1.000.000 (un millón).

Nuestra primera generación tomará valores aleatorios en este intervalo:

np.random.seed(1234) # make results reproducible.


def random_guy(): return np.random.randint(low=-1_000_000, high=1_000_000)


initial_generation_size = 100
first_generation = [random_guy() for _ in range(initial_generation_size)]

Con esto podemos crear nuestro objeto Population. También le pasamos la función a optimizar ($f(x)$ definida más arriba) y le aclaramos a evol si la idea es maximizar o minimizar dicha función:

pop = Population(chromosomes=first_generation, eval_function=f, maximize=True)

Supervivencia del más apto

De nuestra población inicial queremos quedarnos sólo con el 10% que mejor performó, los más aptos. Para esto tenemos la función survive:

survivors = pop.survive(fraction=0.1)

CrosBreeding y Mutaciones

Con los sobrevivientes vamos a crear una nueva generación, producto de breeding y mutaciones aleatorias.

Para esto tenemos que definir tres funciones:

1) una para elegir dos padres del conjunto de supervivientes (lo haremos al azar)

2) otra para definir cómo dos individuos de la población crean un nuevo individuo 😏

3) una última para determinar las mutaciones aleatorias

Entonces:


def pick_parents(population):
    mom, dad = np.random.choice(population, size=2)
    return mom, dad


# the child is the mean between two parents.
def make_child(mom, dad):
    return (mom + dad) / 2


# add gaussian noise with mean zero.
def random_noise(individual, sigma):
    noise = np.random.normal(loc=0, scale=10) * sigma
    return individual + noise


new_generation = survivors.breed(parent_picker=pick_parents, combiner=make_child).mutate(random_noise, sigma=1)

It's evolution, baby

Con la nueva generación podemos repetir el proceso indefinidamente (seleccionar, cruzar, mutar).

Vamos a hacerlo unas cuantas veces y nos quedamos con el individuo más apto de cada generación:

def fittest(generation):
    fit = sorted(generation.evaluate(), key=lambda x: x.fitness)[-1]
    return fit.chromosome


generations = 120
fittest_of_each_gen = []

gen = new_generation
for _ in range(generations):
    survivors = gen.survive(fraction=0.1)
    new_gen = survivors.breed(parent_picker=pick_parents, combiner=make_child).mutate(random_noise, sigma=1)
    fittest_of_each_gen.append(fittest(new_gen))
    gen = new_gen

Analizando los resultados

Volvamos otra vez a la función original y sobre la línea agreguemos los "mejores exponentes" de cada generación.

Las generaciones van de menor a mayor desde el azul (cool) al rojo (warm). También graficamos el error, o sea qué tan alejados estamos del valor máximo real, en función del número de generación:

Como vemos aproximadamente en la generación 100 el valor empieza a converger con el máximo real.

Conclusión

Pero, ¿Tiene sentido utilizar está técnica? después de todo, sabemos que la forma de encontrar el mínimo de una función como

f(x) = 8x - 2x^2

es tomar su derivada

f'(x) = 8 - 4x

y encontrar el valor donde el gradiente es cero, en este caso

f'(x) = 8 - 4 * 2 = 0

La respuesta es que, si bien para este caso es un overkill, los algoritmos genéticos son un método numérico que nos permite encontrar resultados aceptables en funciones no diferenciables o de las que no se conoce un método análitico.

Próximamente

Con esto concluimos esta breve introducción.

En un próximo post, analizaremos como performa este mismo algoritmo genético para resolver el conjunto de ecuaciones que presentó el ministro de educación hace unas semanas: