Merge branch 'feature/price-pipe'

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)
 }

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,
 	}
 }
 

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
+}

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})
+}

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
+}

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
+}

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
+}

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
+}

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)
+	}
+}

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)
 }
 

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
+}

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
+}

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}
+}

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
+}

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'));

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()
+);