InvestflexInvestflexSDK
Introdução

Introdução

investflex-py é um SDK Python tipado para trading algorítmico na plataforma Investflex. Ele permite enviar, editar, monitorar e fazer streaming de ordens algorítmicas na B3 a partir de um único cliente assíncrono.

O que você ganha

  • Modelos de estratégia tipados. Cada algo (TWAP, Iceberg, Spread, Skew, Long & Short, Gradiente, Flex Order, …) é um modelo Pydantic com campos autocompletáveis, validação local e serialização para o protocolo do servidor.
  • Um único cliente async. AsyncInvestflexClient expõe três serviços: algos (ciclo de vida), commands (cancel/suspend/resume) e marketdata (REST + streaming).
  • Streams em tempo real. Atualizações de snapshot e market data (book, trades, definições) chegam por WebSocket com reconexão automática e replay de assinaturas.
  • Idempotência por padrão. Cada criação recebe um ClOrdID e uma Idempotency-Key gerados automaticamente, então retries em verbos idempotentes são seguros.

Execução

Quando você envia um algo pelo SDK, nos garantimos sua execução no colocation da B3. A seguir os ambientes envolvidos na execução dos seus algoritmos:

  1. Seu ambiente. O script Python (seu notebook, um servidor, um Jupyter) monta um objeto de estratégia tipado e envia pela internet usando HTTPS — a mesma conexão segura que o navegador usa em internet banking.
  2. Investflex Cloud. Um serviço gerenciado que autentica você, guarda suas ordens e encaminha tudo para o engine de execução. Ele também publica um feed em tempo real (um WebSocket) que o SDK usa para receber atualizações de snapshot e dados de mercado sem você precisar ficar perguntando "o que mudou?".
  3. Colocação na B3 (São Paulo). O engine algorítmico da Investflex e a infraestrutura de market data ficam no mesmo data center do matching engine da B3. Estar fisicamente perto da bolsa significa que seu algo reage em microssegundos, em vez das dezenas de milissegundos que você perderia na internet pública.

Na prática: você escreve Python, a Investflex leva o pedido com segurança até a bolsa, e o trabalho pesado em tempo real acontece coladinho no matching engine — você só vê o resultado via SDK.

Topologia de execução
Seu ambienteOnde o seu código Python rodaInvestflex CloudGateway público, autenticação e persistênciaColocation B3 — SPMesmo data center do match engineSeu código (SDK)AsyncInvestflexClient + modelos tipadosinvestflex-pyasynciohttpxwebsocketsRuntimePython 3.10+ asyncLinuxmacOSWindowsAPI GatewayREST + WebSocket sobre TLSRESTWebSocketAuth (JWT)Persistência & streamSnapshots, histórico, fan-outMotor de algos17 estratégias em C++TWAPIcebergSniper+14Market dataFeed UMDF da B3, multicastUMDFFASTMBOB3 Match EngineHTTPS · WSS~20–60 msLink privado intra-colo≤1 mssnapshots replicadosSBE Marketdata / Entry order≤1 ms intra-colo · microssegundos até a B3

O motor de algos e o feed de market data ficam dentro do colocation da B3, no mesmo data center do PUMA. Seus comandos chegam por HTTPS/WSS até o gateway público e seguem por link privado até o motor; toda a execução crítica acontece com latência sub-milissegundo.

Tour de 30 segundos

import asyncio
from investflex import AsyncInvestflexClient
from investflex.models.algos import TWAP
from investflex.models.algos.twap import TWAPLeg
from investflex.models.common import OrdType, Side

async def main() -> None:
    async with AsyncInvestflexClient.from_env() as client:
        # 1. Enviar
        twap = TWAP(
            name="petr4-buy-1k",
            init_time="10:00:00",
            end_time="11:00:00",
            legs=[TWAPLeg(symbol="PETR4", side=Side.BUY, quantity=1000,
                          ord_type=OrdType.LIMIT, price_limit=36.00,
                          alloc_account="YOUR_ACCOUNT")],
        )
        handle = await client.algos.create(twap)

        # 2. Inspecionar
        snap = await client.algos.get(handle.clord_id)
        print(snap.status, snap.exec_quantity, "/", snap.total_quantity)

        # 3. Reagir a market data
        last = await client.marketdata.last_trade("PETR4")
        print("last", last.price, "@", last.timestamp)

        # 4. Cancelar
        await client.commands.cancel(handle.clord_id)

asyncio.run(main())

Como a documentação está organizada

Seção O que tem dentro
Getting Started Instalação, opções de login, configuração.
Estratégias Uma página por algo: parâmetros, pernas, edições, ciclo de vida.
Market Data Helpers REST para definições, último trade, lista de papéis.
Snapshots Leitura do estado dos algos — point-in-time, list, diff polling, live view.
Streaming Market data e snapshots em tempo real.

Async-first

O SDK é async-nativo (asyncio + httpx). Todos os exemplos da documentação usam async def main() e await. Se você precisa de um wrapper síncrono para scripts, rode a corrotina com asyncio.run(...).

Convenções

  • Campos de estratégia são snake_case em Python; o SDK serializa para o CamelCase que a plataforma espera.
  • Side, ord-type, time-in-force e similares são enums Python em investflex.models.common.
  • Flags Y / N são passados como as strings literais "Y" e "N".
  • Toda resposta de list/get retorna um StrategySnapshot (ou lista) — veja Snapshots para o formato.

Próximos passos