Séries temporais

import pg from "pg";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

// Um INSERT por lote: os três vetores viram linhas com unnest
async function gravarLote(leituras: { sensorId: number; ts: Date; valor: number }[]) {
  await pool.query(
    `INSERT INTO leituras (sensor_id, ts, valor)
     SELECT * FROM unnest($1::int[], $2::timestamptz[], $3::float8[])`,
    [leituras.map((l) => l.sensorId), leituras.map((l) => l.ts), leituras.map((l) => l.valor)],
  );
}

// Média, máximo e quantidade por hora de um sensor, num intervalo
async function porHora(sensorId: number, inicio: Date, fim: Date) {
  const { rows } = await pool.query(
    `SELECT time_bucket('1 hour', ts) AS hora, round(avg(valor)::numeric, 2) AS media,
            max(valor) AS maximo, count(*) AS n
       FROM leituras
      WHERE sensor_id = $1 AND ts >= $2 AND ts < $3
      GROUP BY hora ORDER BY hora`,
    [sensorId, inicio, fim],
  );
  return rows;
}

// Seis horas de leituras de um sensor, uma por minuto
const inicio = new Date("2026-11-15T00:00:00Z");
const lote = Array.from({ length: 360 }, (_, m) => ({
  sensorId: 42, ts: new Date(inicio.getTime() + m * 60_000), valor: 20 + (m % 60) / 10,
}));
await pool.query("DELETE FROM leituras WHERE sensor_id = 42");
await gravarLote(lote);
for (const r of await porHora(42, inicio, new Date("2026-11-15T06:00:00Z"))) {
  console.log(r.hora.toISOString(), r.media, r.maximo, r.n);
}
await pool.end();

import os
from datetime import datetime, timedelta, timezone

import psycopg

conexao = psycopg.connect(os.environ["DATABASE_URL"], autocommit=True)


def gravar_lote(leituras: list[tuple[int, datetime, float]]) -> None:
    # COPY é a forma mais rápida de gravar muitas linhas no PostgreSQL
    with conexao.cursor().copy("COPY leituras (sensor_id, ts, valor) FROM STDIN") as copia:
        for leitura in leituras:
            copia.write_row(leitura)


def por_hora(sensor_id: int, inicio: datetime, fim: datetime) -> list[tuple]:
    """Média, máximo e quantidade por hora de um sensor, num intervalo."""
    return conexao.execute(
        """SELECT time_bucket('1 hour', ts) AS hora, round(avg(valor)::numeric, 2) AS media,
                  max(valor) AS maximo, count(*) AS n
             FROM leituras
            WHERE sensor_id = %s AND ts >= %s AND ts < %s
            GROUP BY hora ORDER BY hora""",
        (sensor_id, inicio, fim),
    ).fetchall()


# Seis horas de leituras de um sensor, uma por minuto
inicio = datetime(2026, 11, 15, tzinfo=timezone.utc)
lote = [(42, inicio + timedelta(minutes=m), 20 + (m % 60) / 10) for m in range(360)]
conexao.execute("DELETE FROM leituras WHERE sensor_id = 42")
gravar_lote(lote)
for hora, media, maximo, n in por_hora(42, inicio, inicio + timedelta(hours=6)):
    print(hora.astimezone(timezone.utc).isoformat(), media, maximo, n)

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

type Leitura struct {
	SensorID int32
	Ts       time.Time
	Valor    float64
}

// gravarLote usa o protocolo COPY, a forma mais rápida de gravar muitas linhas
func gravarLote(ctx context.Context, db *pgxpool.Pool, leituras []Leitura) error {
	_, err := db.CopyFrom(ctx, pgx.Identifier{"leituras"}, []string{"sensor_id", "ts", "valor"},
		pgx.CopyFromSlice(len(leituras), func(i int) ([]any, error) {
			return []any{leituras[i].SensorID, leituras[i].Ts, leituras[i].Valor}, nil
		}))
	return err
}

type Hora struct {
	Hora   time.Time
	Media  float64
	Maximo float64
	N      int64
}

// porHora devolve média, máximo e quantidade por hora de um sensor, num intervalo
func porHora(
	ctx context.Context, db *pgxpool.Pool, sensor int, inicio, fim time.Time,
) ([]Hora, error) {
	linhas, err := db.Query(ctx, `
		SELECT time_bucket('1 hour', ts) AS hora, round(avg(valor)::numeric, 2)::float8 AS media,
		       max(valor) AS maximo, count(*) AS n
		  FROM leituras
		 WHERE sensor_id = $1 AND ts >= $2 AND ts < $3
		 GROUP BY hora ORDER BY hora`, sensor, inicio, fim)
	if err != nil {
		return nil, err
	}
	return pgx.CollectRows(linhas, pgx.RowToStructByPos[Hora])
}

func main() {
	ctx := context.Background()
	db, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		panic(err)
	}
	defer db.Close()

	// Seis horas de leituras de um sensor, uma por minuto
	inicio := time.Date(2026, 11, 15, 0, 0, 0, 0, time.UTC)
	var lote []Leitura
	for m := range 360 {
		ts := inicio.Add(time.Duration(m) * time.Minute)
		lote = append(lote, Leitura{42, ts, 20 + float64(m%60)/10})
	}
	db.Exec(ctx, "DELETE FROM leituras WHERE sensor_id = 42")
	if err := gravarLote(ctx, db, lote); err != nil {
		panic(err)
	}
	horas, err := porHora(ctx, db, 42, inicio, inicio.Add(6*time.Hour))
	if err != nil {
		panic(err)
	}
	for _, h := range horas {
		fmt.Println(h.Hora.UTC().Format(time.RFC3339), h.Media, h.Maximo, h.N)
	}
}

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;

public class Leituras {
    record Leitura(int sensorId, Instant ts, double valor) {}

    // JDBC batch; com reWriteBatchedInserts=true na URL, o driver junta tudo em poucos comandos
    static void gravarLote(Connection db, List<Leitura> leituras) throws SQLException {
        var sql = "INSERT INTO leituras (sensor_id, ts, valor) VALUES (?, ?, ?)";
        try (var comando = db.prepareStatement(sql)) {
            for (var l : leituras) {
                comando.setInt(1, l.sensorId());
                comando.setObject(2, OffsetDateTime.ofInstant(l.ts(), ZoneOffset.UTC));
                comando.setDouble(3, l.valor());
                comando.addBatch();
            }
            comando.executeBatch();
        }
    }

    // Média, máximo e quantidade por hora de um sensor, num intervalo
    static void porHora(Connection db, int sensorId, Instant inicio, Instant fim)
            throws SQLException {
        var sql = """
            SELECT time_bucket('1 hour', ts) AS hora, round(avg(valor)::numeric, 2) AS media,
                   max(valor) AS maximo, count(*) AS n
              FROM leituras
             WHERE sensor_id = ? AND ts >= ? AND ts < ?
             GROUP BY hora ORDER BY hora""";
        try (var comando = db.prepareStatement(sql)) {
            comando.setInt(1, sensorId);
            comando.setObject(2, OffsetDateTime.ofInstant(inicio, ZoneOffset.UTC));
            comando.setObject(3, OffsetDateTime.ofInstant(fim, ZoneOffset.UTC));
            try (var linhas = comando.executeQuery()) {
                while (linhas.next()) {
                    var hora = linhas.getObject("hora", OffsetDateTime.class).toInstant();
                    System.out.println(hora + " " + linhas.getBigDecimal("media") + " "
                        + linhas.getDouble("maximo") + " " + linhas.getLong("n"));
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        try (var db = DriverManager.getConnection(System.getenv("JDBC_URL"))) {
            // Seis horas de leituras de um sensor, uma por minuto
            var inicio = Instant.parse("2026-11-15T00:00:00Z");
            var lote = new ArrayList<Leitura>();
            for (int m = 0; m < 360; m++) {
                lote.add(new Leitura(42, inicio.plusSeconds(60L * m), 20 + (m % 60) / 10.0));
            }
            db.createStatement().execute("DELETE FROM leituras WHERE sensor_id = 42");
            gravarLote(db, lote);
            porHora(db, 42, inicio, inicio.plusSeconds(6 * 3600));
        }
    }
}

using Npgsql;
using NpgsqlTypes;

await using var db = NpgsqlDataSource.Create(Environment.GetEnvironmentVariable("DATABASE_URL")!);

// COPY em modo binário: a forma mais rápida de gravar muitas linhas no PostgreSQL
async Task GravarLote(IEnumerable<(int SensorId, DateTime Ts, double Valor)> leituras)
{
    await using var conexao = await db.OpenConnectionAsync();
    await using var copia = await conexao.BeginBinaryImportAsync(
        "COPY leituras (sensor_id, ts, valor) FROM STDIN (FORMAT BINARY)");
    foreach (var (sensorId, ts, valor) in leituras)
    {
        await copia.StartRowAsync();
        await copia.WriteAsync(sensorId, NpgsqlDbType.Integer);
        await copia.WriteAsync(ts, NpgsqlDbType.TimestampTz);
        await copia.WriteAsync(valor, NpgsqlDbType.Double);
    }
    await copia.CompleteAsync();
}

// Média, máximo e quantidade por hora de um sensor, num intervalo
async Task PorHora(int sensorId, DateTime inicio, DateTime fim)
{
    await using var comando = db.CreateCommand("""
        SELECT time_bucket('1 hour', ts) AS hora, round(avg(valor)::numeric, 2) AS media,
               max(valor) AS maximo, count(*) AS n
          FROM leituras
         WHERE sensor_id = $1 AND ts >= $2 AND ts < $3
         GROUP BY hora ORDER BY hora
        """);
    comando.Parameters.Add(new NpgsqlParameter { Value = sensorId });
    comando.Parameters.Add(new NpgsqlParameter { Value = inicio });
    comando.Parameters.Add(new NpgsqlParameter { Value = fim });
    await using var linhas = await comando.ExecuteReaderAsync();
    while (await linhas.ReadAsync())
        Console.WriteLine(
            $"{linhas.GetDateTime(0):O} {linhas.GetDecimal(1)} {linhas.GetDouble(2)} {linhas.GetInt64(3)}");
}

// Seis horas de leituras de um sensor, uma por minuto
var inicio = new DateTime(2026, 11, 15, 0, 0, 0, DateTimeKind.Utc);
var lote = Enumerable.Range(0, 360).Select(m => (42, inicio.AddMinutes(m), 20 + (m % 60) / 10.0));
await using (var apagar = db.CreateCommand("DELETE FROM leituras WHERE sensor_id = 42"))
    await apagar.ExecuteNonQueryAsync();
await GravarLote(lote);
await PorHora(42, inicio, inicio.AddHours(6));