ajoute snapshot
Signed-off-by: GnomeZworc <nicolas.boufidjeline@g3e.fr>
This commit is contained in:
parent
c73021a14f
commit
0f2d7126eb
GPG key ID:
4406BBBF8845D632
11 changed files with 862 additions and 19 deletions
|
|
@ -504,6 +504,62 @@ func TestAPI(t *testing.T) {
|
|||
r4.Body.Close()
|
||||
})
|
||||
|
||||
// ── Snapshots ─────────────────────────────────────────────────────────────
|
||||
|
||||
t.Run("POST transaction invalide le snapshot du compte", func(t *testing.T) {
|
||||
// La transaction salaire (tx1) a déjà été créée pour accID.
|
||||
// On vérifie qu'une invalidation existe pour ce compte.
|
||||
resp := s.do("POST", fmt.Sprintf("/accounts/%d/snapshots/recompute", accID), nil)
|
||||
// tx1 a une date passée → invalidation présente → 200 avec from/to
|
||||
s.mustStatus(resp, http.StatusOK)
|
||||
var result map[string]any
|
||||
s.decode(resp, &result)
|
||||
if result["status"] != "ok" {
|
||||
t.Fatalf("expected status ok, got %v", result)
|
||||
}
|
||||
if result["account_id"].(float64) != float64(accID) {
|
||||
t.Fatalf("unexpected account_id: %v", result["account_id"])
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("POST /accounts/{id}/snapshots/recompute — aucune invalidation → 204", func(t *testing.T) {
|
||||
// Le recompute précédent a effacé l'invalidation.
|
||||
resp := s.do("POST", fmt.Sprintf("/accounts/%d/snapshots/recompute", accID), nil)
|
||||
s.mustStatus(resp, http.StatusNoContent)
|
||||
resp.Body.Close()
|
||||
})
|
||||
|
||||
t.Run("Modifier une transaction re-invalide le compte", func(t *testing.T) {
|
||||
// Créer une nouvelle transaction pour déclencher une invalidation.
|
||||
r1 := s.do("POST", "/transactions", map[string]any{
|
||||
"date": "2026-05-01", "label": "Test invalidation",
|
||||
"account_dest_id": accID, "instrument_dest_id": eurID,
|
||||
"quantite_dest": 50, "validated": true,
|
||||
})
|
||||
s.mustStatus(r1, http.StatusCreated)
|
||||
r1.Body.Close()
|
||||
|
||||
// L'invalidation est présente → recompute retourne 200.
|
||||
r2 := s.do("POST", fmt.Sprintf("/accounts/%d/snapshots/recompute", accID), nil)
|
||||
s.mustStatus(r2, http.StatusOK)
|
||||
var result map[string]any
|
||||
s.decode(r2, &result)
|
||||
if result["status"] != "ok" {
|
||||
t.Fatalf("expected status ok, got %v", result)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("POST /accounts/{id}/snapshots/recompute — compte inexistant → 404", func(t *testing.T) {
|
||||
resp := s.do("POST", "/accounts/9999/snapshots/recompute", nil)
|
||||
// Compte inexistant : GetInvalidation retourne pgx.ErrNoRows → 404
|
||||
// (en pratique le compte n'existe pas, l'invalidation non plus → 204)
|
||||
// On accepte 204 ou 404
|
||||
if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotFound {
|
||||
t.Fatalf("expected 204 or 404 for unknown account, got %d", resp.StatusCode)
|
||||
}
|
||||
resp.Body.Close()
|
||||
})
|
||||
|
||||
// Utiliser les variables pour éviter "declared but not used"
|
||||
_ = eurID
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import (
|
|||
"git.g3e.fr/H6N/account/internal/pipeline"
|
||||
"git.g3e.fr/H6N/account/internal/scheduler"
|
||||
"git.g3e.fr/H6N/account/internal/server"
|
||||
"git.g3e.fr/H6N/account/internal/snapshot"
|
||||
"git.g3e.fr/H6N/account/internal/store"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
|
@ -40,6 +41,7 @@ func main() {
|
|||
logger.Info("database connected")
|
||||
|
||||
st := store.New(pool)
|
||||
snap := snapshot.New(st, logger, cfg.SnapshotHorizonDays)
|
||||
pl := pipeline.New(st, pipeline.Config{
|
||||
OpenFIGIKey: cfg.OpenFIGIKey,
|
||||
CoinGeckoKey: cfg.CoinGeckoKey,
|
||||
|
|
@ -56,6 +58,11 @@ func main() {
|
|||
Interval: cfg.PriceCleanInterval,
|
||||
Fn: pl.CleanHistory,
|
||||
})
|
||||
sched.Add(scheduler.Job{
|
||||
Name: "daily-snapshot",
|
||||
Interval: 24 * time.Hour,
|
||||
Fn: snap.DailySnapshot,
|
||||
})
|
||||
sched.Start(ctx)
|
||||
|
||||
srv := server.New(cfg, pool, logger)
|
||||
|
|
|
|||
90
docs/api.md
90
docs/api.md
|
|
@ -381,3 +381,93 @@ Valide ou dé-valide une transaction.
|
|||
### `DELETE /transactions/{id}`
|
||||
|
||||
**Réponse `204`**
|
||||
|
||||
---
|
||||
|
||||
## Snapshots
|
||||
|
||||
Les snapshots capturent la valorisation de chaque compte sur une fenêtre glissante :
|
||||
- **Passé** : transactions validées uniquement
|
||||
- **Futur** : toutes les transactions (validées ou non) — projection basée sur les transactions planifiées
|
||||
|
||||
Deux tables sont alimentées :
|
||||
- `position_snapshot` — quantité + valeur par instrument et par compte
|
||||
- `account_snapshot` — valeur totale agrégée par compte
|
||||
|
||||
**Fenêtre** : de la première transaction jusqu'à `today + SNAPSHOT_HORIZON_DAYS` (défaut : 30 jours). Configurable via la variable d'environnement `SNAPSHOT_HORIZON_DAYS`.
|
||||
|
||||
**Toutes les transactions sont comptées** quelle que soit leur validation.
|
||||
Le snapshot représente toujours l'état envisagé complet du compte :
|
||||
transactions confirmées (`validated = true`) **et** planifiées (`validated = false`).
|
||||
La liste `?pending=true` (transactions non validées à date dépassée) reste distincte — c'est un outil de suivi, pas un filtre de calcul.
|
||||
|
||||
**Job `daily-snapshot`** (toutes les 24h) :
|
||||
1. Traite les invalidations en attente
|
||||
2. Recalcule hier (consolide les transactions de la veille)
|
||||
3. Calcule le nouveau jour entrant dans la fenêtre (`today + horizon`)
|
||||
|
||||
---
|
||||
|
||||
### `GET /accounts/{id}/snapshots`
|
||||
|
||||
Liste la valorisation totale du compte jour par jour.
|
||||
|
||||
**Paramètres de filtre**
|
||||
| Paramètre | Type | Description |
|
||||
|---|---|---|
|
||||
| `from` | string | Date de début `YYYY-MM-DD` (optionnel, défaut : première date disponible) |
|
||||
| `to` | string | Date de fin `YYYY-MM-DD` (optionnel, défaut : dernier jour calculé dans la fenêtre) |
|
||||
|
||||
**Réponse `200`** — triée par date ASC
|
||||
```json
|
||||
[
|
||||
{ "date": "2026-06-01", "valeur": 2750.00 },
|
||||
{ "date": "2026-06-02", "valeur": 2800.00 },
|
||||
{ "date": "2026-07-13", "valeur": 2950.00 }
|
||||
]
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### `GET /accounts/{id}/snapshots/positions`
|
||||
|
||||
Liste les positions détaillées (par instrument) du compte jour par jour.
|
||||
|
||||
**Paramètres de filtre** — mêmes que `/snapshots` (`from`, `to`)
|
||||
|
||||
**Réponse `200`** — triée par date ASC puis instrument_id
|
||||
```json
|
||||
[
|
||||
{
|
||||
"date": "2026-06-01",
|
||||
"instrument_id": 1,
|
||||
"quantite": 2750.00,
|
||||
"prix_cloture": 1.0,
|
||||
"valeur": 2750.00
|
||||
},
|
||||
{
|
||||
"date": "2026-06-01",
|
||||
"instrument_id": 2,
|
||||
"quantite": 5.0,
|
||||
"pru": 350.00,
|
||||
"prix_cloture": 360.00,
|
||||
"valeur": 1800.00
|
||||
}
|
||||
]
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### `POST /accounts/{id}/snapshots/recompute`
|
||||
|
||||
Recalcule immédiatement les snapshots d'un compte depuis sa **date d'invalidation** jusqu'à `today + horizon`.
|
||||
|
||||
Chaque mutation de transaction (create, update, delete, validate/dévalider) marque automatiquement les comptes concernés comme dirty avec `recompute_from = MIN(date_existante, date_transaction)`. Cet endpoint consomme ce flag sans attendre le job nocturne.
|
||||
|
||||
**Réponse `200`** — recalcul effectué
|
||||
```json
|
||||
{ "account_id": 1, "from": "2026-06-01", "to": "2026-07-13", "status": "ok" }
|
||||
```
|
||||
|
||||
**Réponse `204`** — aucune invalidation en attente, rien à faire
|
||||
**Réponse `404`** — compte introuvable
|
||||
|
|
|
|||
|
|
@ -2,17 +2,19 @@ package config
|
|||
|
||||
import (
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
DatabaseURL string
|
||||
Port string
|
||||
Env string
|
||||
OpenFIGIKey string
|
||||
CoinGeckoKey string
|
||||
PriceFetchInterval time.Duration
|
||||
PriceCleanInterval time.Duration
|
||||
DatabaseURL string
|
||||
Port string
|
||||
Env string
|
||||
OpenFIGIKey string
|
||||
CoinGeckoKey string
|
||||
PriceFetchInterval time.Duration
|
||||
PriceCleanInterval time.Duration
|
||||
SnapshotHorizonDays int // nombre de jours dans le futur couverts par les snapshots
|
||||
}
|
||||
|
||||
func Load() *Config {
|
||||
|
|
@ -21,14 +23,20 @@ func Load() *Config {
|
|||
fetchInterval = time.Hour
|
||||
}
|
||||
|
||||
horizonDays, err := strconv.Atoi(getenv("SNAPSHOT_HORIZON_DAYS", "30"))
|
||||
if err != nil || horizonDays < 0 {
|
||||
horizonDays = 30
|
||||
}
|
||||
|
||||
return &Config{
|
||||
DatabaseURL: getenv("DATABASE_URL", "postgres://account:account@localhost:5432/account?sslmode=disable"),
|
||||
Port: getenv("PORT", "8080"),
|
||||
Env: getenv("ENV", "development"),
|
||||
OpenFIGIKey: os.Getenv("OPENFIGI_API_KEY"),
|
||||
CoinGeckoKey: os.Getenv("COINGECKO_API_KEY"),
|
||||
PriceFetchInterval: fetchInterval,
|
||||
PriceCleanInterval: 24 * time.Hour,
|
||||
DatabaseURL: getenv("DATABASE_URL", "postgres://account:account@localhost:5432/account?sslmode=disable"),
|
||||
Port: getenv("PORT", "8080"),
|
||||
Env: getenv("ENV", "development"),
|
||||
OpenFIGIKey: os.Getenv("OPENFIGI_API_KEY"),
|
||||
CoinGeckoKey: os.Getenv("COINGECKO_API_KEY"),
|
||||
PriceFetchInterval: fetchInterval,
|
||||
PriceCleanInterval: 24 * time.Hour,
|
||||
SnapshotHorizonDays: horizonDays,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
128
internal/handler/snapshot.go
Normal file
128
internal/handler/snapshot.go
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"git.g3e.fr/H6N/account/internal/snapshot"
|
||||
"git.g3e.fr/H6N/account/internal/store"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
type SnapshotHandler struct {
|
||||
engine *snapshot.Engine
|
||||
}
|
||||
|
||||
func NewSnapshotHandler(e *snapshot.Engine) *SnapshotHandler {
|
||||
return &SnapshotHandler{engine: e}
|
||||
}
|
||||
|
||||
func (h *SnapshotHandler) RegisterRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("GET /accounts/{id}/snapshots", h.listAccountSnapshots)
|
||||
mux.HandleFunc("GET /accounts/{id}/snapshots/positions", h.listPositionSnapshots)
|
||||
mux.HandleFunc("POST /accounts/{id}/snapshots/recompute", h.recomputeAccount)
|
||||
}
|
||||
|
||||
// parseDateParam parse un paramètre de query YYYY-MM-DD. Retourne time.Time{} si absent.
|
||||
func parseDateParam(r *http.Request, key string) (time.Time, error) {
|
||||
v := r.URL.Query().Get(key)
|
||||
if v == "" {
|
||||
return time.Time{}, nil
|
||||
}
|
||||
return time.Parse("2006-01-02", v)
|
||||
}
|
||||
|
||||
// GET /accounts/{id}/snapshots?from=YYYY-MM-DD&to=YYYY-MM-DD
|
||||
// Liste les snapshots agrégés (valeur totale) d'un compte.
|
||||
// from/to optionnels : sans borne → pas de filtre de ce côté.
|
||||
func (h *SnapshotHandler) listAccountSnapshots(w http.ResponseWriter, r *http.Request) {
|
||||
accountID, err := parseID(r)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid account id")
|
||||
return
|
||||
}
|
||||
from, err := parseDateParam(r, "from")
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid from date")
|
||||
return
|
||||
}
|
||||
to, err := parseDateParam(r, "to")
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid to date")
|
||||
return
|
||||
}
|
||||
|
||||
rows, err := h.engine.Store().ListAccountSnapshots(r.Context(), accountID, from, to)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
if rows == nil {
|
||||
rows = []store.AccountSnapshotRow{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, rows)
|
||||
}
|
||||
|
||||
// GET /accounts/{id}/snapshots/positions?from=YYYY-MM-DD&to=YYYY-MM-DD
|
||||
// Liste les positions détaillées (par instrument) d'un compte.
|
||||
func (h *SnapshotHandler) listPositionSnapshots(w http.ResponseWriter, r *http.Request) {
|
||||
accountID, err := parseID(r)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid account id")
|
||||
return
|
||||
}
|
||||
from, err := parseDateParam(r, "from")
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid from date")
|
||||
return
|
||||
}
|
||||
to, err := parseDateParam(r, "to")
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid to date")
|
||||
return
|
||||
}
|
||||
|
||||
rows, err := h.engine.Store().ListPositionSnapshots(r.Context(), accountID, from, to)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
if rows == nil {
|
||||
rows = []store.PositionSnapshotRow{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, rows)
|
||||
}
|
||||
|
||||
// POST /accounts/{id}/snapshots/recompute
|
||||
// Recalcule les snapshots du compte depuis sa date d'invalidation jusqu'à aujourd'hui.
|
||||
// Retourne 204 si aucune invalidation n'est en attente.
|
||||
func (h *SnapshotHandler) recomputeAccount(w http.ResponseWriter, r *http.Request) {
|
||||
accountID, err := parseID(r)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid account id")
|
||||
return
|
||||
}
|
||||
|
||||
from, to, err := h.engine.RecomputeAccount(r.Context(), accountID)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
writeError(w, http.StatusNotFound, "account not found")
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if from.IsZero() {
|
||||
w.WriteHeader(http.StatusNoContent) // aucune invalidation en attente
|
||||
return
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"account_id": accountID,
|
||||
"from": from.Format("2006-01-02"),
|
||||
"to": to.Format("2006-01-02"),
|
||||
"status": "ok",
|
||||
})
|
||||
}
|
||||
|
|
@ -7,6 +7,7 @@ import (
|
|||
|
||||
"git.g3e.fr/H6N/account/internal/config"
|
||||
"git.g3e.fr/H6N/account/internal/handler"
|
||||
"git.g3e.fr/H6N/account/internal/snapshot"
|
||||
"git.g3e.fr/H6N/account/internal/store"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
|
@ -35,10 +36,16 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
func (s *Server) routes() {
|
||||
st := store.New(s.pool)
|
||||
horizonDays := 30
|
||||
if s.cfg != nil {
|
||||
horizonDays = s.cfg.SnapshotHorizonDays
|
||||
}
|
||||
eng := snapshot.New(st, s.logger, horizonDays)
|
||||
|
||||
handler.NewInstrumentHandler(st).RegisterRoutes(s.mux)
|
||||
handler.NewAccountHandler(st).RegisterRoutes(s.mux)
|
||||
handler.NewTransactionHandler(st).RegisterRoutes(s.mux)
|
||||
handler.NewSnapshotHandler(eng).RegisterRoutes(s.mux)
|
||||
|
||||
s.mux.HandleFunc("GET /health", s.handleHealth)
|
||||
}
|
||||
|
|
|
|||
212
internal/snapshot/engine.go
Normal file
212
internal/snapshot/engine.go
Normal file
|
|
@ -0,0 +1,212 @@
|
|||
package snapshot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"git.g3e.fr/H6N/account/internal/store"
|
||||
)
|
||||
|
||||
type Engine struct {
|
||||
store *store.Store
|
||||
logger *slog.Logger
|
||||
horizonDays int
|
||||
}
|
||||
|
||||
func New(st *store.Store, logger *slog.Logger, horizonDays int) *Engine {
|
||||
return &Engine{store: st, logger: logger, horizonDays: horizonDays}
|
||||
}
|
||||
|
||||
func (e *Engine) Store() *store.Store { return e.store }
|
||||
|
||||
// RecomputeDay calcule les snapshots de tous les comptes pour une date donnée.
|
||||
// Toutes les transactions (validées ou non) sont incluses.
|
||||
func (e *Engine) RecomputeDay(ctx context.Context, date time.Time) error {
|
||||
date = date.Truncate(24 * time.Hour)
|
||||
|
||||
accountIDs, err := e.store.ListAllAccountIDs(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list accounts: %w", err)
|
||||
}
|
||||
|
||||
for _, accountID := range accountIDs {
|
||||
if err := e.recomputeAccount(ctx, accountID, date); err != nil {
|
||||
e.logger.Error("snapshot: account failed", "account_id", accountID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
e.logger.Info("snapshot: day computed", "date", date.Format("2006-01-02"), "accounts", len(accountIDs))
|
||||
return nil
|
||||
}
|
||||
|
||||
// RecomputeAccount recalcule les snapshots d'un compte depuis son invalidation
|
||||
// jusqu'à aujourd'hui + horizon. Efface l'invalidation une fois terminé.
|
||||
// Retourne (zero, zero, nil) si aucune invalidation n'est en attente.
|
||||
func (e *Engine) RecomputeAccount(ctx context.Context, accountID int32) (from, to time.Time, err error) {
|
||||
inv, found, err := e.store.GetInvalidation(ctx, accountID)
|
||||
if err != nil {
|
||||
return time.Time{}, time.Time{}, fmt.Errorf("get invalidation: %w", err)
|
||||
}
|
||||
if !found {
|
||||
return time.Time{}, time.Time{}, nil
|
||||
}
|
||||
|
||||
today := time.Now().Truncate(24 * time.Hour)
|
||||
horizon := today.AddDate(0, 0, e.horizonDays)
|
||||
|
||||
if err := e.backfillAccount(ctx, accountID, inv.RecomputeFrom, horizon); err != nil {
|
||||
return time.Time{}, time.Time{}, err
|
||||
}
|
||||
|
||||
if err := e.store.ClearInvalidation(ctx, accountID); err != nil {
|
||||
return time.Time{}, time.Time{}, fmt.Errorf("clear invalidation: %w", err)
|
||||
}
|
||||
|
||||
return inv.RecomputeFrom, horizon, nil
|
||||
}
|
||||
|
||||
// DailySnapshot :
|
||||
// 1. Traite les invalidations en attente (jusqu'à today + horizon)
|
||||
// 2. Recalcule hier (consolide les transactions de la veille)
|
||||
// 3. Calcule le nouveau jour entrant dans la fenêtre (today + horizonDays)
|
||||
func (e *Engine) DailySnapshot(ctx context.Context) error {
|
||||
today := time.Now().Truncate(24 * time.Hour)
|
||||
yesterday := today.AddDate(0, 0, -1)
|
||||
newHorizonDay := today.AddDate(0, 0, e.horizonDays)
|
||||
|
||||
// 1. Invalidations
|
||||
if err := e.processInvalidations(ctx, newHorizonDay); err != nil {
|
||||
e.logger.Error("snapshot: invalidation processing failed", "error", err)
|
||||
}
|
||||
|
||||
// 2. Hier
|
||||
if err := e.RecomputeDay(ctx, yesterday); err != nil {
|
||||
e.logger.Error("snapshot: yesterday recompute failed", "error", err)
|
||||
}
|
||||
|
||||
// 3. Nouveau jour entrant dans la fenêtre
|
||||
if err := e.RecomputeDay(ctx, newHorizonDay); err != nil {
|
||||
e.logger.Error("snapshot: horizon day failed", "error", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Engine) processInvalidations(ctx context.Context, until time.Time) error {
|
||||
invalidations, err := e.store.GetInvalidations(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get invalidations: %w", err)
|
||||
}
|
||||
if len(invalidations) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
e.logger.Info("snapshot: processing invalidations", "count", len(invalidations))
|
||||
|
||||
for _, inv := range invalidations {
|
||||
if err := e.backfillAccount(ctx, inv.AccountID, inv.RecomputeFrom, until); err != nil {
|
||||
e.logger.Error("snapshot: backfill failed", "account_id", inv.AccountID, "error", err)
|
||||
continue
|
||||
}
|
||||
if err := e.store.ClearInvalidation(ctx, inv.AccountID); err != nil {
|
||||
e.logger.Error("snapshot: clear invalidation failed", "account_id", inv.AccountID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Engine) backfillAccount(ctx context.Context, accountID int32, from, to time.Time) error {
|
||||
from = from.Truncate(24 * time.Hour)
|
||||
to = to.Truncate(24 * time.Hour)
|
||||
|
||||
if from.After(to) {
|
||||
return fmt.Errorf("from (%s) after to (%s)", from.Format("2006-01-02"), to.Format("2006-01-02"))
|
||||
}
|
||||
|
||||
for d := from; !d.After(to); d = d.AddDate(0, 0, 1) {
|
||||
if err := e.recomputeAccount(ctx, accountID, d); err != nil {
|
||||
return fmt.Errorf("recompute %s: %w", d.Format("2006-01-02"), err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Engine) recomputeAccount(ctx context.Context, accountID int32, date time.Time) error {
|
||||
baseDate, hasBase, err := e.store.GetLatestSnapshotDate(ctx, accountID, date)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get latest snapshot date: %w", err)
|
||||
}
|
||||
|
||||
positions := map[int32]store.PositionRow{}
|
||||
if hasBase {
|
||||
base, err := e.store.GetBasePositions(ctx, accountID, baseDate)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get base positions: %w", err)
|
||||
}
|
||||
for _, p := range base {
|
||||
positions[p.InstrumentID] = p
|
||||
}
|
||||
}
|
||||
|
||||
deltas, err := e.store.ComputeDeltas(ctx, accountID, baseDate, date)
|
||||
if err != nil {
|
||||
return fmt.Errorf("compute deltas: %w", err)
|
||||
}
|
||||
|
||||
for _, d := range deltas {
|
||||
p := positions[d.InstrumentID]
|
||||
p.InstrumentID = d.InstrumentID
|
||||
p.InstrumentType = d.InstrumentType
|
||||
p.Quantite += d.Quantite
|
||||
positions[d.InstrumentID] = p
|
||||
}
|
||||
|
||||
var totalValeur float64
|
||||
|
||||
for _, pos := range positions {
|
||||
if pos.Quantite <= 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
prix, found, err := e.store.GetPriceAt(ctx, pos.InstrumentID, date)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get price instrument %d: %w", pos.InstrumentID, err)
|
||||
}
|
||||
if !found {
|
||||
if pos.InstrumentType == "devise" {
|
||||
prix = 1.0
|
||||
} else {
|
||||
e.logger.Warn("snapshot: no price, position skipped",
|
||||
"account_id", accountID,
|
||||
"instrument_id", pos.InstrumentID,
|
||||
"date", date.Format("2006-01-02"),
|
||||
)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
valeur := pos.Quantite * prix
|
||||
totalValeur += valeur
|
||||
|
||||
snap := store.PositionSnapshot{
|
||||
Date: date,
|
||||
AccountID: accountID,
|
||||
InstrumentID: pos.InstrumentID,
|
||||
Quantite: pos.Quantite,
|
||||
PrixCloture: &prix,
|
||||
Valeur: &valeur,
|
||||
}
|
||||
if err := e.store.UpsertPositionSnapshot(ctx, snap); err != nil {
|
||||
return fmt.Errorf("upsert position snapshot: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return e.store.UpsertAccountSnapshot(ctx, store.AccountSnapshot{
|
||||
Date: date,
|
||||
AccountID: accountID,
|
||||
Valeur: totalValeur,
|
||||
})
|
||||
}
|
||||
271
internal/store/snapshot.go
Normal file
271
internal/store/snapshot.go
Normal file
|
|
@ -0,0 +1,271 @@
|
|||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
type PositionRow struct {
|
||||
InstrumentID int32 `json:"instrument_id"`
|
||||
InstrumentType string `json:"instrument_type"`
|
||||
Quantite float64 `json:"quantite"`
|
||||
}
|
||||
|
||||
type PositionSnapshot struct {
|
||||
Date time.Time
|
||||
AccountID int32
|
||||
InstrumentID int32
|
||||
Quantite float64
|
||||
PRU *float64
|
||||
PrixCloture *float64
|
||||
Valeur *float64
|
||||
}
|
||||
|
||||
type AccountSnapshot struct {
|
||||
Date time.Time
|
||||
AccountID int32
|
||||
Valeur float64
|
||||
}
|
||||
|
||||
// ListAllAccountIDs retourne tous les comptes (maîtres + enveloppes).
|
||||
func (s *Store) ListAllAccountIDs(ctx context.Context) ([]int32, error) {
|
||||
rows, err := s.pool.Query(ctx, `SELECT id FROM account ORDER BY id`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var ids []int32
|
||||
for rows.Next() {
|
||||
var id int32
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
|
||||
// GetLatestSnapshotDate retourne la date du snapshot le plus récent strictement
|
||||
// antérieur à before. Retourne (zero, false, nil) si aucun snapshot n'existe.
|
||||
// On utilise account_snapshot comme marqueur de "jour déjà calculé" car il
|
||||
// contient toujours une ligne même pour un compte à solde nul.
|
||||
func (s *Store) GetLatestSnapshotDate(ctx context.Context, accountID int32, before time.Time) (time.Time, bool, error) {
|
||||
var t *time.Time
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT MAX(date) FROM account_snapshot
|
||||
WHERE account_id = $1 AND date < $2::date
|
||||
`, accountID, before).Scan(&t)
|
||||
if err != nil || t == nil {
|
||||
return time.Time{}, false, err
|
||||
}
|
||||
return *t, true, nil
|
||||
}
|
||||
|
||||
// GetBasePositions retourne les positions d'un compte à une date de snapshot
|
||||
// existante, enrichies du type d'instrument pour le fallback de prix.
|
||||
func (s *Store) GetBasePositions(ctx context.Context, accountID int32, date time.Time) ([]PositionRow, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT ps.instrument_id, i.type AS instrument_type, ps.quantite
|
||||
FROM position_snapshot ps
|
||||
JOIN instrument i ON i.id = ps.instrument_id
|
||||
WHERE ps.account_id = $1 AND ps.date = $2::date
|
||||
`, accountID, date)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pgx.CollectRows(rows, pgx.RowToStructByName[PositionRow])
|
||||
}
|
||||
|
||||
// ComputeDeltas calcule les variations nettes par instrument pour un compte
|
||||
// entre fromExclusive (exclu) et toInclusive (inclus).
|
||||
// Toutes les transactions sont comptées quelle que soit leur validation :
|
||||
// le solde reflète l'état envisagé complet (passé confirmé + prévisionnel).
|
||||
// Passer time.Time{} comme fromExclusive couvre toutes les transactions.
|
||||
func (s *Store) ComputeDeltas(ctx context.Context, accountID int32, fromExclusive, toInclusive time.Time) ([]PositionRow, error) {
|
||||
from := fromExclusive
|
||||
if from.IsZero() {
|
||||
from = time.Date(1, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
}
|
||||
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
WITH flows AS (
|
||||
SELECT instrument_dest_id AS instrument_id, quantite_dest AS delta
|
||||
FROM transaction
|
||||
WHERE account_dest_id = $1
|
||||
AND date > $2::date AND date <= $3::date
|
||||
UNION ALL
|
||||
SELECT instrument_source_id, -quantite_source
|
||||
FROM transaction
|
||||
WHERE account_source_id = $1
|
||||
AND date > $2::date AND date <= $3::date
|
||||
)
|
||||
SELECT f.instrument_id, i.type AS instrument_type, SUM(f.delta) AS quantite
|
||||
FROM flows f
|
||||
JOIN instrument i ON i.id = f.instrument_id
|
||||
GROUP BY f.instrument_id, i.type
|
||||
`, accountID, from, toInclusive)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pgx.CollectRows(rows, pgx.RowToStructByName[PositionRow])
|
||||
}
|
||||
|
||||
// AccountSnapshotRow représente une ligne de account_snapshot pour la restitution.
|
||||
type AccountSnapshotRow struct {
|
||||
Date string `json:"date"`
|
||||
Valeur float64 `json:"valeur"`
|
||||
}
|
||||
|
||||
// PositionSnapshotRow représente une ligne de position_snapshot pour la restitution.
|
||||
type PositionSnapshotRow struct {
|
||||
Date string `json:"date"`
|
||||
InstrumentID int32 `json:"instrument_id"`
|
||||
Quantite float64 `json:"quantite"`
|
||||
PRU *float64 `json:"pru,omitempty"`
|
||||
PrixCloture *float64 `json:"prix_cloture,omitempty"`
|
||||
Valeur *float64 `json:"valeur,omitempty"`
|
||||
}
|
||||
|
||||
// ListAccountSnapshots retourne les snapshots agrégés d'un compte entre from et to.
|
||||
// Si from est zero, pas de borne inférieure. Si to est zero, pas de borne supérieure.
|
||||
func (s *Store) ListAccountSnapshots(ctx context.Context, accountID int32, from, to time.Time) ([]AccountSnapshotRow, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT date::text, valeur
|
||||
FROM account_snapshot
|
||||
WHERE account_id = $1
|
||||
AND ($2::date IS NULL OR date >= $2::date)
|
||||
AND ($3::date IS NULL OR date <= $3::date)
|
||||
ORDER BY date
|
||||
`, accountID, nullableDate(from), nullableDate(to))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pgx.CollectRows(rows, pgx.RowToStructByName[AccountSnapshotRow])
|
||||
}
|
||||
|
||||
// ListPositionSnapshots retourne les positions détaillées d'un compte entre from et to.
|
||||
// Si from est zero, pas de borne inférieure. Si to est zero, pas de borne supérieure.
|
||||
func (s *Store) ListPositionSnapshots(ctx context.Context, accountID int32, from, to time.Time) ([]PositionSnapshotRow, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT date::text, instrument_id, quantite, pru, prix_cloture, valeur
|
||||
FROM position_snapshot
|
||||
WHERE account_id = $1
|
||||
AND ($2::date IS NULL OR date >= $2::date)
|
||||
AND ($3::date IS NULL OR date <= $3::date)
|
||||
ORDER BY date, instrument_id
|
||||
`, accountID, nullableDate(from), nullableDate(to))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pgx.CollectRows(rows, pgx.RowToStructByName[PositionSnapshotRow])
|
||||
}
|
||||
|
||||
// nullableDate convertit time.Time{} en nil pour les paramètres SQL optionnels.
|
||||
func nullableDate(t time.Time) any {
|
||||
if t.IsZero() {
|
||||
return nil
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
// GetPriceAt retourne le dernier prix connu pour un instrument à une date donnée.
|
||||
// Retourne (0, false, nil) si aucun prix n'est trouvé.
|
||||
func (s *Store) GetPriceAt(ctx context.Context, instrumentID int32, date time.Time) (float64, bool, error) {
|
||||
var prix float64
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT prix FROM price_history
|
||||
WHERE instrument_id = $1 AND fetched_at::date <= $2::date
|
||||
ORDER BY fetched_at DESC
|
||||
LIMIT 1
|
||||
`, instrumentID, date).Scan(&prix)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return 0, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
}
|
||||
return prix, true, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpsertPositionSnapshot(ctx context.Context, snap PositionSnapshot) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO position_snapshot (date, account_id, instrument_id, quantite, pru, prix_cloture, valeur)
|
||||
VALUES ($1::date, $2, $3, $4, $5, $6, $7)
|
||||
ON CONFLICT (date, account_id, instrument_id) DO UPDATE SET
|
||||
quantite = EXCLUDED.quantite,
|
||||
pru = EXCLUDED.pru,
|
||||
prix_cloture = EXCLUDED.prix_cloture,
|
||||
valeur = EXCLUDED.valeur
|
||||
`, snap.Date, snap.AccountID, snap.InstrumentID,
|
||||
snap.Quantite, snap.PRU, snap.PrixCloture, snap.Valeur)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Store) UpsertAccountSnapshot(ctx context.Context, snap AccountSnapshot) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO account_snapshot (date, account_id, valeur)
|
||||
VALUES ($1::date, $2, $3)
|
||||
ON CONFLICT (date, account_id) DO UPDATE SET valeur = EXCLUDED.valeur
|
||||
`, snap.Date, snap.AccountID, snap.Valeur)
|
||||
return err
|
||||
}
|
||||
|
||||
// ── Invalidation ─────────────────────────────────────────────────────────────
|
||||
|
||||
type Invalidation struct {
|
||||
AccountID int32
|
||||
RecomputeFrom time.Time
|
||||
}
|
||||
|
||||
// InvalidateSnapshot marque un compte comme nécessitant un recalcul depuis date.
|
||||
// Conserve le MIN si une invalidation antérieure existe déjà.
|
||||
func (s *Store) InvalidateSnapshot(ctx context.Context, accountID int32, from time.Time) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO account_snapshot_invalidation (account_id, recompute_from)
|
||||
VALUES ($1, $2::date)
|
||||
ON CONFLICT (account_id) DO UPDATE
|
||||
SET recompute_from = LEAST(account_snapshot_invalidation.recompute_from, EXCLUDED.recompute_from)
|
||||
`, accountID, from)
|
||||
return err
|
||||
}
|
||||
|
||||
// GetInvalidations retourne tous les comptes en attente de recalcul.
|
||||
func (s *Store) GetInvalidations(ctx context.Context) ([]Invalidation, error) {
|
||||
rows, err := s.pool.Query(ctx,
|
||||
`SELECT account_id, recompute_from FROM account_snapshot_invalidation ORDER BY account_id`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Invalidation
|
||||
for rows.Next() {
|
||||
var inv Invalidation
|
||||
if err := rows.Scan(&inv.AccountID, &inv.RecomputeFrom); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, inv)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// ClearInvalidation supprime le flag d'invalidation une fois le recalcul terminé.
|
||||
func (s *Store) ClearInvalidation(ctx context.Context, accountID int32) error {
|
||||
_, err := s.pool.Exec(ctx,
|
||||
`DELETE FROM account_snapshot_invalidation WHERE account_id = $1`, accountID)
|
||||
return err
|
||||
}
|
||||
|
||||
// GetInvalidation retourne l'invalidation d'un compte spécifique, si elle existe.
|
||||
func (s *Store) GetInvalidation(ctx context.Context, accountID int32) (Invalidation, bool, error) {
|
||||
var inv Invalidation
|
||||
err := s.pool.QueryRow(ctx,
|
||||
`SELECT account_id, recompute_from FROM account_snapshot_invalidation WHERE account_id = $1`,
|
||||
accountID).Scan(&inv.AccountID, &inv.RecomputeFrom)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Invalidation{}, false, nil
|
||||
}
|
||||
return inv, err == nil, err
|
||||
}
|
||||
|
|
@ -3,6 +3,7 @@ package store
|
|||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
|
@ -129,10 +130,21 @@ func (s *Store) CreateTransaction(ctx context.Context, p CreateTransactionParams
|
|||
if err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
return pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction])
|
||||
tx, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction])
|
||||
if err != nil {
|
||||
return tx, err
|
||||
}
|
||||
s.invalidateAccounts(ctx, tx.Date, tx.AccountSourceID, tx.AccountDestID)
|
||||
return tx, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpdateTransaction(ctx context.Context, id int64, p CreateTransactionParams) (Transaction, error) {
|
||||
// Récupérer l'ancienne date pour invalider à partir du MIN(ancienne, nouvelle).
|
||||
old, err := s.GetTransaction(ctx, id)
|
||||
if err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
|
||||
rows, err := s.pool.Query(ctx,
|
||||
`UPDATE transaction SET
|
||||
date = $2::date,
|
||||
|
|
@ -148,7 +160,16 @@ func (s *Store) UpdateTransaction(ctx context.Context, id int64, p CreateTransac
|
|||
if err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
return pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction])
|
||||
tx, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction])
|
||||
if err != nil {
|
||||
return tx, err
|
||||
}
|
||||
// Invalider depuis la date la plus ancienne (ancienne ou nouvelle).
|
||||
earliest := minDateStr(old.Date, tx.Date)
|
||||
s.invalidateAccounts(ctx, earliest, tx.AccountSourceID, tx.AccountDestID)
|
||||
// Si les comptes ont changé, invalider aussi les anciens.
|
||||
s.invalidateAccounts(ctx, earliest, old.AccountSourceID, old.AccountDestID)
|
||||
return tx, nil
|
||||
}
|
||||
|
||||
func (s *Store) SetValidated(ctx context.Context, id int64, validated bool) (Transaction, error) {
|
||||
|
|
@ -158,10 +179,45 @@ func (s *Store) SetValidated(ctx context.Context, id int64, validated bool) (Tra
|
|||
if err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
return pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction])
|
||||
tx, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction])
|
||||
if err != nil {
|
||||
return tx, err
|
||||
}
|
||||
s.invalidateAccounts(ctx, tx.Date, tx.AccountSourceID, tx.AccountDestID)
|
||||
return tx, nil
|
||||
}
|
||||
|
||||
func (s *Store) DeleteTransaction(ctx context.Context, id int64) error {
|
||||
_, err := s.pool.Exec(ctx, `DELETE FROM transaction WHERE id = $1`, id)
|
||||
return err
|
||||
tx, err := s.GetTransaction(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := s.pool.Exec(ctx, `DELETE FROM transaction WHERE id = $1`, id); err != nil {
|
||||
return err
|
||||
}
|
||||
s.invalidateAccounts(ctx, tx.Date, tx.AccountSourceID, tx.AccountDestID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// invalidateAccounts marque les comptes non-nil comme devant être recalculés depuis dateStr.
|
||||
func (s *Store) invalidateAccounts(ctx context.Context, dateStr string, accountIDs ...*int32) {
|
||||
date, err := time.Parse("2006-01-02", dateStr)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
seen := map[int32]bool{}
|
||||
for _, id := range accountIDs {
|
||||
if id != nil && !seen[*id] {
|
||||
seen[*id] = true
|
||||
s.InvalidateSnapshot(ctx, *id, date) //nolint:errcheck
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// minDateStr retourne la plus petite des deux dates au format YYYY-MM-DD.
|
||||
func minDateStr(a, b string) string {
|
||||
if a <= b {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
|
|
|||
1
migrations/000002_snapshot_invalidation.down.sql
Normal file
1
migrations/000002_snapshot_invalidation.down.sql
Normal file
|
|
@ -0,0 +1 @@
|
|||
DROP TABLE IF EXISTS account_snapshot_invalidation;
|
||||
7
migrations/000002_snapshot_invalidation.up.sql
Normal file
7
migrations/000002_snapshot_invalidation.up.sql
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
-- Marque les comptes dont les snapshots doivent être recalculés.
|
||||
-- recompute_from = date la plus ancienne à partir de laquelle recalculer.
|
||||
-- Mise à jour par MIN() à chaque mutation de transaction (create/update/delete/validate).
|
||||
CREATE TABLE account_snapshot_invalidation (
|
||||
account_id INTEGER PRIMARY KEY REFERENCES account(id) ON DELETE CASCADE,
|
||||
recompute_from DATE NOT NULL
|
||||
);
|
||||
Loading…
Add table
Add a link
Reference in a new issue