Indicador PFE (Eficiencia Fractal Polarizada) Para 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:

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 decimalespfe_suavizado[-1]: último valor (actual)









