diff --git a/cmd/api/main.go b/cmd/api/main.go index fe69697..445afc2 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -11,7 +11,10 @@ import ( "time" "git.g3e.fr/H6N/account/internal/config" + "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/store" "github.com/jackc/pgx/v5/pgxpool" ) @@ -20,7 +23,9 @@ func main() { cfg := config.Load() - ctx := context.Background() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + pool, err := pgxpool.New(ctx, cfg.DatabaseURL) if err != nil { logger.Error("failed to create db pool", "error", err) @@ -34,6 +39,25 @@ func main() { } logger.Info("database connected") + st := store.New(pool) + pl := pipeline.New(st, pipeline.Config{ + OpenFIGIKey: cfg.OpenFIGIKey, + CoinGeckoKey: cfg.CoinGeckoKey, + }, logger) + + sched := scheduler.New(logger) + sched.Add(scheduler.Job{ + Name: "fetch-prices", + Interval: cfg.PriceFetchInterval, + Fn: pl.FetchAll, + }) + sched.Add(scheduler.Job{ + Name: "clean-price-history", + Interval: cfg.PriceCleanInterval, + Fn: pl.CleanHistory, + }) + sched.Start(ctx) + srv := server.New(cfg, pool, logger) httpServer := &http.Server{ @@ -57,7 +81,8 @@ func main() { <-quit logger.Info("shutting down") - shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() + cancel() + shutdownCtx, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() httpServer.Shutdown(shutdownCtx) } diff --git a/internal/config/config.go b/internal/config/config.go index 967d158..d7d9019 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,18 +1,34 @@ package config -import "os" +import ( + "os" + "time" +) type Config struct { - DatabaseURL string - Port string - Env string + DatabaseURL string + Port string + Env string + OpenFIGIKey string + CoinGeckoKey string + PriceFetchInterval time.Duration + PriceCleanInterval time.Duration } func Load() *Config { + fetchInterval, err := time.ParseDuration(getenv("PRICE_FETCH_INTERVAL", "1h")) + if err != nil { + fetchInterval = time.Hour + } + return &Config{ - DatabaseURL: getenv("DATABASE_URL", "postgres://account:account@localhost:5432/account?sslmode=disable"), - Port: getenv("PORT", "8080"), - Env: getenv("ENV", "development"), + 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, } } diff --git a/internal/handler/instrument.go b/internal/handler/instrument.go new file mode 100644 index 0000000..86491b6 --- /dev/null +++ b/internal/handler/instrument.go @@ -0,0 +1,116 @@ +package handler + +import ( + "encoding/json" + "errors" + "net/http" + "strconv" + + "git.g3e.fr/H6N/account/internal/store" + "github.com/jackc/pgx/v5" +) + +type InstrumentHandler struct { + store *store.Store +} + +func NewInstrumentHandler(s *store.Store) *InstrumentHandler { + return &InstrumentHandler{store: s} +} + +func (h *InstrumentHandler) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /instruments", h.list) + mux.HandleFunc("POST /instruments", h.create) + mux.HandleFunc("GET /instruments/{id}", h.get) + mux.HandleFunc("PUT /instruments/{id}", h.update) + mux.HandleFunc("DELETE /instruments/{id}", h.delete) +} + +func (h *InstrumentHandler) list(w http.ResponseWriter, r *http.Request) { + instruments, err := h.store.ListInstruments(r.Context()) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to list instruments") + return + } + writeJSON(w, http.StatusOK, instruments) +} + +func (h *InstrumentHandler) get(w http.ResponseWriter, r *http.Request) { + id, err := parseID(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid id") + return + } + instrument, err := h.store.GetInstrument(r.Context(), id) + if errors.Is(err, pgx.ErrNoRows) { + writeError(w, http.StatusNotFound, "instrument not found") + return + } + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to get instrument") + return + } + writeJSON(w, http.StatusOK, instrument) +} + +func (h *InstrumentHandler) create(w http.ResponseWriter, r *http.Request) { + var p store.CreateInstrumentParams + if err := json.NewDecoder(r.Body).Decode(&p); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + if p.Type == "" || p.Code == "" || p.Name == "" { + writeError(w, http.StatusBadRequest, "type, code and name are required") + return + } + if p.DeviseCotation == "" { + p.DeviseCotation = "EUR" + } + instrument, err := h.store.CreateInstrument(r.Context(), p) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to create instrument") + return + } + writeJSON(w, http.StatusCreated, instrument) +} + +func (h *InstrumentHandler) update(w http.ResponseWriter, r *http.Request) { + id, err := parseID(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid id") + return + } + var p store.CreateInstrumentParams + if err := json.NewDecoder(r.Body).Decode(&p); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + instrument, err := h.store.UpdateInstrument(r.Context(), id, p) + if errors.Is(err, pgx.ErrNoRows) { + writeError(w, http.StatusNotFound, "instrument not found") + return + } + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to update instrument") + return + } + writeJSON(w, http.StatusOK, instrument) +} + +func (h *InstrumentHandler) delete(w http.ResponseWriter, r *http.Request) { + id, err := parseID(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid id") + return + } + if err := h.store.DeleteInstrument(r.Context(), id); err != nil { + writeError(w, http.StatusInternalServerError, "failed to delete instrument") + return + } + w.WriteHeader(http.StatusNoContent) +} + +func parseID(r *http.Request) (int32, error) { + v, err := strconv.Atoi(r.PathValue("id")) + return int32(v), err +} diff --git a/internal/handler/response.go b/internal/handler/response.go new file mode 100644 index 0000000..43a60fc --- /dev/null +++ b/internal/handler/response.go @@ -0,0 +1,16 @@ +package handler + +import ( + "encoding/json" + "net/http" +) + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(v) +} + +func writeError(w http.ResponseWriter, status int, msg string) { + writeJSON(w, status, map[string]string{"error": msg}) +} diff --git a/internal/pipeline/coingecko.go b/internal/pipeline/coingecko.go new file mode 100644 index 0000000..20d9a00 --- /dev/null +++ b/internal/pipeline/coingecko.go @@ -0,0 +1,88 @@ +package pipeline + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" +) + +// ResolveCryptoID résout un ticker crypto (ex: "BTC") en ID CoinGecko (ex: "bitcoin"). +func ResolveCryptoID(ctx context.Context, apiKey, ticker string) (string, error) { + url := fmt.Sprintf("https://api.coingecko.com/api/v3/search?query=%s", ticker) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return "", err + } + if apiKey != "" { + req.Header.Set("x-cg-pro-api-key", apiKey) + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var result struct { + Coins []struct { + ID string `json:"id"` + Symbol string `json:"symbol"` + } `json:"coins"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", fmt.Errorf("decode coingecko search: %w", err) + } + + // Le premier résultat dont le symbol correspond exactement est le bon + upper := strings.ToUpper(ticker) + for _, coin := range result.Coins { + if strings.ToUpper(coin.Symbol) == upper { + return coin.ID, nil + } + } + return "", fmt.Errorf("no coingecko match for ticker %s", ticker) +} + +// 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) { + if len(coinIDs) == 0 { + return map[string]float64{}, nil + } + + baseURL := "https://api.coingecko.com/api/v3/simple/price" + if apiKey != "" { + baseURL = "https://pro-api.coingecko.com/api/v3/simple/price" + } + + url := fmt.Sprintf("%s?ids=%s&vs_currencies=eur", baseURL, strings.Join(coinIDs, ",")) + + 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 : {"bitcoin": {"eur": 45000.0}, "ethereum": {"eur": 2500.0}} + var raw map[string]map[string]float64 + if err := json.NewDecoder(resp.Body).Decode(&raw); err != nil { + return nil, fmt.Errorf("decode coingecko response: %w", err) + } + + prices := make(map[string]float64, len(raw)) + for id, currencies := range raw { + prices[id] = currencies["eur"] + } + return prices, nil +} diff --git a/internal/pipeline/openfigi.go b/internal/pipeline/openfigi.go new file mode 100644 index 0000000..5626f63 --- /dev/null +++ b/internal/pipeline/openfigi.go @@ -0,0 +1,75 @@ +package pipeline + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" +) + +// exchSuffix mappe le code bourse OpenFIGI vers le suffixe Yahoo Finance +var exchSuffix = map[string]string{ + "FP": ".PA", // Euronext Paris + "GY": ".DE", // Xetra + "LN": ".L", // London + "SM": ".MC", // Madrid + "IM": ".MI", // Milan + "NA": ".AS", // Amsterdam + "BB": ".BR", // Bruxelles + "UN": "", // NYSE + "UQ": "", // NASDAQ + "UW": "", // NASDAQ (alt) +} + +type openFIGIRequest struct { + IDType string `json:"idType"` + IDValue string `json:"idValue"` +} + +type openFIGIData struct { + Ticker string `json:"ticker"` + ExchCode string `json:"exchCode"` + Name string `json:"name"` +} + +type openFIGIResult struct { + Data []openFIGIData `json:"data"` + Error string `json:"error"` +} + +// ResolveISIN résout un ISIN en ticker Yahoo Finance via OpenFIGI. +func ResolveISIN(ctx context.Context, apiKey, isin string) (string, error) { + body, _ := json.Marshal([]openFIGIRequest{{IDType: "ID_ISIN", IDValue: isin}}) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + "https://api.openfigi.com/v3/mapping", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json") + if apiKey != "" { + req.Header.Set("X-OPENFIGI-APIKEY", apiKey) + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var results []openFIGIResult + if err := json.NewDecoder(resp.Body).Decode(&results); err != nil { + return "", fmt.Errorf("decode openfigi response: %w", err) + } + if len(results) == 0 || len(results[0].Data) == 0 { + return "", fmt.Errorf("no result for ISIN %s", isin) + } + if results[0].Error != "" { + return "", fmt.Errorf("openfigi error: %s", results[0].Error) + } + + d := results[0].Data[0] + suffix := exchSuffix[d.ExchCode] + return d.Ticker + suffix, nil +} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go new file mode 100644 index 0000000..04b5f35 --- /dev/null +++ b/internal/pipeline/pipeline.go @@ -0,0 +1,133 @@ +package pipeline + +import ( + "context" + "log/slog" + "time" + + "git.g3e.fr/H6N/account/internal/store" +) + +type Config struct { + OpenFIGIKey string + CoinGeckoKey string +} + +type Pipeline struct { + store *store.Store + cfg Config + logger *slog.Logger +} + +func New(s *store.Store, cfg Config, logger *slog.Logger) *Pipeline { + return &Pipeline{store: s, cfg: cfg, logger: logger} +} + +// FetchAll récupère les prix de tous les instruments et les écrit dans price_history. +func (p *Pipeline) FetchAll(ctx context.Context) error { + instruments, err := p.store.ListInstruments(ctx) + if err != nil { + return err + } + + now := time.Now().UTC() + + // Séparer les instruments par type pour batch CoinGecko + var cryptoIDs []string + cryptoByID := make(map[string]store.Instrument) + + for _, inst := range instruments { + switch inst.Type { + case "devise": + if err := p.store.UpsertPrice(ctx, inst.ID, now, 1.0); err != nil { + p.logger.Error("upsert devise price", "code", inst.Code, "error", err) + } + + case "action", "etf": + ticker, err := p.resolveTicker(ctx, inst) + if err != nil { + p.logger.Warn("could not resolve ticker", "instrument", inst.Code, "error", err) + continue + } + price, err := FetchYahooPrice(ctx, ticker) + if err != nil { + p.logger.Warn("yahoo fetch failed", "ticker", ticker, "error", err) + continue + } + if err := p.store.UpsertPrice(ctx, inst.ID, now, price); err != nil { + p.logger.Error("upsert price", "instrument", inst.Code, "error", err) + } + p.logger.Info("price fetched", "instrument", inst.Code, "ticker", ticker, "price", price) + + case "crypto": + coinID, err := p.resolveCoinID(ctx, inst) + if err != nil { + p.logger.Warn("could not resolve coingecko id", "instrument", inst.Code, "error", err) + continue + } + cryptoIDs = append(cryptoIDs, coinID) + cryptoByID[coinID] = inst + } + } + + // Batch fetch crypto + if len(cryptoIDs) > 0 { + prices, err := FetchCoinGeckoPrices(ctx, p.cfg.CoinGeckoKey, cryptoIDs) + if err != nil { + p.logger.Error("coingecko fetch failed", "error", err) + } else { + for coinID, price := range prices { + inst := cryptoByID[coinID] + if err := p.store.UpsertPrice(ctx, inst.ID, now, price); err != nil { + p.logger.Error("upsert crypto price", "instrument", inst.Code, "error", err) + continue + } + p.logger.Info("price fetched", "instrument", inst.Code, "coin_id", coinID, "price", price) + } + } + } + + 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 { + p.logger.Error("clean price history failed", "error", err) + return err + } + p.logger.Info("price history cleaned") + return nil +} + +// resolveTicker retourne le ticker Yahoo Finance, en le résolvant via OpenFIGI si nécessaire. +func (p *Pipeline) resolveTicker(ctx context.Context, inst store.Instrument) (string, error) { + ticker, err := p.store.GetTicker(ctx, inst.ID) + if err == nil { + return ticker, nil + } + + resolved, err := ResolveISIN(ctx, p.cfg.OpenFIGIKey, inst.Code) + if err != nil { + return "", err + } + + _ = p.store.UpsertTicker(ctx, inst.ID, resolved) + return resolved, nil +} + +// resolveCoinID retourne l'ID CoinGecko, en le résolvant via l'API search si nécessaire. +func (p *Pipeline) resolveCoinID(ctx context.Context, inst store.Instrument) (string, error) { + coinID, err := p.store.GetTicker(ctx, inst.ID) + if err == nil { + return coinID, nil + } + + resolved, err := ResolveCryptoID(ctx, p.cfg.CoinGeckoKey, inst.Code) + if err != nil { + return "", err + } + + _ = p.store.UpsertTicker(ctx, inst.ID, resolved) + return resolved, nil +} diff --git a/internal/pipeline/yahoo.go b/internal/pipeline/yahoo.go new file mode 100644 index 0000000..f08eb68 --- /dev/null +++ b/internal/pipeline/yahoo.go @@ -0,0 +1,55 @@ +package pipeline + +import ( + "context" + "encoding/json" + "fmt" + "net/http" +) + +type yahooChart struct { + Chart struct { + Result []struct { + Meta struct { + RegularMarketPrice float64 `json:"regularMarketPrice"` + Currency string `json:"currency"` + } `json:"meta"` + } `json:"result"` + Error *struct{ Description string } `json:"error"` + } `json:"chart"` +} + +// FetchYahooPrice récupère le dernier prix connu pour un ticker Yahoo Finance. +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) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + 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 +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go new file mode 100644 index 0000000..8ff5ae0 --- /dev/null +++ b/internal/scheduler/scheduler.go @@ -0,0 +1,60 @@ +package scheduler + +import ( + "context" + "log/slog" + "time" +) + +type Job struct { + Name string + Interval time.Duration + Fn func(ctx context.Context) error +} + +type Scheduler struct { + jobs []Job + logger *slog.Logger +} + +func New(logger *slog.Logger) *Scheduler { + return &Scheduler{logger: logger} +} + +func (s *Scheduler) Add(job Job) { + s.jobs = append(s.jobs, job) +} + +// Start lance tous les jobs en arrière-plan. Chaque job est exécuté immédiatement +// puis répété selon son intervalle. S'arrête proprement à l'annulation du contexte. +func (s *Scheduler) Start(ctx context.Context) { + for _, job := range s.jobs { + go s.run(ctx, job) + } +} + +func (s *Scheduler) run(ctx context.Context, job Job) { + s.logger.Info("scheduler: starting job", "name", job.Name, "interval", job.Interval) + + s.execute(ctx, job) + + ticker := time.NewTicker(job.Interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + s.logger.Info("scheduler: job stopped", "name", job.Name) + return + case <-ticker.C: + s.execute(ctx, job) + } + } +} + +func (s *Scheduler) execute(ctx context.Context, job Job) { + s.logger.Info("scheduler: running job", "name", job.Name) + if err := job.Fn(ctx); err != nil { + s.logger.Error("scheduler: job failed", "name", job.Name, "error", err) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index e49b697..fee7822 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -6,6 +6,8 @@ import ( "net/http" "git.g3e.fr/H6N/account/internal/config" + "git.g3e.fr/H6N/account/internal/handler" + "git.g3e.fr/H6N/account/internal/store" "github.com/jackc/pgx/v5/pgxpool" ) @@ -32,6 +34,10 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { } func (s *Server) routes() { + st := store.New(s.pool) + + handler.NewInstrumentHandler(st).RegisterRoutes(s.mux) + s.mux.HandleFunc("GET /health", s.handleHealth) } diff --git a/internal/store/instrument.go b/internal/store/instrument.go new file mode 100644 index 0000000..f6876ec --- /dev/null +++ b/internal/store/instrument.go @@ -0,0 +1,78 @@ +package store + +import ( + "context" + + "github.com/jackc/pgx/v5" +) + +type Instrument struct { + ID int32 `json:"id"` + Type string `json:"type"` + Code string `json:"code"` + Name string `json:"name"` + DeviseCotation string `json:"devise_cotation"` +} + +type CreateInstrumentParams struct { + Type string `json:"type"` + Code string `json:"code"` + Name string `json:"name"` + DeviseCotation string `json:"devise_cotation"` +} + +func (s *Store) ListInstruments(ctx context.Context) ([]Instrument, error) { + rows, err := s.pool.Query(ctx, + `SELECT id, type, code, name, devise_cotation FROM instrument ORDER BY type, code`) + if err != nil { + return nil, err + } + return pgx.CollectRows(rows, pgx.RowToStructByName[Instrument]) +} + +func (s *Store) GetInstrument(ctx context.Context, id int32) (Instrument, error) { + rows, err := s.pool.Query(ctx, + `SELECT id, type, code, name, devise_cotation FROM instrument WHERE id = $1`, id) + if err != nil { + return Instrument{}, err + } + return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument]) +} + +func (s *Store) GetInstrumentByCode(ctx context.Context, code string) (Instrument, error) { + rows, err := s.pool.Query(ctx, + `SELECT id, type, code, name, devise_cotation FROM instrument WHERE code = $1`, code) + if err != nil { + return Instrument{}, err + } + return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument]) +} + +func (s *Store) CreateInstrument(ctx context.Context, p CreateInstrumentParams) (Instrument, error) { + rows, err := s.pool.Query(ctx, + `INSERT INTO instrument (type, code, name, devise_cotation) + VALUES ($1, $2, $3, $4) + RETURNING id, type, code, name, devise_cotation`, + p.Type, p.Code, p.Name, p.DeviseCotation) + if err != nil { + return Instrument{}, err + } + return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument]) +} + +func (s *Store) UpdateInstrument(ctx context.Context, id int32, p CreateInstrumentParams) (Instrument, error) { + rows, err := s.pool.Query(ctx, + `UPDATE instrument SET type = $2, code = $3, name = $4, devise_cotation = $5 + WHERE id = $1 + RETURNING id, type, code, name, devise_cotation`, + id, p.Type, p.Code, p.Name, p.DeviseCotation) + if err != nil { + return Instrument{}, err + } + return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument]) +} + +func (s *Store) DeleteInstrument(ctx context.Context, id int32) error { + _, err := s.pool.Exec(ctx, `DELETE FROM instrument WHERE id = $1`, id) + return err +} diff --git a/internal/store/price_history.go b/internal/store/price_history.go new file mode 100644 index 0000000..165ab92 --- /dev/null +++ b/internal/store/price_history.go @@ -0,0 +1,41 @@ +package store + +import ( + "context" + "time" +) + +func (s *Store) UpsertPrice(ctx context.Context, instrumentID int32, fetchedAt time.Time, prix float64) error { + _, err := s.pool.Exec(ctx, + `INSERT INTO price_history (instrument_id, fetched_at, prix) + VALUES ($1, $2, $3) + ON CONFLICT (instrument_id, fetched_at) DO UPDATE SET prix = EXCLUDED.prix`, + instrumentID, fetchedAt, prix) + return err +} + +// GetLastPrice retourne le dernier prix connu pour un instrument. +func (s *Store) GetLastPrice(ctx context.Context, instrumentID int32) (float64, time.Time, error) { + var prix float64 + var fetchedAt time.Time + err := s.pool.QueryRow(ctx, + `SELECT prix, fetched_at FROM price_history + WHERE instrument_id = $1 + ORDER BY fetched_at DESC LIMIT 1`, + instrumentID).Scan(&prix, &fetchedAt) + return prix, fetchedAt, err +} + +// CleanPastDays supprime pour chaque jour passé toutes les entrées sauf la dernière par instrument. +func (s *Store) CleanPastDays(ctx context.Context) error { + _, err := s.pool.Exec(ctx, ` + DELETE FROM price_history ph + WHERE ph.fetched_at::date < CURRENT_DATE + AND ph.fetched_at < ( + SELECT MAX(ph2.fetched_at) + FROM price_history ph2 + WHERE ph2.instrument_id = ph.instrument_id + AND ph2.fetched_at::date = ph.fetched_at::date + )`) + return err +} diff --git a/internal/store/store.go b/internal/store/store.go new file mode 100644 index 0000000..6a0eded --- /dev/null +++ b/internal/store/store.go @@ -0,0 +1,11 @@ +package store + +import "github.com/jackc/pgx/v5/pgxpool" + +type Store struct { + pool *pgxpool.Pool +} + +func New(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} diff --git a/internal/store/ticker_cache.go b/internal/store/ticker_cache.go new file mode 100644 index 0000000..ca5b9ca --- /dev/null +++ b/internal/store/ticker_cache.go @@ -0,0 +1,31 @@ +package store + +import ( + "context" + + "github.com/jackc/pgx/v5" +) + +// GetTicker retourne le ticker (ou CoinGecko ID) mis en cache pour un instrument. +// Retourne pgx.ErrNoRows si absent. +func (s *Store) GetTicker(ctx context.Context, instrumentID int32) (string, error) { + var ticker string + err := s.pool.QueryRow(ctx, + `SELECT ticker FROM instrument_ticker_cache WHERE instrument_id = $1`, + instrumentID).Scan(&ticker) + return ticker, err +} + +// UpsertTicker met à jour ou insère le ticker pour un instrument. +func (s *Store) UpsertTicker(ctx context.Context, instrumentID int32, ticker string) error { + _, err := s.pool.Exec(ctx, + `INSERT INTO instrument_ticker_cache (instrument_id, ticker, fetched_at) + VALUES ($1, $2, NOW()) + ON CONFLICT (instrument_id) DO UPDATE SET ticker = EXCLUDED.ticker, fetched_at = NOW()`, + instrumentID, ticker) + return err +} + +func isNotFound(err error) bool { + return err == pgx.ErrNoRows +} diff --git a/migrations/000002_price_history_intraday.down.sql b/migrations/000002_price_history_intraday.down.sql new file mode 100644 index 0000000..42b971f --- /dev/null +++ b/migrations/000002_price_history_intraday.down.sql @@ -0,0 +1,10 @@ +DROP TABLE IF EXISTS instrument_ticker_cache; +DROP TABLE IF EXISTS price_history; + +CREATE TABLE price_history ( + instrument_id INTEGER NOT NULL REFERENCES instrument(id), + date DATE NOT NULL, + prix NUMERIC(24, 8) NOT NULL, + PRIMARY KEY (instrument_id, date) +); +SELECT create_hypertable('price_history', by_range('date')); diff --git a/migrations/000002_price_history_intraday.up.sql b/migrations/000002_price_history_intraday.up.sql new file mode 100644 index 0000000..49bc9f0 --- /dev/null +++ b/migrations/000002_price_history_intraday.up.sql @@ -0,0 +1,19 @@ +-- Recréation de price_history avec support intraday (plusieurs prix par jour) +-- Le job de nettoyage consolide à J-1 en gardant uniquement le dernier fetch +DROP TABLE IF EXISTS price_history; + +CREATE TABLE price_history ( + instrument_id INTEGER NOT NULL REFERENCES instrument(id), + fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + prix NUMERIC(24, 8) NOT NULL, + PRIMARY KEY (instrument_id, fetched_at) +); +SELECT create_hypertable('price_history', by_range('fetched_at')); +CREATE INDEX idx_price_history_instrument ON price_history(instrument_id, fetched_at DESC); + +-- Cache OpenFIGI (ISIN → ticker Yahoo Finance) et CoinGecko (ticker → coin ID) +CREATE TABLE instrument_ticker_cache ( + instrument_id INTEGER PRIMARY KEY REFERENCES instrument(id) ON DELETE CASCADE, + ticker TEXT NOT NULL, + fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +);