Dashboard de Vendas em Tempo Real com Python, FastAPI e WebSockets

Cara Core Informática 82 seguidores
25 de agosto de 2025

Este artigo apresenta um sistema completo de dashboard de vendas que atualiza automaticamente em tempo real. Usando FastAPI, WebSockets e Chart.js v4.4.1, construímos uma solução profissional para monitorar vendas e estoque de maneira contínua.

Arquitetura do Sistema

• Monitoramento automático de arquivo CSV (watchdog + polling)
• API REST para snapshot e histórico
• Canal WebSocket para atualizações em tempo real
• Interface responsiva com Chart.js (v4.4.1)
• Fallback inteligente para polling
• Logging estruturado JSON com rotação
• Layout "dashboard-layout" focado em visualização

Objetivos

• Comparar vendas em janelas de tempo
• Identificar produtos com maior e menor giro
• Antecipar reposição de estoque
• Expor métricas de forma simples e clara

Componentes da Interface

• Cards de métricas (vendas totais, linhas, última atualização)
• Gráfico de pizza principal (distribuição por produto)
• Gráficos de linha (histórico de vendas e estoque)
• Sidebar de tabelas (vendas e estoque por produto)
• Status de conexão (WebSocket / Polling)
• Atualização reativa em segundos após alterações

Estrutura de Dados

O arquivo CSV segue a estrutura:

timestamp,produto,vendas,estoque
2025-08-15 14:30:00,Produto A,120,45
2025-08-15 14:33:00,Produto B,85,32
2025-08-15 14:36:00,Produto C,95,28
2025-08-15 14:39:00,Produto D,110,15
2025-08-15 14:42:00,Produto E,75,22

Configuração Rápida

git clone https://github.com/chmulato/cara-core-dashboard.git
cd cara-core-dashboard
pip install -r requirements.txt
python main.py

Backend (Trecho Simplificado)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio

app = FastAPI()

class Manager:
    def __init__(self):
        self.clients: list[WebSocket] = []
        self._lock = asyncio.Lock()

    async def connect(self, ws: WebSocket):
        await ws.accept()
        async with self._lock:
            self.clients.append(ws)

    async def disconnect(self, ws: WebSocket):
        async with self._lock:
            if ws in self.clients:
                self.clients.remove(ws)

manager = Manager()

@app.websocket('/ws')
async def ws_endpoint(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            await ws.receive_text()  # keep-alive
    except WebSocketDisconnect:
        await manager.disconnect(ws)

Frontend (Trecho)

const state = { ws:null, charts:{}, reconnect:0, maxReconnect:10 };
function connect(){
  state.ws = new WebSocket(`ws://${location.host}/ws`);
  state.ws.onopen = () => { state.reconnect = 0; };
  state.ws.onclose = () => {
    if(state.reconnect < state.maxReconnect){
      setTimeout(connect, 1500); state.reconnect++; return;
    }
    // fallback: iniciar polling
  };
}
connect();

Layout (CSS Core)

.dashboard-layout{display:grid;grid-template-columns:1fr 300px;gap:2rem}
.chart-pizza{background:linear-gradient(135deg,#667eea,#764ba2);color:#fff}
.chart-container{background:#fff;border:1px solid #e5e7eb;border-radius:12px;padding:1.25rem}
.chart-row{display:grid;grid-template-columns:1fr 1fr;gap:1.25rem}
@media (max-width:780px){.dashboard-layout{grid-template-columns:1fr}.chart-row{grid-template-columns:1fr}}

Estrutura do Projeto

cara-core-dashboard/
├── main.py
├── app/
│   ├── main.py
│   ├── data_loader.py
│   ├── logging_setup.py
│   ├── sample_data.csv
│   ├── templates/index.html
│   └── static/
├── src/
│   ├── generate_batch_data.py
│   ├── update_simulator.py
│   └── quick_demo_data.py
├── tests/
├── requirements.txt
├── Dockerfile
└── README.md

Execução & Deploy

python main.py                          # Porta 8000 (ou próxima livre)
LOG_LEVEL=DEBUG LOG_FORMAT=plain python main.py  # Logs detalhados
docker build -t dashboard-vendas .
docker run -p 8001:8000 dashboard-vendas

Checklist

• ✅ CSV padronizado
• ✅ Watchdog + Polling
• ✅ WebSocket com reconexão
• ✅ Gráficos Chart.js
• ✅ Layout responsivo
• ✅ Logging estruturado
• ✅ Dockerfile

Repositório

Hashtags

Contato

• E-mail: suporte@caracore.com.br
• Site: www.caracore.com.br
• LinkedIn: Cara Core Informática