Docker y docker-compose: imágenes reproducibles con uv.

from pathlib import Path

RAIZ = Path("proyecto")
ARCHIVOS = {
    "pyproject.toml": '[project]\nname = "bot"\nversion = "1.4.0"\n'
                      'requires-python = ">=3.14,<3.15"\n',
    "uv.lock": "version = 1\n# pandas 3.0.6 ...\n",
    "src/bot/__init__.py": "",
    "src/bot/__main__.py": "from bot.bucle import main\nmain()\n",
    "src/bot/bucle.py": "def main():\n    ...\n" + "#\n" * 3000,
    "src/bot/estrategia.py": "def senal(p):\n    ...\n" + "#\n" * 4500,
    "src/bot/salud.py": "# ver la celda del latido\n" + "#\n" * 900,
    "tests/test_estrategia.py": "def test_senal():\n    ...\n",
    "README.md": "# Bot\n",
    ".env": "EXCHANGE_API_KEY=clave-de-juguete-123\n",
    "secretos/db_password.txt": "juguete-db\n",
    "secretos/exchange_api_key.txt": "clave-de-juguete-123\n",
    "secretos/grafana_admin.txt": "juguete-grafana\n",
    "prometheus.yml": "scrape_configs: []\n",
    "data/ticks.parquet": "x" * 2_000_000,
    ".git/objects/pack/pack.pack": "x" * 3_000_000,
}
for ruta, texto in ARCHIVOS.items():
    p = RAIZ / ruta
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(texto)

total = sum(len(t.encode()) for t in ARCHIVOS.values())
print(len(ARCHIVOS), "ficheros,", f"{total / 1e6:.2f} MB")

DF_INGENUO = """\
FROM python:3.14
WORKDIR /app
COPY . .
RUN pip install uv && uv sync
ENV EXCHANGE_API_KEY=clave-de-juguete-123
CMD ["uv", "run", "python", "-m", "bot"]
"""

import re

def leer_dockerfile(texto):
    """Instrucciones con su línea, etapa, opciones y argumentos."""
    instr, buf, n0, etapa = [], "", None, -1
    for n, linea in enumerate(texto.splitlines(), 1):
        s = linea.strip()
        if not s or s.startswith("#"):
            continue                   # vacías y comentarios
        n0 = n0 or n
        if s.endswith("\\"):
            buf += s[:-1].strip() + " "
            continue
        buf += s
        kw, _, args = buf.partition(" ")
        opciones = {}
        args = args.strip()
        while args.startswith("--"):
            tok, _, args = args.partition(" ")
            k, _, v = tok[2:].partition("=")
            opciones.setdefault(k, []).append(v)
            args = args.strip()
        kw = kw.upper()
        etapa += kw == "FROM"
        instr.append({"linea": n0, "etapa": etapa, "kw": kw,
                      "op": opciones, "args": args})
        buf, n0 = "", None
    return instr

for i in leer_dockerfile(DF_INGENUO):
    print(i["linea"], i["etapa"], i["kw"], i["args"])

IMG = {
    "PY": "python:3.14.9-slim-trixie@sha256:"
          "b347e2dcbc7f0b6f9fb7182e9da6a187"
          "fb10ae715e0692518c1aed0eec94a700",
    "UV": "ghcr.io/astral-sh/uv:0.13.5@sha256:"
          "32a7bb39b496babbd997b0a39a4d66c1"
          "26bdaf5661b302bf6f06abda8deca6ff",
}
DF_BUENO = """\
# syntax=docker/dockerfile:1
# --- etapa 1: dependencias (uv no llega a la imagen final)
FROM {PY} AS builder
COPY --from={UV} /uv /uvx /bin/
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy \\
    UV_PYTHON_DOWNLOADS=0 UV_NO_DEV=1
WORKDIR /app
RUN --mount=type=cache,target=/root/.cache/uv \\
    --mount=type=bind,source=uv.lock,target=uv.lock \\
    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \\
    uv sync --locked --no-install-project --no-editable

# --- etapa 2: Python, el entorno y el código
FROM {PY}
RUN groupadd --system --gid 10001 bot \\
 && useradd --system --uid 10001 --gid bot --no-create-home bot
WORKDIR /app
COPY --from=builder --chown=bot:bot /app/.venv /app/.venv
COPY --chown=bot:bot src/ /app/src/
ENV PATH="/app/.venv/bin:$PATH" PYTHONPATH=/app/src \\
    PYTHONUNBUFFERED=1
USER bot
HEALTHCHECK --interval=30s --timeout=5s --start-period=60s \\
    --retries=3 CMD ["python", "-m", "bot.salud"]
CMD ["python", "-m", "bot"]
""".format(**IMG)

(RAIZ / "Dockerfile").write_text(DF_BUENO)
print(len(leer_dockerfile(DF_BUENO)), "instrucciones en",
      1 + max(i["etapa"] for i in leer_dockerfile(DF_BUENO)),
      "etapas")

DIGEST = re.compile(r"@sha256:[0-9a-f]{64}$")
SECRETO = re.compile(r"KEY|SECRET|TOKEN|PASS", re.I)
INSTALA = re.compile(r"uv sync|pip install")

def revisar_dockerfile(texto):
    ins = leer_dockerfile(texto)
    etapas = {i["args"].split()[-1].lower() for i in ins
              if i["kw"] == "FROM" and " as " in i["args"].lower()}
    ultima = max(i["etapa"] for i in ins)
    out = []

    def mal(i, cod, msg):
        out.append((i["linea"], cod, msg))

    instalo = set()                     # etapas que ya instalaron
    for i in ins:
        kw, a, op = i["kw"], i["args"], i["op"]
        imgs = [a.split()[0]] if kw == "FROM" else op.get("from", [])
        for img in imgs:
            if img.lower() not in etapas and not DIGEST.search(img):
                mal(i, "DF01", f"imagen sin digest: {img}")
        if kw in ("ENV", "ARG"):
            for k in re.findall(r"(\w+)=", a) or a.split()[:1]:
                if SECRETO.search(k):
                    mal(i, "DF06", f"secreto en {kw}: {k}")
        if kw == "COPY" and a.split()[0] == "." \
                and i["etapa"] not in instalo:
            mal(i, "DF07", "COPY . antes de instalar dependencias")
        if kw == "RUN" and INSTALA.search(a):
            instalo.add(i["etapa"])
            if "pip install" in a:
                mal(i, "DF03", "pip install sin lockfile")
            if "uv sync" in a and not re.search(
                    r"--(locked|frozen)\b", a):
                mal(i, "DF03", "uv sync sin --locked ni --frozen")
            if "uv sync" in a and not any(
                    "type=cache" in m and ".cache/uv" in m
                    for m in op.get("mount", [])):
                mal(i, "DF08", "uv sync sin caché montada")
            if i["etapa"] == ultima:
                mal(i, "DF04", "dependencias instaladas en la "
                               "imagen final")
    final = [i for i in ins if i["etapa"] == ultima]
    users = [i["args"].split(":")[0] for i in final
             if i["kw"] == "USER"]
    if not users or users[-1] in ("root", "0"):
        mal(final[0], "DF02", "la imagen final corre como root")
    if not any(i["kw"] == "HEALTHCHECK" for i in final):
        mal(final[0], "DF05", "sin HEALTHCHECK")
    return sorted(out)

for nombre, df in (("ingenuo", DF_INGENUO), ("bueno", DF_BUENO)):
    h = revisar_dockerfile(df)
    print(f"{nombre}: {len(h)} hallazgos")
    for linea, cod, msg in h:
        print(f"  línea {linea:>2} {cod} {msg}")

def patron_a_regex(p):
    p = p.strip("/")
    r = re.escape(p).replace(r"\*\*/", "(?:.*/)?")
    r = r.replace(r"\*\*", ".*").replace(r"\*", "[^/]*")
    return re.compile(r.replace(r"\?", "[^/]") + "(?:/.*)?$")

def contexto(raiz, ignorar=""):
    reglas = [(l.startswith("!"), patron_a_regex(l.lstrip("!")))
              for l in ignorar.splitlines()
              if l.strip() and not l.startswith("#")]
    dentro = {}
    for f in sorted(p for p in raiz.rglob("*") if p.is_file()):
        rel = f.relative_to(raiz).as_posix()
        fuera = False
        for negar, rx in reglas:
            if rx.match(rel):
                fuera = not negar
        if not fuera:
            dentro[rel] = f.stat().st_size
    return dentro

def revisar_contexto(dentro):
    return [("DF09", f) for f in dentro
            if f == ".env" or f.startswith("secretos/")]

IGNORAR = """\
.git
.venv
**/__pycache__
.env
secretos/
data/
tests/
*.md
Dockerfile
"""
(RAIZ / ".dockerignore").write_text(IGNORAR)
for nombre, ign in (("sin .dockerignore", ""), ("con él", IGNORAR)):
    c = contexto(RAIZ, ign)
    print(f"{nombre}: {len(c)} ficheros, "
          f"{sum(c.values()) / 1e6:.2f} MB,",
          revisar_contexto(c))
print(sorted(contexto(RAIZ, IGNORAR)))

from importlib import metadata as md
from packaging.requirements import Requirement

def cierre(raices):
    """Paquetes instalados que arrastran `raices` (sin extras)."""
    vistos, pila = set(), list(raices)
    while pila:
        n = pila.pop().lower().replace("_", "-")
        if n in vistos:
            continue
        try:
            d = md.distribution(n)
        except md.PackageNotFoundError:
            continue
        vistos.add(n)
        for r in d.requires or []:
            req = Requirement(r)
            if req.marker is None or req.marker.evaluate({"extra": ""}):
                pila.append(req.name)
    return vistos

def bytes_instalados(n):
    return sum(f.size or 0 for f in md.distribution(n).files or [])

DEPS = cierre(["numpy", "pandas", "pyarrow", "ccxt", "pydantic",
               "pyyaml", "requests"])
tam = {n: bytes_instalados(n) for n in DEPS}
DEPS_MB = round(sum(tam.values()) / 1e6, 1)
print(len(DEPS), "paquetes,", DEPS_MB, "MB instalados")
for n, b in sorted(tam.items(), key=lambda x: -x[1])[:4]:
    print(f"  {n:<10} {b / 1e6:6.1f} MB")

import hashlib

def h(*partes):
    return hashlib.sha256("|".join(partes).encode()).hexdigest()

def claves(df, repo, ignorar):
    """Clave y peso (MB) de cada capa de la imagen final."""
    ctx = {f: t for f, t in repo.items()
           if contexto_virtual(f, ignorar)}
    ins = leer_dockerfile(df)
    ultima = max(i["etapa"] for i in ins)
    nombre, fin, previa, capas = {}, {}, {}, []
    for i in ins:
        kw, a, op, e = i["kw"], i["args"], i["op"], i["etapa"]
        lee = []
        if kw == "FROM":
            if " as " in a.lower():
                nombre[e] = a.split()[-1]
            clave = h(a)
        else:
            if kw == "COPY" and "from" in op:
                lee = [fin.get(op["from"][0], op["from"][0])]
            elif kw == "COPY":
                src = a.split()[-2].rstrip("/")
                lee = [h(f, t) for f, t in sorted(ctx.items())
                       if src == "." or f.startswith(src + "/")]
            for m in op.get("mount", []):
                if "type=bind" in m:
                    s = re.search(r"source=([^,]+)", m).group(1)
                    lee.append(h(s, ctx.get(s, "")))
            clave = h(previa[e], kw, a, str(op), *lee)
        previa[e] = clave
        if e < ultima:                  # etapa intermedia
            fin[nombre.get(e, str(e))] = clave
            continue
        peso = 0.0                      # MB de la capa
        if kw == "RUN" and INSTALA.search(a):
            peso = DEPS_MB
        elif kw == "COPY" and "/app/.venv" in a:
            peso = DEPS_MB
        elif kw == "COPY":
            peso = sum(len(t) for f, t in ctx.items()
                       if src == "." or
                       f.startswith(src + "/")) / 1e6
        capas.append((clave, peso))
    return capas

def contexto_virtual(f, ignorar):
    reglas = [(l.startswith("!"), patron_a_regex(l.lstrip("!")))
              for l in ignorar.splitlines() if l.strip()]
    fuera = False
    for negar, rx in reglas:
        if rx.match(f):
            fuera = not negar
    return not fuera

COMMITS = [
    ("estrategia: filtro de volumen", "src/bot/estrategia.py"),
    ("tests del filtro", "tests/test_estrategia.py"),
    ("estrategia: umbral", "src/bot/estrategia.py"),
    ("README", "README.md"),
    ("bucle: reconexión", "src/bot/bucle.py"),
    ("datos de ejemplo", "data/ticks.parquet"),
    ("pandas 3.0.6 a 3.0.7", "uv.lock"),
    ("estrategia: corrige signo", "src/bot/estrategia.py"),
    ("salud: latido a 90 s", "src/bot/salud.py"),
    ("estrategia: stop", "src/bot/estrategia.py"),
]
repo = dict(ARCHIVOS)
antes = {"ingenuo": claves(DF_INGENUO, repo, ""),
         "bueno": claves(DF_BUENO, repo, IGNORAR)}
FILAS = []
for n, (msg, f) in enumerate(COMMITS, 1):
    repo[f] = repo[f] + f"# cambio {n}\n"
    fila = [n, msg]
    for nombre, df, ign in (("ingenuo", DF_INGENUO, ""),
                            ("bueno", DF_BUENO, IGNORAR)):
        ahora = claves(df, repo, ign)
        viejas = {c for c, _ in antes[nombre]}
        fila.append(round(sum(p for c, p in ahora
                              if c not in viejas), 3))
        antes[nombre] = ahora
    FILAS.append(fila)
    print(f"{n:>2} {msg:<29} {fila[2]:>7.1f} {fila[3]:>7.3f}")
tot_i = sum(f[2] for f in FILAS)
tot_b = sum(f[3] for f in FILAS)
print(f"total MB nuevos: ingenuo {tot_i:.1f}, bueno {tot_b:.1f}")

import os

LATIDO = Path("tmp/latido")
MAX_EDAD = 90                           # segundos

def comprobar(ruta, ahora, max_edad=MAX_EDAD):
    """0 = sano, 1 = enfermo (el código que espera HEALTHCHECK)."""
    try:
        edad = ahora - ruta.stat().st_mtime
    except FileNotFoundError:
        return 1
    return 0 if edad <= max_edad else 1

LATIDO.parent.mkdir(exist_ok=True)
LATIDO.touch()
T0 = 1_760_000_000                      # reloj inyectado (epoch)
os.utime(LATIDO, (T0, T0))
for t in (T0 + 30, T0 + 90, T0 + 91):
    print(f"latido hace {t - T0:>3} s -> {comprobar(LATIDO, t)}")
print("sin fichero ->", comprobar(Path("tmp/no-existe"), T0))

def peor_deteccion(max_edad, intervalo, reintentos, timeout):
    """Del último latido a `unhealthy`, en el peor caso."""
    return max_edad + intervalo * reintentos + timeout

print("peor caso:", peor_deteccion(MAX_EDAD, 30, 3, 5), "s")

import yaml

IMG |= {
    "PG": "postgres:18.6-trixie@sha256:"
          "edb0c08dc33befd0262ab3b5cf1c4f3f"
          "8cfd8b09e88c05b45adee29d6761ddb6",
    "PROM": "prom/prometheus:v3.15.1@sha256:"
            "fa3302279710be2f21a8301ec17a8b4f"
            "2301fd829bd113839b3b6b02b6f191a2",
    "GRAF": "grafana/grafana:13.2.4@sha256:"
            "dbbc38ea21103cefb50396340cd88afe"
            "c779c7e070b0a7bf5a75b4ce46651d3f",
}
COMPOSE_INGENUO = """\
services:
  db:
    image: postgres:latest
    environment:
      POSTGRES_PASSWORD: juguete-db
    ports: ["5432:5432"]
  bot:
    build: .
    environment:
      EXCHANGE_API_KEY: clave-de-juguete-123
    depends_on: [db]
  grafana:
    image: grafana/grafana
    ports: ["3000:3000"]
"""
COMPOSE_BUENO = """\
name: bot-trading
services:
  db:
    image: {PG}
    environment:
      POSTGRES_USER: bot
      POSTGRES_DB: trading
      POSTGRES_PASSWORD_FILE: /run/secrets/db_password
    secrets: [db_password]
    volumes: ["pgdata:/var/lib/postgresql"]
    networks: [interna]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U bot -d trading"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: unless-stopped
  bot:
    build: .
    image: bot-trading:1.4.0
    environment:
      BOT_MODO: paper
      DB_HOST: db
    secrets: [db_password, exchange_api_key]
    depends_on:
      db: {{condition: service_healthy}}
    networks: [interna, salida]
    read_only: true
    tmpfs: [/tmp]
    stop_grace_period: 30s
    restart: unless-stopped
  prometheus:
    image: {PROM}
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - promdata:/prometheus
    networks: [interna]
    restart: unless-stopped
  grafana:
    image: {GRAF}
    environment:
      GF_SECURITY_ADMIN_PASSWORD__FILE: /run/secrets/grafana_admin
    secrets: [grafana_admin]
    ports: ["127.0.0.1:3000:3000"]
    networks: [interna]
    depends_on:
      prometheus: {{condition: service_started}}
    restart: unless-stopped
networks:
  interna: {{internal: true}}
  salida: {{}}
volumes:
  pgdata: {{}}
  promdata: {{}}
secrets:
  db_password: {{file: ./secretos/db_password.txt}}
  exchange_api_key: {{file: ./secretos/exchange_api_key.txt}}
  grafana_admin: {{file: ./secretos/grafana_admin.txt}}
""".format(**IMG)

(RAIZ / "compose.yaml").write_text(COMPOSE_BUENO)
c = yaml.safe_load(COMPOSE_BUENO)
print(list(c["services"]), list(c["secrets"]))

def revisar_compose(texto, raiz):
    c = yaml.safe_load(texto)
    svc, out = c["services"], []
    declarados = set(c.get("secrets", {}))
    for n, s in svc.items():
        img = s.get("image", "")
        if "build" not in s and not DIGEST.search(img):
            out.append((n, "DC01", f"imagen sin digest: {img}"))
        env = s.get("environment", {})
        if isinstance(env, list):
            env = dict(e.split("=", 1) for e in env)
        for k in env:
            if SECRETO.search(k) and not k.endswith("FILE"):
                out.append((n, "DC02", f"secreto en environment: {k}"))
        for p in s.get("ports", []):
            if not str(p).startswith("127.0.0.1:"):
                out.append((n, "DC03", f"puerto en todas las "
                                       f"interfaces: {p}"))
        dep = s.get("depends_on", {})
        if isinstance(dep, list):
            dep = {d: {"condition": "service_started"} for d in dep}
        for d, cond in dep.items():
            sano = "healthcheck" in svc.get(d, {})
            if sano and cond.get("condition") != "service_healthy":
                out.append((n, "DC04", f"no espera a que {d} "
                                       f"esté sano"))
            if not sano and d == "db":
                out.append((d, "DC05", "sin healthcheck y otro "
                                       "servicio depende de él"))
        if "restart" not in s:
            out.append((n, "DC06", "sin política de reinicio"))
        for sec in s.get("secrets", []):
            f = c.get("secrets", {}).get(sec, {}).get("file")
            if sec not in declarados or not (raiz / f).is_file():
                out.append((n, "DC07", f"secreto sin fichero: {sec}"))
    return sorted(set(out))

for nombre, txt in (("ingenuo", COMPOSE_INGENUO),
                    ("bueno", COMPOSE_BUENO)):
    h = revisar_compose(txt, RAIZ)
    print(f"{nombre}: {len(h)} hallazgos")
    for servicio, cod, msg in h:
        print(f"  {servicio:<8} {cod} {msg}")

from graphlib import TopologicalSorter

svc = yaml.safe_load(COMPOSE_BUENO)["services"]
grafo = {n: set(s.get("depends_on", {})) for n, s in svc.items()}
ts = TopologicalSorter(grafo)
ts.prepare()
tandas = []
while ts.is_active():
    listos = sorted(ts.get_ready())
    tandas.append(listos)
    ts.done(*listos)
print("arranque:", " -> ".join("+".join(t) for t in tandas))
print("parada:  ", " -> ".join("+".join(t) for t in tandas[::-1]))

import subprocess

def docker(*args):
    r = subprocess.run(["docker", *args], cwd=RAIZ,
                       capture_output=True, text=True)
    if r.returncode:
        raise RuntimeError(r.stderr.strip())
    return r.stdout.strip()

docker("buildx", "build", "-t", "bot-trading:1.4.0", ".")
docker("compose", "config", "--quiet")
docker("compose", "up", "-d", "--wait")
print(docker("compose", "ps", "--format", "json"))