Project structure, configuration and quality: from notebook to repository.

from pathlib import Path

RAIZ = Path("ppt-bot").resolve()

def escribir(ruta, texto):
    p = RAIZ / ruta
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(texto.lstrip("\n"))

escribir("pyproject.toml", '''
[project]
name = "ppt-bot"
version = "0.1.0"
requires-python = ">=3.14"
dependencies = ["pydantic-settings>=2.16", "pyyaml>=6.0"]

[dependency-groups]
dev = ["pytest>=9.1", "ruff>=0.17", "pre-commit>=4.7"]

[tool.ruff]
line-length = 72
target-version = "py314"

[tool.ruff.lint]
select = ["E", "F", "W", "I", "UP", "B", "DTZ", "S", "SIM",
          "PD", "NPY"]

[tool.ruff.lint.per-file-ignores]
"tests/*" = ["S101"]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
addopts = "-p no:cacheprovider"
''')
escribir("src/ppt_bot/__init__.py", "")
escribir("config/base.yaml", '''
modo: paper
riesgo:
  nocional_max_orden: 50000
  perdida_diaria_max: 2000
broker:
  nombre: simulado
''')
escribir("config/paper.yaml", '''
# paper: same as base.yaml
''')
escribir("config/real.yaml", '''
modo: real
riesgo:
  nocional_max_orden: 20000
broker:
  nombre: alpaca
  cuenta: U-0418272
''')
print(sorted(str(p.relative_to(RAIZ)) for p in RAIZ.rglob("*")
             if p.is_file()))

escribir("src/ppt_bot/dominio.py", '''
"""Domain: orders, fills and portfolio. No I/O, no clock."""

import math
from dataclasses import dataclass, field


@dataclass(frozen=True)
class Orden:
    clave: str  # idempotency: derived from the intent (4.5.5)
    instrumento: str
    cantidad: int  # > 0 buy, < 0 sell


@dataclass(frozen=True)
class Ejecucion:
    id: str
    orden: str
    instrumento: str
    cantidad: int
    precio: float
    comision: float = 0.0


def ordenes_para_objetivo(objetivo, posiciones, precios, fecha, lote=1):
    """Orders that take each position to the target notional,
    in whole lots, rounding toward zero."""
    ordenes = []
    for ins in sorted(objetivo):
        x = objetivo[ins] / precios[ins]
        deseada = math.copysign(math.floor(abs(x) / lote) * lote, x)
        delta = int(deseada) - posiciones.get(ins, 0)
        if delta:
            ordenes.append(Orden(f"{fecha}-{ins}", ins, delta))
    return ordenes


@dataclass
class Cartera:
    efectivo: float
    posiciones: dict = field(default_factory=dict)
    vistas: set = field(default_factory=set)

    def aplicar(self, e):
        """Applies a fill only once (by its id)."""
        if e.id in self.vistas:
            return False
        self.vistas.add(e.id)
        q = self.posiciones.get(e.instrumento, 0) + e.cantidad
        self.posiciones[e.instrumento] = q
        self.efectivo -= e.cantidad * e.precio + e.comision
        return True

    def valor(self, precios):
        return self.efectivo + sum(
            q * precios[i] for i, q in self.posiciones.items()
        )
''')
escribir("src/ppt_bot/puertos.py", '''
"""What the service needs from the outside (7.1.3)."""

from typing import Protocol

from ppt_bot.dominio import Ejecucion, Orden


class OrdenRechazada(Exception):
    pass


class Broker(Protocol):
    def enviar(self, orden: Orden) -> list[Ejecucion]: ...
''')
escribir("src/ppt_bot/broker_sim.py", '''
"""Simulated broker for tests: deterministic and configurable."""

from ppt_bot.dominio import Ejecucion
from ppt_bot.puertos import OrdenRechazada


class BrokerSimulado:
    def __init__(
        self,
        precios,
        partes=1,
        llena=1.0,
        comision=0.0,
        rechaza=(),
        duplica=False,
    ):
        self.precios, self.partes, self.llena = precios, partes, llena
        self.comision, self.rechaza = comision, set(rechaza)
        self.duplica, self.enviadas = duplica, []

    def enviar(self, orden):
        if orden.instrumento in self.rechaza:
            raise OrdenRechazada(orden.clave)
        self.enviadas.append(orden)
        n = int(abs(orden.cantidad) * self.llena)
        trozos = [n // self.partes] * self.partes
        trozos[-1] += n - sum(trozos)
        signo = 1 if orden.cantidad > 0 else -1
        ejecs = [
            Ejecucion(
                f"{orden.clave}.{i:02d}",
                orden.clave,
                orden.instrumento,
                signo * q,
                self.precios[orden.instrumento],
                self.comision,
            )
            for i, q in enumerate(trozos, 1)
            if q
        ]
        if self.duplica and ejecs:
            ejecs.append(ejecs[-1])  # the broker resends the last one
        return ejecs
''')

escribir("src/ppt_bot/servicio.py", '''
"""Use case: rebalance the portfolio through a broker."""

from datetime import UTC, datetime

from ppt_bot.dominio import ordenes_para_objetivo
from ppt_bot.puertos import OrdenRechazada


class LimiteExcedido(Exception):
    pass


def ahora():
    return datetime.now(UTC)


def rebalancear(
    broker,
    cartera,
    objetivo,
    precios,
    riesgo,
    fecha,
    valor_inicio,
    lote=1,
    reloj=ahora,
    historial=None,
):
    perdida = valor_inicio - cartera.valor(precios)
    if perdida > riesgo.perdida_diaria_max:
        raise LimiteExcedido(f"daily loss {perdida:.0f}")
    informe = {
        "hora": reloj().isoformat(),
        "enviadas": 0,
        "rechazadas": 0,
    }
    ordenes = ordenes_para_objetivo(
        objetivo, cartera.posiciones, precios, fecha, lote
    )
    for o in ordenes:
        nocional = abs(o.cantidad) * precios[o.instrumento]
        if nocional > riesgo.nocional_max_orden:
            raise LimiteExcedido(f"{o.clave}: {nocional:.0f}")
        try:
            ejecuciones = broker.enviar(o)
        except OrdenRechazada:
            informe["rechazadas"] += 1
            continue
        for e in ejecuciones:
            cartera.aplicar(e)
        informe["enviadas"] += 1
    if historial is None:
        historial = []
    historial.append(informe)
    return informe
''')
escribir("src/ppt_bot/adaptadores.py", '''
"""Each broker's messages -> domain fills."""

from ppt_bot.dominio import Ejecucion


def de_alpaca(msg, orden):
    out = []
    for m in msg["trade_updates"]:
        d = m["data"]
        if d["event"] not in ("partial_fill", "fill"):
            continue
        o = d["order"]
        signo = 1 if o["side"] == "buy" else -1
        out.append(
            Ejecucion(
                d["execution_id"],
                o["client_order_id"],
                o["symbol"],
                signo * int(d["qty"]),
                float(d["price"]),
            )
        )
    return out


def de_ib(msg, orden):
    out = []
    for f in msg["fills"]:
        x, c = f["execution"], f["commissionReport"]
        signo = 1 if x["side"] == "BOT" else -1
        out.append(
            Ejecucion(
                x["execId"],
                x["orderRef"],
                orden.instrumento,
                signo * int(x["shares"]),
                x["price"],
                c["commission"],
            )
        )
    return out


def de_tradier(msg, orden):
    o = msg["order"]
    signo = 1 if o["side"] == "buy" else -1
    return [
        Ejecucion(
            str(o["id"]),
            orden.clave,
            o["symbol"],
            signo * int(o["exec_quantity"]),
            o["avg_fill_price"],
        )
    ]


ADAPTADORES = {
    "alpaca": de_alpaca,
    "interactive_brokers": de_ib,
    "tradier": de_tradier,
}
''')

escribir("src/ppt_bot/config.py", '''
"""Configuration: YAML per profile, environment and validation."""

import os
from pathlib import Path
from typing import Literal

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    SecretStr,
    model_validator,
)
from pydantic_settings import (
    BaseSettings,
    SettingsConfigDict,
    YamlConfigSettingsSource,
)

CONFIG = Path("config")


class Riesgo(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)
    nocional_max_orden: float = Field(gt=0)
    perdida_diaria_max: float = Field(gt=0)


class Broker(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)
    nombre: Literal["simulado", "alpaca", "ib"]
    cuenta: str = "PAPER"
    api_key: SecretStr | None = None


class Ajustes(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="PPT_",
        env_nested_delimiter="__",
        extra="forbid",
        frozen=True,
    )
    modo: Literal["paper", "real"] = "paper"
    riesgo: Riesgo
    broker: Broker

    @model_validator(mode="after")
    def real_exige_broker_real(self):
        if self.modo == "real" and self.broker.nombre == "simulado":
            raise ValueError("real mode with the simulated broker")
        if self.modo == "real" and self.broker.api_key is None:
            raise ValueError("real mode without api_key")
        return self

    @classmethod
    def settings_customise_sources(
        cls, settings_cls, init_settings, env_settings, **_
    ):
        """Sources from highest to lowest priority."""
        return (init_settings, env_settings, fuente_yaml(settings_cls))


def ficheros_yaml():
    perfil = os.environ.get("PPT_PERFIL", "paper")
    return [CONFIG / "base.yaml", CONFIG / f"{perfil}.yaml"]


def fuente_yaml(settings_cls):
    return YamlConfigSettingsSource(
        settings_cls, yaml_file=ficheros_yaml(), deep_merge=True
    )
''')

import hashlib
import json
import os
import sys

sys.path.insert(0, str(RAIZ / "src"))
os.chdir(RAIZ)
from pydantic_settings import EnvSettingsSource

from ppt_bot import config as cfg

def hojas(d, prefijo=()):
    for k, v in d.items():
        if isinstance(v, dict):
            yield from hojas(v, prefijo + (k,))
        else:
            yield prefijo + (k,), v

def procedencia():
    fuentes = [("env", EnvSettingsSource(cfg.Ajustes)()),
               ("yaml", cfg.fuente_yaml(cfg.Ajustes)())]
    final = cfg.Ajustes().model_dump(mode="json")
    for ruta, valor in hojas(final):
        origen = "default"
        for nombre, datos in fuentes:
            d = datos
            for k in ruta:
                d = d.get(k, {}) if isinstance(d, dict) else {}
            if d != {}:
                origen = nombre
                break
        print(f"  {'.'.join(ruta):28} {str(valor):12} {origen}")

def huella(ajustes):
    texto = json.dumps(ajustes.model_dump(mode="json"),
                       sort_keys=True)
    return hashlib.sha256(texto.encode()).hexdigest()[:12]

print("paper profile")
procedencia()
os.environ |= {"PPT_PERFIL": "real",
               "PPT_BROKER__API_KEY": "toy-key",
               "PPT_RIESGO__NOCIONAL_MAX_ORDEN": "15000"}
print("real profile, with two environment variables")
procedencia()
real = cfg.Ajustes()
print(repr(real.broker.api_key), "| fingerprint:", huella(real))

from pydantic import ValidationError
from pydantic_settings import YamlConfigSettingsSource

def intentar(crear):
    try:
        crear()
        print("  loads")
    except ValidationError as e:
        for err in e.errors():
            campo = ".".join(map(str, err["loc"])) or "-"
            print(f"  {campo}: {err['msg']}")

for k in ("PPT_RIESGO__NOCIONAL_MAX_ORDEN",):
    del os.environ[k]

class SinFusion(cfg.Ajustes):
    @classmethod
    def settings_customise_sources(cls, settings_cls, **_):
        return (YamlConfigSettingsSource(
            settings_cls, yaml_file=cfg.ficheros_yaml()),)

print("real.yaml without deep merge:")
intentar(SinFusion)

original = Path("config/real.yaml").read_text()
Path("config/real.yaml").write_text(original.replace(
    "riesgo:\n", "riesgo:\n  perdida_diaria_maxima: 1500\n"))
print("typo in a key:")
intentar(cfg.Ajustes)
Path("config/real.yaml").write_text(original)
print("real mode with the simulated broker:")
intentar(lambda: cfg.Ajustes(broker={"nombre": "simulado"}))

import shutil
import subprocess

escribir("notebook/senal.py", '''
import os, json
from datetime import datetime
import numpy as np
import pandas as pd


def registrar(orden, historial=[]):
    historial.append((datetime.now(), orden))
    return historial


def tamano(capital, precio, riesgo=0.01):
    assert precio > 0
    return int(capital * riesgo / precio)


def cargar(ruta):
    try:
        df = pd.read_csv(ruta)
    except:
        return None
    df.dropna(inplace=True)
    ruido = np.random.normal(size=len(df))
    if df.empty == True:
        print(f"empty")
    return df, ruido


def umbrales():
    fs = []
    for k in range(3):
        fs.append(lambda x: x * k)
    return fs
''')
r = subprocess.run(["ruff", "check", "--no-cache", "--output-format",
                    "concise", "notebook"], capture_output=True,
                   text=True, cwd=RAIZ)
print(r.stdout)

shutil.rmtree(RAIZ / "notebook")  # not part of the repository
for cmd in (["ruff", "check", "--no-cache", "src"],
            ["ruff", "format", "--no-cache", "--check", "src"]):
    r = subprocess.run(cmd, capture_output=True, text=True, cwd=RAIZ)
    print(" ".join(cmd[:2]), "->", r.returncode, r.stdout.strip())

escribir("tests/conftest.py", '''
import json
from datetime import UTC, datetime
from pathlib import Path

import pytest

from ppt_bot.config import Riesgo
from ppt_bot.dominio import Cartera


@pytest.fixture
def riesgo():
    return Riesgo(nocional_max_orden=50_000, perdida_diaria_max=2_000)


@pytest.fixture
def cartera():
    return Cartera(efectivo=100_000.0)


@pytest.fixture
def reloj():
    return lambda: datetime(2026, 1, 2, 15, 30, tzinfo=UTC)


@pytest.fixture(scope="session")
def grabados():
    ruta = Path(__file__).parent / "datos" / "brokers4_ejemplos.json"
    return json.loads(ruta.read_text())
''')
escribir("tests/test_dominio.py", '''
import pytest

from ppt_bot.broker_sim import BrokerSimulado
from ppt_bot.dominio import ordenes_para_objetivo
from ppt_bot.servicio import LimiteExcedido, rebalancear

PRECIOS = {"ORSC": 234.84, "ORUS": 716.82}


def correr(broker, cartera, objetivo, riesgo, reloj):
    return rebalancear(
        broker,
        cartera,
        objetivo,
        PRECIOS,
        riesgo,
        "2026-01-02",
        valor_inicio=100_000.0,
        reloj=reloj,
    )


@pytest.mark.parametrize(
    ("nocional", "lote", "esperada"),
    [
        (10_000, 1, 42),
        (10_000, 10, 40),
        (-10_000, 10, -40),
        (100, 1, 0),
    ],
)
def test_lotes_hacia_cero(nocional, lote, esperada):
    o = ordenes_para_objetivo(
        {"ORSC": nocional}, {}, PRECIOS, "d", lote
    )
    assert sum(x.cantidad for x in o) == esperada


def test_parciales_y_comision(riesgo, cartera, reloj):
    b = BrokerSimulado(PRECIOS, partes=3, comision=0.35)
    correr(b, cartera, {"ORSC": 10_000}, riesgo, reloj)
    assert cartera.posiciones == {"ORSC": 42}
    assert cartera.efectivo == pytest.approx(
        100_000 - 42 * 234.84 - 3 * 0.35
    )


def test_ejecucion_repetida_cuenta_una_vez(riesgo, cartera, reloj):
    b = BrokerSimulado(PRECIOS, partes=2, duplica=True)
    correr(b, cartera, {"ORSC": 10_000}, riesgo, reloj)
    assert cartera.posiciones == {"ORSC": 42}


def test_limite_de_nocional(riesgo, cartera, reloj):
    b = BrokerSimulado(PRECIOS)
    with pytest.raises(LimiteExcedido):
        correr(b, cartera, {"ORUS": 60_000}, riesgo, reloj)
    assert b.enviadas == []


def test_rechazo_no_mueve_la_cartera(riesgo, cartera, reloj):
    b = BrokerSimulado(PRECIOS, rechaza={"ORUS"})
    inf = correr(
        b, cartera, {"ORSC": 10_000, "ORUS": 10_000}, riesgo, reloj
    )
    assert inf["rechazadas"] == 1
    assert cartera.posiciones == {"ORSC": 42}


def test_hora_del_reloj_inyectado(riesgo, cartera, reloj):
    inf = correr(BrokerSimulado(PRECIOS), cartera, {}, riesgo, reloj)
    assert inf["hora"] == "2026-01-02T15:30:00+00:00"
''')

escribir("tests/test_contrato.py", '''
import pytest

from ppt_bot.adaptadores import ADAPTADORES
from ppt_bot.dominio import Orden

ORDEN = Orden("R00005", "ORSC", 434)


@pytest.mark.parametrize("broker", sorted(ADAPTADORES))
def test_la_misma_orden_en_cada_broker(broker, grabados):
    ejecs = ADAPTADORES[broker](grabados[broker], ORDEN)
    total = sum(e.cantidad for e in ejecs)
    vwap = sum(e.cantidad * e.precio for e in ejecs) / total
    assert {e.instrumento for e in ejecs} == {"ORSC"}
    assert all(isinstance(e.cantidad, int) for e in ejecs)
    assert total == ORDEN.cantidad
    assert vwap == pytest.approx(234.8353, abs=5e-5)
    assert len({e.id for e in ejecs}) == len(ejecs)
''')
(Path("tests") / "datos").mkdir(exist_ok=True)
shutil.copy(Path("..") / "brokers4_ejemplos.json",
            Path("tests") / "datos")

import re

ENTORNO = os.environ | {"PYTHONPATH": "src",
                        "PYTHONDONTWRITEBYTECODE": "1"}

def ejecutar(*cmd):
    r = subprocess.run(cmd, capture_output=True, text=True,
                       env=ENTORNO)
    salida = re.sub(r" in \d+\.\d+s", "", r.stdout + r.stderr)
    return r.returncode, salida.strip()

codigo, salida = ejecutar("python", "-m", "pytest", "-v")
print(codigo)
print("\n".join(l.split(" ", 1)[0].split("::", 1)[1]
                for l in salida.splitlines() if "PASSED" in l))
print(salida.splitlines()[-1].strip("= "))

escribir("scripts/validar_config.py", '''
"""Loads each config/ profile with Ajustes (pre-commit and CI)."""

import os
import sys

from pydantic import ValidationError

from ppt_bot.config import Ajustes

os.environ.setdefault("PPT_BROKER__API_KEY", "validation-only")
fallos = 0
for perfil in ("paper", "real"):
    os.environ["PPT_PERFIL"] = perfil
    try:
        Ajustes()
    except ValidationError as e:
        fallos += 1
        for err in e.errors():
            campo = ".".join(map(str, err["loc"])) or "-"
            print(f"{perfil}: {campo}: {err['msg']}")
sys.exit(1 if fallos else 0)
''')
escribir(".pre-commit-config.yaml", '''
repos:
  - repo: local
    hooks:
      - id: ruff
        name: ruff check
        entry: ruff check --no-cache --output-format concise
        language: system
        types: [python]
        require_serial: true
      - id: ruff-format
        name: ruff format
        entry: ruff format --no-cache --check
        language: system
        types: [python]
        require_serial: true
      - id: config
        name: validate config
        entry: env PYTHONPATH=src python scripts/validar_config.py
        language: system
        pass_filenames: false
        always_run: true
      - id: pytest
        name: pytest
        entry: python -m pytest -q
        language: system
        pass_filenames: false
        always_run: true
''')
escribir(".gitignore", "__pycache__/\n")
for k in ("PPT_PERFIL", "PPT_BROKER__API_KEY"):
    os.environ.pop(k, None)
ENTORNO = os.environ | {"PYTHONPATH": "src",
                        "PYTHONDONTWRITEBYTECODE": "1"}

ejecutar("git", "init", "-q")
print(ejecutar("pre-commit", "install")[1])
servicio = Path("src/ppt_bot/servicio.py")
limpio = servicio.read_text()
servicio.write_text(limpio.replace(
    "datetime.now(UTC)", "datetime.now()"))
ejecutar("git", "add", "-A")
codigo, salida = ejecutar("git", "commit", "-qm", "Rebalancing bot")
print("commit:", codigo)
print(salida)

servicio.write_text(limpio)
ejecutar("git", "add", "-A")
codigo, salida = ejecutar("git", "commit", "-qm", "Rebalancing bot")
print("commit:", codigo)
print(salida)
print(ejecutar("git", "log", "--oneline")[1])

from contextlib import contextmanager

S, DOM = "src/ppt_bot/servicio.py", "src/ppt_bot/dominio.py"
AD, REAL = "src/ppt_bot/adaptadores.py", "config/real.yaml"
LIMITE = '''        if nocional > riesgo.nocional_max_orden:
            raise LimiteExcedido(f"{o.clave}: {nocional:.0f}")
'''
PERDIDA = '''    perdida = valor_inicio - cartera.valor(precios)
    if perdida > riesgo.perdida_diaria_max:
        raise LimiteExcedido(f"daily loss {perdida:.0f}")
'''
DEFECTOS = {
    "D01 naive datetime": [(S, "now(UTC)", "now()")],
    "D02 limit with assert": [(S, LIMITE, "        assert nocional"
                               " <= riesgo.nocional_max_orden\n")],
    "D03 bare except": [(S, "except OrdenRechazada:", "except:")],
    "D04 historial=[]": [(S, "historial=None,", "historial=[],")],
    "D05 lots with round": [(DOM, "math.floor(abs(x) / lote)",
                             "round(abs(x) / lote)")],
    "D06 commission added": [(DOM, "e.precio + e.comision",
                              "e.precio - e.comision")],
    "D07 no deduplication": [(DOM, "        if e.id in self.vistas:\n"
                              "            return False\n", "")],
    "D08 cumulative quantity": [(AD, 'int(d["qty"])',
                                 'int(o["filled_qty"])')],
    "D09 side BOT/BUY": [(AD, '"BOT"', '"BUY"')],
    "D10 no daily loss limit": [(S, PERDIDA, "")],
    "C01 modo: live": [(REAL, "modo: real", "modo: live")],
    "C02 20.000 in YAML": [(REAL, "20000", "20.000")],
    "C03 typo in key": [(REAL, "nocional_max_orden",
                         "nocional_maximo_orden")],
    "C04 real with simulated": [(REAL, "nombre: alpaca",
                                 "nombre: simulado")],
}
CAPAS = {
    "ruff": ["ruff", "check", "--no-cache", "-q", "."],
    "unit": ["python", "-m", "pytest", "tests/test_dominio.py"],
    "contract": ["python", "-m", "pytest", "tests/test_contrato.py"],
    "config": ["python", "scripts/validar_config.py"],
}

@contextmanager
def sembrado(cambios):
    copias = {}
    for ruta, antes, despues in cambios:
        p = Path(ruta)
        copias.setdefault(p, p.read_text())
        assert p.read_text().count(antes) == 1, (ruta, antes)
        p.write_text(p.read_text().replace(antes, despues))
    try:
        yield
    finally:
        for p, texto in copias.items():
            p.write_text(texto)

assert all(ejecutar(*cmd)[0] == 0 for cmd in CAPAS.values())
matriz = {}
for nombre, cambios in DEFECTOS.items():
    with sembrado(cambios):
        matriz[nombre] = [ejecutar(*c)[0] != 0
                          for c in CAPAS.values()]
print(f"{'':24}" + "".join(f"{c:>11}" for c in CAPAS))
for nombre, fila in matriz.items():
    print(f"{nombre:24}" + "".join(
        f"{'X' if x else '·':>11}" for x in fila))
totales = [sum(f[i] for f in matriz.values())
           for i in range(len(CAPAS))]
nadie = [n for n, f in matriz.items() if not any(f)]
print("per layer:", dict(zip(CAPAS, totales)))
print("none:", nadie)