Snapshots

Um snapshot é uma visão point-in-time de um algo: status, total alvo, quanto já executou, preço médio e o detalhamento por perna. Snapshots são como o SDK expõe tudo que a engine sabe sobre uma estratégia em execução.

Pegar um

O caso mais simples: ler um algo pelo seu ClOrdID.

async with AsyncInvestflexClient.from_env() as client:
    snap = await client.algos.get(handle.clord_id)
    print(snap.status.name, snap.output.executed_qtty, "/", snap.output.target_qtty)

Campos de StrategySnapshot:

Campo	Tipo	Observações
clord_id	str	O ClOrdID do algo.
strategy_code	str | None	Código no fio (ex.: twap, iceberg).
name	str | None	O rótulo que você enviou.
status	AlgoStatus	Enum: PENDING_NEW, RUNNING, SUSPENDED, DONE, CANCELED, REJECTED.
output	StrategyOutput	Estatísticas agregadas de execução.
legs_output	list de LegOutput	Uma entrada por perna.
raw	dict | None	Envelope original, para campos que o SDK ainda não modela.

StrategyOutput:

Campo	Tipo	Observações
avg_price	float | None	Preço médio de execução.
executed_qtty	int | None	Quantidade já executada.
sent_orders	int | None	Ordens enviadas à bolsa.
executed_orders	int | None	Ordens totalmente executadas.
cum_qty_perc	float | None	Progresso em % (0..100).
target_qtty	int | None	Quantidade alvo.

LegOutput: leg_avg_px, leg_cum_qty_perc, leg_exec_qty, leg_last_px, leg_last_qty, leg_ref_id.

Listar vários

list() retorna uma página de snapshots. Você pode filtrar por status, código de estratégia ou conta.

from investflex.models.common import AlgoStatus

async with AsyncInvestflexClient.from_env() as client:
    running = await client.algos.list(status=AlgoStatus.RUNNING)
    for s in running:
        print(s.clord_id, s.strategy_code, s.status.name)

Filtros:

• status — valor AlgoStatus.
• strategy_code — código no fio, ex.: "twap".
• account — conta da corretora.

Diff polling

Se você quer manter uma visão local sincronizada sem assinar um WebSocket, use snapshots_since() para um diff polling barato. Cada chamada retorna um batch e um token de continuação; passe esse token na próxima chamada para receber apenas o que mudou.

token: str | None = None
async with AsyncInvestflexClient.from_env() as client:
    while True:
        batch = await client.algos.snapshots_since(token)
        for snap in batch.snapshots:
            print(snap.clord_id, "→", snap.status.name)
        token = batch.next_token
        await asyncio.sleep(1)

SnapshotBatch:

Campo	Tipo	Observações
snapshots	list de StrategySnapshot	Algos que mudaram desde token.
next_token	str | None	Passe na próxima chamada para o próximo diff.

Live view

live_view() é um helper de alto nível: um dict[str, StrategySnapshot] auto-mantido que se inicializa via list() e permanece atualizado pelo stream de snapshots — re-inicializando automaticamente após uma reconexão.

async with AsyncInvestflexClient.from_env() as client:
    async with client.algos.live_view() as view:
        async for snapshots in view:
            print(len(snapshots), "algos:")
            for cid, s in snapshots.items():
                print(" ", cid, s.status.name, s.output.cum_qty_perc, "%")

A view emite um novo dict toda vez que algo muda. Passe list_kwargs={"status": AlgoStatus.RUNNING, ...} para filtrar a inicialização e a visão em memória.

Para streaming bruto (sem a abstração de live-dict), veja Streaming.