Pipelines — Go Taps, Meltano & Nomad Orchestration

import requests

symbols = load_universe()  # 10,000 symbols

for sym in symbols:
    resp = requests.get(f"https://query2.finance.yahoo.com/v8/finance/chart/{sym}")
    store(resp.json())      # ~2s per symbol (network round-trip + parsing)

# Total: 10,000 × 2s = 20,000s ≈ 5.5 hours
# The market opens at 09:30. Your data is stale by noon.

CREATE SCHEMA raw_zone;

-- ─── Reference data: instrument metadata from any provider ────────────────────
CREATE TABLE raw_zone.instruments (
    id              BIGSERIAL    PRIMARY KEY,
    provider_name   VARCHAR(50)  NOT NULL,   -- 'YAHOO', 'BVC', 'STOCKANALYSIS', 'BINANCE'
    external_id     VARCHAR(100),            -- provider-native ID (BVC Drupal node, Yahoo ticker)
    symbol          VARCHAR(50),
    isin            VARCHAR(12),
    instrument_type VARCHAR(50),             -- 'EQUITY', 'CRYPTO', 'ETF', 'FUTURE'
    raw_payload     JSONB        NOT NULL,   -- exact API response, verbatim
    ingested_at     TIMESTAMPTZ  NOT NULL DEFAULT now(),

    -- Idempotency: re-running the tap on the same day is a no-op
    CONSTRAINT uq_raw_instrument UNIQUE (provider_name, external_id, ingested_at::date)
);

CREATE INDEX idx_raw_instr_provider ON raw_zone.instruments (provider_name, ingested_at DESC);
CREATE INDEX idx_raw_instr_isin     ON raw_zone.instruments (isin) WHERE isin IS NOT NULL;

-- ─── OHLCV time-series — partitioned by year for parallel COPY ────────────────
CREATE TABLE raw_zone.market_data (
    provider_name  VARCHAR(50)  NOT NULL,
    symbol         VARCHAR(50)  NOT NULL,
    trade_date     DATE         NOT NULL,   -- DATE not TIMESTAMP: avoids TZ drift bugs
    open_price     NUMERIC(20,8),
    high_price     NUMERIC(20,8),
    low_price      NUMERIC(20,8),
    close_price    NUMERIC(20,8),
    adj_close      NUMERIC(20,8),
    volume         NUMERIC(20,8),
    raw_payload    JSONB,                   -- full JSON kept for debugging
    ingested_at    TIMESTAMPTZ  NOT NULL DEFAULT now(),
    PRIMARY KEY (provider_name, symbol, trade_date)
) PARTITION BY RANGE (trade_date);

-- Year partitions (Go partition manager auto-creates future ones — see Part 1)
CREATE TABLE raw_zone.market_data_y2000 PARTITION OF raw_zone.market_data
    FOR VALUES FROM ('2000-01-01') TO ('2001-01-01');
-- ... remaining year partitions created by partition_manager.go

-- ─── Watermarks: Singer STATE persisted per (provider, stream) ─────────────────
-- Meltano writes these after each successful run; the Go tap reads them on startup
CREATE TABLE raw_zone.watermarks (
    provider_name   VARCHAR(50)   NOT NULL,
    stream_name     VARCHAR(100)  NOT NULL,
    bookmark_value  TEXT          NOT NULL,   -- ISO-8601 date or opaque cursor token
    updated_at      TIMESTAMPTZ   NOT NULL DEFAULT now(),
    PRIMARY KEY (provider_name, stream_name)
);

package singer

import (
	"encoding/json"
	"fmt"
	"os"
	"sync"
)

// mu serializes all writes to stdout.
// Multiple goroutines produce records concurrently;
// stdout is a single sequential stream that must not be interleaved.
var mu sync.Mutex

type schemaMsg struct {
	Type          string         `json:"type"`
	Stream        string         `json:"stream"`
	Schema        map[string]any `json:"schema"`
	KeyProperties []string       `json:"key_properties"`
}

type recordMsg struct {
	Type   string `json:"type"`
	Stream string `json:"stream"`
	Record any    `json:"record"`
}

type stateMsg struct {
	Type  string `json:"type"`
	Value any    `json:"value"`
}

// EmitSchema writes a SCHEMA message. Call once per stream before any records.
func EmitSchema(stream string, schema map[string]any, keyProps []string) error {
	mu.Lock()
	defer mu.Unlock()
	msg := schemaMsg{Type: "SCHEMA", Stream: stream, Schema: schema, KeyProperties: keyProps}
	b, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("marshal schema %s: %w", stream, err)
	}
	_, err = fmt.Fprintln(os.Stdout, string(b))
	return err
}

// EmitRecord writes a RECORD message. Thread-safe: safe to call from goroutines.
func EmitRecord(stream string, record any) error {
	mu.Lock()
	defer mu.Unlock()
	msg := recordMsg{Type: "RECORD", Stream: stream, Record: record}
	b, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("marshal record: %w", err)
	}
	_, err = fmt.Fprintln(os.Stdout, string(b))
	return err
}

// EmitState writes a STATE message at end of run.
// Meltano persists this value; the next run receives it via --state flag.
func EmitState(state any) error {
	mu.Lock()
	defer mu.Unlock()
	msg := stateMsg{Type: "STATE", Value: state}
	b, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("marshal state: %w", err)
	}
	_, err = fmt.Fprintln(os.Stdout, string(b))
	return err
}

package yahoo

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"sync"
	"time"

	"tap-marketdata/internal/singer"
)

// OHLCVRecord is the canonical row emitted for every OHLCV bar.
type OHLCVRecord struct {
	ProviderName string  `json:"provider_name"`
	Symbol       string  `json:"symbol"`
	TradeDate    string  `json:"trade_date"`   // "YYYY-MM-DD"
	OpenPrice    float64 `json:"open_price"`
	HighPrice    float64 `json:"high_price"`
	LowPrice     float64 `json:"low_price"`
	ClosePrice   float64 `json:"close_price"`
	AdjClose     float64 `json:"adj_close"`
	Volume       float64 `json:"volume"`
}

// Extractor fetches daily OHLCV from Yahoo Finance v8 chart API.
type Extractor struct {
	client      *http.Client
	concurrency int
	fromDate    string // watermark start date "YYYY-MM-DD"
}

func New(fromDate string, concurrency int) *Extractor {
	return &Extractor{
		client: &http.Client{
			Timeout: 15 * time.Second,
			Transport: &http.Transport{
				MaxIdleConnsPerHost: concurrency,
				IdleConnTimeout:     90 * time.Second,
			},
		},
		concurrency: concurrency,
		fromDate:    fromDate,
	}
}

// Run fetches all symbols concurrently within a semaphore-bounded goroutine pool.
// Returns the latest trade_date successfully fetched (used to update STATE).
func (e *Extractor) Run(ctx context.Context, symbols []string) (string, error) {
	sem := make(chan struct{}, e.concurrency)
	var wg sync.WaitGroup
	var mu sync.Mutex
	var errs []error
	latestDate := e.fromDate

	for _, sym := range symbols {
		wg.Add(1)
		go func(ticker string) {
			defer wg.Done()
			// Acquire semaphore slot — blocks when pool is full
			select {
			case sem <- struct{}{}:
				defer func() { <-sem }()
			case <-ctx.Done():
				return // Context cancelled — bail out cleanly
			}

			date, err := e.fetchOne(ticker)
			mu.Lock()
			defer mu.Unlock()
			if err != nil {
				// Non-fatal: delisted symbols, holidays, API blips are expected
				slog.Warn("yahoo fetch failed", "symbol", ticker, "err", err)
				errs = append(errs, err)
				return
			}
			if date > latestDate {
				latestDate = date
			}
		}(sym)
	}
	wg.Wait()

	if len(errs) > 0 {
		// Partial failures are normal — log count and continue
		slog.Warn("partial fetch errors", "count", len(errs), "total", len(symbols))
	}
	return latestDate, nil
}

// yahooChartResponse mirrors the JSON structure of Yahoo Finance v8 chart API.
type yahooChartResponse struct {
	Chart struct {
		Result []struct {
			Timestamp  []int64 `json:"timestamp"`
			Indicators struct {
				Quote []struct {
					Open   []float64 `json:"open"`
					High   []float64 `json:"high"`
					Low    []float64 `json:"low"`
					Close  []float64 `json:"close"`
					Volume []float64 `json:"volume"`
				} `json:"quote"`
				Adjclose []struct {
					Adjclose []float64 `json:"adjclose"`
				} `json:"adjclose"`
			} `json:"indicators"`
		} `json:"result"`
		Error *struct {
			Code        string `json:"code"`
			Description string `json:"description"`
		} `json:"error"`
	} `json:"chart"`
}

func (e *Extractor) fetchOne(ticker string) (latestDate string, err error) {
	// Fetch 10 days to cover the 3-day lookback + current week
	url := fmt.Sprintf(
		"https://query2.finance.yahoo.com/v8/finance/chart/%s?range=10d&interval=1d&includeAdjustedClose=true",
		ticker,
	)
	req, _ := http.NewRequest(http.MethodGet, url, nil)
	// Yahoo rate-limits automated bots — a realistic User-Agent reduces 429 errors
	req.Header.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36")
	req.Header.Set("Accept", "application/json")

	resp, err := e.client.Do(req)
	if err != nil {
		return "", fmt.Errorf("http: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusTooManyRequests {
		time.Sleep(5 * time.Second) // back off before returning error (caller may retry)
		return "", fmt.Errorf("rate limited: %s", ticker)
	}
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("HTTP %d: %s", resp.StatusCode, ticker)
	}

	var result yahooChartResponse
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return "", fmt.Errorf("decode %s: %w", ticker, err)
	}
	if result.Chart.Error != nil {
		// Yahoo returns HTTP 200 with an error body for unknown tickers
		return "", fmt.Errorf("yahoo error %s: %s", result.Chart.Error.Code, result.Chart.Error.Description)
	}
	if len(result.Chart.Result) == 0 || len(result.Chart.Result[0].Indicators.Quote) == 0 {
		return "", nil // Delisted or no data — skip silently
	}

	r := result.Chart.Result[0]
	q := r.Indicators.Quote[0]

	for i, ts := range r.Timestamp {
		if i >= len(q.Close) || q.Close[i] == 0 {
			continue // Skip empty bars (market closures, suspended instruments)
		}
		date := time.Unix(ts, 0).UTC().Format("2006-01-02")
		if date < e.fromDate {
			continue // Before watermark: already ingested
		}

		adjClose := 0.0
		if len(r.Indicators.Adjclose) > 0 && i < len(r.Indicators.Adjclose[0].Adjclose) {
			adjClose = r.Indicators.Adjclose[0].Adjclose[i]
		}

		if err := singer.EmitRecord("raw_market_data", OHLCVRecord{
			ProviderName: "YAHOO", Symbol: ticker, TradeDate: date,
			OpenPrice: q.Open[i], HighPrice: q.High[i], LowPrice: q.Low[i],
			ClosePrice: q.Close[i], AdjClose: adjClose, Volume: q.Volume[i],
		}); err != nil {
			return "", err
		}
		if date > latestDate {
			latestDate = date
		}
	}
	return latestDate, nil
}

package bvc

import (
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"time"

	"tap-marketdata/internal/singer"
)

// InstrumentRecord is the canonical row emitted for reference data streams.
type InstrumentRecord struct {
	ProviderName   string `json:"provider_name"`
	ExternalID     string `json:"external_id"`
	Symbol         string `json:"symbol"`
	ISIN           string `json:"isin"`
	Name           string `json:"name"`
	InstrumentType string `json:"instrument_type"`
	Currency       string `json:"currency"`
	RawPayload     any    `json:"raw_payload"` // stored verbatim as JSONB
}

// Extractor fetches the complete Casablanca Bourse instrument listing.
// BVC has strict rate limits: max ~240 req/hour. Pagination via JSON:API.
type Extractor struct {
	client  *http.Client
	baseURL string
	delay   time.Duration
}

func New() *Extractor {
	return &Extractor{
		client:  &http.Client{Timeout: 30 * time.Second},
		baseURL: "https://api.casablanca-bourse.com/fr/api/bourse_data/instrument?page[limit]=200",
		delay:   250 * time.Millisecond,
	}
}

type bvcPage struct {
	Data []struct {
		ID         string `json:"id"`
		Attributes struct {
			CodeISIN string `json:"codeISIN"`
			Symbol   string `json:"symbol"`
			Libelle  string `json:"libelle"` // French security name
			Secteur  string `json:"secteur"`
		} `json:"attributes"`
	} `json:"data"`
	Links struct {
		Next *struct{ Href string `json:"href"` } `json:"next"`
	} `json:"links"`
}

// Run paginates through the full BVC listing, emitting one RECORD per instrument.
// BVC lists ~75 companies, so sequential pagination (no concurrency needed) suffices.
func (b *Extractor) Run() error {
	nextURL := b.baseURL
	for page := 1; nextURL != ""; page++ {
		slog.Info("fetching BVC page", "page", page)

		resp, err := b.client.Get(nextURL)
		if err != nil {
			return fmt.Errorf("BVC page %d: %w", page, err)
		}
		body, _ := io.ReadAll(resp.Body)
		resp.Body.Close()

		if resp.StatusCode != http.StatusOK {
			return fmt.Errorf("BVC HTTP %d on page %d", resp.StatusCode, page)
		}

		var result bvcPage
		if err := json.Unmarshal(body, &result); err != nil {
			return fmt.Errorf("BVC parse page %d: %w", page, err)
		}

		for _, item := range result.Data {
			attr := item.Attributes
			if err := singer.EmitRecord("raw_instruments", InstrumentRecord{
				ProviderName:   "BVC",
				ExternalID:     item.ID,
				Symbol:         attr.Symbol,
				ISIN:           attr.CodeISIN,
				Name:           attr.Libelle,
				InstrumentType: "EQUITY",
				Currency:       "MAD",
				RawPayload:     attr, // full attributes blob → stored as JSONB
			}); err != nil {
				return err
			}
		}

		if result.Links.Next != nil && result.Links.Next.Href != "" {
			nextURL = result.Links.Next.Href
			time.Sleep(b.delay) // respect BVC rate limit strictly
		} else {
			nextURL = ""
		}
	}
	return nil
}

package main

import (
	"context"
	"encoding/json"
	"flag"
	"log/slog"
	"os"
	"strings"
	"time"

	"tap-marketdata/internal/bvc"
	"tap-marketdata/internal/singer"
	"tap-marketdata/internal/yahoo"
)

// tapState is the Singer STATE value persisted between runs.
type tapState struct {
	Bookmarks map[string]bookmark `json:"bookmarks"`
}

type bookmark struct {
	LastFetchedDate string `json:"last_fetched_date"`
}

func main() {
	stateFile := flag.String("state", "", "Path to Singer state file from previous run")
	allocation := flag.String("allocation", "us_stocks", "us_stocks|eu_stocks|crypto|bvc")
	flag.Parse()

	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo})))

	// Load prior state; bookmarks are empty on the very first run
	state := tapState{Bookmarks: make(map[string]bookmark)}
	if *stateFile != "" {
		if data, err := os.ReadFile(*stateFile); err == nil {
			_ = json.Unmarshal(data, &state)
		}
	}

	// Determine watermark; subtract 3 days for late volume/adjclose corrections
	fromDate := "2000-01-01" // default: full history on first run
	if bk, ok := state.Bookmarks["raw_market_data"]; ok && bk.LastFetchedDate != "" {
		t, err := time.Parse("2006-01-02", bk.LastFetchedDate)
		if err == nil {
			fromDate = t.AddDate(0, 0, -3).Format("2006-01-02")
		}
	}

	emitSchemas() // Declare all stream schemas once before any RECORD messages

	ctx := context.Background()
	var latestDate string

	switch strings.ToLower(*allocation) {
	case "bvc":
		// BVC is small (~75 instruments) — instruments only, no OHLCV tap yet
		if err := bvc.New().Run(); err != nil {
			slog.Error("BVC extraction failed", "err", err)
			os.Exit(1)
		}

	default: // us_stocks, eu_stocks, crypto, global_etfs
		concurrency := 50
		if *allocation == "crypto" {
			concurrency = 30 // Binance/Yahoo crypto endpoints are stricter on rate limits
		}
		symbols := loadSymbolList(*allocation)
		ext := yahoo.New(fromDate, concurrency)
		var err error
		if latestDate, err = ext.Run(ctx, symbols); err != nil {
			slog.Error("Yahoo extraction failed", "allocation", *allocation, "err", err)
			os.Exit(1)
		}
	}

	// Update and persist the bookmark for the next incremental run
	if latestDate != "" {
		bk := state.Bookmarks["raw_market_data"]
		bk.LastFetchedDate = latestDate
		state.Bookmarks["raw_market_data"] = bk
		if err := singer.EmitState(state); err != nil {
			slog.Error("emit state failed", "err", err)
		}
	}

	slog.Info("tap complete", "allocation", *allocation, "latest_date", latestDate)
}

func loadSymbolList(allocation string) []string {
	// In production: read from a flat file populated nightly by the screener
	// Alternative: query the instruments table directly
	data, err := os.ReadFile("/etc/tap-marketdata/symbols_" + allocation + ".txt")
	if err != nil {
		slog.Warn("symbol list not found", "allocation", allocation, "err", err)
		return nil
	}
	return strings.Fields(string(data))
}

version: 1
default_environment: production

plugins:
  extractors:
    - name: tap-marketdata-go
      namespace: tap_marketdata
      executable: /usr/local/bin/tap-marketdata-go   # Our compiled Go binary
      capabilities:
        - state      # Supports --state flag (incremental runs)
        - catalog    # Supports --catalog flag (schema declaration)
      settings:
        - name: allocation
          kind: string
          env: ALLOCATION_TARGET
        - name: concurrency
          kind: integer
          value: 50
          env: TAP_CONCURRENCY

  loaders:
    - name: target-postgres
      variant: transferwise
      pip_url: pipelinewise-target-postgres
      config:
        host: ${POSTGRES_HOST}
        port: 5432
        dbname: ${POSTGRES_DB}
        user: ${POSTGRES_USER}
        password: ${POSTGRES_PASSWORD}
        default_target_schema: raw_zone
        batch_size_rows: 10000     # Bulk COPY every 10K records
        flush_all_streams: true
        add_metadata_columns: true # Adds _sdc_extracted_at, _sdc_received_at

schedules:
  - name: daily-us-stocks
    extractor: tap-marketdata-go
    loader: target-postgres
    transform: skip   # dbt runs as a separate Nomad job — see Part 3
    interval: "@daily"
    env:
      ALLOCATION_TARGET: us_stocks

  - name: daily-crypto
    extractor: tap-marketdata-go
    loader: target-postgres
    transform: skip
    interval: "@daily"
    env:
      ALLOCATION_TARGET: crypto
      TAP_CONCURRENCY: "30"

# ─── Stage 1: Compile Go binary ───────────────────────────────────────────────
FROM golang:1.22-alpine AS go-builder
WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download
COPY . .
# CGO_ENABLED=0: fully static binary — runs on any Linux image, no libc dependency
RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \
    go build -ldflags="-s -w" -o tap-marketdata-go ./cmd/tap

# ─── Stage 2: Meltano runtime (Python + target-postgres) ──────────────────────
FROM meltano/meltano:v3.3.0-python3.11
WORKDIR /project
RUN meltano add loader target-postgres --variant transferwise
COPY --from=go-builder /app/tap-marketdata-go /usr/local/bin/
RUN chmod +x /usr/local/bin/tap-marketdata-go
COPY meltano.yml .
COPY config/symbols/ /etc/tap-marketdata/
RUN meltano install      # Creates Python venv for target-postgres
ENV ALLOCATION_TARGET=us_stocks
ENTRYPOINT ["meltano", "run"]
CMD ["tap-marketdata-go", "target-postgres"]

variable "allocation" {
  type    = string
  default = "us_stocks"
}

variable "tap_image" {
  type    = string
  default = "registry.internal.org/data/tap-marketdata:latest"
}

job "marketdata-ingest-${var.allocation}" {
  datacenters = ["dc1"]
  type        = "batch"

  # Mon–Fri at 23:00 UTC (after US and EU market close)
  periodic {
    crons            = ["0 23 * * 1-5"]
    prohibit_overlap = true  # Never start a second instance while one is running
    timezone         = "UTC"
  }

  reschedule {
    attempts       = 3
    interval       = "1h"
    delay          = "5m"
    delay_function = "exponential"
    max_delay      = "30m"
    unlimited      = false
  }

  group "elt" {
    count = 1

    task "meltano-tap" {
      driver = "docker"

      config {
        image   = var.tap_image
        command = "meltano"
        args    = ["run", "tap-marketdata-go", "target-postgres"]
      }

      resources {
        cpu    = 500   # MHz — Go tap <100 MHz; Meltano Python overhead ~400 MHz
        memory = 512   # MB  — Python ~256 MB; 50 goroutines ~100 MB
      }

      # Nomad injects secrets from Vault — no plaintext credentials in job specs
      template {
        data = <<EOH
{{ with secret "secret/data/dailytickers/postgres" }}
POSTGRES_HOST="{{ .Data.data.host }}"
POSTGRES_DB="{{ .Data.data.dbname }}"
POSTGRES_USER="{{ .Data.data.user }}"
POSTGRES_PASSWORD="{{ .Data.data.password }}"
{{ end }}
EOH
        destination = "secrets/db.env"
        env         = true
      }

      env {
        ALLOCATION_TARGET = var.allocation
        TAP_CONCURRENCY   = "50"
      }

      logs {
        max_files     = 7
        max_file_size = 50  # MB
      }
    }
  }
}

IMAGE="registry.internal.org/data/tap-marketdata:$(git rev-parse --short HEAD)"
docker build -t "$IMAGE" . && docker push "$IMAGE"

for alloc in us_stocks eu_stocks crypto bvc; do
  nomad job run \
    -var="allocation=${alloc}" \
    -var="tap_image=${IMAGE}" \
    jobs/marketdata-ingest.nomad
  echo "✓ ${alloc} deployed"
done

# Monitor a running job
nomad alloc logs -f "$(nomad job allocs -json marketdata-ingest-us_stocks | jq -r '.[0].ID')"

// validateBar blocks catastrophically invalid data before it enters the database.
// Comprehensive validation (range checks, cross-column assertions) belongs in dbt.
func validateBar(symbol, date string, open, high, low, close float64) error {
    if close <= 0 || open <= 0 {
        return fmt.Errorf("non-positive price: %s %s close=%.4f", symbol, date, close)
    }
    if high < low {
        return fmt.Errorf("high < low: %s %s high=%.4f low=%.4f", symbol, date, high, low)
    }
    if close > high || close < low {
        return fmt.Errorf("close outside high-low band: %s %s", symbol, date)
    }
    // Flag but don't reject extreme moves — may be a split (Part 4 handles these)
    if open > 0 {
        ret := (close - open) / open
        if ret > 0.80 || ret < -0.80 {
            slog.Warn("extreme daily return — possible split/error",
                "symbol", symbol, "date", date, "return_pct", ret*100)
        }
    }
    return nil
}