Alpaca, Tradier, MetaTrader 5 con Python y OANDA v20.

import os

URLS = {
    "alpaca": {"papel": "https://paper-api.alpaca.markets",
               "real": "https://api.alpaca.markets"},
    "tradier": {"papel": "https://sandbox.tradier.com/v1",
                "real": "https://api.tradier.com/v1"},
    "oanda": {"papel": "https://api-fxpractice.oanda.com",
              "real": "https://api-fxtrade.oanda.com"},
}

class ModoNoPermitido(RuntimeError):
    pass

def url_base(broker, modo):
    if modo == "real" and os.environ.get("PPT_REAL") != "si":
        raise ModoNoPermitido(
            f"{broker}: modo real sin PPT_REAL=si en el entorno")
    return URLS[broker][modo]

for b in URLS:
    print(f"{b:<8} papel -> {url_base(b, 'papel')}")
try:
    url_base("alpaca", "real")
except ModoNoPermitido as e:
    print("bloqueado:", e)

import os
from alpaca.trading.client import TradingClient
from alpaca.trading.enums import OrderSide, TimeInForce
from alpaca.trading.requests import MarketOrderRequest
from alpaca.trading.stream import TradingStream

clave = os.environ["APCA_API_KEY_ID"]
secreto = os.environ["APCA_API_SECRET_KEY"]
cliente = TradingClient(clave, secreto, paper=True)
orden = cliente.submit_order(MarketOrderRequest(
    symbol="SPY", qty=434, side=OrderSide.BUY,
    time_in_force=TimeInForce.DAY, client_order_id="R00005"))
print(orden.status, orden.client_order_id)

async def al_recibir(evento):               # el manejador es async
    print(evento.event, evento.order.filled_qty, evento.price)

stream = TradingStream(clave, secreto, paper=True)
stream.subscribe_trade_updates(al_recibir)
stream.run()                                # bloquea: bucle propio

import json
from pathlib import Path

EJ = json.loads(Path("brokers4_ejemplos.json").read_text())
ORDER_DOC = set("""id client_order_id created_at updated_at
submitted_at filled_at expired_at expires_at canceled_at failed_at
replaced_at replaced_by replaces asset_id symbol asset_class notional
qty filled_qty filled_avg_price order_class order_type type side
position_intent time_in_force limit_price stop_price status
extended_hours legs trail_percent trail_price hwm ratio_qty group_id
group_role""".split())
final = EJ["alpaca"]["orden_final"]
print("en el ejemplo y no en el esquema:", set(final) - ORDER_DOC)
print("en el esquema y no en el ejemplo:",
      sorted(ORDER_DOC - set(final)))
tipos = {type(final[k]).__name__ for k in
         ("qty", "filled_qty", "filled_avg_price")}
print("tipo de qty, filled_qty y filled_avg_price:", tipos)
for m in EJ["alpaca"]["trade_updates"]:
    d = m["data"]
    print(f"{d['event']:<13}", {k: d[k] for k in
          ("qty", "price", "position_qty") if k in d})

import asyncio
from dataclasses import dataclass, field, replace

@dataclass
class Orden:
    id: str
    cid: str
    simbolo: str
    lado: str
    cantidad: int
    tipo: str
    tif: str
    creada: str
    enviada: str
    cuenta: str
    estado: str = "new"
    ejecutada: int = 0
    efectivo: float = 0.0
    ultima: str = ""
    ejecuciones: list = field(default_factory=list)

    @property
    def media(self):
        if not self.ejecutada:
            return None
        return self.efectivo / self.ejecutada

class Nucleo:
    """Órdenes y ejecuciones; publica cada evento a los suscriptores.
    Sin reloj: las horas salen del guion de cada orden."""

    def __init__(self, poder_compra, guiones):
        self.poder_compra = poder_compra
        self.guiones = guiones    # cid -> (alta, envío, ejecuciones)
        self.ordenes, self.por_cid = {}, {}
        self.subs = {}                    # cuenta -> colas asyncio
        self.posicion = {}                # (cuenta, símbolo) -> títulos

    def suscribir(self, cuenta):
        cola = asyncio.Queue()
        self.subs.setdefault(cuenta, []).append(cola)
        return cola

    async def cerrar(self):
        """Avisa a los suscriptores de que el servidor se apaga."""
        for colas in self.subs.values():
            for cola in colas:
                await cola.put(None)

    async def publicar(self, tipo, o, **extra):
        foto = replace(o, ejecuciones=list(o.ejecuciones))  # copia
        for cola in self.subs.get(o.cuenta, []):
            await cola.put((tipo, foto, extra))

    def alta(self, cuenta, oid, cid, simbolo, lado, cantidad, tipo,
             tif):
        creada, enviada, _ = self.guiones[cid]
        o = Orden(oid, cid, simbolo, lado, cantidad, tipo, tif, creada,
                  enviada, cuenta)
        self.ordenes[oid], self.por_cid[cid] = o, oid
        return o

    async def ejecutar(self, o):
        _, _, ejec = self.guiones[o.cid]
        await self.publicar("new", o, ts=o.enviada)
        for ts, q, px, eid in ejec:
            o.ejecutada += q
            o.efectivo += q * px
            o.ultima = ts
            signo = 1 if o.lado == "buy" else -1
            k = (o.cuenta, o.simbolo)
            self.posicion[k] = self.posicion.get(k, 0) + signo * q
            o.ejecuciones.append((ts, q, px, eid))
            o.estado = ("filled" if o.ejecutada == o.cantidad
                        else "partially_filled")
            tipo = "fill" if o.estado == "filled" else "partial_fill"
            await self.publicar(tipo, o, ts=ts, qty=q, price=px,
                                eid=eid)

R5 = EJ["alpaca"]["trade_updates"]
guion_r5 = (R5[0]["data"]["order"]["created_at"],
            R5[0]["data"]["order"]["submitted_at"],
            [(m["data"]["timestamp"], int(m["data"]["qty"]),
              float(m["data"]["price"]), m["data"]["execution_id"])
             for m in R5[1:]])
nucleo = Nucleo(poder_compra=1_000_000.0,
                guiones={"R00005": guion_r5})
print("alta y envío:", guion_r5[0][11:], guion_r5[1][11:])
print("ejecuciones del guion:", [(q, px) for _, q, px, _ in
                                 guion_r5[2]])

from aiohttp import web, WSMsgType

CLAVES_ALPACA = {"PK-FALSA-PAPEL": "FALSA-SECRETO-PAPEL"}
TIPOS = {"market", "limit", "stop", "stop_limit", "trailing_stop"}
TIFS = {"day", "gtc", "opg", "cls", "ioc", "fok"}
PRECIO_REF = {"ORSC": 234.80}             # para el poder de compra

def orden_alpaca(o):
    texto = lambda x: None if x is None else f"{x:.4f}"
    return {"id": o.id, "client_order_id": o.cid,
            "created_at": o.creada,
            "updated_at": o.ultima or o.enviada,
            "submitted_at": o.enviada,
            "filled_at": o.ultima if o.estado == "filled" else None,
            "symbol": o.simbolo, "qty": str(o.cantidad),
            "filled_qty": str(o.ejecutada),
            "filled_avg_price": texto(o.media), "type": o.tipo,
            "side": o.lado, "time_in_force": o.tif,
            "status": o.estado}

def error(http, codigo, mensaje):
    return web.json_response({"code": codigo, "message": mensaje},
                             status=http)

async def alpaca_post(req):
    if CLAVES_ALPACA.get(req.headers.get("APCA-API-KEY-ID")) != \
            req.headers.get("APCA-API-SECRET-KEY"):
        return error(401, 40110000, "request is not authorized")
    d = await req.json()
    if d.get("type") not in TIPOS or d.get("time_in_force") not in TIFS:
        return error(422, 42210000, "invalid type or time_in_force")
    if d.get("client_order_id") in nucleo.por_cid:
        return error(422, 42210000, "client_order_id must be unique")
    qty = int(d["qty"])
    if qty * PRECIO_REF[d["symbol"]] > nucleo.poder_compra:
        return error(403, 40310000, "insufficient buying power")
    o = nucleo.alta("alpaca", "6f1c2a9e-4b7d-4c1a-9e55-000000039595",
                    d["client_order_id"], d["symbol"], d["side"], qty,
                    d["type"], d["time_in_force"])
    asyncio.get_running_loop().create_task(nucleo.ejecutar(o))
    return web.json_response(orden_alpaca(o))

async def alpaca_por_cid(req):
    oid = nucleo.por_cid.get(req.query["client_order_id"])
    if oid is None:
        return error(404, 40410000, "order not found")
    return web.json_response(orden_alpaca(nucleo.ordenes[oid]))

async def alpaca_stream(req):
    ws = web.WebSocketResponse()
    await ws.prepare(req)
    auth = json.loads((await ws.receive()).data)
    ok = CLAVES_ALPACA.get(auth.get("key")) == auth.get("secret")
    await ws.send_json({"stream": "authorization", "data": {
        "status": "authorized" if ok else "unauthorized",
        "action": "authenticate"}})
    if not ok:
        await ws.close()
        return ws
    pedido = json.loads((await ws.receive()).data)
    cola = nucleo.suscribir("alpaca")     # antes de confirmar
    await ws.send_json({"stream": "listening",
                        "data": pedido["data"]})
    try:
        await emitir_alpaca(ws, cola)
    except ConnectionResetError:          # el cliente se fue
        pass
    finally:
        nucleo.subs["alpaca"].remove(cola)
    return ws

async def emitir_alpaca(ws, cola):
    while (evento := await cola.get()) is not None:
        tipo, o, x = evento
        data = {"event": tipo, "timestamp": x["ts"],
                "order": orden_alpaca(o)}
        if "qty" in x:
            data |= {"execution_id": x["eid"],
                     "price": f"{x['price']:.2f}", "qty": str(x["qty"]),
                     "position_qty": str(
                         nucleo.posicion["alpaca", o.simbolo])}
        await ws.send_json({"stream": "trade_updates", "data": data})

from aiohttp import ClientSession

app = web.Application()
app.router.add_post("/v2/orders", alpaca_post)
app.router.add_get("/v2/orders:by_client_order_id", alpaca_por_cid)
app.router.add_get("/stream", alpaca_stream)
corredor = web.AppRunner(app)
await corredor.setup()
sitio = web.TCPSite(corredor, "127.0.0.1", 0)
await sitio.start()
BASE = f"http://127.0.0.1:{sitio._server.sockets[0].getsockname()[1]}"
CAB = {"APCA-API-KEY-ID": "PK-FALSA-PAPEL",
       "APCA-API-SECRET-KEY": "FALSA-SECRETO-PAPEL"}
FINALES = {"filled", "canceled", "expired", "rejected"}

sesion = ClientSession()
ws = await sesion.ws_connect(BASE + "/stream")
await ws.send_json({"action": "auth", "key": CAB["APCA-API-KEY-ID"],
                    "secret": CAB["APCA-API-SECRET-KEY"]})
print((await ws.receive_json())["data"]["status"])
await ws.send_json({"action": "listen",
                    "data": {"streams": ["trade_updates"]}})
print((await ws.receive_json())["stream"])

cuerpo = {"symbol": "ORSC", "qty": "434", "side": "buy",
          "type": "market", "time_in_force": "day",
          "client_order_id": "R00005"}
async with sesion.post(BASE + "/v2/orders", json=cuerpo,
                       headers=CAB) as r:
    print("POST:", r.status, (await r.json())["status"])

acum, efectivo, eventos = 0, 0.0, []
while True:
    d = (await ws.receive_json())["data"]
    if "qty" in d:
        acum += int(d["qty"])
        efectivo += int(d["qty"]) * float(d["price"])
    eventos.append(d)
    print(f"{d['timestamp'][11:23]} {d['event']:<13}"
          f" ejecutadas {acum:>3}  media "
          f"{efectivo / acum if acum else 0:.4f}"
          f"  (bróker: {d['order']['filled_avg_price']})")
    if d["order"]["status"] in FINALES:
        break

async with sesion.get(BASE + "/v2/orders:by_client_order_id",
                      params={"client_order_id": "R00005"},
                      headers=CAB) as r:
    consultada = await r.json()
comunes = sorted(set(consultada) & set(final))
distintos = [k for k in comunes if consultada[k] != final[k]]
print(len(comunes), "campos comunes; distintos:", distintos)
print("reconstruida:", acum, f"{efectivo / acum:.4f}")
print("consultada:  ", consultada["filled_qty"],
      consultada["filled_avg_price"], consultada["status"])

malas = {
    "tif gtd": cuerpo | {"client_order_id": "R9001",
                         "time_in_force": "gtd"},
    "10 000 títulos": cuerpo | {"client_order_id": "R9002",
                               "qty": "10000"},
    "R00005 otra vez": cuerpo,
}
for nombre, c in malas.items():
    async with sesion.post(BASE + "/v2/orders", json=c,
                           headers=CAB) as r:
        print(f"{nombre:<16} {r.status} {(await r.json())['message']}")

import json
import os
from urllib.parse import urlencode
from urllib.request import Request, urlopen

BASE = "https://sandbox.tradier.com/v1"        # papel
CAB = {"Authorization": f"Bearer {os.environ['TRADIER_TOKEN']}",
       "Accept": "application/json"}
cuenta = os.environ["TRADIER_CUENTA"]
datos = urlencode({"class": "equity", "symbol": "SPY", "side": "buy",
                   "quantity": 434, "type": "market",
                   "duration": "day", "tag": "R00005"}).encode()
req = Request(f"{BASE}/accounts/{cuenta}/orders", data=datos,
              headers=CAB, method="POST")
with urlopen(req, timeout=10) as r:
    oid = json.load(r)["order"]["id"]
with urlopen(Request(f"{BASE}/accounts/{cuenta}/orders/{oid}",
                     headers=CAB), timeout=10) as r:
    print(json.load(r)["order"]["status"])

TOKEN_TRADIER = "FALSA-TOKEN-SANDBOX"
EJ_T = EJ["tradier"]["order"]
nucleo.guiones["R00006"] = guion_r5      # la misma orden, otra cuenta
IDS_T = {}                               # id del núcleo -> id Tradier

def orden_tradier(o):
    ult = o.ejecuciones[-1] if o.ejecuciones else (None, 0, 0.0, None)
    return {"id": IDS_T[o.id], "type": o.tipo, "symbol": o.simbolo,
            "side": o.lado, "quantity": float(o.cantidad),
            "status": o.estado, "duration": o.tif,
            "avg_fill_price": round(o.media or 0.0, 4),
            "exec_quantity": float(o.ejecutada),
            "last_fill_price": ult[2],
            "last_fill_quantity": float(ult[1]),
            "remaining_quantity": float(o.cantidad - o.ejecutada),
            "create_date": o.creada[:23] + "Z",
            "transaction_date": (o.ultima or o.creada)[:23] + "Z",
            "class": "equity"}

def autorizado(req):
    return req.headers.get("Authorization") == f"Bearer {TOKEN_TRADIER}"

async def tradier_post(req):
    if not autorizado(req):
        return web.Response(status=401, text="Invalid Access Token")
    f = await req.post()                  # formulario, no JSON
    o = nucleo.alta("tradier", "t-" + f["tag"], f["tag"], f["symbol"],
                    f["side"], int(f["quantity"]), f["type"],
                    f["duration"])
    IDS_T[o.id] = EJ_T["id"]
    asyncio.get_running_loop().create_task(nucleo.ejecutar(o))
    return web.json_response({"order": {"id": IDS_T[o.id],
                                        "status": "ok"}})

async def tradier_get(req):
    inv = {v: k for k, v in IDS_T.items()}
    o = nucleo.ordenes[inv[int(req.match_info["oid"])]]
    return web.json_response({"order": orden_tradier(o)})

async def tradier_sesion(req):
    return web.json_response({"stream": {
        "url": "wss://ws.tradier.com/v1/accounts/events",
        "sessionid": "sesion-falsa-1"}})

CAMPOS_EVENTO = ("status", "type", "avg_fill_price", "exec_quantity",
                 "last_fill_quantity", "remaining_quantity",
                 "transaction_date", "create_date")

async def tradier_eventos(req):
    ws = web.WebSocketResponse()
    await ws.prepare(req)
    pedido = json.loads((await ws.receive()).data)
    if pedido.get("sessionid") != "sesion-falsa-1":
        await ws.close()
        return ws
    cola = nucleo.suscribir("tradier")
    await ws.send_json({"event": "heartbeat", "status": "active",
                        "timestamp": 1760448900})
    try:
        while (evento := await cola.get()) is not None:
            tipo, o, x = evento
            t = orden_tradier(o)
            if tipo == "new":
                t["status"] = "open"
            await ws.send_json({"id": t["id"], "event": "order",
                                **{k: t[k] for k in CAMPOS_EVENTO},
                                "account": "FALSA-1"})
    except ConnectionResetError:
        pass
    finally:
        nucleo.subs["tradier"].remove(cola)
    return ws

app_t = web.Application()
app_t.router.add_post("/v1/accounts/{cuenta}/orders", tradier_post)
app_t.router.add_get("/v1/accounts/{cuenta}/orders/{oid}",
                     tradier_get)
app_t.router.add_post("/v1/accounts/events/session", tradier_sesion)
app_t.router.add_get("/v1/accounts/events", tradier_eventos)
corredor_t = web.AppRunner(app_t)
await corredor_t.setup()
sitio_t = web.TCPSite(corredor_t, "127.0.0.1", 0)
await sitio_t.start()
BASE_T = ("http://127.0.0.1:"
          f"{sitio_t._server.sockets[0].getsockname()[1]}/v1")
GET_DOC = set("""id type symbol side quantity status duration
avg_fill_price exec_quantity last_fill_price last_fill_quantity
remaining_quantity create_date transaction_date class""".split())
print("ejemplo frente a la documentación:", set(EJ_T) ^ GET_DOC)
print("tipos:", {type(EJ_T[k]).__name__ for k in
                 ("quantity", "avg_fill_price", "exec_quantity")})

CAB_T = {"Authorization": f"Bearer {TOKEN_TRADIER}",
         "Accept": "application/json"}
async with sesion.post(BASE_T + "/accounts/events/session",
                       headers=CAB_T) as r:
    sid = (await r.json())["stream"]["sessionid"]
ws_t = await sesion.ws_connect(BASE_T + "/accounts/events")
await ws_t.send_json({"events": ["order"], "sessionid": sid,
                      "excludeAccounts": []})
print("primer mensaje:", (await ws_t.receive_json())["event"])
formulario = {"class": "equity", "symbol": "ORSC", "side": "buy",
              "quantity": "434", "type": "market", "duration": "day",
              "tag": "R00006"}
async with sesion.post(BASE_T + "/accounts/FALSA-1/orders",
                       data=formulario, headers=CAB_T) as r:
    alta = (await r.json())["order"]
print("POST:", alta)

previo = 0.0
while True:
    e = await ws_t.receive_json()
    caja = e["avg_fill_price"] * e["exec_quantity"]
    ultimo = ((caja - previo) / e["last_fill_quantity"]
              if e["last_fill_quantity"] else None)
    previo = caja
    print(f"{e['status']:<17} exec {e['exec_quantity']:>5}"
          f"  media {e['avg_fill_price']:<9}"
          f"  último precio deducido {ultimo and round(ultimo, 4)}")
    if e["status"] in FINALES:
        break
async with sesion.get(f"{BASE_T}/accounts/FALSA-1/orders/"
                      f"{alta['id']}", headers=CAB_T) as r:
    t_final = (await r.json())["order"]
print("iguales al ejemplo:", all(t_final[k] == EJ_T[k] for k in (
    "id", "quantity", "status", "avg_fill_price", "exec_quantity",
    "last_fill_price", "last_fill_quantity", "remaining_quantity",
    "create_date", "transaction_date")))

from decimal import Decimal

ESTADOS = {"alpaca": {"new": "aceptada", "accepted": "aceptada",
                      "partially_filled": "parcial",
                      "filled": "ejecutada", "canceled": "cancelada",
                      "expired": "cancelada", "rejected": "rechazada"},
           "tradier": {"open": "aceptada", "pending": "aceptada",
                       "partially_filled": "parcial",
                       "filled": "ejecutada", "canceled": "cancelada",
                       "expired": "cancelada", "rejected": "rechazada"}}

@dataclass(frozen=True)
class Ejecucion:
    broker: str
    orden: str
    cantidad: int
    precio: Decimal
    acumulada: int
    estado: str

def desde_alpaca(d):
    return Ejecucion("alpaca", d["order"]["client_order_id"],
                     int(d["qty"]), Decimal(d["price"]),
                     int(d["order"]["filled_qty"]),
                     ESTADOS["alpaca"][d["order"]["status"]])

def desde_tradier(e, precio):
    return Ejecucion("tradier", str(e["id"]),
                     int(e["last_fill_quantity"]),
                     Decimal(str(precio)).quantize(Decimal("0.01")),
                     int(e["exec_quantity"]),
                     ESTADOS["tradier"][e["status"]])

print(desde_alpaca(eventos[-1]))
print(desde_tradier({"id": 228005, "status": "filled",
                     "last_fill_quantity": 18.0,
                     "exec_quantity": 434.0}, 234.8099))

TIF_OANDA = {"MARKET": {"FOK", "IOC"},
             "LIMIT": {"GTC", "GTD", "GFD", "FOK", "IOC"},
             "STOP": {"GTC", "GTD", "GFD", "FOK", "IOC"},
             "MARKET_IF_TOUCHED": {"GTC", "GTD", "GFD", "FOK", "IOC"}}

def validar_oanda(o):
    errores = []
    if o["timeInForce"] not in TIF_OANDA[o["type"]]:
        errores.append(f"{o['type']} no admite {o['timeInForce']}")
    if o["timeInForce"] == "GTD" and "gtdTime" not in o:
        errores.append("GTD sin gtdTime")
    if o["type"] != "MARKET" and "price" not in o:
        errores.append("falta price")
    u, p = float(o["units"]), float(o.get("price", "nan"))
    sl = float(o.get("stopLossOnFill", {}).get("price", "nan"))
    if u > 0 and sl >= p or u < 0 and sl <= p:
        errores.append("stop del lado equivocado")
    return errores

limite = {"type": "LIMIT", "instrument": "EUR_USD", "units": "10000",
          "price": "1.16300", "timeInForce": "GTD",
          "gtdTime": "2025-10-15T20:00:00.000000000Z",
          "positionFill": "DEFAULT",
          "stopLossOnFill": {"price": "1.15900"},
          "takeProfitOnFill": {"price": "1.17100"},
          "clientExtensions": {"id": "R4-fx-0001", "tag": "rebalanceo"}}
print(validar_oanda(limite))
print(validar_oanda(limite | {"type": "MARKET"}))
print(validar_oanda(limite | {"stopLossOnFill": {"price": "1.16500"}}))
print(json.dumps({"order": limite})[:72] + "…")

FILL_DOC = set("""id time userID accountID batchID requestID type
orderID clientOrderID instrument units gainQuoteHomeConversionFactor
lossQuoteHomeConversionFactor homeConversionFactors price fullVWAP
fullPrice reason pl quotePL financing baseFinancing quoteFinancing
commission guaranteedExecutionFee quoteGuaranteedExecutionFee
accountBalance tradeOpened tradesClosed tradeReduced
halfSpreadCost""".split())
OBSOLETOS = {"price", "gainQuoteHomeConversionFactor",
             "lossQuoteHomeConversionFactor"}
tx = EJ["oanda_v20"]["transaction"]
print("no documentados:", set(tx) - FILL_DOC)
print("obsoletos usados:", sorted(OBSOLETOS & set(tx)))
print("precio oficial (tradeOpened):", tx["tradeOpened"]["price"],
      "| fullVWAP:", tx["fullVWAP"], "| tipo:",
      type(tx["units"]).__name__)

fill = {k: v for k, v in tx.items() if k != "requestedUnits"}
T0 = 1_760_448_900                     # 2025-10-14 13:35:00 UTC

def hora(s):
    import datetime as dt
    t = dt.datetime.fromtimestamp(T0 + s, dt.UTC)
    return t.strftime("%Y-%m-%dT%H:%M:%S.000000000Z")

TRANS = [
    (0.9, {"id": "6409", "type": "MARKET_ORDER",
           "instrument": "EUR_USD", "units": "10000"}),
    (1.0, fill),
    (7.2, {"id": "6411", "type": "STOP_LOSS_ORDER", "tradeID": "6410",
           "price": "1.15900"}),
    (11.6, {"id": "6412", "type": "ORDER_CANCEL", "orderID": "6411",
            "reason": "CLIENT_REQUEST"}),
]
PERDIDA = "6411"

def linea_tiempo(fin=20):
    sucesos = [(s, t) for s, t in TRANS]
    sucesos += [(s, None) for s in range(5, fin + 1, 5)]
    ultima = "6408"
    for s, t in sorted(sucesos, key=lambda x: x[0]):
        if t is None:
            yield s, {"type": "HEARTBEAT", "lastTransactionID": ultima,
                      "time": hora(s)}
        else:
            ultima = t["id"]
            yield s, t | {"time": hora(s)}

async def oanda_stream(req):
    r = web.StreamResponse()
    await r.prepare(req)
    for s, msg in linea_tiempo():
        if msg.get("id") != PERDIDA:
            await r.write((json.dumps(msg) + "\n").encode())
    return r

async def oanda_sinceid(req):
    desde = int(req.query["id"])
    return web.json_response({"transactions": [
        t | {"time": hora(s)} for s, t in TRANS
        if int(t["id"]) > desde],
        "lastTransactionID": TRANS[-1][1]["id"]})

app_o = web.Application()
app_o.router.add_get("/v3/accounts/{a}/transactions/stream",
                     oanda_stream)
app_o.router.add_get("/v3/accounts/{a}/transactions/sinceid",
                     oanda_sinceid)
corredor_o = web.AppRunner(app_o)
await corredor_o.setup()
sitio_o = web.TCPSite(corredor_o, "127.0.0.1", 0)
await sitio_o.start()
BASE_O = ("http://127.0.0.1:"
          f"{sitio_o._server.sockets[0].getsockname()[1]}"
          "/v3/accounts/101-004-0418272-001")
print([(s, m.get("id", m["type"])) for s, m in linea_tiempo()])

vistas, ultima = [], 6408
async with sesion.get(BASE_O + "/transactions/stream") as r:
    async for crudo in r.content:
        m = json.loads(crudo)
        if m["type"] != "HEARTBEAT":
            if m["id"] in vistas:
                print("  repetida", m["id"],
                      "por el stream: descartada")
                continue
            ultima = int(m["id"])
            vistas.append(m["id"])
            continue
        if int(m["lastTransactionID"]) > ultima:
            print(f"{m['time'][11:19]} latido {m['lastTransactionID']}"
                  f" > última vista {ultima}: pido sinceid")
            async with sesion.get(BASE_O + "/transactions/sinceid",
                                  params={"id": str(ultima)}) as s:
                for t in (await s.json())["transactions"]:
                    if t["id"] not in vistas:
                        vistas.append(t["id"])
                        print("  recuperada", t["id"], t["type"])
                    ultima = max(ultima, int(t["id"]))
print("transacciones procesadas:", vistas)

import MetaTrader5 as mt5                      # solo Windows

if not mt5.initialize(login=int(os.environ["MT5_LOGIN"]),
                      password=os.environ["MT5_PASSWORD"],
                      server=os.environ["MT5_SERVIDOR"]):
    raise RuntimeError(mt5.last_error())
cuenta = mt5.account_info()
assert cuenta.trade_mode == mt5.ACCOUNT_TRADE_MODE_DEMO   # papel
peticion = {
    "action": mt5.TRADE_ACTION_PENDING, "symbol": "EURUSD",
    "volume": 0.10, "type": mt5.ORDER_TYPE_BUY_LIMIT,
    "price": 1.16300, "sl": 1.15900, "tp": 1.17100,
    "type_time": mt5.ORDER_TIME_SPECIFIED,
    "expiration": 1760558400,                  # segundos Unix
    "type_filling": mt5.ORDER_FILLING_RETURN,
    "magic": 418272, "comment": "R4-fx-0001",
}
chequeo = mt5.order_check(peticion)            # margen, sin enviar
print(chequeo.retcode, chequeo.margin, chequeo.margin_free)
resultado = mt5.order_send(peticion)
propias = [o for o in mt5.orders_get(symbol="EURUSD")
           if o.magic == 418272]               # solo las del sistema
posiciones = mt5.positions_get(symbol="EURUSD")
mt5.shutdown()

RESULT_DOC = {"retcode", "deal", "order", "volume", "price", "bid",
              "ask", "comment", "request_id", "retcode_external",
              "request"}
REQUEST_DOC = {"action", "magic", "order", "symbol", "volume", "price",
               "stoplimit", "sl", "tp", "deviation", "type",
               "type_filling", "type_time", "expiration", "comment",
               "position", "position_by"}
CONST = {"TRADE_ACTION_DEAL": 1, "TRADE_ACTION_PENDING": 5,
         "TRADE_ACTION_SLTP": 6, "ORDER_TYPE_BUY": 0,
         "ORDER_TYPE_BUY_LIMIT": 2, "ORDER_FILLING_IOC": 1,
         "ORDER_FILLING_RETURN": 2, "TRADE_RETCODE_DONE": 10009}
res = EJ["metatrader5"]["resultado"]
print(set(res) == RESULT_DOC, set(res["request"]) == REQUEST_DOC)
nombre = {v: k for k, v in CONST.items() if k.startswith("TRADE_A")}
print("acción:", nombre[res["request"]["action"]],
      "| tipo compra:",
      res["request"]["type"] == CONST["ORDER_TYPE_BUY"],
      "| relleno IOC:",
      res["request"]["type_filling"] == CONST["ORDER_FILLING_IOC"],
      "| hecho:", res["retcode"] == CONST["TRADE_RETCODE_DONE"])
print(f"precio {res['price']} = ask {res['ask']}; horquilla "
      f"{(res['ask'] - res['bid']) * 1e5:.1f} puntos")

import numpy as np

rng = np.random.default_rng(4403)
llegadas = np.cumsum(rng.exponential(250, 400))      # ms
ticks = [(int(t), round(1.16450 + 1e-5 * k, 5))
         for k, t in enumerate(llegadas[llegadas < 60_000])]

def symbol_info_tick(ahora):
    previos = [t for t in ticks if t[0] <= ahora]
    return previos[-1] if previos else None

def copy_ticks_from(desde, cuantos=1000, ahora=None):
    return [t for t in ticks if desde <= t[0] <= ahora][:cuantos]

vistos_sondeo = {symbol_info_tick(ms) for ms in range(500, 60_001, 500)}
vistos_sondeo.discard(None)
vistos_copia, ultimo = [], -1
for ms in range(500, 60_001, 500):
    nuevos = [t for t in copy_ticks_from(ultimo, ahora=ms)
              if t[0] > ultimo]
    vistos_copia += nuevos
    if nuevos:
        ultimo = nuevos[-1][0]
print(f"ticks del minuto: {len(ticks)}")
print(f"sondeo de symbol_info_tick cada 500 ms: {len(vistos_sondeo)}")
print(f"copy_ticks_from desde el último time_msc: {len(vistos_copia)}")

await nucleo.cerrar()          # el servidor suelta los websockets
await ws.close()
await ws_t.close()
await sesion.close()
for c in (corredor, corredor_t, corredor_o):
    await c.cleanup()
print("simuladores cerrados")