Kit de código del trader — Python y Pine Script

Entrada de S2 y S3: lee exportación F, comprueba encabezados y número de campos y limpia columna r_net . Nombres duplicados, encabezado vacío y campos de más o menos detienen importación; no elige entre dos columnas de resultado ni mueve datos al índice. Represente campo vacío con separador, por ejemplo ;A+;;0; una fila truncada es distinta. Configure separador y decimal según variante (PL: punto y coma y coma; EN: coma y punto) y codificación: el generador F usa UTF-8 y «CSV UTF-8» de Excel también puede leerse como UTF-8; exportación antigua con página polaca puede necesitar cp1250. Compruebe formato, pues idioma de interfaz no determina codificación. Un r_net vacío y un close_time_utc vacío significan operación aún abierta y se omite; hora de cierre sin resultado es registro incompleto y detiene cálculo. Valor no vacío ilegible como número (incluidos «NA» o guion) detiene y señala registro; también operación cerrada sin hora de cierre en formato F (2026-07-08 13:15), pues habría que adivinar orden de la curva.

import pandas as pd

CSV_PATH = "dziennik.csv"   # exportación del diario (estructura del anexo F)
CSV_SEP  = ";"              # ";" para CSV con configuración PL, "," para EN
DECIMAL  = ","              # "," para CSV con configuración PL, "." para EN
ENCODING = "utf-8-sig"      # el generador del anexo F escribe UTF-8 (admite BOM); un CSV antiguo con página de códigos polaca puede requerir "cp1250"

REQUIRED = ["r_net", "setup", "setup_class", "error_code", "close_time_utc"]

# Leer un archivo local y validar su estructura ANTES de crear un DataFrame.
# pandas.read_csv puede renombrar cabeceras duplicadas o inferir un índice de campos adicionales.
import csv

try:
    with open(CSV_PATH, encoding=ENCODING, newline="") as source:
        reader = csv.reader(source, delimiter=CSV_SEP, strict=True)
        header = next(reader, None)
        if not header:
            raise SystemExit("No se puede leer el CSV: la cabecera está vacía.")
        if any(not name or name != name.strip() for name in header):
            raise SystemExit("Nombre de columna vacío o con espacios en los extremos — corregir la cabecera CSV.")
        duplicates = sorted({name for name in header if header.count(name) > 1})
        if duplicates:
            raise SystemExit(f"Columnas duplicadas: {duplicates} — resolverlas en el diario.")
        rows, record_indexes = [], []
        for index, row in enumerate(reader):
            if not row:                              # un registro CSV completamente vacío
                continue
            if len(row) != len(header):
                raise SystemExit(f"Registro {index + 1}: se esperaban {len(header)} campos, se obtuvieron {len(row)} "
                                 f"— revisar separadores y comillas; no se descartó ningún campo.")
            rows.append(row)
            record_indexes.append(index)
    df = pd.DataFrame(rows, columns=header, index=record_indexes, dtype="string").replace("", pd.NA)
except UnicodeDecodeError:
    raise SystemExit(f"No se puede decodificar {CSV_PATH} como {ENCODING} — revisar UTF-8 o la antigua cp1250.") from None
except (OSError, csv.Error) as exc:
    raise SystemExit(f"No se puede leer {CSV_PATH}: {exc}") from None

missing = [c for c in REQUIRED if c not in df.columns]
if missing:
    raise SystemExit(f"Faltan columnas: {missing} — contrastar cabeceras con el anexo F y el ajuste CSV_SEP.")

raw = df["r_net"]                                   # se permite un resultado vacío solo si la hora de cierre también está vacía
txt = raw.astype("string").str.strip().str.replace(DECIMAL, ".", regex=False)   # interpretar nosotros la columna: una celda incorrecta no debe ocultar las legibles
r   = pd.to_numeric(txt, errors="coerce").astype("Float64")  # evitar desbordamientos de enteros en las sumas posteriores
r   = r.where(r.abs() != float("inf"))              # inf tampoco es un múltiplo R válido
invalid = raw.notna() & r.isna()                    # rellenado, pero no legible como número
if invalid.any():
    raise SystemExit(f"r_net no válido en los registros {(df.index[invalid] + 1).tolist()} "
                     f"(registro 1 = primera fila bajo la cabecera) — revisar valores y DECIMAL; "
                     f"NA, guiones e infinito no son números.")

blank = df.isna().all(axis=1)                       # fila totalmente vacía = fila sin usar de la hoja (exportación XLSX completa), no una operación
df, r = df[~blank], r[~blank]
has_close = df["close_time_utc"].fillna("").str.strip().ne("")
incomplete = r.isna() & has_close
if incomplete.any():
    raise SystemExit(f"Falta r_net en los registros cerrados {(df.index[incomplete] + 1).tolist()} "
                     f"— completar el resultado neto antes de calcular las métricas.")
open_trades = int(r.isna().sum())
df = df.assign(r_net=r).dropna(subset=["r_net"])   # índice conservado = números de registro para el mensaje siguiente

# la curva y el drawdown requieren orden cronológico (anexo F: por hora de cierre); una operación cerrada sin hora de cierre legible detiene el script
close_text = df["close_time_utc"].fillna("").str.strip()
has_time = close_text.str.fullmatch(r"\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?(?:Z|[+-]\d{2}:\d{2})?")
closed = pd.to_datetime(close_text.where(has_time), errors="coerce", format="ISO8601", utc=True)   # el anexo F escribe 2026-07-08 13:15; el formato fijo independiza la comprobación del orden de filas; utc=True evita fallos por sufijos inesperados "Z"/"+00:00"
if closed.isna().any():
    raise SystemExit(f"close_time_utc ilegible en los registros {(df.index[closed.isna()] + 1).tolist()} "
                     f"— cada operación cerrada necesita una hora de cierre en el formato del anexo F (2026-07-08 13:15); corregir el diario, no el script.")
df = df.assign(_t=closed).sort_values("_t", kind="stable").drop(columns="_t").reset_index(drop=True)

print(f"Cargadas {len(df)} operaciones cerradas ({open_trades} fila(s) omitida(s): sin hora de cierre ni r_net).")

Obtiene: un DataFrame, df, preparado para siguientes fragmentos y ordenado por cierre (curva y drawdown siguen ese orden), más número de abiertas omitidas (fila vacía es registro enteramente vacío, como al exportar toda XLSX, y no se cuenta; con datos pero sin resultado ni cierre se omite como abierta; entrada cerrada sin resultado detiene). Si lista registros, corrija filas del diario, no el script; si afecta toda la columna revise también constante DECIMAL . Numeración empieza tras encabezado e incluye registros vacíos; un campo entre comillas puede contener separador o salto de línea y sigue siendo un registro. Cada operación debe aparecer una vez: compruebe identificadores trade_id antes de combinar exportaciones; S1 no elimina duplicados. Fecha sin hora no basta para reconstruir secuencia. Hora sin zona se trata como UTC; un desplazamiento ISO 8601 se convierte a UTC. Para horas idénticas se conserva orden del archivo; establézcalo en el diario, pues afecta al drawdown.

Mismas definiciones que P5 y panel de la hoja: tasa de acierto incluye equilibrio en denominador; factor de beneficio devuelve mensaje en vez de dividir por cero; esperanza es media de R neto (r_net); drawdown se mide desde curva acumulada inicial de 0 R (primera pérdida ya es drawdown, como fórmula F). Ejecute tras S1. Comparten definiciones, pero política de error difiere de P5: S1 detiene análisis por resultado inválido de cerrada; P5 ordena omitirla y contarla. Compare solo tras establecer qué operaciones incluyen ambas muestras.

import math

def metrics(r):
    "Métricas principales del diario sobre una serie de múltiplos R netos.\n    Definiciones (como en el panel del anexo F):\n    - tasa de acierto: operaciones con r > 0 entre TODAS las válidas (break-even permanece en el denominador)\n    - factor de beneficio: suma de r positivos entre abs(suma de r negativos)\n    - esperanza matemática: media aritmética de r (neta de costes)\n    - drawdown máximo: mayor caída de la curva R acumulada desde su máximo móvil,\n      con inicio en 0 R (una primera pérdida de -1 R ya supone 1 R de drawdown)"
    n = len(r)
    if n == 0:
        raise SystemExit("No hay operaciones válidas — nada que medir.")
    try:
        r = r.astype(float)
    except (TypeError, ValueError):
        raise SystemExit("Las métricas requieren valores R netos numéricos — ejecutar S1 primero.") from None
    if not r.map(math.isfinite).all():
        raise SystemExit("Las métricas requieren valores R netos finitos — ejecutar S1 primero.")
    wins   = r[r > 0]
    losses = r[r < 0]
    equity   = r.cumsum()
    peak     = equity.cummax().clip(lower=0)   # anexo F: MAX(0; máximo móvil)
    drawdown = peak - equity
    try:
        total = math.fsum(r)
        pf_value = None if losses.empty else math.fsum(wins) / abs(math.fsum(losses))
    except (OverflowError, ValueError, ZeroDivisionError):
        raise SystemExit("Rango numérico superado — revisar los valores R; no se devolvieron métricas.") from None
    if (not equity.map(math.isfinite).all() or not drawdown.map(math.isfinite).all()
            or (pf_value is not None and not math.isfinite(pf_value))):
        raise SystemExit("Rango numérico superado — revisar los valores R; no se devolvieron métricas.")
    pf = "sin pérdidas en la muestra" if pf_value is None else round(pf_value, 2)
    return {
        "trades":         n,
        "win_rate":       round(len(wins) / n, 3),
        "profit_factor":  pf,
        "expectancy_R":   round(total / n, 3),
        "sum_R":          round(total, 2),
        "max_drawdown_R": round(drawdown.max(), 2),
    }

for key, value in metrics(df["r_net"]).items():
    print(f"{key:>16}: {value}")

Obtiene seis métricas semanales; sin pérdidas PF es mensaje, no número. Si excede rango de cálculo, detiene en lugar de devolver inf, NaN o un drawdown aparentemente cero. WR = 0.4 significa 40%. PF se calcula en R, no PLN; con riesgo monetario variable puede diferir. Costes ya incluidos en r_net ; S2 no los resta otra vez. No espere evaluación de ventaja solo de la media ni por superar 20 o 50 operaciones. Definiciones del panel en complemento F y plan de muestra, costes y validación en complemento N.

Dibuja curva acumulada R y drawdown del máximo debajo y guarda PNG junto al CSV indicado por CSV_PATH (S1), independientemente de carpeta de ejecución. Revela secuencia y agrupaciones de pérdidas ocultas en agregados. Es suma de R, no saldo en PLN ni equity porcentual: con riesgo monetario variable no reproduce cambios de saldo. No incluye depósitos, retiradas ni posiciones abiertas. Ejecute tras S1.

import math
import matplotlib.pyplot as plt
from pathlib import Path

if len(df) == 0:
    raise SystemExit("No hay operaciones cerradas — nada que representar.")

equity   = df["r_net"].astype(float).cumsum()
drawdown = equity.cummax().clip(lower=0) - equity   # misma definición que S2 y el anexo F (la curva empieza en 0 R)

if not equity.map(math.isfinite).all() or not drawdown.map(math.isfinite).all():
    raise SystemExit("Rango numérico superado — revisar los valores R; no se guardó ningún gráfico.")

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(9, 6), sharex=True,
                               gridspec_kw={"height_ratios": [3, 1]})
x = [0, *(equity.index + 1)]                        # operación 0 = punto inicial de 0 R desde el que se mide el drawdown
ax1.plot(x, [0.0, *equity.values])
ax1.set_ylabel("resultado (R)")
ax1.set_title("Curva de resultados — R netos acumulados")
ax1.grid(True, alpha=0.3)

ax2.fill_between(x, [0.0, *(-drawdown.values)], 0, alpha=0.5)
ax2.set_ylabel("drawdown (R)")
ax2.set_xlabel("operación n.º")
ax2.grid(True, alpha=0.3)

fig.tight_layout()
PNG_PATH = Path(CSV_PATH).with_name("equity_curve.png")   # junto al archivo del diario, sea cual sea el directorio de trabajo
fig.savefig(PNG_PATH, dpi=150)
plt.close(fig)
print(f"Guardado: {PNG_PATH}")

Obtiene un equity_curve.png junto al diario; curva comienza en 0 R (operación 0), base de drawdown. El panel inferior muestra caída de R acumulado desde máximo (inicial 0 R, como S2 y hoja F) de toda la historia, no pérdida semanal. Compruebe límite semanal del plan E por separado, semana a semana, con su unidad y regla de reinicio.

Guarde como dziennik.csv (UTF-8) junto al script S1+S2+S3 y ejecute. Solo incluye columnas necesarias para S1; exportación completa F (26 columnas) se carga igual. Última fila es operación abierta (r_net vacío); primeras dos cerradas pierden, por lo que drawdown parte de 0 R.

trade_id;close_time_utc;setup;setup_class;r_net;error_code
2026-041;2026-07-08 13:15;wybicie-azja-londyn;A+;-1,00;5
2026-042;2026-07-09 16:40;powrot-do-sredniej;B;-1,08;3
2026-043;2026-07-10 09:05;wybicie-azja-londyn;A+;1,83;0
2026-044;2026-07-13 12:20;wybicie-azja-londyn;B;0,00;0
2026-045;2026-07-14 14:30;powrot-do-sredniej;C;2,40;0
2026-046;;wybicie-azja-londyn;B;;

Salida esperada (Python 3.12.10, pandas 3.0.5, matplotlib 3.11.1; ejecutado el 17 de septiembre de 2026):

Cargadas 5 operaciones cerradas (1 fila(s) omitida(s): sin hora de cierre ni r_net).
          trades: 5
        win_rate: 0.4
   profit_factor: 2.03
    expectancy_R: 0.43
           sum_R: 2.15
  max_drawdown_R: 2.08
Guardado: equity_curve.png

Compruebe a mano: ganancias 1.83 + 2.40 = 4.23 R, pérdidas 1.00 + 1.08 = 2.08 R → PF 2.03; total 2.15 R / 5 = 0.43 R; curva −1.00 → −2.08 → −0.25 → −0.25 → 2.15, con máximo inicial 0 R → drawdown 2.08 R (sin punto 0 R sería 1.08 R: distinción de definición F). La salida corresponde exactamente a este archivo y versiones; si difiere, revise en orden datos (separador, decimal, codificación y filas), constantes S1, código y versiones.

Volumen en lotes para stop k × ATR: riesgo de capital y porcentaje, distancia de ATR y redondeo ABAJO al paso de lotes de su cuenta (parámetros lot_step y min_lot , ambos 0.01 por defecto), para no superar riesgo PREVISTO. Pérdida real puede ser mayor: no incluye deslizamiento, huecos, spread, comisión ni swap. Capital y valor del pip deben estar en misma divisa; compruebe aparte volumen máximo y margen en especificación. Introduzca ATR en pips (EURUSD 0.0080 → 80; USDJPY 0.80 → 80; tamaño del pip en perfil B); el valor para 1 lote está en especificación o calculadora C.

import math
from decimal import Decimal
from fractions import Fraction

def atr_position_size(capital, risk_pct, atr_pips, atr_mult, pip_value_per_lot,
                      lot_step=0.01, min_lot=0.01):
    "Volumen en lotes para un stop de atr_mult × ATR.\n    atr_pips: ATR en PIPS, no en unidades de precio (EURUSD ATR 0.0080 -> 80; USDJPY ATR 0.80 -> 80 — tamaño de pip del par).\n    pip_value_per_lot: valor de 1 pip para 1.00 lote, en divisa de la cuenta\n    (p. ej., ~10 USD para EURUSD en cuenta USD — revisar especificaciones).\n    lot_step / min_lot: paso y mínimo de volumen de TU cuenta (0.01 es frecuente, no universal).\n    Redondea HACIA ABAJO a múltiplos de lot_step medidos desde cero. La aritmética de fracciones\n    conserva las entradas decimales sin depender de la precisión global de Decimal.\n    El riesgo planificado no es la pérdida máxima: el modelo no incluye slippage, gaps, spread ni swap."
    inputs = (capital, risk_pct, atr_pips, atr_mult, pip_value_per_lot, lot_step, min_lot)
    if any(isinstance(x, bool) or not isinstance(x, (int, float, Decimal)) for x in inputs):
        raise ValueError("Usar valores numéricos, no texto ni booleanos (los decimales de Python usan punto).")
    decimals = tuple(Decimal(str(x)) for x in inputs)
    if not all(x.is_finite() and x > 0 for x in decimals):
        raise ValueError("Todas las entradas deben ser números finitos y positivos.")
    c, p, a, k, v, step, minimum = map(Fraction, decimals)
    if p > 100:
        raise ValueError("risk_pct es un porcentaje del capital (1.0 = 1%).")
    risk_amount = c * p / 100
    sl_pips     = a * k
    lots        = math.floor(risk_amount / (sl_pips * v * step)) * step
    if lots < minimum:
        return {"warning": "No hay volumen admisible para este lot_step y min_lot dentro del presupuesto de riesgo."}
    exact = {"risk_amount": risk_amount, "sl_pips": sl_pips,
             "lots": lots, "planned_risk": lots * sl_pips * v}
    try:
        result = {key: float(value) for key, value in exact.items()}
    except OverflowError:
        raise ValueError("Las entradas superan el rango numérico de salida.") from None
    if not all(math.isfinite(value) and value > 0 for value in result.values()):
        raise ValueError("Las entradas superan el rango numérico de salida.")
    return result  # flotantes para mostrar; sin redondeo adicional del stop ni de importes monetarios


print(atr_position_size(capital=10_000, risk_pct=1.0,
                        atr_pips=80, atr_mult=1.5, pip_value_per_lot=10.0))
# {'risk_amount': 100.0, 'sl_pips': 120.0, 'lots': 0.08, 'planned_risk': 96.0}

Obtiene lotes redondeados abajo y riesgo PREVISTO resultante (sin costes, ≤ presupuesto; pérdida real puede ser mayor). El paso se mide desde cero; confirme esa cuadrícula en su cuenta. Stop y dinero no se redondean más: ATR de 0.26 pips × 1.25 da 0.325 pips, no 0.3. Se devuelven valores float para lectura y los pasos se calculan con fracciones exactas; compruebe aparte incremento mínimo del precio. Si ajusta stop a ese incremento, recalcule volumen. No espere decisión de entrada: calcula «cuánto», nunca «si entrar».

Herramienta de conciencia de volatilidad intradía: en barras reales muestra qué porcentaje del ATR diario ya se ha «usado» y alerta al cruzar umbral, momento de consultar plan, no perseguir precio. Obtiene datos diarios sin lookahead (sin futuro); en históricas, request.security mantiene último valor diario confirmado hasta final de siguiente periodo D y muestra el nuevo en última barra intradía. En tiempo real obtiene rango y ATR en desarrollo de vela D actual, por lo que lecturas previas pueden cambiar al recargar (repintado). Día significa vela D del proveedor, no día civil Europe/Warsaw. ATR del denominador también cambia durante el día; 100% no limita el movimiento posible. No reconstruye cruces sucesivos de hoy en barras históricas.

//@version=6
indicator("Rango diario frente a ATR (herramienta de contexto)", overlay=false)

if not timeframe.isintraday
    runtime.error("Usar un gráfico intradía temporal inferior a 1D.")

atrLen  = input.int(14, "Longitud ATR", minval=2)
warnPct = input.int(80, "Alertar cuando el rango diario alcance % del ATR", minval=10, maxval=200)

// Histórico: último valor D confirmado, actualizado al FINAL de cada periodo D.
// Tiempo real: valores D en formación; las lecturas pasadas pueden repintarse al recargar.
dayHigh = request.security(syminfo.tickerid, "D", high,            lookahead=barmerge.lookahead_off)
dayLow  = request.security(syminfo.tickerid, "D", low,             lookahead=barmerge.lookahead_off)
dayAtr  = request.security(syminfo.tickerid, "D", ta.atr(atrLen),  lookahead=barmerge.lookahead_off)

rangePct = dayAtr > 0 ? 100.0 * (dayHigh - dayLow) / dayAtr : na

plot(rangePct, "Rango diario como % del ATR", linewidth=2)
hline(100, "100% del ATR")
hline(warnPct, "Nivel de alerta", linestyle=hline.style_dotted)

alertcondition(ta.crossover(rangePct, warnPct),
     title="Rango alcanzó % del ATR",
     message="El rango diario alcanzó tu % del ATR — releer el plan antes de cualquier nueva entrada.")

Obtiene panel inferior y condición de alerta; créela en plataforma tras añadir indicador y seleccione «Rango alcanzó % de ATR». Use velas intradía temporales estándar; Heikin Ashi/Renko pueden diferir. Lectura ausente (na) significa falta ATR o datos, no 0%. Condición exige pasar de valor ≤ umbral a > umbral; permanecer encima no repite alerta. «Una vez por cierre de barra» limita a cierre intradía, no confirma toda vela D. Recree alerta tras cambiar parámetros. No espere señal direccional: lectura alta dice «ya ocurrió movimiento»; playbook decide actuación.

Sección 3 del plan E dibujada: resalta horas en que SE PERMITE abrir y alerta al cerrar ventana. Introduzca horas Europe/Warsaw; identificador IANA incluye CET/CEST. Por defecto lunes a viernes; opción amplía a todos los días. Solo confirma horario del plan, no otras condiciones. Use velas intradía temporales estándar.

//@version=6
indicator("Ventanas de sesión del plan de trading", overlay=true)

if not timeframe.isintraday
    runtime.error("Usar un gráfico intradía temporal inferior a 1D.")

tz   = input.string("Europe/Warsaw", "Zona horaria de las ventanas")
win1 = input.session("0900-1200", "Ventana 1")
win2 = input.session("1400-1700", "Ventana 2")

weekdaysOnly = input.bool(true, "Solo de lunes a viernes")
days = weekdaysOnly ? ":23456" : ":1234567"

inWin = not na(time(timeframe.period, win1 + days, tz)) or not na(time(timeframe.period, win2 + days, tz))

bgcolor(inWin ? color.new(color.teal, 88) : na, title="Dentro de una ventana del plan")

windowJustClosed = inWin[1] and not inWin
alertcondition(windowJustClosed,
     title="Ventana del plan cerrada",
     message="Tu ventana operativa acaba de cerrarse — gestionar solo posiciones abiertas, sin nuevas entradas (plan, sección 3).")

Obtiene sombreado del horario y condición de cierre («Ventana del plan cerrada»: cree alerta en plataforma tras añadir indicador). Detecta primera vela fuera tras otra dentro, por lo que no garantiza aviso exacto al cerrar: sin vela nueva ni actualización no comprueba nada. Use marco con aperturas/cierres alineados a límites. En M15, ventana 09:00–12:00 y vela 12:00 disponible, sombreado termina 12:00 y condición activa en esa primera vela exterior. «Una vez por barra» puede avisar en primera actualización; «Una vez por cierre» solo al cerrar. Ventanas solapadas forman unión: fin de primera no cierra segunda. Recree tras cambiar horas. No bloquea órdenes: recuerda plan; lista previa lo exige.