From 28e082f0fce8d280e2d1915fd5d709740642eb73 Mon Sep 17 00:00:00 2001 From: GnomeZworc Date: Sun, 14 Jun 2026 13:41:42 +0200 Subject: [PATCH] add correct backfile Signed-off-by: GnomeZworc --- cmd/api/main.go | 2 +- internal/handler/snapshot.go | 28 +++++++++++ internal/pipeline/coingecko.go | 43 +++++++++++++++++ internal/pipeline/pipeline.go | 49 +++++++++++++++++++ internal/pipeline/yahoo.go | 87 +++++++++++++++++++++++++++------- internal/server/server.go | 9 +++- internal/snapshot/engine.go | 79 +++++++++++++++++++++++------- internal/store/snapshot.go | 55 +++++++++++++++++++++ 8 files changed, 317 insertions(+), 35 deletions(-) diff --git a/cmd/api/main.go b/cmd/api/main.go index bcf971c..2a8ed58 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -41,11 +41,11 @@ 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, }, logger) + snap := snapshot.New(st, logger, cfg.SnapshotHorizonDays).WithBackfiller(pl) sched := scheduler.New(logger) sched.Add(scheduler.Job{ diff --git a/internal/handler/snapshot.go b/internal/handler/snapshot.go index 7d51ab2..41a4a7f 100644 --- a/internal/handler/snapshot.go +++ b/internal/handler/snapshot.go @@ -42,6 +42,7 @@ 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) + mux.HandleFunc("GET /snapshots/pending", h.listPending) } // parseDateParam parse un paramètre de query YYYY-MM-DD. Retourne time.Time{} si absent. @@ -53,6 +54,33 @@ func parseDateParam(r *http.Request, key string) (time.Time, error) { return time.Parse("2006-01-02", v) } +// GET /snapshots/pending +// Liste les comptes de l'owner ayant une invalidation en attente. +func (h *SnapshotHandler) listPending(w http.ResponseWriter, r *http.Request) { + ownerID, ok := requireOwner(w, r) + if !ok { + return + } + invalidations, err := h.engine.Store().GetInvalidationsForOwner(r.Context(), ownerID) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + type row struct { + AccountID int32 `json:"account_id"` + RecomputeFrom string `json:"recompute_from"` + } + out := make([]row, 0, len(invalidations)) + for _, inv := range invalidations { + out = append(out, row{ + AccountID: inv.AccountID, + RecomputeFrom: inv.RecomputeFrom.Format("2006-01-02"), + }) + } + writeJSON(w, http.StatusOK, out) +} + // 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é. diff --git a/internal/pipeline/coingecko.go b/internal/pipeline/coingecko.go index 20d9a00..2bd472d 100644 --- a/internal/pipeline/coingecko.go +++ b/internal/pipeline/coingecko.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "strings" + "time" ) // ResolveCryptoID résout un ticker crypto (ex: "BTC") en ID CoinGecko (ex: "bitcoin"). @@ -46,6 +47,48 @@ func ResolveCryptoID(ctx context.Context, apiKey, ticker string) (string, error) return "", fmt.Errorf("no coingecko match for ticker %s", ticker) } +// FetchCoinGeckoHistory récupère les prix EUR journaliers entre from et to pour un coin. +func FetchCoinGeckoHistory(ctx context.Context, apiKey, coinID string, from, to time.Time) ([]PricePoint, error) { + baseURL := "https://api.coingecko.com/api/v3" + if apiKey != "" { + baseURL = "https://pro-api.coingecko.com/api/v3" + } + + url := fmt.Sprintf("%s/coins/%s/market_chart/range?vs_currency=eur&from=%d&to=%d", + baseURL, coinID, from.Unix(), to.Unix()) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + if apiKey != "" { + req.Header.Set("x-cg-pro-api-key", apiKey) + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + // Réponse : {"prices": [[timestamp_ms, price], ...]} + var raw struct { + Prices [][2]float64 `json:"prices"` + } + if err := json.NewDecoder(resp.Body).Decode(&raw); err != nil { + return nil, fmt.Errorf("decode coingecko history: %w", err) + } + + points := make([]PricePoint, 0, len(raw.Prices)) + for _, p := range raw.Prices { + points = append(points, PricePoint{ + At: time.UnixMilli(int64(p[0])).UTC(), + Price: p[1], + }) + } + return points, nil +} + // FetchCoinGeckoPrices récupère les prix EUR pour une liste d'IDs CoinGecko. // Le ticker stocké dans instrument_ticker_cache doit être l'ID CoinGecko (ex: "bitcoin", "ethereum"). func FetchCoinGeckoPrices(ctx context.Context, apiKey string, coinIDs []string) (map[string]float64, error) { diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 04b5f35..3139c28 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -2,6 +2,7 @@ package pipeline import ( "context" + "fmt" "log/slog" "time" @@ -90,6 +91,54 @@ func (p *Pipeline) FetchAll(ctx context.Context) error { return nil } +// BackfillInstrument récupère l'historique de prix d'un instrument entre from et to +// et upserte chaque point dans price_history. Idempotent. +func (p *Pipeline) BackfillInstrument(ctx context.Context, instrumentID int32, from, to time.Time) error { + inst, err := p.store.GetInstrument(ctx, instrumentID) + if err != nil { + return fmt.Errorf("get instrument %d: %w", instrumentID, err) + } + + switch inst.Type { + case "devise": + return nil // EUR toujours = 1, pas de backfill nécessaire + + case "action", "etf": + ticker, err := p.resolveTicker(ctx, inst) + if err != nil { + return fmt.Errorf("resolve ticker for %s: %w", inst.Code, err) + } + points, err := FetchYahooHistory(ctx, ticker, from, to) + if err != nil { + return fmt.Errorf("yahoo history %s: %w", ticker, err) + } + for _, pt := range points { + if err := p.store.UpsertPrice(ctx, inst.ID, pt.At, pt.Price); err != nil { + p.logger.Error("upsert historical price", "instrument", inst.Code, "date", pt.At, "error", err) + } + } + p.logger.Info("backfill done", "instrument", inst.Code, "points", len(points)) + + case "crypto": + coinID, err := p.resolveCoinID(ctx, inst) + if err != nil { + return fmt.Errorf("resolve coin id for %s: %w", inst.Code, err) + } + points, err := FetchCoinGeckoHistory(ctx, p.cfg.CoinGeckoKey, coinID, from, to) + if err != nil { + return fmt.Errorf("coingecko history %s: %w", coinID, err) + } + for _, pt := range points { + if err := p.store.UpsertPrice(ctx, inst.ID, pt.At, pt.Price); err != nil { + p.logger.Error("upsert historical price", "instrument", inst.Code, "date", pt.At, "error", err) + } + } + p.logger.Info("backfill done", "instrument", inst.Code, "points", len(points)) + } + + return nil +} + // CleanHistory supprime les prix intraday des jours passés, ne gardant que le dernier par instrument. func (p *Pipeline) CleanHistory(ctx context.Context) error { if err := p.store.CleanPastDays(ctx); err != nil { diff --git a/internal/pipeline/yahoo.go b/internal/pipeline/yahoo.go index f08eb68..d750576 100644 --- a/internal/pipeline/yahoo.go +++ b/internal/pipeline/yahoo.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "net/http" + "time" ) type yahooChart struct { @@ -14,6 +15,12 @@ type yahooChart struct { RegularMarketPrice float64 `json:"regularMarketPrice"` Currency string `json:"currency"` } `json:"meta"` + Timestamps []int64 `json:"timestamp"` + Indicators struct { + Quote []struct { + Close []*float64 `json:"close"` + } `json:"quote"` + } `json:"indicators"` } `json:"result"` Error *struct{ Description string } `json:"error"` } `json:"chart"` @@ -23,33 +30,79 @@ type yahooChart struct { func FetchYahooPrice(ctx context.Context, ticker string) (float64, error) { url := fmt.Sprintf( "https://query1.finance.yahoo.com/v8/finance/chart/%s?interval=1d&range=1d", ticker) + return fetchYahoo(ctx, url, ticker) +} - req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) +// FetchYahooHistory récupère les prix de clôture journaliers entre from et to. +// Retourne une slice de (time, price) pour chaque jour ayant une clôture non nulle. +func FetchYahooHistory(ctx context.Context, ticker string, from, to time.Time) ([]PricePoint, error) { + url := fmt.Sprintf( + "https://query1.finance.yahoo.com/v8/finance/chart/%s?interval=1d&period1=%d&period2=%d", + ticker, from.Unix(), to.Unix()) + + chart, err := fetchYahooChart(ctx, url, ticker) + if err != nil { + return nil, err + } + if len(chart.Chart.Result) == 0 || len(chart.Chart.Result[0].Indicators.Quote) == 0 { + return nil, nil + } + + res := chart.Chart.Result[0] + closes := res.Indicators.Quote[0].Close + var points []PricePoint + for i, ts := range res.Timestamps { + if i >= len(closes) || closes[i] == nil || *closes[i] == 0 { + continue + } + points = append(points, PricePoint{ + At: time.Unix(ts, 0).UTC(), + Price: *closes[i], + }) + } + return points, nil +} + +func fetchYahoo(ctx context.Context, url, ticker string) (float64, error) { + chart, err := fetchYahooChart(ctx, url, ticker) if err != nil { return 0, err } - req.Header.Set("User-Agent", "Mozilla/5.0") - - resp, err := http.DefaultClient.Do(req) - if err != nil { - return 0, err - } - defer resp.Body.Close() - - var chart yahooChart - if err := json.NewDecoder(resp.Body).Decode(&chart); err != nil { - return 0, fmt.Errorf("decode yahoo response: %w", err) - } - if chart.Chart.Error != nil { - return 0, fmt.Errorf("yahoo error: %s", chart.Chart.Error.Description) - } if len(chart.Chart.Result) == 0 { return 0, fmt.Errorf("no result for ticker %s", ticker) } - price := chart.Chart.Result[0].Meta.RegularMarketPrice if price == 0 { return 0, fmt.Errorf("zero price for ticker %s", ticker) } return price, nil } + +func fetchYahooChart(ctx context.Context, url, ticker string) (*yahooChart, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + req.Header.Set("User-Agent", "Mozilla/5.0") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var chart yahooChart + if err := json.NewDecoder(resp.Body).Decode(&chart); err != nil { + return nil, fmt.Errorf("decode yahoo response: %w", err) + } + if chart.Chart.Error != nil { + return nil, fmt.Errorf("yahoo error: %s", chart.Chart.Error.Description) + } + return &chart, nil +} + +// PricePoint est un prix à un instant donné, partagé entre les providers. +type PricePoint struct { + At time.Time + Price float64 +} diff --git a/internal/server/server.go b/internal/server/server.go index 15ddec6..805320a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -8,6 +8,7 @@ import ( "git.g3e.fr/H6N/account/internal/auth" "git.g3e.fr/H6N/account/internal/config" "git.g3e.fr/H6N/account/internal/handler" + "git.g3e.fr/H6N/account/internal/pipeline" "git.g3e.fr/H6N/account/internal/snapshot" "git.g3e.fr/H6N/account/internal/store" "github.com/jackc/pgx/v5/pgxpool" @@ -43,10 +44,16 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (s *Server) routes() { st := store.New(s.pool) horizonDays := 30 + var pipelineCfg pipeline.Config if s.cfg != nil { horizonDays = s.cfg.SnapshotHorizonDays + pipelineCfg = pipeline.Config{ + OpenFIGIKey: s.cfg.OpenFIGIKey, + CoinGeckoKey: s.cfg.CoinGeckoKey, + } } - eng := snapshot.New(st, s.logger, horizonDays) + pl := pipeline.New(st, pipelineCfg, s.logger) + eng := snapshot.New(st, s.logger, horizonDays).WithBackfiller(pl) handler.NewInstrumentHandler(st).RegisterRoutes(s.mux) handler.NewAccountHandler(st).RegisterRoutes(s.mux) diff --git a/internal/snapshot/engine.go b/internal/snapshot/engine.go index 1059df0..f72127c 100644 --- a/internal/snapshot/engine.go +++ b/internal/snapshot/engine.go @@ -9,16 +9,28 @@ import ( "git.g3e.fr/H6N/account/internal/store" ) +// PriceBackfiller est implémenté par le pipeline pour récupérer l'historique de prix manquant. +type PriceBackfiller interface { + BackfillInstrument(ctx context.Context, instrumentID int32, from, to time.Time) error +} + type Engine struct { store *store.Store logger *slog.Logger horizonDays int + backfiller PriceBackfiller } func New(st *store.Store, logger *slog.Logger, horizonDays int) *Engine { return &Engine{store: st, logger: logger, horizonDays: horizonDays} } +// WithBackfiller active le backfill automatique des prix historiques manquants. +func (e *Engine) WithBackfiller(b PriceBackfiller) *Engine { + e.backfiller = b + return e +} + func (e *Engine) Store() *store.Store { return e.store } // RecomputeDay calcule les snapshots de tous les comptes pour une date donnée. @@ -126,8 +138,11 @@ func (e *Engine) backfillAccount(ctx context.Context, accountID int32, from, to return fmt.Errorf("from (%s) after to (%s)", from.Format("2006-01-02"), to.Format("2006-01-02")) } + // backfilledInstruments évite de re-fetcher le même instrument plusieurs fois dans ce run. + backfilledInstruments := make(map[int32]bool) + for d := from; !d.After(to); d = d.AddDate(0, 0, 1) { - if err := e.recomputeAccount(ctx, accountID, d); err != nil { + if err := e.recomputeAccountWithBackfill(ctx, accountID, d, backfilledInstruments); err != nil { return fmt.Errorf("recompute %s: %w", d.Format("2006-01-02"), err) } } @@ -135,11 +150,20 @@ func (e *Engine) backfillAccount(ctx context.Context, accountID int32, from, to } func (e *Engine) recomputeAccount(ctx context.Context, accountID int32, date time.Time) error { + return e.recomputeAccountWithBackfill(ctx, accountID, date, nil) +} + +func (e *Engine) recomputeAccountWithBackfill(ctx context.Context, accountID int32, date time.Time, backfilled map[int32]bool) error { baseDate, hasBase, err := e.store.GetLatestSnapshotDate(ctx, accountID, date) if err != nil { return fmt.Errorf("get latest snapshot date: %w", err) } + prus, err := e.store.ComputePRUs(ctx, accountID, date) + if err != nil { + return fmt.Errorf("compute PRUs: %w", err) + } + positions := map[int32]store.PositionRow{} if hasBase { base, err := e.store.GetBasePositions(ctx, accountID, baseDate) @@ -175,30 +199,53 @@ func (e *Engine) recomputeAccount(ctx context.Context, accountID int32, date tim 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 - } + if !found && pos.InstrumentType == "devise" { + prix = 1.0 + found = true } - valeur := pos.Quantite * prix - totalValeur += valeur + // Backfill automatique si prix manquant et backfiller disponible. + // from = date - 1 jour (marge timezone), to = maintenant (inclut aujourd'hui). + if !found && e.backfiller != nil && (backfilled == nil || !backfilled[pos.InstrumentID]) { + from := date.AddDate(0, 0, -1) + now := time.Now().UTC() + if err := e.backfiller.BackfillInstrument(ctx, pos.InstrumentID, from, now); err != nil { + e.logger.Warn("snapshot: backfill échoué", + "instrument_id", pos.InstrumentID, "error", err) + } else { + if backfilled != nil { + backfilled[pos.InstrumentID] = true + } + // Retry après backfill + prix, found, err = e.store.GetPriceAt(ctx, pos.InstrumentID, date) + if err != nil { + return fmt.Errorf("get price instrument %d (post-backfill): %w", pos.InstrumentID, err) + } + } + } snap := store.PositionSnapshot{ Date: date, AccountID: accountID, InstrumentID: pos.InstrumentID, Quantite: pos.Quantite, - PrixCloture: &prix, - Valeur: &valeur, } + if pru, ok := prus[pos.InstrumentID]; ok { + snap.PRU = &pru + } + if found { + valeur := pos.Quantite * prix + snap.PrixCloture = &prix + snap.Valeur = &valeur + totalValeur += valeur + } else { + e.logger.Warn("snapshot: no price, valeur non calculée", + "account_id", accountID, + "instrument_id", pos.InstrumentID, + "date", date.Format("2006-01-02"), + ) + } + if err := e.store.UpsertPositionSnapshot(ctx, snap); err != nil { return fmt.Errorf("upsert position snapshot: %w", err) } diff --git a/internal/store/snapshot.go b/internal/store/snapshot.go index 25d6d23..c4ee352 100644 --- a/internal/store/snapshot.go +++ b/internal/store/snapshot.go @@ -171,6 +171,37 @@ func nullableDate(t time.Time) any { return t } +// ComputePRUs retourne le PRU (coût moyen pondéré) par instrument pour un compte, +// calculé sur toutes les transactions d'achat (account_dest_id) jusqu'à asOf inclus. +// PRU = sum(quantite_source) / sum(quantite_dest) +// Retourne uniquement les instruments qui ont des achats. +func (s *Store) ComputePRUs(ctx context.Context, accountID int32, asOf time.Time) (map[int32]float64, error) { + rows, err := s.pool.Query(ctx, ` + SELECT instrument_dest_id, + SUM(quantite_source) / NULLIF(SUM(quantite_dest), 0) AS pru + FROM transaction + WHERE account_dest_id = $1 + AND date <= $2::date + AND quantite_source IS NOT NULL + GROUP BY instrument_dest_id + `, accountID, asOf) + if err != nil { + return nil, err + } + defer rows.Close() + + result := make(map[int32]float64) + for rows.Next() { + var instrumentID int32 + var pru float64 + if err := rows.Scan(&instrumentID, &pru); err != nil { + return nil, err + } + result[instrumentID] = pru + } + return result, rows.Err() +} + // 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) { @@ -251,6 +282,30 @@ func (s *Store) GetInvalidations(ctx context.Context) ([]Invalidation, error) { return out, rows.Err() } +// GetInvalidationsForOwner retourne les invalidations des comptes appartenant à ownerID. +func (s *Store) GetInvalidationsForOwner(ctx context.Context, ownerID int32) ([]Invalidation, error) { + rows, err := s.pool.Query(ctx, ` + SELECT ai.account_id, ai.recompute_from + FROM account_snapshot_invalidation ai + JOIN account a ON a.id = ai.account_id + WHERE a.owner_id = $1 + ORDER BY ai.account_id + `, ownerID) + 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,