Indicador PFE (Eficiencia Fractal Polarizada) Para Python

0
11
Indicador PFE de Python

En este artículo vamos a mostrar como programar el indicador PFE (Eficiencia Fractal Polarizada) con el lenguaje de programación Python, mostrando su versatilidad que permite la aplicación de múltiples metodologías de análisis del mercado y trading. Lo que hace este código es tomar datos de precios de precios de un archivo .csv, calcular el indicador PFE y luego trazarlo justo debajo de un gráfico de precios para encontrar sus señales.

Antes de mostrar este código vamos a explicar en que consiste el indicador PFE:

¿Qué es el indicador PFE?

El indicador de Eficiencia Fractal Polarizada (PFE, por sus siglas en inglés) mide la eficiencia de los movimientos de precios basándose en conceptos de la geometría fractal y la teoría del caos. Cuanto más lineal y eficiente es el movimiento del precio, menor es la distancia que los precios deben recorrer entre dos puntos y, por lo tanto, más eficiente es el movimiento.

El indicador PFE mide cuán tendencial o congestionada está la acción del precio. Lecturas del PFE por encima de cero indican que la tendencia es alcista. Cuanto más alta sea la lectura, más “tendencioso” y eficiente será el movimiento ascendente. Lecturas del PFE por debajo de cero significan que la tendencia es bajista. Cuanto más baja sea la lectura, más “tendencioso” y eficiente será el movimiento descendente. Las lecturas cercanas a cero indican un movimiento irregular y menos eficiente, con un equilibrio entre las fuerzas de la oferta y la demanda.

Diversas plataformas de trading utilizan un PFE suavizado a corto plazo con un intervalo intermedio de retrospectiva y genera señales de compra/venta basándose en cruces de umbrales del PFE.

Pueden encontrar más información del PFE en el siguiente artículo: Indicador Eficiencia Fractal Polarizada

Código del indicador PFE para Python

Primero vamos a mostrar el código completo y luego vamos a explicar en detalle cada una de sus partes.

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

def calcular_pfe(precios, periodo=10, suavizado=5):
    """
    Calcula el indicador Eficiencia Fractal Polarizada (PFE)
    
    Parámetros:
    -----------
    precios : array-like
        Serie de precios (generalmente precios de cierre)
    periodo : int
        Periodo para el cálculo del PFE (default: 10)
    suavizado : int
        Periodo para la EMA de suavizado (default: 5)
    
    Returns:
    --------
    pfe : numpy array
        Serie del indicador PFE
    pfe_suavizado : numpy array
        Serie del PFE suavizado con EMA
    """
    precios = np.array(precios)
    n = len(precios)
    pfe = np.full(n, np.nan)
    
    # Calcular PFE para cada punto
    for i in range(periodo, n):
        # Precio actual y precio hace n periodos
        precio_actual = precios[i]
        precio_anterior = precios[i - periodo]
        
        # 1. Distancia lineal (directa)
        distancia_lineal = np.sqrt((precio_actual - precio_anterior)**2 + periodo**2)
        
        # 2. Distancia fractal (suma de segmentos)
        distancia_fractal = 0
        for j in range(i - periodo + 1, i + 1):
            distancia_fractal += np.sqrt((precios[j] - precios[j-1])**2 + 1)
        
        # 3. Eficiencia fractal
        eficiencia = distancia_lineal / distancia_fractal if distancia_fractal != 0 else 0
        
        # 4. Determinar signo (polaridad)
        signo = 1 if precio_actual > precio_anterior else -1
        
        # 5. Calcular PFE
        pfe[i] = signo * eficiencia * 100
    
    # Aplicar EMA de suavizado
    pfe_suavizado = calcular_ema(pfe, suavizado)
    
    return pfe, pfe_suavizado

def calcular_ema(datos, periodo):
    """
    Calcula la Media Móvil Exponencial (EMA)
    """
    datos = pd.Series(datos)
    ema = datos.ewm(span=periodo, adjust=False).mean()
    return ema.values

# Ejemplo de uso
if __name__ == "__main__":
    # ===== CONFIGURACIÓN =====
    # Cambiar esta ruta por tu archivo CSV
    archivo_csv = 'precios.csv'
    
    # Nombre de la columna de precios (comunes: 'Close', 'close', 'Cierre', 'precio')
    columna_precio = 'Close'
    
    # Parámetros del PFE
    periodo_pfe = 10
    periodo_suavizado = 5
    
    # ===== LEER DATOS DEL CSV =====
    try:
        df = pd.read_csv(archivo_csv)
        print(f"✓ Archivo leído correctamente: {archivo_csv}")
        print(f"✓ Total de registros: {len(df)}")
        print(f"\nColumnas disponibles: {list(df.columns)}")
        
        # Verificar que la columna existe
        if columna_precio not in df.columns:
            print(f"\n⚠ Advertencia: La columna '{columna_precio}' no existe.")
            print(f"Columnas disponibles: {list(df.columns)}")
            # Intentar encontrar columna de cierre común
            for col in ['Close', 'close', 'Cierre', 'precio', 'Precio']:
                if col in df.columns:
                    columna_precio = col
                    print(f"→ Usando columna '{columna_precio}' en su lugar.")
                    break
        
        precios = df[columna_precio].values
        
        # Si hay columna de fecha, usarla
        if 'Date' in df.columns or 'date' in df.columns or 'Fecha' in df.columns:
            col_fecha = 'Date' if 'Date' in df.columns else ('date' if 'date' in df.columns else 'Fecha')
            fechas = pd.to_datetime(df[col_fecha])
        else:
            fechas = range(len(precios))
        
    except FileNotFoundError:
        print(f"\n⚠ Error: No se encontró el archivo '{archivo_csv}'")
        print("→ Generando datos de ejemplo en su lugar...\n")
        
        # Generar datos de ejemplo
        np.random.seed(42)
        n_puntos = 200
        tendencia = np.linspace(100, 120, n_puntos)
        ruido = np.random.normal(0, 2, n_puntos)
        precios = tendencia + ruido
        fechas = pd.date_range('2024-01-01', periods=n_puntos)
    
    # Calcular PFE
    pfe, pfe_suavizado = calcular_pfe(precios, periodo=periodo_pfe, suavizado=periodo_suavizado)
    
    # Visualizar resultados
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), sharex=True)
    
    # Gráfico de precios
    ax1.plot(fechas, precios, label='Precio', color='blue', linewidth=1.5)
    ax1.set_ylabel('Precio', fontsize=12)
    ax1.set_title('Precio y PFE (Eficiencia Fractal Polarizada)', fontsize=14, fontweight='bold')
    ax1.legend()
    ax1.grid(True, alpha=0.3)
    
    # Gráfico del PFE
    ax2.plot(fechas, pfe, label='PFE', color='gray', alpha=0.5, linewidth=1)
    ax2.plot(fechas, pfe_suavizado, label='PFE Suavizado (EMA)', color='purple', linewidth=2)
    ax2.axhline(y=0, color='black', linestyle='--', linewidth=1, alpha=0.5)
    ax2.axhline(y=50, color='green', linestyle=':', alpha=0.5)
    ax2.axhline(y=-50, color='red', linestyle=':', alpha=0.5)
    ax2.fill_between(fechas, 0, pfe_suavizado, 
                      where=(pfe_suavizado > 0), alpha=0.3, color='green', label='Zona alcista')
    ax2.fill_between(fechas, 0, pfe_suavizado, 
                      where=(pfe_suavizado < 0), alpha=0.3, color='red', label='Zona bajista')
    ax2.set_xlabel('Periodo', fontsize=12)
    ax2.set_ylabel('PFE (%)', fontsize=12)
    ax2.legend()
    ax2.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()
    
    # Mostrar estadísticas
    print("=" * 50)
    print("ESTADÍSTICAS DEL PFE")
    print("=" * 50)
    print(f"PFE Máximo: {np.nanmax(pfe):.2f}")
    print(f"PFE Mínimo: {np.nanmin(pfe):.2f}")
    print(f"PFE Promedio: {np.nanmean(pfe):.2f}")
    print(f"PFE Actual: {pfe_suavizado[-1]:.2f}")
    print("=" * 50)

Cuando aplicamos este indicador usando datos de precios de Bitcoin obtenemos la siguiente imagen:

Indicador PFE de Python

Con respecto a las partes del código, estas tienen las siguientes características:

Importación de Librerías

import numpy as np 
import pandas as pd 
import matplotlib.pyplot as plt
  • numpy: Para cálculos matemáticos y manejo de arrays
  • pandas: Para leer archivos CSV y manipular datos en tablas
  • matplotlib: Para crear gráficos y visualizaciones

Función Principal: calcular_pfe()

Esta función representa el corazón del indicador. Vamos paso a paso:

def calcular_pfe(precios, periodo=10, suavizado=5):

Parámetros:

  • precios: Lista o array con los precios de cierre
  • periodo: Cuántos periodos atrás mirar (por defecto 10)
  • suavizado: Periodos para suavizar el resultado (por defecto 5)

Preparación de datos

precios = np.array(precios)
n = len(precios)
pfe = np.full(n, np.nan)
  • Convierte los precios a un array de numpy
  • n guarda cuántos precios tenemos
  • pfe crea un array lleno de NaN (valores vacíos) del mismo tamaño
  • Los primeros valores serán NaN porque necesitamos datos históricos para calcular

Bucle principal del cálculo

for i in range(periodo, n):

Empieza desde periodo (ej: 10) porque necesitamos 10 precios anteriores para calcular.

Paso 1: Distancia Lineal

precio_actual = precios[i]
precio_anterior = precios[i - periodo]

distancia_lineal = np.sqrt((precio_actual - precio_anterior)**2 + periodo**2)

¿Qué hace?

  • Toma el precio actual y el precio de hace N periodos
  • Calcula la distancia directa en “línea recta” usando el teorema de Pitágoras
  • Es como medir con una regla: del punto A al punto B directamente

Ejemplo:

  • Si el precio era 100 hace 10 días y ahora es 110
  • Distancia = √[(110-100)² + 10²] = √[100 + 100] = 14.14

Paso 2: Distancia Fractal

distancia_fractal = 0
for j in range(i - periodo + 1, i + 1):
    distancia_fractal += np.sqrt((precios[j] - precios[j-1])**2 + 1)
```

**¿Qué hace?**
- Suma TODOS los pequeños movimientos día a día
- Es como medir el camino real que recorrió el precio
- Incluye todas las subidas y bajadas intermedias

**Ejemplo visual:**
```
Distancia lineal:  A ---------> B  (línea recta)
Distancia fractal: A /\/\/\/\-> B  (camino zigzag real)

Paso 3: Eficiencia

eficiencia = distancia_lineal / distancia_fractal if distancia_fractal != 0 else 0

¿Qué mide?

  • Divide la distancia directa entre el camino real
  • Si el precio fue directo: eficiencia cercana a 1 (100%)
  • Si el precio zigzagueó mucho: eficiencia baja (cercana a 0)

Ejemplo:

  • Distancia lineal = 14.14
  • Distancia fractal = 30 (muchos zigzags)
  • Eficiencia = 14.14 / 30 = 0.47 (47%)

Paso 4: Polaridad (Signo)

signo = 1 if precio_actual > precio_anterior else -1

¿Qué hace?

  • Si el precio subió: signo = +1 (alcista)
  • Si el precio bajó: signo = -1 (bajista)
  • Esto indica la DIRECCIÓN del movimiento

Paso 5: PFE Final

pfe[i] = signo * eficiencia * 100

¿Qué resulta?

  • Multiplica: dirección × eficiencia × 100
  • Resultado positivo: Tendencia alcista eficiente
  • Resultado negativo: Tendencia bajista eficiente
  • Cercano a ±100: Movimiento muy directo (fuerte)
  • Cercano a 0: Movimiento errático (débil)

Paso 6: Suavizado con EMA

pfe_suavizado = calcular_ema(pfe, suavizado)
return pfe, pfe_suavizado

Aplica una media móvil exponencial para reducir el “ruido” y hacer el indicador más legible.

Función calcular_ema()

def calcular_ema(datos, periodo):
    datos = pd.Series(datos)
    ema = datos.ewm(span=periodo, adjust=False).mean()
    return ema.values

¿Qué hace este código?

  • Calcula la Media Móvil Exponencial
  • Da más peso a los datos recientes
  • span=periodo: cuántos periodos considerar
  • Suaviza las fluctuaciones del PFE

Lectura del Archivo CSV

archivo_csv = 'precios.csv'
columna_precio = 'Close'
periodo_pfe = 10
periodo_suavizado = 5

Estas son variables de configuración que puedes cambiar fácilmente.

Bloque try-except

try:
    df = pd.read_csv(archivo_csv)
    print(f"✓ Archivo leído correctamente: {archivo_csv}")
  • Intenta leer el archivo CSV
  • Si tiene éxito, muestra un mensaje
  • Si falla, ejecuta el bloque except

Verificación de columnas

if columna_precio not in df.columns:
    print(f"\n⚠ Advertencia: La columna '{columna_precio}' no existe.")
    for col in ['Close', 'close', 'Cierre', 'precio', 'Precio']:
        if col in df.columns:
            columna_precio = col
            print(f"→ Usando columna '{columna_precio}' en su lugar.")
            break

¿Qué hace?

  • Verifica si existe la columna especificada
  • Si no existe, busca automáticamente nombres comunes
  • Esto hace el código más flexible

Extracción de datos

precios = df[columna_precio].values

if 'Date' in df.columns or 'date' in df.columns or 'Fecha' in df.columns:
    col_fecha = 'Date' if 'Date' in df.columns else ('date' if 'date' in df.columns else 'Fecha')
    fechas = pd.to_datetime(df[col_fecha])
else:
    fechas = range(len(precios))

¿Qué hace este código?

  • Extrae los precios como un array
  • Busca columnas de fechas con nombres comunes
  • Si encuentra fechas, las convierte al formato datetime
  • Si no hay fechas, usa números secuenciales (0, 1, 2, 3…)

Manejo de errores

except FileNotFoundError:
    print(f"\n⚠ Error: No se encontró el archivo '{archivo_csv}'")
    print("→ Generando datos de ejemplo en su lugar...\n")
    
    # Generar datos de ejemplo
    np.random.seed(42)
    n_puntos = 200
    tendencia = np.linspace(100, 120, n_puntos)
    ruido = np.random.normal(0, 2, n_puntos)
    precios = tendencia + ruido
    fechas = pd.date_range('2024-01-01', periods=n_puntos)

¿Qué hace este código?

  • Si no encuentra el archivo, NO se detiene el programa
  • En su lugar, genera datos de ejemplo para que puedas probar
  • linspace: crea una tendencia alcista de 100 a 120
  • normal: añade ruido aleatorio (volatilidad)

Visualización con Matplotlib

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), sharex=True)
  • Crea una figura con 2 gráficos apilados verticalmente
  • figsize=(12, 8): tamaño en pulgadas
  • sharex=True: ambos gráficos comparten el eje X (fechas)

Gráfico de Precios (superior)

ax1.plot(fechas, precios, label='Precio', color='blue', linewidth=1.5)
ax1.set_ylabel('Precio', fontsize=12)
ax1.set_title('Precio y PFE (Eficiencia Fractal Polarizada)', fontsize=14, fontweight='bold')
ax1.legend()
ax1.grid(True, alpha=0.3)
  • Dibuja la línea de precios en azul
  • Añade título, etiquetas y cuadrícula
  • alpha=0.3: transparencia del 30% en la cuadrícula

Gráfico del PFE (inferior)

ax2.plot(fechas, pfe, label='PFE', color='gray', alpha=0.5, linewidth=1)
ax2.plot(fechas, pfe_suavizado, label='PFE Suavizado (EMA)', color='purple', linewidth=2)
  • Dibuja el PFE original en gris (transparente)
  • Dibuja el PFE suavizado en morado (más grueso y visible)

Líneas de referencia

ax2.axhline(y=0, color='black', linestyle='--', linewidth=1, alpha=0.5)
ax2.axhline(y=50, color='green', linestyle=':', alpha=0.5)
ax2.axhline(y=-50, color='red', linestyle=':', alpha=0.5)
  • y=0: Línea central (neutro)
  • y=50: Nivel de sobrecompra (alcista fuerte)
  • y=-50: Nivel de sobreventa (bajista fuerte)

Áreas coloreadas

ax2.fill_between(fechas, 0, pfe_suavizado, 
                 where=(pfe_suavizado > 0), alpha=0.3, color='green', label='Zona alcista')
ax2.fill_between(fechas, 0, pfe_suavizado, 
                 where=(pfe_suavizado < 0), alpha=0.3, color='red', label='Zona bajista')

¿Qué hace este código?

  • Rellena el área entre 0 y el PFE
  • Verde cuando PFE > 0 (alcista)
  • Rojo cuando PFE < 0 (bajista)
  • Hace visual la dirección de la tendencia

Estadísticas Finales del Indicador

print("=" * 50)
print("ESTADÍSTICAS DEL PFE")
print("=" * 50)
print(f"PFE Máximo: {np.nanmax(pfe):.2f}")
print(f"PFE Mínimo: {np.nanmin(pfe):.2f}")
print(f"PFE Promedio: {np.nanmean(pfe):.2f}")
print(f"PFE Actual: {pfe_suavizado[-1]:.2f}")
  • np.nanmax(): máximo ignorando valores NaN
  • :.2f: formato con 2 decimales
  • pfe_suavizado[-1]: último valor (actual)

 

TagsPython
Raul Canessa

Leave a reply